jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 21:27:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use service_calls fixture in homeassistant tests (#120922)

Browse Source
This commit is contained in:
epenet 2024-07-01 17:26:41 +02:00 committed by GitHub
parent 2815c43f3e
commit c8bb64882e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 515 additions and 471 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

113
tests/components/homeassistant/triggers/test_event.py
View File

@ -7,28 +7,24 @@ from homeassistant.const import ATTR_ENTITY_ID, ENTITY_MATCH_ALL, SERVICE_TURN_O
from homeassistant.core import Context, HomeAssistant, ServiceCall
from homeassistant.setup import async_setup_component
from tests.common import async_mock_service, mock_component
from tests.common import mock_component
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
@pytest.fixture
def context_with_user():
def context_with_user() -> Context:
"""Create a context with default user_id."""
return Context(user_id="test_user_id")
@pytest.fixture(autouse=True)
def setup_comp(hass):
def setup_comp(hass: HomeAssistant) -> None:
"""Initialize components."""
mock_component(hass, "group")
async def test_if_fires_on_event(hass: HomeAssistant, calls: list[ServiceCall]) -> None:
async def test_if_fires_on_event(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test the firing of events."""
context = Context()
@ -48,8 +44,8 @@ async def test_if_fires_on_event(hass: HomeAssistant, calls: list[ServiceCall])
hass.bus.async_fire("test_event", context=context)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].context.parent_id == context.id
assert len(service_calls) == 1
assert service_calls[0].context.parent_id == context.id
await hass.services.async_call(
automation.DOMAIN,
@ -57,15 +53,16 @@ async def test_if_fires_on_event(hass: HomeAssistant, calls: list[ServiceCall])
{ATTR_ENTITY_ID: ENTITY_MATCH_ALL},
blocking=True,
)
assert len(service_calls) == 2
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["id"] == 0
assert len(service_calls) == 2
assert service_calls[0].data["id"] == 0
async def test_if_fires_on_templated_event(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test the firing of events."""
context = Context()
@ -84,8 +81,8 @@ async def test_if_fires_on_templated_event(
hass.bus.async_fire("test_event", context=context)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].context.parent_id == context.id
assert len(service_calls) == 1
assert service_calls[0].context.parent_id == context.id
await hass.services.async_call(
automation.DOMAIN,
@ -93,14 +90,15 @@ async def test_if_fires_on_templated_event(
{ATTR_ENTITY_ID: ENTITY_MATCH_ALL},
blocking=True,
)
assert len(service_calls) == 2
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 2
async def test_if_fires_on_multiple_events(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test the firing of events."""
context = Context()
@ -123,13 +121,13 @@ async def test_if_fires_on_multiple_events(
await hass.async_block_till_done()
hass.bus.async_fire("test2_event", context=context)
await hass.async_block_till_done()
assert len(calls) == 2
assert calls[0].context.parent_id == context.id
assert calls[1].context.parent_id == context.id
assert len(service_calls) == 2
assert service_calls[0].context.parent_id == context.id
assert service_calls[1].context.parent_id == context.id
async def test_if_fires_on_event_extra_data(
hass: HomeAssistant, calls: list[ServiceCall], context_with_user: Context
hass: HomeAssistant, service_calls: list[ServiceCall], context_with_user: Context
) -> None:
"""Test the firing of events still matches with event data and context."""
assert await async_setup_component(
@ -146,7 +144,7 @@ async def test_if_fires_on_event_extra_data(
"test_event", {"extra_key": "extra_data"}, context=context_with_user
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
await hass.services.async_call(
automation.DOMAIN,
@ -154,14 +152,15 @@ async def test_if_fires_on_event_extra_data(
{ATTR_ENTITY_ID: ENTITY_MATCH_ALL},
blocking=True,
)
assert len(service_calls) == 2
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 2
async def test_if_fires_on_event_with_data_and_context(
hass: HomeAssistant, calls: list[ServiceCall], context_with_user: Context
hass: HomeAssistant, service_calls: list[ServiceCall], context_with_user: Context
) -> None:
"""Test the firing of events with data and context."""
assert await async_setup_component(
@ -189,7 +188,7 @@ async def test_if_fires_on_event_with_data_and_context(
context=context_with_user,
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
hass.bus.async_fire(
"test_event",
@ -197,18 +196,18 @@ async def test_if_fires_on_event_with_data_and_context(
context=context_with_user,
)
await hass.async_block_till_done()
assert len(calls) == 1 # No new call
assert len(service_calls) == 1 # No new call
hass.bus.async_fire(
"test_event",
{"some_attr": "some_value", "another": "value", "second_attr": "second_value"},
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_event_with_templated_data_and_context(
hass: HomeAssistant, calls: list[ServiceCall], context_with_user: Context
hass: HomeAssistant, service_calls: list[ServiceCall], context_with_user: Context
) -> None:
"""Test the firing of events with templated data and context."""
assert await async_setup_component(
@ -241,7 +240,7 @@ async def test_if_fires_on_event_with_templated_data_and_context(
context=context_with_user,
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
hass.bus.async_fire(
"test_event",
@ -249,18 +248,18 @@ async def test_if_fires_on_event_with_templated_data_and_context(
context=context_with_user,
)
await hass.async_block_till_done()
assert len(calls) == 1 # No new call
assert len(service_calls) == 1 # No new call
hass.bus.async_fire(
"test_event",
{"attr_1": "milk", "another": "value", "attr_2": "beer"},
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_event_with_empty_data_and_context_config(
hass: HomeAssistant, calls: list[ServiceCall], context_with_user: Context
hass: HomeAssistant, service_calls: list[ServiceCall], context_with_user: Context
) -> None:
"""Test the firing of events with empty data and context config.
@ -289,11 +288,11 @@ async def test_if_fires_on_event_with_empty_data_and_context_config(
context=context_with_user,
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_event_with_nested_data(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test the firing of events with nested data.
@ -319,11 +318,11 @@ async def test_if_fires_on_event_with_nested_data(
"test_event", {"parent_attr": {"some_attr": "some_value", "another": "value"}}
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_event_with_empty_data(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test the firing of events with empty data.
@ -345,11 +344,11 @@ async def test_if_fires_on_event_with_empty_data(
)
hass.bus.async_fire("test_event", {"any_attr": {}})
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_sample_zha_event(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test the firing of events with a sample zha event.
@ -390,7 +389,7 @@ async def test_if_fires_on_sample_zha_event(
},
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
hass.bus.async_fire(
"zha_event",
@ -404,11 +403,11 @@ async def test_if_fires_on_sample_zha_event(
},
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_not_fires_if_event_data_not_matches(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test firing of event if no data match."""
assert await async_setup_component(
@ -428,11 +427,11 @@ async def test_if_not_fires_if_event_data_not_matches(
hass.bus.async_fire("test_event", {"some_attr": "some_other_value"})
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async def test_if_not_fires_if_event_context_not_matches(
hass: HomeAssistant, calls: list[ServiceCall], context_with_user: Context
hass: HomeAssistant, service_calls: list[ServiceCall], context_with_user: Context
) -> None:
"""Test firing of event if no context match."""
assert await async_setup_component(
@ -452,11 +451,11 @@ async def test_if_not_fires_if_event_context_not_matches(
hass.bus.async_fire("test_event", {}, context=context_with_user)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async def test_if_fires_on_multiple_user_ids(
hass: HomeAssistant, calls: list[ServiceCall], context_with_user: Context
hass: HomeAssistant, service_calls: list[ServiceCall], context_with_user: Context
) -> None:
"""Test the firing of event when the trigger has multiple user ids.
@ -481,11 +480,11 @@ async def test_if_fires_on_multiple_user_ids(
hass.bus.async_fire("test_event", {}, context=context_with_user)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_event_data_with_list(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test the (non)firing of event when the data schema has lists."""
assert await async_setup_component(
@ -506,17 +505,17 @@ async def test_event_data_with_list(
hass.bus.async_fire("test_event", {"some_attr": [1, 2]})
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
# don't match a single value
hass.bus.async_fire("test_event", {"some_attr": 1})
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
# don't match a containing list
hass.bus.async_fire("test_event", {"some_attr": [1, 2, 3]})
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
@pytest.mark.parametrize(
@ -524,7 +523,7 @@ async def test_event_data_with_list(
)
async def test_state_reported_event(
hass: HomeAssistant,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
caplog: pytest.LogCaptureFixture,
event_type: str | list[str],
) -> None:
@ -547,7 +546,7 @@ async def test_state_reported_event(
hass.bus.async_fire("test_event", context=context)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
assert (
"Unnamed automation failed to setup triggers and has been disabled: Can't "
"listen to state_reported in event trigger for dictionary value @ "
@ -556,7 +555,9 @@ async def test_state_reported_event(
async def test_templated_state_reported_event(
hass: HomeAssistant, calls: list[ServiceCall], caplog: pytest.LogCaptureFixture
hass: HomeAssistant,
service_calls: list[ServiceCall],
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test triggering on state reported event."""
context = Context()
@ -578,7 +579,7 @@ async def test_templated_state_reported_event(
hass.bus.async_fire("test_event", context=context)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
assert (
"Got error 'Can't listen to state_reported in event trigger' "
"when setting up triggers for automation 0" in caplog.text

325
tests/components/homeassistant/triggers/test_numeric_state.py
View File

@ -23,22 +23,11 @@ from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from tests.common import (
assert_setup_component,
async_fire_time_changed,
async_mock_service,
mock_component,
)
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
from tests.common import assert_setup_component, async_fire_time_changed, mock_component
@pytest.fixture(autouse=True)
async def setup_comp(hass):
async def setup_comp(hass: HomeAssistant) -> None:
"""Initialize components."""
mock_component(hass, "group")
await async_setup_component(
@ -63,7 +52,7 @@ async def setup_comp(hass):
"below", [10, "input_number.value_10", "number.value_10", "sensor.value_10"]
)
async def test_if_not_fires_on_entity_removal(
hass: HomeAssistant, calls: list[ServiceCall], below: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], below: int | str
) -> None:
"""Test the firing with removed entity."""
hass.states.async_set("test.entity", 11)
@ -86,14 +75,14 @@ async def test_if_not_fires_on_entity_removal(
# Entity disappears
hass.states.async_remove("test.entity")
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
@pytest.mark.parametrize(
"below", [10, "input_number.value_10", "number.value_10", "sensor.value_10"]
)
async def test_if_fires_on_entity_change_below(
hass: HomeAssistant, calls: list[ServiceCall], below: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], below: int | str
) -> None:
"""Test the firing with changed entity."""
hass.states.async_set("test.entity", 11)
@ -120,8 +109,8 @@ async def test_if_fires_on_entity_change_below(
# 9 is below 10
hass.states.async_set("test.entity", 9, context=context)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].context.parent_id == context.id
assert len(service_calls) == 1
assert service_calls[0].context.parent_id == context.id
# Set above 12 so the automation will fire again
hass.states.async_set("test.entity", 12)
@ -132,10 +121,12 @@ async def test_if_fires_on_entity_change_below(
{ATTR_ENTITY_ID: ENTITY_MATCH_ALL},
blocking=True,
)
assert len(service_calls) == 2
hass.states.async_set("test.entity", 9)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["id"] == 0
assert len(service_calls) == 2
assert service_calls[0].data["id"] == 0
@pytest.mark.parametrize(
@ -144,7 +135,7 @@ async def test_if_fires_on_entity_change_below(
async def test_if_fires_on_entity_change_below_uuid(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
below: int | str,
) -> None:
"""Test the firing with changed entity specified by registry entry id."""
@ -177,8 +168,8 @@ async def test_if_fires_on_entity_change_below_uuid(
# 9 is below 10
hass.states.async_set("test.entity", 9, context=context)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].context.parent_id == context.id
assert len(service_calls) == 1
assert service_calls[0].context.parent_id == context.id
# Set above 12 so the automation will fire again
hass.states.async_set("test.entity", 12)
@ -189,17 +180,19 @@ async def test_if_fires_on_entity_change_below_uuid(
{ATTR_ENTITY_ID: ENTITY_MATCH_ALL},
blocking=True,
)
assert len(service_calls) == 2
hass.states.async_set("test.entity", 9)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["id"] == 0
assert len(service_calls) == 2
assert service_calls[0].data["id"] == 0
@pytest.mark.parametrize(
"below", [10, "input_number.value_10", "number.value_10", "sensor.value_10"]
)
async def test_if_fires_on_entity_change_over_to_below(
hass: HomeAssistant, calls: list[ServiceCall], below: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], below: int | str
) -> None:
"""Test the firing with changed entity."""
hass.states.async_set("test.entity", 11)
@ -223,14 +216,14 @@ async def test_if_fires_on_entity_change_over_to_below(
# 9 is below 10
hass.states.async_set("test.entity", 9)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
@pytest.mark.parametrize(
"below", [10, "input_number.value_10", "number.value_10", "sensor.value_10"]
)
async def test_if_fires_on_entities_change_over_to_below(
hass: HomeAssistant, calls: list[ServiceCall], below: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], below: int | str
) -> None:
"""Test the firing with changed entities."""
hass.states.async_set("test.entity_1", 11)
@ -255,17 +248,17 @@ async def test_if_fires_on_entities_change_over_to_below(
# 9 is below 10
hass.states.async_set("test.entity_1", 9)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
hass.states.async_set("test.entity_2", 9)
await hass.async_block_till_done()
assert len(calls) == 2
assert len(service_calls) == 2
@pytest.mark.parametrize(
"below", [10, "input_number.value_10", "number.value_10", "sensor.value_10"]
)
async def test_if_not_fires_on_entity_change_below_to_below(
hass: HomeAssistant, calls: list[ServiceCall], below: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], below: int | str
) -> None:
"""Test the firing with changed entity."""
context = Context()
@ -290,25 +283,25 @@ async def test_if_not_fires_on_entity_change_below_to_below(
# 9 is below 10 so this should fire
hass.states.async_set("test.entity", 9, context=context)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].context.parent_id == context.id
assert len(service_calls) == 1
assert service_calls[0].context.parent_id == context.id
# already below so should not fire again
hass.states.async_set("test.entity", 5)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
# still below so should not fire again
hass.states.async_set("test.entity", 3)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
@pytest.mark.parametrize(
"below", [10, "input_number.value_10", "number.value_10", "sensor.value_10"]
)
async def test_if_not_below_fires_on_entity_change_to_equal(
hass: HomeAssistant, calls: list[ServiceCall], below: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], below: int | str
) -> None:
"""Test the firing with changed entity."""
hass.states.async_set("test.entity", 11)
@ -332,14 +325,14 @@ async def test_if_not_below_fires_on_entity_change_to_equal(
# 10 is not below 10 so this should not fire again
hass.states.async_set("test.entity", 10)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
@pytest.mark.parametrize(
"below", [10, "input_number.value_10", "number.value_10", "sensor.value_10"]
)
async def test_if_not_fires_on_initial_entity_below(
hass: HomeAssistant, calls: list[ServiceCall], below: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], below: int | str
) -> None:
"""Test the firing when starting with a match."""
hass.states.async_set("test.entity", 9)
@ -363,14 +356,14 @@ async def test_if_not_fires_on_initial_entity_below(
# Do not fire on first update when initial state was already below
hass.states.async_set("test.entity", 8)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
@pytest.mark.parametrize(
"above", [10, "input_number.value_10", "number.value_10", "sensor.value_10"]
)
async def test_if_not_fires_on_initial_entity_above(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], above: int | str
) -> None:
"""Test the firing when starting with a match."""
hass.states.async_set("test.entity", 11)
@ -394,14 +387,14 @@ async def test_if_not_fires_on_initial_entity_above(
# Do not fire on first update when initial state was already above
hass.states.async_set("test.entity", 12)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
@pytest.mark.parametrize(
"above", [10, "input_number.value_10", "number.value_10", "sensor.value_10"]
)
async def test_if_fires_on_entity_change_above(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], above: int | str
) -> None:
"""Test the firing with changed entity."""
hass.states.async_set("test.entity", 9)
@ -424,11 +417,11 @@ async def test_if_fires_on_entity_change_above(
# 11 is above 10
hass.states.async_set("test.entity", 11)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_entity_unavailable_at_startup(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test the firing with changed entity at startup."""
assert await async_setup_component(
@ -448,12 +441,12 @@ async def test_if_fires_on_entity_unavailable_at_startup(
# 11 is above 10
hass.states.async_set("test.entity", 11)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
@pytest.mark.parametrize("above", [10, "input_number.value_10"])
async def test_if_fires_on_entity_change_below_to_above(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], above: int | str
) -> None:
"""Test the firing with changed entity."""
# set initial state
@ -478,12 +471,12 @@ async def test_if_fires_on_entity_change_below_to_above(
# 11 is above 10 and 9 is below
hass.states.async_set("test.entity", 11)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
@pytest.mark.parametrize("above", [10, "input_number.value_10"])
async def test_if_not_fires_on_entity_change_above_to_above(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], above: int | str
) -> None:
"""Test the firing with changed entity."""
# set initial state
@ -508,17 +501,17 @@ async def test_if_not_fires_on_entity_change_above_to_above(
# 12 is above 10 so this should fire
hass.states.async_set("test.entity", 12)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
# already above, should not fire again
hass.states.async_set("test.entity", 15)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
@pytest.mark.parametrize("above", [10, "input_number.value_10"])
async def test_if_not_above_fires_on_entity_change_to_equal(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], above: int | str
) -> None:
"""Test the firing with changed entity."""
# set initial state
@ -543,7 +536,7 @@ async def test_if_not_above_fires_on_entity_change_to_equal(
# 10 is not above 10 so this should not fire again
hass.states.async_set("test.entity", 10)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
@pytest.mark.parametrize(
@ -556,7 +549,10 @@ async def test_if_not_above_fires_on_entity_change_to_equal(
],
)
async def test_if_fires_on_entity_change_below_range(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str, below: int | str
hass: HomeAssistant,
service_calls: list[ServiceCall],
above: int | str,
below: int | str,
) -> None:
"""Test the firing with changed entity."""
hass.states.async_set("test.entity", 11)
@ -580,7 +576,7 @@ async def test_if_fires_on_entity_change_below_range(
# 9 is below 10
hass.states.async_set("test.entity", 9)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
@pytest.mark.parametrize(
@ -593,7 +589,10 @@ async def test_if_fires_on_entity_change_below_range(
],
)
async def test_if_fires_on_entity_change_below_above_range(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str, below: int | str
hass: HomeAssistant,
service_calls: list[ServiceCall],
above: int | str,
below: int | str,
) -> None:
"""Test the firing with changed entity."""
assert await async_setup_component(
@ -614,7 +613,7 @@ async def test_if_fires_on_entity_change_below_above_range(
# 4 is below 5
hass.states.async_set("test.entity", 4)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
@pytest.mark.parametrize(
@ -627,7 +626,10 @@ async def test_if_fires_on_entity_change_below_above_range(
],
)
async def test_if_fires_on_entity_change_over_to_below_range(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str, below: int | str
hass: HomeAssistant,
service_calls: list[ServiceCall],
above: int | str,
below: int | str,
) -> None:
"""Test the firing with changed entity."""
hass.states.async_set("test.entity", 11)
@ -652,7 +654,7 @@ async def test_if_fires_on_entity_change_over_to_below_range(
# 9 is below 10
hass.states.async_set("test.entity", 9)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
@pytest.mark.parametrize(
@ -665,7 +667,10 @@ async def test_if_fires_on_entity_change_over_to_below_range(
],
)
async def test_if_fires_on_entity_change_over_to_below_above_range(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str, below: int | str
hass: HomeAssistant,
service_calls: list[ServiceCall],
above: int | str,
below: int | str,
) -> None:
"""Test the firing with changed entity."""
hass.states.async_set("test.entity", 11)
@ -690,12 +695,12 @@ async def test_if_fires_on_entity_change_over_to_below_above_range(
# 4 is below 5 so it should not fire
hass.states.async_set("test.entity", 4)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
@pytest.mark.parametrize("below", [100, "input_number.value_100"])
async def test_if_not_fires_if_entity_not_match(
hass: HomeAssistant, calls: list[ServiceCall], below: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], below: int | str
) -> None:
"""Test if not fired with non matching entity."""
assert await async_setup_component(
@ -715,11 +720,13 @@ async def test_if_not_fires_if_entity_not_match(
hass.states.async_set("test.entity", 11)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async def test_if_not_fires_and_warns_if_below_entity_unknown(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, calls: list[ServiceCall]
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
service_calls: list[ServiceCall],
) -> None:
"""Test if warns with unknown below entity."""
assert await async_setup_component(
@ -742,7 +749,7 @@ async def test_if_not_fires_and_warns_if_below_entity_unknown(
hass.states.async_set("test.entity", 1)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
assert len(caplog.record_tuples) == 1
assert caplog.record_tuples[0][1] == logging.WARNING
@ -750,7 +757,7 @@ async def test_if_not_fires_and_warns_if_below_entity_unknown(
@pytest.mark.parametrize("below", [10, "input_number.value_10"])
async def test_if_fires_on_entity_change_below_with_attribute(
hass: HomeAssistant, calls: list[ServiceCall], below: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], below: int | str
) -> None:
"""Test attributes change."""
hass.states.async_set("test.entity", 11, {"test_attribute": 11})
@ -773,12 +780,12 @@ async def test_if_fires_on_entity_change_below_with_attribute(
# 9 is below 10
hass.states.async_set("test.entity", 9, {"test_attribute": 11})
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
@pytest.mark.parametrize("below", [10, "input_number.value_10"])
async def test_if_not_fires_on_entity_change_not_below_with_attribute(
hass: HomeAssistant, calls: list[ServiceCall], below: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], below: int | str
) -> None:
"""Test attributes."""
assert await async_setup_component(
@ -798,12 +805,12 @@ async def test_if_not_fires_on_entity_change_not_below_with_attribute(
# 11 is not below 10
hass.states.async_set("test.entity", 11, {"test_attribute": 9})
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
@pytest.mark.parametrize("below", [10, "input_number.value_10"])
async def test_if_fires_on_attribute_change_with_attribute_below(
hass: HomeAssistant, calls: list[ServiceCall], below: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], below: int | str
) -> None:
"""Test attributes change."""
hass.states.async_set("test.entity", "entity", {"test_attribute": 11})
@ -827,12 +834,12 @@ async def test_if_fires_on_attribute_change_with_attribute_below(
# 9 is below 10
hass.states.async_set("test.entity", "entity", {"test_attribute": 9})
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
@pytest.mark.parametrize("below", [10, "input_number.value_10"])
async def test_if_not_fires_on_attribute_change_with_attribute_not_below(
hass: HomeAssistant, calls: list[ServiceCall], below: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], below: int | str
) -> None:
"""Test attributes change."""
assert await async_setup_component(
@ -853,12 +860,12 @@ async def test_if_not_fires_on_attribute_change_with_attribute_not_below(
# 11 is not below 10
hass.states.async_set("test.entity", "entity", {"test_attribute": 11})
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
@pytest.mark.parametrize("below", [10, "input_number.value_10"])
async def test_if_not_fires_on_entity_change_with_attribute_below(
hass: HomeAssistant, calls: list[ServiceCall], below: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], below: int | str
) -> None:
"""Test attributes change."""
assert await async_setup_component(
@ -879,12 +886,12 @@ async def test_if_not_fires_on_entity_change_with_attribute_below(
# 11 is not below 10, entity state value should not be tested
hass.states.async_set("test.entity", "9", {"test_attribute": 11})
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
@pytest.mark.parametrize("below", [10, "input_number.value_10"])
async def test_if_not_fires_on_entity_change_with_not_attribute_below(
hass: HomeAssistant, calls: list[ServiceCall], below: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], below: int | str
) -> None:
"""Test attributes change."""
assert await async_setup_component(
@ -905,12 +912,12 @@ async def test_if_not_fires_on_entity_change_with_not_attribute_below(
# 11 is not below 10, entity state value should not be tested
hass.states.async_set("test.entity", "entity")
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
@pytest.mark.parametrize("below", [10, "input_number.value_10"])
async def test_fires_on_attr_change_with_attribute_below_and_multiple_attr(
hass: HomeAssistant, calls: list[ServiceCall], below: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], below: int | str
) -> None:
"""Test attributes change."""
hass.states.async_set(
@ -937,12 +944,12 @@ async def test_fires_on_attr_change_with_attribute_below_and_multiple_attr(
"test.entity", "entity", {"test_attribute": 9, "not_test_attribute": 11}
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
@pytest.mark.parametrize("below", [10, "input_number.value_10"])
async def test_template_list(
hass: HomeAssistant, calls: list[ServiceCall], below: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], below: int | str
) -> None:
"""Test template list."""
hass.states.async_set("test.entity", "entity", {"test_attribute": [11, 15, 11]})
@ -965,12 +972,12 @@ async def test_template_list(
# 3 is below 10
hass.states.async_set("test.entity", "entity", {"test_attribute": [11, 15, 3]})
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
@pytest.mark.parametrize("below", [10.0, "input_number.value_10"])
async def test_template_string(
hass: HomeAssistant, calls: list[ServiceCall], below: float | str
hass: HomeAssistant, service_calls: list[ServiceCall], below: float | str
) -> None:
"""Test template string."""
assert await async_setup_component(
@ -1004,15 +1011,15 @@ async def test_template_string(
await hass.async_block_till_done()
hass.states.async_set("test.entity", "test state 2", {"test_attribute": "0.9"})
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
assert (
calls[0].data["some"]
service_calls[0].data["some"]
== f"numeric_state - test.entity - {below} - None - test state 1 - test state 2"
)
async def test_not_fires_on_attr_change_with_attr_not_below_multiple_attr(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test if not fired changed attributes."""
assert await async_setup_component(
@ -1035,7 +1042,7 @@ async def test_not_fires_on_attr_change_with_attr_not_below_multiple_attr(
"test.entity", "entity", {"test_attribute": 11, "not_test_attribute": 9}
)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
@pytest.mark.parametrize(
@ -1048,7 +1055,10 @@ async def test_not_fires_on_attr_change_with_attr_not_below_multiple_attr(
],
)
async def test_if_action(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str, below: int | str
hass: HomeAssistant,
service_calls: list[ServiceCall],
above: int | str,
below: int | str,
) -> None:
"""Test if action."""
entity_id = "domain.test_entity"
@ -1073,19 +1083,19 @@ async def test_if_action(
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
hass.states.async_set(entity_id, 8)
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
hass.states.async_set(entity_id, 9)
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 2
assert len(service_calls) == 2
@pytest.mark.parametrize(
@ -1098,7 +1108,7 @@ async def test_if_action(
],
)
async def test_if_fails_setup_bad_for(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str, below: int | str
hass: HomeAssistant, above: int | str, below: int | str
) -> None:
"""Test for setup failure for bad for."""
hass.states.async_set("test.entity", 5)
@ -1124,9 +1134,7 @@ async def test_if_fails_setup_bad_for(
assert hass.states.get("automation.automation_0").state == STATE_UNAVAILABLE
async def test_if_fails_setup_for_without_above_below(
hass: HomeAssistant, calls: list[ServiceCall]
) -> None:
async def test_if_fails_setup_for_without_above_below(hass: HomeAssistant) -> None:
"""Test for setup failures for missing above or below."""
with assert_setup_component(1, automation.DOMAIN):
assert await async_setup_component(
@ -1158,7 +1166,7 @@ async def test_if_fails_setup_for_without_above_below(
async def test_if_not_fires_on_entity_change_with_for(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
above: int | str,
below: int | str,
) -> None:
@ -1187,7 +1195,7 @@ async def test_if_not_fires_on_entity_change_with_for(
freezer.tick(timedelta(seconds=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
@pytest.mark.parametrize(
@ -1200,7 +1208,10 @@ async def test_if_not_fires_on_entity_change_with_for(
],
)
async def test_if_not_fires_on_entities_change_with_for_after_stop(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str, below: int | str
hass: HomeAssistant,
service_calls: list[ServiceCall],
above: int | str,
below: int | str,
) -> None:
"""Test for not firing on entities change with for after stop."""
hass.states.async_set("test.entity_1", 0)
@ -1232,7 +1243,7 @@ async def test_if_not_fires_on_entities_change_with_for_after_stop(
await hass.async_block_till_done()
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
hass.states.async_set("test.entity_1", 15)
hass.states.async_set("test.entity_2", 15)
@ -1246,9 +1257,11 @@ async def test_if_not_fires_on_entities_change_with_for_after_stop(
{ATTR_ENTITY_ID: ENTITY_MATCH_ALL},
blocking=True,
)
assert len(service_calls) == 2
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 2
@pytest.mark.parametrize(
@ -1263,7 +1276,7 @@ async def test_if_not_fires_on_entities_change_with_for_after_stop(
async def test_if_fires_on_entity_change_with_for_attribute_change(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
above: int | str,
below: int | str,
) -> None:
@ -1294,11 +1307,11 @@ async def test_if_fires_on_entity_change_with_for_attribute_change(
async_fire_time_changed(hass)
hass.states.async_set("test.entity", 9, attributes={"mock_attr": "attr_change"})
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
freezer.tick(timedelta(seconds=4))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
@pytest.mark.parametrize(
@ -1311,7 +1324,10 @@ async def test_if_fires_on_entity_change_with_for_attribute_change(
],
)
async def test_if_fires_on_entity_change_with_for(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str, below: int | str
hass: HomeAssistant,
service_calls: list[ServiceCall],
above: int | str,
below: int | str,
) -> None:
"""Test for firing on entity change with for."""
hass.states.async_set("test.entity", 0)
@ -1338,12 +1354,12 @@ async def test_if_fires_on_entity_change_with_for(
await hass.async_block_till_done()
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
@pytest.mark.parametrize("above", [10, "input_number.value_10"])
async def test_wait_template_with_trigger(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], above: int | str
) -> None:
"""Test using wait template with 'trigger.entity_id'."""
hass.states.async_set("test.entity", "0")
@ -1381,8 +1397,8 @@ async def test_wait_template_with_trigger(
hass.states.async_set("test.entity", "12")
hass.states.async_set("test.entity", "8")
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "numeric_state - test.entity - 12"
assert len(service_calls) == 1
assert service_calls[0].data["some"] == "numeric_state - test.entity - 12"
@pytest.mark.parametrize(
@ -1397,7 +1413,7 @@ async def test_wait_template_with_trigger(
async def test_if_fires_on_entities_change_no_overlap(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
above: int | str,
below: int | str,
) -> None:
@ -1432,16 +1448,16 @@ async def test_if_fires_on_entities_change_no_overlap(
freezer.tick(timedelta(seconds=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "test.entity_1"
assert len(service_calls) == 1
assert service_calls[0].data["some"] == "test.entity_1"
hass.states.async_set("test.entity_2", 9)
await hass.async_block_till_done()
freezer.tick(timedelta(seconds=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 2
assert calls[1].data["some"] == "test.entity_2"
assert len(service_calls) == 2
assert service_calls[1].data["some"] == "test.entity_2"
@pytest.mark.parametrize(
@ -1456,7 +1472,7 @@ async def test_if_fires_on_entities_change_no_overlap(
async def test_if_fires_on_entities_change_overlap(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
above: int | str,
below: int | str,
) -> None:
@ -1500,18 +1516,18 @@ async def test_if_fires_on_entities_change_overlap(
async_fire_time_changed(hass)
hass.states.async_set("test.entity_2", 9)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
freezer.tick(timedelta(seconds=3))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "test.entity_1"
assert len(service_calls) == 1
assert service_calls[0].data["some"] == "test.entity_1"
freezer.tick(timedelta(seconds=3))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 2
assert calls[1].data["some"] == "test.entity_2"
assert len(service_calls) == 2
assert service_calls[1].data["some"] == "test.entity_2"
@pytest.mark.parametrize(
@ -1524,7 +1540,10 @@ async def test_if_fires_on_entities_change_overlap(
],
)
async def test_if_fires_on_change_with_for_template_1(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str, below: int | str
hass: HomeAssistant,
service_calls: list[ServiceCall],
above: int | str,
below: int | str,
) -> None:
"""Test for firing on change with for template."""
hass.states.async_set("test.entity", 0)
@ -1549,10 +1568,10 @@ async def test_if_fires_on_change_with_for_template_1(
hass.states.async_set("test.entity", 9)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
@pytest.mark.parametrize(
@ -1565,7 +1584,10 @@ async def test_if_fires_on_change_with_for_template_1(
],
)
async def test_if_fires_on_change_with_for_template_2(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str, below: int | str
hass: HomeAssistant,
service_calls: list[ServiceCall],
above: int | str,
below: int | str,
) -> None:
"""Test for firing on change with for template."""
hass.states.async_set("test.entity", 0)
@ -1590,10 +1612,10 @@ async def test_if_fires_on_change_with_for_template_2(
hass.states.async_set("test.entity", 9)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
@pytest.mark.parametrize(
@ -1606,7 +1628,10 @@ async def test_if_fires_on_change_with_for_template_2(
],
)
async def test_if_fires_on_change_with_for_template_3(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str, below: int | str
hass: HomeAssistant,
service_calls: list[ServiceCall],
above: int | str,
below: int | str,
) -> None:
"""Test for firing on change with for template."""
hass.states.async_set("test.entity", 0)
@ -1631,14 +1656,14 @@ async def test_if_fires_on_change_with_for_template_3(
hass.states.async_set("test.entity", 9)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_not_fires_on_error_with_for_template(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for not firing on error with for template."""
hass.states.async_set("test.entity", 0)
@ -1662,17 +1687,17 @@ async def test_if_not_fires_on_error_with_for_template(
hass.states.async_set("test.entity", 101)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=3))
hass.states.async_set("test.entity", "unavailable")
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=3))
hass.states.async_set("test.entity", 101)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
@pytest.mark.parametrize(
@ -1685,7 +1710,7 @@ async def test_if_not_fires_on_error_with_for_template(
],
)
async def test_invalid_for_template(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str, below: int | str
hass: HomeAssistant, above: int | str, below: int | str
) -> None:
"""Test for invalid for template."""
hass.states.async_set("test.entity", 0)
@ -1726,7 +1751,7 @@ async def test_invalid_for_template(
async def test_if_fires_on_entities_change_overlap_for_template(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
above: int | str,
below: int | str,
) -> None:
@ -1773,22 +1798,22 @@ async def test_if_fires_on_entities_change_overlap_for_template(
async_fire_time_changed(hass)
hass.states.async_set("test.entity_2", 9)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
freezer.tick(timedelta(seconds=3))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "test.entity_1 - 0:00:05"
assert len(service_calls) == 1
assert service_calls[0].data["some"] == "test.entity_1 - 0:00:05"
freezer.tick(timedelta(seconds=3))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
freezer.tick(timedelta(seconds=5))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 2
assert calls[1].data["some"] == "test.entity_2 - 0:00:10"
assert len(service_calls) == 2
assert service_calls[1].data["some"] == "test.entity_2 - 0:00:10"
async def test_below_above(hass: HomeAssistant) -> None:
@ -1823,7 +1848,7 @@ async def test_schema_unacceptable_entities(hass: HomeAssistant) -> None:
@pytest.mark.parametrize("above", [3, "input_number.value_3"])
async def test_attribute_if_fires_on_entity_change_with_both_filters(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], above: int | str
) -> None:
"""Test for firing if both filters are match attribute."""
hass.states.async_set("test.entity", "bla", {"test-measurement": 1})
@ -1847,12 +1872,12 @@ async def test_attribute_if_fires_on_entity_change_with_both_filters(
hass.states.async_set("test.entity", "bla", {"test-measurement": 4})
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
@pytest.mark.parametrize("above", [3, "input_number.value_3"])
async def test_attribute_if_not_fires_on_entities_change_with_for_after_stop(
hass: HomeAssistant, calls: list[ServiceCall], above: int | str
hass: HomeAssistant, service_calls: list[ServiceCall], above: int | str
) -> None:
"""Test for not firing on entity change with for after stop trigger."""
hass.states.async_set("test.entity", "bla", {"test-measurement": 1})
@ -1880,10 +1905,10 @@ async def test_attribute_if_not_fires_on_entities_change_with_for_after_stop(
hass.states.async_set("test.entity", "bla", {"test-measurement": 4})
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
@pytest.mark.parametrize(
@ -1893,7 +1918,7 @@ async def test_attribute_if_not_fires_on_entities_change_with_for_after_stop(
async def test_variables_priority(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
above: int,
below: int,
) -> None:
@ -1941,17 +1966,17 @@ async def test_variables_priority(
async_fire_time_changed(hass)
hass.states.async_set("test.entity_2", 9)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
freezer.tick(timedelta(seconds=3))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "test.entity_1 - 0:00:05"
assert len(service_calls) == 1
assert service_calls[0].data["some"] == "test.entity_1 - 0:00:05"
@pytest.mark.parametrize("multiplier", [1, 5])
async def test_template_variable(
hass: HomeAssistant, calls: list[ServiceCall], multiplier: int
hass: HomeAssistant, service_calls: list[ServiceCall], multiplier: int
) -> None:
"""Test template variable."""
hass.states.async_set("test.entity", "entity", {"test_attribute": [11, 15, 11]})
@ -1976,6 +2001,6 @@ async def test_template_variable(
hass.states.async_set("test.entity", "entity", {"test_attribute": [11, 15, 3]})
await hass.async_block_till_done()
if multiplier * 3 < 10:
assert len(calls) == 1
assert len(service_calls) == 1
else:
assert len(calls) == 0
assert len(service_calls) == 0

363
tests/components/homeassistant/triggers/test_state.py
View File

@ -19,29 +19,18 @@ from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from tests.common import (
assert_setup_component,
async_fire_time_changed,
async_mock_service,
mock_component,
)
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
from tests.common import assert_setup_component, async_fire_time_changed, mock_component
@pytest.fixture(autouse=True)
def setup_comp(hass):
def setup_comp(hass: HomeAssistant) -> None:
"""Initialize components."""
mock_component(hass, "group")
hass.states.async_set("test.entity", "hello")
async def test_if_fires_on_entity_change(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing on entity change."""
context = Context()
@ -74,9 +63,12 @@ async def test_if_fires_on_entity_change(
hass.states.async_set("test.entity", "world", context=context)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].context.parent_id == context.id
assert calls[0].data["some"] == "state - test.entity - hello - world - None - 0"
assert len(service_calls) == 1
assert service_calls[0].context.parent_id == context.id
assert (
service_calls[0].data["some"]
== "state - test.entity - hello - world - None - 0"
)
await hass.services.async_call(
automation.DOMAIN,
@ -84,13 +76,16 @@ async def test_if_fires_on_entity_change(
{ATTR_ENTITY_ID: ENTITY_MATCH_ALL},
blocking=True,
)
assert len(service_calls) == 2
hass.states.async_set("test.entity", "planet")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 2
async def test_if_fires_on_entity_change_uuid(
hass: HomeAssistant, entity_registry: er.EntityRegistry, calls: list[ServiceCall]
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing on entity change."""
context = Context()
@ -130,9 +125,11 @@ async def test_if_fires_on_entity_change_uuid(
hass.states.async_set("test.beer", "world", context=context)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].context.parent_id == context.id
assert calls[0].data["some"] == "state - test.beer - hello - world - None - 0"
assert len(service_calls) == 1
assert service_calls[0].context.parent_id == context.id
assert (
service_calls[0].data["some"] == "state - test.beer - hello - world - None - 0"
)
await hass.services.async_call(
automation.DOMAIN,
@ -140,13 +137,14 @@ async def test_if_fires_on_entity_change_uuid(
{ATTR_ENTITY_ID: ENTITY_MATCH_ALL},
blocking=True,
)
assert len(service_calls) == 2
hass.states.async_set("test.beer", "planet")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 2
async def test_if_fires_on_entity_change_with_from_filter(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing on entity change with filter."""
assert await async_setup_component(
@ -167,11 +165,11 @@ async def test_if_fires_on_entity_change_with_from_filter(
hass.states.async_set("test.entity", "world")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_entity_change_with_not_from_filter(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing on entity change inverse filter."""
assert await async_setup_component(
@ -193,15 +191,15 @@ async def test_if_fires_on_entity_change_with_not_from_filter(
# Do not fire from hello
hass.states.async_set("test.entity", "world")
await hass.async_block_till_done()
assert not calls
assert not service_calls
hass.states.async_set("test.entity", "universum")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_entity_change_with_to_filter(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing on entity change with to filter."""
assert await async_setup_component(
@ -222,11 +220,11 @@ async def test_if_fires_on_entity_change_with_to_filter(
hass.states.async_set("test.entity", "world")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_entity_change_with_not_to_filter(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing on entity change with to filter."""
assert await async_setup_component(
@ -248,15 +246,15 @@ async def test_if_fires_on_entity_change_with_not_to_filter(
# Do not fire to world
hass.states.async_set("test.entity", "world")
await hass.async_block_till_done()
assert not calls
assert not service_calls
hass.states.async_set("test.entity", "universum")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_entity_change_with_from_filter_all(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing on entity change with filter."""
assert await async_setup_component(
@ -278,11 +276,11 @@ async def test_if_fires_on_entity_change_with_from_filter_all(
hass.states.async_set("test.entity", "world")
hass.states.async_set("test.entity", "world", {"attribute": 5})
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_entity_change_with_to_filter_all(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing on entity change with to filter."""
assert await async_setup_component(
@ -304,11 +302,11 @@ async def test_if_fires_on_entity_change_with_to_filter_all(
hass.states.async_set("test.entity", "world")
hass.states.async_set("test.entity", "world", {"attribute": 5})
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_attribute_change_with_to_filter(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for not firing on attribute change."""
assert await async_setup_component(
@ -330,11 +328,11 @@ async def test_if_fires_on_attribute_change_with_to_filter(
hass.states.async_set("test.entity", "world", {"test_attribute": 11})
hass.states.async_set("test.entity", "world", {"test_attribute": 12})
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_entity_change_with_both_filters(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing if both filters are a non match."""
assert await async_setup_component(
@ -356,11 +354,11 @@ async def test_if_fires_on_entity_change_with_both_filters(
hass.states.async_set("test.entity", "world")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_entity_change_with_not_from_to(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing if not from doesn't match and to match."""
assert await async_setup_component(
@ -383,31 +381,31 @@ async def test_if_fires_on_entity_change_with_not_from_to(
# We should not trigger from hello
hass.states.async_set("test.entity", "world")
await hass.async_block_till_done()
assert not calls
assert not service_calls
# We should not trigger to != galaxy
hass.states.async_set("test.entity", "world")
await hass.async_block_till_done()
assert not calls
assert not service_calls
# We should trigger to galaxy
hass.states.async_set("test.entity", "galaxy")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
# We should not trigger from milky way
hass.states.async_set("test.entity", "milky_way")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
# We should trigger to universe
hass.states.async_set("test.entity", "universe")
await hass.async_block_till_done()
assert len(calls) == 2
assert len(service_calls) == 2
async def test_if_fires_on_entity_change_with_from_not_to(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing if not from doesn't match and to match."""
assert await async_setup_component(
@ -430,31 +428,31 @@ async def test_if_fires_on_entity_change_with_from_not_to(
# We should trigger to world from hello
hass.states.async_set("test.entity", "world")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
# Reset back to hello, should not trigger
hass.states.async_set("test.entity", "hello")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
# We should not trigger to galaxy
hass.states.async_set("test.entity", "galaxy")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
# We should trigger form galaxy to milky way
hass.states.async_set("test.entity", "milky_way")
await hass.async_block_till_done()
assert len(calls) == 2
assert len(service_calls) == 2
# We should not trigger to universe
hass.states.async_set("test.entity", "universe")
await hass.async_block_till_done()
assert len(calls) == 2
assert len(service_calls) == 2
async def test_if_not_fires_if_to_filter_not_match(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for not firing if to filter is not a match."""
assert await async_setup_component(
@ -476,11 +474,11 @@ async def test_if_not_fires_if_to_filter_not_match(
hass.states.async_set("test.entity", "moon")
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async def test_if_not_fires_if_from_filter_not_match(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for not firing if from filter is not a match."""
hass.states.async_set("test.entity", "bye")
@ -504,11 +502,11 @@ async def test_if_not_fires_if_from_filter_not_match(
hass.states.async_set("test.entity", "world")
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async def test_if_not_fires_if_entity_not_match(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for not firing if entity is not matching."""
assert await async_setup_component(
@ -525,10 +523,10 @@ async def test_if_not_fires_if_entity_not_match(
hass.states.async_set("test.entity", "world")
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async def test_if_action(hass: HomeAssistant, calls: list[ServiceCall]) -> None:
async def test_if_action(hass: HomeAssistant, service_calls: list[ServiceCall]) -> None:
"""Test for to action."""
entity_id = "domain.test_entity"
test_state = "new_state"
@ -551,18 +549,16 @@ async def test_if_action(hass: HomeAssistant, calls: list[ServiceCall]) -> None:
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
hass.states.async_set(entity_id, test_state + "something")
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fails_setup_if_to_boolean_value(
hass: HomeAssistant, calls: list[ServiceCall]
) -> None:
async def test_if_fails_setup_if_to_boolean_value(hass: HomeAssistant) -> None:
"""Test for setup failure for boolean to."""
with assert_setup_component(1, automation.DOMAIN):
assert await async_setup_component(
@ -582,9 +578,7 @@ async def test_if_fails_setup_if_to_boolean_value(
assert hass.states.get("automation.automation_0").state == STATE_UNAVAILABLE
async def test_if_fails_setup_if_from_boolean_value(
hass: HomeAssistant, calls: list[ServiceCall]
) -> None:
async def test_if_fails_setup_if_from_boolean_value(hass: HomeAssistant) -> None:
"""Test for setup failure for boolean from."""
with assert_setup_component(1, automation.DOMAIN):
assert await async_setup_component(
@ -604,9 +598,7 @@ async def test_if_fails_setup_if_from_boolean_value(
assert hass.states.get("automation.automation_0").state == STATE_UNAVAILABLE
async def test_if_fails_setup_bad_for(
hass: HomeAssistant, calls: list[ServiceCall]
) -> None:
async def test_if_fails_setup_bad_for(hass: HomeAssistant) -> None:
"""Test for setup failure for bad for."""
with assert_setup_component(1, automation.DOMAIN):
assert await async_setup_component(
@ -628,7 +620,7 @@ async def test_if_fails_setup_bad_for(
async def test_if_not_fires_on_entity_change_with_for(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for not firing on entity change with for."""
assert await async_setup_component(
@ -654,11 +646,11 @@ async def test_if_not_fires_on_entity_change_with_for(
await hass.async_block_till_done()
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async def test_if_not_fires_on_entities_change_with_for_after_stop(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for not firing on entity change with for after stop trigger."""
assert await async_setup_component(
@ -686,7 +678,7 @@ async def test_if_not_fires_on_entities_change_with_for_after_stop(
await hass.async_block_till_done()
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
hass.states.async_set("test.entity_1", "world_no")
hass.states.async_set("test.entity_2", "world_no")
@ -700,14 +692,17 @@ async def test_if_not_fires_on_entities_change_with_for_after_stop(
{ATTR_ENTITY_ID: ENTITY_MATCH_ALL},
blocking=True,
)
assert len(service_calls) == 2
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 2
async def test_if_fires_on_entity_change_with_for_attribute_change(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing on entity change with for and attribute change."""
assert await async_setup_component(
@ -735,15 +730,17 @@ async def test_if_fires_on_entity_change_with_for_attribute_change(
"test.entity", "world", attributes={"mock_attr": "attr_change"}
)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
freezer.tick(timedelta(seconds=4))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_entity_change_with_for_multiple_force_update(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing on entity change with for and force update."""
assert await async_setup_component(
@ -770,15 +767,15 @@ async def test_if_fires_on_entity_change_with_for_multiple_force_update(
async_fire_time_changed(hass)
hass.states.async_set("test.force_entity", "world", None, True)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
freezer.tick(timedelta(seconds=4))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_entity_change_with_for(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing on entity change with for."""
assert await async_setup_component(
@ -802,11 +799,11 @@ async def test_if_fires_on_entity_change_with_for(
await hass.async_block_till_done()
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_entity_change_with_for_without_to(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing on entity change with for."""
assert await async_setup_component(
@ -830,22 +827,24 @@ async def test_if_fires_on_entity_change_with_for_without_to(
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=2))
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
hass.states.async_set("test.entity", "world")
await hass.async_block_till_done()
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=4))
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_does_not_fires_on_entity_change_with_for_without_to_2(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing on entity change with for."""
assert await async_setup_component(
@ -871,11 +870,11 @@ async def test_if_does_not_fires_on_entity_change_with_for_without_to_2(
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async def test_if_fires_on_entity_creation_and_removal(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing on entity creation and removal, with to/from constraints."""
# set automations for multiple combinations to/from
@ -917,32 +916,32 @@ async def test_if_fires_on_entity_creation_and_removal(
# automation with match_all triggers on creation
hass.states.async_set("test.entity_0", "any", context=context_0)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].context.parent_id == context_0.id
assert len(service_calls) == 1
assert service_calls[0].context.parent_id == context_0.id
# create entities, trigger on test.entity_2 ('to' matches, no 'from')
hass.states.async_set("test.entity_1", "hello", context=context_1)
hass.states.async_set("test.entity_2", "world", context=context_2)
await hass.async_block_till_done()
assert len(calls) == 2
assert calls[1].context.parent_id == context_2.id
assert len(service_calls) == 2
assert service_calls[1].context.parent_id == context_2.id
# removal of both, trigger on test.entity_1 ('from' matches, no 'to')
assert hass.states.async_remove("test.entity_1", context=context_1)
assert hass.states.async_remove("test.entity_2", context=context_2)
await hass.async_block_till_done()
assert len(calls) == 3
assert calls[2].context.parent_id == context_1.id
assert len(service_calls) == 3
assert service_calls[2].context.parent_id == context_1.id
# automation with match_all triggers on removal
assert hass.states.async_remove("test.entity_0", context=context_0)
await hass.async_block_till_done()
assert len(calls) == 4
assert calls[3].context.parent_id == context_0.id
assert len(service_calls) == 4
assert service_calls[3].context.parent_id == context_0.id
async def test_if_fires_on_for_condition(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing if condition is on."""
point1 = dt_util.utcnow()
@ -971,17 +970,17 @@ async def test_if_fires_on_for_condition(
# not enough time has passed
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
# Time travel 10 secs into the future
mock_utcnow.return_value = point2
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_for_condition_attribute_change(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing if condition is on with attribute change."""
point1 = dt_util.utcnow()
@ -1011,7 +1010,7 @@ async def test_if_fires_on_for_condition_attribute_change(
# not enough time has passed
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
# Still not enough time has passed, but an attribute is changed
mock_utcnow.return_value = point2
@ -1020,18 +1019,16 @@ async def test_if_fires_on_for_condition_attribute_change(
)
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
# Enough time has now passed
mock_utcnow.return_value = point3
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fails_setup_for_without_time(
hass: HomeAssistant, calls: list[ServiceCall]
) -> None:
async def test_if_fails_setup_for_without_time(hass: HomeAssistant) -> None:
"""Test for setup failure if no time is provided."""
with assert_setup_component(1, automation.DOMAIN):
assert await async_setup_component(
@ -1053,9 +1050,7 @@ async def test_if_fails_setup_for_without_time(
assert hass.states.get("automation.automation_0").state == STATE_UNAVAILABLE
async def test_if_fails_setup_for_without_entity(
hass: HomeAssistant, calls: list[ServiceCall]
) -> None:
async def test_if_fails_setup_for_without_entity(hass: HomeAssistant) -> None:
"""Test for setup failure if no entity is provided."""
with assert_setup_component(1, automation.DOMAIN):
assert await async_setup_component(
@ -1077,7 +1072,7 @@ async def test_if_fails_setup_for_without_entity(
async def test_wait_template_with_trigger(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test using wait template with 'trigger.entity_id'."""
assert await async_setup_component(
@ -1113,12 +1108,14 @@ async def test_wait_template_with_trigger(
hass.states.async_set("test.entity", "world")
hass.states.async_set("test.entity", "hello")
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "state - test.entity - hello - world"
assert len(service_calls) == 1
assert service_calls[0].data["some"] == "state - test.entity - hello - world"
async def test_if_fires_on_entities_change_no_overlap(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing on entities change with no overlap."""
assert await async_setup_component(
@ -1146,20 +1143,22 @@ async def test_if_fires_on_entities_change_no_overlap(
freezer.tick(timedelta(seconds=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "test.entity_1"
assert len(service_calls) == 1
assert service_calls[0].data["some"] == "test.entity_1"
hass.states.async_set("test.entity_2", "world")
await hass.async_block_till_done()
freezer.tick(timedelta(seconds=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 2
assert calls[1].data["some"] == "test.entity_2"
assert len(service_calls) == 2
assert service_calls[1].data["some"] == "test.entity_2"
async def test_if_fires_on_entities_change_overlap(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing on entities change with overlap."""
assert await async_setup_component(
@ -1196,22 +1195,22 @@ async def test_if_fires_on_entities_change_overlap(
async_fire_time_changed(hass)
hass.states.async_set("test.entity_2", "world")
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
freezer.tick(timedelta(seconds=3))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "test.entity_1"
assert len(service_calls) == 1
assert service_calls[0].data["some"] == "test.entity_1"
freezer.tick(timedelta(seconds=3))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 2
assert calls[1].data["some"] == "test.entity_2"
assert len(service_calls) == 2
assert service_calls[1].data["some"] == "test.entity_2"
async def test_if_fires_on_change_with_for_template_1(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing on change with for template."""
assert await async_setup_component(
@ -1232,14 +1231,14 @@ async def test_if_fires_on_change_with_for_template_1(
hass.states.async_set("test.entity", "world")
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_change_with_for_template_2(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing on change with for template."""
assert await async_setup_component(
@ -1260,14 +1259,14 @@ async def test_if_fires_on_change_with_for_template_2(
hass.states.async_set("test.entity", "world")
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_change_with_for_template_3(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing on change with for template."""
assert await async_setup_component(
@ -1288,14 +1287,14 @@ async def test_if_fires_on_change_with_for_template_3(
hass.states.async_set("test.entity", "world")
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_change_with_for_template_4(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing on change with for template."""
assert await async_setup_component(
@ -1317,14 +1316,14 @@ async def test_if_fires_on_change_with_for_template_4(
hass.states.async_set("test.entity", "world")
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_on_change_from_with_for(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing on change with from/for."""
assert await async_setup_component(
@ -1351,11 +1350,11 @@ async def test_if_fires_on_change_from_with_for(
await hass.async_block_till_done()
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(minutes=1))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_not_fires_on_change_from_with_for(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing on change with from/for."""
assert await async_setup_component(
@ -1382,12 +1381,10 @@ async def test_if_not_fires_on_change_from_with_for(
await hass.async_block_till_done()
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(minutes=1))
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async def test_invalid_for_template_1(
hass: HomeAssistant, calls: list[ServiceCall]
) -> None:
async def test_invalid_for_template_1(hass: HomeAssistant) -> None:
"""Test for invalid for template."""
assert await async_setup_component(
hass,
@ -1412,7 +1409,9 @@ async def test_invalid_for_template_1(
async def test_if_fires_on_entities_change_overlap_for_template(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing on entities change with overlap and for template."""
assert await async_setup_component(
@ -1452,26 +1451,26 @@ async def test_if_fires_on_entities_change_overlap_for_template(
async_fire_time_changed(hass)
hass.states.async_set("test.entity_2", "world")
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
freezer.tick(timedelta(seconds=3))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "test.entity_1 - 0:00:05"
assert len(service_calls) == 1
assert service_calls[0].data["some"] == "test.entity_1 - 0:00:05"
freezer.tick(timedelta(seconds=3))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
freezer.tick(timedelta(seconds=5))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 2
assert calls[1].data["some"] == "test.entity_2 - 0:00:10"
assert len(service_calls) == 2
assert service_calls[1].data["some"] == "test.entity_2 - 0:00:10"
async def test_attribute_if_fires_on_entity_change_with_both_filters(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing if both filters are match attribute."""
hass.states.async_set("test.entity", "bla", {"name": "hello"})
@ -1496,11 +1495,11 @@ async def test_attribute_if_fires_on_entity_change_with_both_filters(
hass.states.async_set("test.entity", "bla", {"name": "world"})
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_attribute_if_fires_on_entity_where_attr_stays_constant(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing if attribute stays the same."""
hass.states.async_set("test.entity", "bla", {"name": "hello", "other": "old_value"})
@ -1524,21 +1523,21 @@ async def test_attribute_if_fires_on_entity_where_attr_stays_constant(
# Leave all attributes the same
hass.states.async_set("test.entity", "bla", {"name": "hello", "other": "old_value"})
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
# Change the untracked attribute
hass.states.async_set("test.entity", "bla", {"name": "hello", "other": "new_value"})
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
# Change the tracked attribute
hass.states.async_set("test.entity", "bla", {"name": "world", "other": "old_value"})
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_attribute_if_fires_on_entity_where_attr_stays_constant_filter(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing if attribute stays the same."""
hass.states.async_set("test.entity", "bla", {"name": "other_name"})
@ -1565,25 +1564,25 @@ async def test_attribute_if_fires_on_entity_where_attr_stays_constant_filter(
"test.entity", "bla", {"name": "best_name", "other": "old_value"}
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
# Change the untracked attribute
hass.states.async_set(
"test.entity", "bla", {"name": "best_name", "other": "new_value"}
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
# Change the tracked attribute
hass.states.async_set(
"test.entity", "bla", {"name": "other_name", "other": "old_value"}
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_attribute_if_fires_on_entity_where_attr_stays_constant_all(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing if attribute stays the same."""
hass.states.async_set("test.entity", "bla", {"name": "hello", "other": "old_value"})
@ -1610,25 +1609,25 @@ async def test_attribute_if_fires_on_entity_where_attr_stays_constant_all(
"test.entity", "bla", {"name": "name_1", "other": "old_value"}
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
# Change the untracked attribute
hass.states.async_set(
"test.entity", "bla", {"name": "name_1", "other": "new_value"}
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
# Change the tracked attribute
hass.states.async_set(
"test.entity", "bla", {"name": "name_2", "other": "old_value"}
)
await hass.async_block_till_done()
assert len(calls) == 2
assert len(service_calls) == 2
async def test_attribute_if_not_fires_on_entities_change_with_for_after_stop(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for not firing on entity change with for after stop trigger."""
hass.states.async_set("test.entity", "bla", {"name": "hello"})
@ -1658,33 +1657,33 @@ async def test_attribute_if_not_fires_on_entities_change_with_for_after_stop(
# Test that the for-check works
hass.states.async_set("test.entity", "bla", {"name": "world"})
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=2))
hass.states.async_set("test.entity", "bla", {"name": "world", "something": "else"})
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
# Now remove state while inside "for"
hass.states.async_set("test.entity", "bla", {"name": "hello"})
hass.states.async_set("test.entity", "bla", {"name": "world"})
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
hass.states.async_remove("test.entity")
await hass.async_block_till_done()
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_attribute_if_fires_on_entity_change_with_both_filters_boolean(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for firing if both filters are match attribute."""
hass.states.async_set("test.entity", "bla", {"happening": False})
@ -1709,11 +1708,13 @@ async def test_attribute_if_fires_on_entity_change_with_both_filters_boolean(
hass.states.async_set("test.entity", "bla", {"happening": True})
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_variables_priority(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test an externally defined trigger variable is overridden."""
assert await async_setup_component(
@ -1754,19 +1755,19 @@ async def test_variables_priority(
async_fire_time_changed(hass)
hass.states.async_set("test.entity_2", "world")
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
freezer.tick(timedelta(seconds=3))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "test.entity_1 - 0:00:05"
assert len(service_calls) == 1
assert service_calls[0].data["some"] == "test.entity_1 - 0:00:05"
freezer.tick(timedelta(seconds=3))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
freezer.tick(timedelta(seconds=5))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 2
assert calls[1].data["some"] == "test.entity_2 - 0:00:10"
assert len(service_calls) == 2
assert service_calls[1].data["some"] == "test.entity_2 - 0:00:10"

110
tests/components/homeassistant/triggers/test_time.py
View File

@ -20,28 +20,19 @@ from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from tests.common import (
assert_setup_component,
async_fire_time_changed,
async_mock_service,
mock_component,
)
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
from tests.common import assert_setup_component, async_fire_time_changed, mock_component
@pytest.fixture(autouse=True)
def setup_comp(hass):
def setup_comp(hass: HomeAssistant) -> None:
"""Initialize components."""
mock_component(hass, "group")
async def test_if_fires_using_at(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing at."""
now = dt_util.now()
@ -71,9 +62,9 @@ async def test_if_fires_using_at(
async_fire_time_changed(hass, trigger_dt + timedelta(seconds=1))
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "time - 5"
assert calls[0].data["id"] == 0
assert len(service_calls) == 1
assert service_calls[0].data["some"] == "time - 5"
assert service_calls[0].data["id"] == 0
@pytest.mark.parametrize(
@ -82,7 +73,7 @@ async def test_if_fires_using_at(
async def test_if_fires_using_at_input_datetime(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
calls: list[ServiceCall],
service_calls: list[ServiceCall],
has_date,
has_time,
) -> None:
@ -132,9 +123,9 @@ async def test_if_fires_using_at_input_datetime(
async_fire_time_changed(hass, trigger_dt + timedelta(seconds=1))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 2
assert (
calls[0].data["some"]
service_calls[1].data["some"]
== f"time-{trigger_dt.day}-{trigger_dt.hour}-input_datetime.trigger"
)
@ -152,20 +143,23 @@ async def test_if_fires_using_at_input_datetime(
},
blocking=True,
)
assert len(service_calls) == 3
await hass.async_block_till_done()
async_fire_time_changed(hass, trigger_dt + timedelta(seconds=1))
await hass.async_block_till_done()
assert len(calls) == 2
assert len(service_calls) == 4
assert (
calls[1].data["some"]
service_calls[3].data["some"]
== f"time-{trigger_dt.day}-{trigger_dt.hour}-input_datetime.trigger"
)
async def test_if_fires_using_multiple_at(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing at."""
@ -195,18 +189,20 @@ async def test_if_fires_using_multiple_at(
async_fire_time_changed(hass, trigger_dt + timedelta(seconds=1))
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "time - 5"
assert len(service_calls) == 1
assert service_calls[0].data["some"] == "time - 5"
async_fire_time_changed(hass, trigger_dt + timedelta(hours=1, seconds=1))
await hass.async_block_till_done()
assert len(calls) == 2
assert calls[1].data["some"] == "time - 6"
assert len(service_calls) == 2
assert service_calls[1].data["some"] == "time - 6"
async def test_if_not_fires_using_wrong_at(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""YAML translates time values to total seconds.
@ -242,10 +238,12 @@ async def test_if_not_fires_using_wrong_at(
)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
async def test_if_action_before(hass: HomeAssistant, calls: list[ServiceCall]) -> None:
async def test_if_action_before(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for if action before."""
assert await async_setup_component(
hass,
@ -267,16 +265,18 @@ async def test_if_action_before(hass: HomeAssistant, calls: list[ServiceCall]) -
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
with patch("homeassistant.helpers.condition.dt_util.now", return_value=after_10):
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_action_after(hass: HomeAssistant, calls: list[ServiceCall]) -> None:
async def test_if_action_after(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for if action after."""
assert await async_setup_component(
hass,
@ -298,17 +298,17 @@ async def test_if_action_after(hass: HomeAssistant, calls: list[ServiceCall]) ->
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 0
with patch("homeassistant.helpers.condition.dt_util.now", return_value=after_10):
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_action_one_weekday(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for if action with one weekday."""
assert await async_setup_component(
@ -332,17 +332,17 @@ async def test_if_action_one_weekday(
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
with patch("homeassistant.helpers.condition.dt_util.now", return_value=tuesday):
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_action_list_weekday(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test for action with a list of weekdays."""
assert await async_setup_component(
@ -367,19 +367,19 @@ async def test_if_action_list_weekday(
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
with patch("homeassistant.helpers.condition.dt_util.now", return_value=tuesday):
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 2
assert len(service_calls) == 2
with patch("homeassistant.helpers.condition.dt_util.now", return_value=wednesday):
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 2
assert len(service_calls) == 2
async def test_untrack_time_change(hass: HomeAssistant) -> None:
@ -416,7 +416,9 @@ async def test_untrack_time_change(hass: HomeAssistant) -> None:
async def test_if_fires_using_at_sensor(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing at sensor time."""
now = dt_util.now()
@ -452,9 +454,9 @@ async def test_if_fires_using_at_sensor(
async_fire_time_changed(hass, trigger_dt + timedelta(seconds=1))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
assert (
calls[0].data["some"]
service_calls[0].data["some"]
== f"time-{trigger_dt.day}-{trigger_dt.hour}-sensor.next_alarm"
)
@ -470,9 +472,9 @@ async def test_if_fires_using_at_sensor(
async_fire_time_changed(hass, trigger_dt + timedelta(seconds=1))
await hass.async_block_till_done()
assert len(calls) == 2
assert len(service_calls) == 2
assert (
calls[1].data["some"]
service_calls[1].data["some"]
== f"time-{trigger_dt.day}-{trigger_dt.hour}-sensor.next_alarm"
)
@ -494,7 +496,7 @@ async def test_if_fires_using_at_sensor(
await hass.async_block_till_done()
# We should not have listened to anything
assert len(calls) == 2
assert len(service_calls) == 2
# Now without device class
hass.states.async_set(
@ -513,7 +515,7 @@ async def test_if_fires_using_at_sensor(
await hass.async_block_till_done()
# We should not have listened to anything
assert len(calls) == 2
assert len(service_calls) == 2
@pytest.mark.parametrize(
@ -544,7 +546,7 @@ def test_schema_invalid(conf) -> None:
async def test_datetime_in_past_on_load(
hass: HomeAssistant, calls: list[ServiceCall]
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test time trigger works if input_datetime is in past."""
await async_setup_component(
@ -566,6 +568,7 @@ async def test_datetime_in_past_on_load(
},
blocking=True,
)
assert len(service_calls) == 1
await hass.async_block_till_done()
assert await async_setup_component(
@ -587,7 +590,7 @@ async def test_datetime_in_past_on_load(
async_fire_time_changed(hass, now)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(service_calls) == 1
await hass.services.async_call(
"input_datetime",
@ -598,13 +601,14 @@ async def test_datetime_in_past_on_load(
},
blocking=True,
)
assert len(service_calls) == 2
await hass.async_block_till_done()
async_fire_time_changed(hass, future + timedelta(seconds=1))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 3
assert (
calls[0].data["some"]
service_calls[2].data["some"]
== f"time-{future.day}-{future.hour}-input_datetime.my_trigger"
)

75
tests/components/homeassistant/triggers/test_time_pattern.py
View File

@ -13,23 +13,19 @@ from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from tests.common import async_fire_time_changed, async_mock_service, mock_component
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
from tests.common import async_fire_time_changed, mock_component
@pytest.fixture(autouse=True)
def setup_comp(hass):
def setup_comp(hass: HomeAssistant) -> None:
"""Initialize components."""
mock_component(hass, "group")
async def test_if_fires_when_hour_matches(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing if hour is matching."""
now = dt_util.utcnow()
@ -58,7 +54,8 @@ async def test_if_fires_when_hour_matches(
async_fire_time_changed(hass, now.replace(year=now.year + 2, day=1, hour=0))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
assert service_calls[0].data["id"] == 0
await hass.services.async_call(
automation.DOMAIN,
@ -66,15 +63,17 @@ async def test_if_fires_when_hour_matches(
{ATTR_ENTITY_ID: ENTITY_MATCH_ALL},
blocking=True,
)
assert len(service_calls) == 2
async_fire_time_changed(hass, now.replace(year=now.year + 1, day=1, hour=0))
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["id"] == 0
assert len(service_calls) == 2
async def test_if_fires_when_minute_matches(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing if minutes are matching."""
now = dt_util.utcnow()
@ -101,11 +100,13 @@ async def test_if_fires_when_minute_matches(
async_fire_time_changed(hass, now.replace(year=now.year + 2, day=1, minute=0))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_when_second_matches(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing if seconds are matching."""
now = dt_util.utcnow()
@ -132,11 +133,13 @@ async def test_if_fires_when_second_matches(
async_fire_time_changed(hass, now.replace(year=now.year + 2, day=1, second=0))
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_when_second_as_string_matches(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing if seconds are matching."""
now = dt_util.utcnow()
@ -165,11 +168,13 @@ async def test_if_fires_when_second_as_string_matches(
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_when_all_matches(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing if everything matches."""
now = dt_util.utcnow()
@ -198,11 +203,13 @@ async def test_if_fires_when_all_matches(
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_periodic_seconds(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing periodically every second."""
now = dt_util.utcnow()
@ -231,11 +238,13 @@ async def test_if_fires_periodic_seconds(
)
await hass.async_block_till_done()
assert len(calls) >= 1
assert len(service_calls) >= 1
async def test_if_fires_periodic_minutes(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing periodically every minute."""
@ -265,11 +274,13 @@ async def test_if_fires_periodic_minutes(
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_if_fires_periodic_hours(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing periodically every hour."""
now = dt_util.utcnow()
@ -298,11 +309,13 @@ async def test_if_fires_periodic_hours(
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async def test_default_values(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, calls: list[ServiceCall]
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
service_calls: list[ServiceCall],
) -> None:
"""Test for firing at 2 minutes every hour."""
now = dt_util.utcnow()
@ -326,24 +339,24 @@ async def test_default_values(
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async_fire_time_changed(
hass, now.replace(year=now.year + 2, day=1, hour=1, minute=2, second=1)
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(service_calls) == 1
async_fire_time_changed(
hass, now.replace(year=now.year + 2, day=1, hour=2, minute=2, second=0)
)
await hass.async_block_till_done()
assert len(calls) == 2
assert len(service_calls) == 2
async def test_invalid_schemas(hass: HomeAssistant, calls: list[ServiceCall]) -> None:
async def test_invalid_schemas() -> None:
"""Test invalid schemas."""
schemas = (
None,