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 core platform tests [a-l] (#120904)

Browse Source
This commit is contained in:
epenet 2024-07-01 19:27:50 +02:00 committed by GitHub
parent c2dc9e9b67
commit 07f095aa42
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
20 changed files with 316 additions and 473 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
tests/components/binary_sensor/test_device_condition.py
View File

@ -22,7 +22,6 @@ from tests.common import (
MockConfigEntry, MockConfigEntry,
async_get_device_automation_capabilities, async_get_device_automation_capabilities,
async_get_device_automations, async_get_device_automations,
async_mock_service,
setup_test_component_platform, setup_test_component_platform,
) )
@ -32,12 +31,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder.""" """Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_get_conditions( async def test_get_conditions(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
@ -239,7 +232,7 @@ async def test_if_state(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
mock_binary_sensor_entities: dict[str, MockBinarySensor], mock_binary_sensor_entities: dict[str, MockBinarySensor],
) -> None: ) -> None:
"""Test for turn_on and turn_off conditions.""" """Test for turn_on and turn_off conditions."""
@ -308,26 +301,26 @@ async def test_if_state(
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get(entry.entity_id).state == STATE_ON assert hass.states.get(entry.entity_id).state == STATE_ON
assert len(calls) == 0 assert len(service_calls) == 0
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "is_on event - test_event1" assert service_calls[0].data["some"] == "is_on event - test_event1"
hass.states.async_set(entry.entity_id, STATE_OFF) hass.states.async_set(entry.entity_id, STATE_OFF)
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert calls[1].data["some"] == "is_off event - test_event2" assert service_calls[1].data["some"] == "is_off event - test_event2"
async def test_if_state_legacy( async def test_if_state_legacy(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
mock_binary_sensor_entities: dict[str, MockBinarySensor], mock_binary_sensor_entities: dict[str, MockBinarySensor],
) -> None: ) -> None:
"""Test for turn_on and turn_off conditions.""" """Test for turn_on and turn_off conditions."""
@ -375,19 +368,19 @@ async def test_if_state_legacy(
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get(entry.entity_id).state == STATE_ON assert hass.states.get(entry.entity_id).state == STATE_ON
assert len(calls) == 0 assert len(service_calls) == 0
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "is_on event - test_event1" assert service_calls[0].data["some"] == "is_on event - test_event1"
async def test_if_fires_on_for_condition( async def test_if_fires_on_for_condition(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
mock_binary_sensor_entities: dict[str, MockBinarySensor], mock_binary_sensor_entities: dict[str, MockBinarySensor],
) -> None: ) -> None:
"""Test for firing if condition is on with delay.""" """Test for firing if condition is on with delay."""
@ -439,26 +432,26 @@ async def test_if_fires_on_for_condition(
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get(entry.entity_id).state == STATE_ON assert hass.states.get(entry.entity_id).state == STATE_ON
assert len(calls) == 0 assert len(service_calls) == 0
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
# Time travel 10 secs into the future # Time travel 10 secs into the future
time_freeze.move_to(point2) time_freeze.move_to(point2)
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
hass.states.async_set(entry.entity_id, STATE_OFF) hass.states.async_set(entry.entity_id, STATE_OFF)
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
# Time travel 20 secs into the future # Time travel 20 secs into the future
time_freeze.move_to(point3) time_freeze.move_to(point3)
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "is_off event - test_event1" assert service_calls[0].data["some"] == "is_off event - test_event1"

38
tests/components/binary_sensor/test_device_trigger.py
View File

@ -22,7 +22,6 @@ from tests.common import (
async_fire_time_changed, async_fire_time_changed,
async_get_device_automation_capabilities, async_get_device_automation_capabilities,
async_get_device_automations, async_get_device_automations,
async_mock_service,
setup_test_component_platform, setup_test_component_platform,
) )
@ -32,12 +31,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder.""" """Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_get_triggers( async def test_get_triggers(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
@ -240,7 +233,7 @@ async def test_if_fires_on_state_change(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
mock_binary_sensor_entities: dict[str, MockBinarySensor], mock_binary_sensor_entities: dict[str, MockBinarySensor],
) -> None: ) -> None:
"""Test for on and off triggers firing.""" """Test for on and off triggers firing."""
@ -313,21 +306,22 @@ async def test_if_fires_on_state_change(
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get(entry.entity_id).state == STATE_ON assert hass.states.get(entry.entity_id).state == STATE_ON
assert len(calls) == 0 assert len(service_calls) == 0
hass.states.async_set(entry.entity_id, STATE_OFF) hass.states.async_set(entry.entity_id, STATE_OFF)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"not_bat_low device - {entry.entity_id} - on - off - None" == f"not_bat_low device - {entry.entity_id} - on - off - None"
) )
hass.states.async_set(entry.entity_id, STATE_ON) hass.states.async_set(entry.entity_id, STATE_ON)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert ( assert (
calls[1].data["some"] == f"bat_low device - {entry.entity_id} - off - on - None" service_calls[1].data["some"]
== f"bat_low device - {entry.entity_id} - off - on - None"
) )
@ -335,7 +329,7 @@ async def test_if_fires_on_state_change_with_for(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
mock_binary_sensor_entities: dict[str, MockBinarySensor], mock_binary_sensor_entities: dict[str, MockBinarySensor],
) -> None: ) -> None:
"""Test for triggers firing with delay.""" """Test for triggers firing with delay."""
@ -388,17 +382,17 @@ async def test_if_fires_on_state_change_with_for(
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get(entry.entity_id).state == STATE_ON assert hass.states.get(entry.entity_id).state == STATE_ON
assert len(calls) == 0 assert len(service_calls) == 0
hass.states.async_set(entry.entity_id, STATE_OFF) hass.states.async_set(entry.entity_id, STATE_OFF)
await hass.async_block_till_done() 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)) async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
await hass.async_block_till_done() await hass.async_block_till_done()
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"turn_off device - {entry.entity_id} - on - off - 0:00:05" == f"turn_off device - {entry.entity_id} - on - off - 0:00:05"
) )
@ -407,7 +401,7 @@ async def test_if_fires_on_state_change_legacy(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
mock_binary_sensor_entities: dict[str, MockBinarySensor], mock_binary_sensor_entities: dict[str, MockBinarySensor],
) -> None: ) -> None:
"""Test for triggers firing.""" """Test for triggers firing."""
@ -459,12 +453,12 @@ async def test_if_fires_on_state_change_legacy(
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get(entry.entity_id).state == STATE_ON assert hass.states.get(entry.entity_id).state == STATE_ON
assert len(calls) == 0 assert len(service_calls) == 0
hass.states.async_set(entry.entity_id, STATE_OFF) hass.states.async_set(entry.entity_id, STATE_OFF)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"turn_off device - {entry.entity_id} - on - off - None" == f"turn_off device - {entry.entity_id} - on - off - None"
) )

24
tests/components/button/test_device_trigger.py
View File

@ -13,17 +13,7 @@ from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.helpers import device_registry as dr, entity_registry as er from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from tests.common import ( from tests.common import MockConfigEntry, async_get_device_automations
MockConfigEntry,
async_get_device_automations,
async_mock_service,
)
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_get_triggers( async def test_get_triggers(
@ -109,7 +99,7 @@ async def test_if_fires_on_state_change(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off triggers firing.""" """Test for turn_on and turn_off triggers firing."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -158,9 +148,9 @@ async def test_if_fires_on_state_change(
# Test triggering device trigger with a to state # Test triggering device trigger with a to state
hass.states.async_set(entry.entity_id, "2021-01-01T23:59:59+00:00") hass.states.async_set(entry.entity_id, "2021-01-01T23:59:59+00:00")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"to - device - {entry.entity_id} - unknown - 2021-01-01T23:59:59+00:00 - None - 0" == f"to - device - {entry.entity_id} - unknown - 2021-01-01T23:59:59+00:00 - None - 0"
) )
@ -169,7 +159,7 @@ async def test_if_fires_on_state_change_legacy(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off triggers firing.""" """Test for turn_on and turn_off triggers firing."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -218,8 +208,8 @@ async def test_if_fires_on_state_change_legacy(
# Test triggering device trigger with a to state # Test triggering device trigger with a to state
hass.states.async_set(entry.entity_id, "2021-01-01T23:59:59+00:00") hass.states.async_set(entry.entity_id, "2021-01-01T23:59:59+00:00")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"to - device - {entry.entity_id} - unknown - 2021-01-01T23:59:59+00:00 - None - 0" == f"to - device - {entry.entity_id} - unknown - 2021-01-01T23:59:59+00:00 - None - 0"
) )

34
tests/components/climate/test_device_condition.py
View File

@ -17,11 +17,7 @@ from homeassistant.helpers import (
from homeassistant.helpers.entity_registry import RegistryEntryHider from homeassistant.helpers.entity_registry import RegistryEntryHider
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from tests.common import ( from tests.common import MockConfigEntry, async_get_device_automations
MockConfigEntry,
async_get_device_automations,
async_mock_service,
)
@pytest.fixture(autouse=True, name="stub_blueprint_populate") @pytest.fixture(autouse=True, name="stub_blueprint_populate")
@ -29,12 +25,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder.""" """Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
@pytest.mark.parametrize( @pytest.mark.parametrize(
("set_state", "features_reg", "features_state", "expected_condition_types"), ("set_state", "features_reg", "features_state", "expected_condition_types"),
[ [
@ -151,7 +141,7 @@ async def test_if_state(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off conditions.""" """Test for turn_on and turn_off conditions."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -220,7 +210,7 @@ async def test_if_state(
# Should not fire, entity doesn't exist yet # Should not fire, entity doesn't exist yet
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
hass.states.async_set( hass.states.async_set(
entry.entity_id, entry.entity_id,
@ -232,8 +222,8 @@ async def test_if_state(
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "is_hvac_mode - event - test_event1" assert service_calls[0].data["some"] == "is_hvac_mode - event - test_event1"
hass.states.async_set( hass.states.async_set(
entry.entity_id, entry.entity_id,
@ -246,13 +236,13 @@ async def test_if_state(
# Should not fire # Should not fire
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert calls[1].data["some"] == "is_preset_mode - event - test_event2" assert service_calls[1].data["some"] == "is_preset_mode - event - test_event2"
hass.states.async_set( hass.states.async_set(
entry.entity_id, entry.entity_id,
@ -265,14 +255,14 @@ async def test_if_state(
# Should not fire # Should not fire
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
async def test_if_state_legacy( async def test_if_state_legacy(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off conditions.""" """Test for turn_on and turn_off conditions."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -323,8 +313,8 @@ async def test_if_state_legacy(
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "is_hvac_mode - event - test_event1" assert service_calls[0].data["some"] == "is_hvac_mode - event - test_event1"
@pytest.mark.parametrize( @pytest.mark.parametrize(

32
tests/components/climate/test_device_trigger.py
View File

@ -23,11 +23,7 @@ from homeassistant.helpers import (
from homeassistant.helpers.entity_registry import RegistryEntryHider from homeassistant.helpers.entity_registry import RegistryEntryHider
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from tests.common import ( from tests.common import MockConfigEntry, async_get_device_automations
MockConfigEntry,
async_get_device_automations,
async_mock_service,
)
@pytest.fixture(autouse=True, name="stub_blueprint_populate") @pytest.fixture(autouse=True, name="stub_blueprint_populate")
@ -35,12 +31,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder.""" """Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_get_triggers( async def test_get_triggers(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
@ -151,7 +141,7 @@ async def test_if_fires_on_state_change(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off triggers firing.""" """Test for turn_on and turn_off triggers firing."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -236,8 +226,8 @@ async def test_if_fires_on_state_change(
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "hvac_mode_changed" assert service_calls[0].data["some"] == "hvac_mode_changed"
# Fake that the temperature is changing # Fake that the temperature is changing
hass.states.async_set( hass.states.async_set(
@ -250,8 +240,8 @@ async def test_if_fires_on_state_change(
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert calls[1].data["some"] == "current_temperature_changed" assert service_calls[1].data["some"] == "current_temperature_changed"
# Fake that the humidity is changing # Fake that the humidity is changing
hass.states.async_set( hass.states.async_set(
@ -264,15 +254,15 @@ async def test_if_fires_on_state_change(
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 3 assert len(service_calls) == 3
assert calls[2].data["some"] == "current_humidity_changed" assert service_calls[2].data["some"] == "current_humidity_changed"
async def test_if_fires_on_state_change_legacy( async def test_if_fires_on_state_change_legacy(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off triggers firing.""" """Test for turn_on and turn_off triggers firing."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -329,8 +319,8 @@ async def test_if_fires_on_state_change_legacy(
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "hvac_mode_changed" assert service_calls[0].data["some"] == "hvac_mode_changed"
async def test_get_trigger_capabilities_hvac_mode(hass: HomeAssistant) -> None: async def test_get_trigger_capabilities_hvac_mode(hass: HomeAssistant) -> None:

79
tests/components/cover/test_device_condition.py
View File

@ -26,7 +26,6 @@ from tests.common import (
MockConfigEntry, MockConfigEntry,
async_get_device_automation_capabilities, async_get_device_automation_capabilities,
async_get_device_automations, async_get_device_automations,
async_mock_service,
setup_test_component_platform, setup_test_component_platform,
) )
@ -36,12 +35,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder.""" """Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
@pytest.mark.parametrize( @pytest.mark.parametrize(
("set_state", "features_reg", "features_state", "expected_condition_types"), ("set_state", "features_reg", "features_state", "expected_condition_types"),
[ [
@ -359,7 +352,7 @@ async def test_if_state(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off conditions.""" """Test for turn_on and turn_off conditions."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -473,36 +466,36 @@ async def test_if_state(
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "is_open - event - test_event1" assert service_calls[0].data["some"] == "is_open - event - test_event1"
hass.states.async_set(entry.entity_id, STATE_CLOSED) hass.states.async_set(entry.entity_id, STATE_CLOSED)
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert calls[1].data["some"] == "is_closed - event - test_event2" assert service_calls[1].data["some"] == "is_closed - event - test_event2"
hass.states.async_set(entry.entity_id, STATE_OPENING) hass.states.async_set(entry.entity_id, STATE_OPENING)
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event3") hass.bus.async_fire("test_event3")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 3 assert len(service_calls) == 3
assert calls[2].data["some"] == "is_opening - event - test_event3" assert service_calls[2].data["some"] == "is_opening - event - test_event3"
hass.states.async_set(entry.entity_id, STATE_CLOSING) hass.states.async_set(entry.entity_id, STATE_CLOSING)
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event4") hass.bus.async_fire("test_event4")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 4 assert len(service_calls) == 4
assert calls[3].data["some"] == "is_closing - event - test_event4" assert service_calls[3].data["some"] == "is_closing - event - test_event4"
async def test_if_state_legacy( async def test_if_state_legacy(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off conditions.""" """Test for turn_on and turn_off conditions."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -550,15 +543,15 @@ async def test_if_state_legacy(
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "is_open - event - test_event1" assert service_calls[0].data["some"] == "is_open - event - test_event1"
async def test_if_position( async def test_if_position(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
caplog: pytest.LogCaptureFixture, caplog: pytest.LogCaptureFixture,
mock_cover_entities: list[MockCover], mock_cover_entities: list[MockCover],
) -> None: ) -> None:
@ -676,10 +669,10 @@ async def test_if_position(
await hass.async_block_till_done() await hass.async_block_till_done()
hass.bus.async_fire("test_event3") hass.bus.async_fire("test_event3")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 3 assert len(service_calls) == 3
assert calls[0].data["some"] == "is_pos_gt_45 - event - test_event1" assert service_calls[0].data["some"] == "is_pos_gt_45 - event - test_event1"
assert calls[1].data["some"] == "is_pos_lt_90 - event - test_event2" assert service_calls[1].data["some"] == "is_pos_lt_90 - event - test_event2"
assert calls[2].data["some"] == "is_pos_gt_45_lt_90 - event - test_event3" assert service_calls[2].data["some"] == "is_pos_gt_45_lt_90 - event - test_event3"
hass.states.async_set( hass.states.async_set(
ent.entity_id, STATE_CLOSED, attributes={"current_position": 45} ent.entity_id, STATE_CLOSED, attributes={"current_position": 45}
@ -690,9 +683,9 @@ async def test_if_position(
await hass.async_block_till_done() await hass.async_block_till_done()
hass.bus.async_fire("test_event3") hass.bus.async_fire("test_event3")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 5 assert len(service_calls) == 5
assert calls[3].data["some"] == "is_pos_not_gt_45 - event - test_event1" assert service_calls[3].data["some"] == "is_pos_not_gt_45 - event - test_event1"
assert calls[4].data["some"] == "is_pos_lt_90 - event - test_event2" assert service_calls[4].data["some"] == "is_pos_lt_90 - event - test_event2"
hass.states.async_set( hass.states.async_set(
ent.entity_id, STATE_CLOSED, attributes={"current_position": 90} ent.entity_id, STATE_CLOSED, attributes={"current_position": 90}
@ -701,14 +694,14 @@ async def test_if_position(
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
hass.bus.async_fire("test_event3") hass.bus.async_fire("test_event3")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 6 assert len(service_calls) == 6
assert calls[5].data["some"] == "is_pos_gt_45 - event - test_event1" assert service_calls[5].data["some"] == "is_pos_gt_45 - event - test_event1"
hass.states.async_set(ent.entity_id, STATE_UNAVAILABLE, attributes={}) hass.states.async_set(ent.entity_id, STATE_UNAVAILABLE, attributes={})
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 7 assert len(service_calls) == 7
assert calls[6].data["some"] == "is_pos_not_gt_45 - event - test_event1" assert service_calls[6].data["some"] == "is_pos_not_gt_45 - event - test_event1"
for record in caplog.records: for record in caplog.records:
assert record.levelname in ("DEBUG", "INFO") assert record.levelname in ("DEBUG", "INFO")
@ -718,7 +711,7 @@ async def test_if_tilt_position(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
caplog: pytest.LogCaptureFixture, caplog: pytest.LogCaptureFixture,
mock_cover_entities: list[MockCover], mock_cover_entities: list[MockCover],
) -> None: ) -> None:
@ -836,10 +829,10 @@ async def test_if_tilt_position(
await hass.async_block_till_done() await hass.async_block_till_done()
hass.bus.async_fire("test_event3") hass.bus.async_fire("test_event3")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 3 assert len(service_calls) == 3
assert calls[0].data["some"] == "is_pos_gt_45 - event - test_event1" assert service_calls[0].data["some"] == "is_pos_gt_45 - event - test_event1"
assert calls[1].data["some"] == "is_pos_lt_90 - event - test_event2" assert service_calls[1].data["some"] == "is_pos_lt_90 - event - test_event2"
assert calls[2].data["some"] == "is_pos_gt_45_lt_90 - event - test_event3" assert service_calls[2].data["some"] == "is_pos_gt_45_lt_90 - event - test_event3"
hass.states.async_set( hass.states.async_set(
ent.entity_id, STATE_CLOSED, attributes={"current_tilt_position": 45} ent.entity_id, STATE_CLOSED, attributes={"current_tilt_position": 45}
@ -850,9 +843,9 @@ async def test_if_tilt_position(
await hass.async_block_till_done() await hass.async_block_till_done()
hass.bus.async_fire("test_event3") hass.bus.async_fire("test_event3")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 5 assert len(service_calls) == 5
assert calls[3].data["some"] == "is_pos_not_gt_45 - event - test_event1" assert service_calls[3].data["some"] == "is_pos_not_gt_45 - event - test_event1"
assert calls[4].data["some"] == "is_pos_lt_90 - event - test_event2" assert service_calls[4].data["some"] == "is_pos_lt_90 - event - test_event2"
hass.states.async_set( hass.states.async_set(
ent.entity_id, STATE_CLOSED, attributes={"current_tilt_position": 90} ent.entity_id, STATE_CLOSED, attributes={"current_tilt_position": 90}
@ -863,14 +856,14 @@ async def test_if_tilt_position(
await hass.async_block_till_done() await hass.async_block_till_done()
hass.bus.async_fire("test_event3") hass.bus.async_fire("test_event3")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 6 assert len(service_calls) == 6
assert calls[5].data["some"] == "is_pos_gt_45 - event - test_event1" assert service_calls[5].data["some"] == "is_pos_gt_45 - event - test_event1"
hass.states.async_set(ent.entity_id, STATE_UNAVAILABLE, attributes={}) hass.states.async_set(ent.entity_id, STATE_UNAVAILABLE, attributes={})
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 7 assert len(service_calls) == 7
assert calls[6].data["some"] == "is_pos_not_gt_45 - event - test_event1" assert service_calls[6].data["some"] == "is_pos_not_gt_45 - event - test_event1"
for record in caplog.records: for record in caplog.records:
assert record.levelname in ("DEBUG", "INFO") assert record.levelname in ("DEBUG", "INFO")

77
tests/components/cover/test_device_trigger.py
View File

@ -29,7 +29,6 @@ from tests.common import (
async_fire_time_changed, async_fire_time_changed,
async_get_device_automation_capabilities, async_get_device_automation_capabilities,
async_get_device_automations, async_get_device_automations,
async_mock_service,
setup_test_component_platform, setup_test_component_platform,
) )
@ -39,12 +38,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder.""" """Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
@pytest.mark.parametrize( @pytest.mark.parametrize(
("set_state", "features_reg", "features_state", "expected_trigger_types"), ("set_state", "features_reg", "features_state", "expected_trigger_types"),
[ [
@ -381,7 +374,7 @@ async def test_if_fires_on_state_change(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for state triggers firing.""" """Test for state triggers firing."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -496,36 +489,36 @@ async def test_if_fires_on_state_change(
# Fake that the entity is opened. # Fake that the entity is opened.
hass.states.async_set(entry.entity_id, STATE_OPEN) hass.states.async_set(entry.entity_id, STATE_OPEN)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"opened - device - {entry.entity_id} - closed - open - None" == f"opened - device - {entry.entity_id} - closed - open - None"
) )
# Fake that the entity is closed. # Fake that the entity is closed.
hass.states.async_set(entry.entity_id, STATE_CLOSED) hass.states.async_set(entry.entity_id, STATE_CLOSED)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert ( assert (
calls[1].data["some"] service_calls[1].data["some"]
== f"closed - device - {entry.entity_id} - open - closed - None" == f"closed - device - {entry.entity_id} - open - closed - None"
) )
# Fake that the entity is opening. # Fake that the entity is opening.
hass.states.async_set(entry.entity_id, STATE_OPENING) hass.states.async_set(entry.entity_id, STATE_OPENING)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 3 assert len(service_calls) == 3
assert ( assert (
calls[2].data["some"] service_calls[2].data["some"]
== f"opening - device - {entry.entity_id} - closed - opening - None" == f"opening - device - {entry.entity_id} - closed - opening - None"
) )
# Fake that the entity is closing. # Fake that the entity is closing.
hass.states.async_set(entry.entity_id, STATE_CLOSING) hass.states.async_set(entry.entity_id, STATE_CLOSING)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 4 assert len(service_calls) == 4
assert ( assert (
calls[3].data["some"] service_calls[3].data["some"]
== f"closing - device - {entry.entity_id} - opening - closing - None" == f"closing - device - {entry.entity_id} - opening - closing - None"
) )
@ -534,7 +527,7 @@ async def test_if_fires_on_state_change_legacy(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for state triggers firing.""" """Test for state triggers firing."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -583,9 +576,9 @@ async def test_if_fires_on_state_change_legacy(
# Fake that the entity is opened. # Fake that the entity is opened.
hass.states.async_set(entry.entity_id, STATE_OPEN) hass.states.async_set(entry.entity_id, STATE_OPEN)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"opened - device - {entry.entity_id} - closed - open - None" == f"opened - device - {entry.entity_id} - closed - open - None"
) )
@ -594,7 +587,7 @@ async def test_if_fires_on_state_change_with_for(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for triggers firing with delay.""" """Test for triggers firing with delay."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -640,17 +633,17 @@ async def test_if_fires_on_state_change_with_for(
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
hass.states.async_set(entry.entity_id, STATE_OPEN) hass.states.async_set(entry.entity_id, STATE_OPEN)
await hass.async_block_till_done() 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)) async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
await hass.async_block_till_done() await hass.async_block_till_done()
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"turn_off device - {entry.entity_id} - closed - open - 0:00:05" == f"turn_off device - {entry.entity_id} - closed - open - 0:00:05"
) )
@ -660,7 +653,7 @@ async def test_if_fires_on_position(
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
mock_cover_entities: list[MockCover], mock_cover_entities: list[MockCover],
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for position triggers.""" """Test for position triggers."""
setup_test_component_platform(hass, DOMAIN, mock_cover_entities) setup_test_component_platform(hass, DOMAIN, mock_cover_entities)
@ -769,9 +762,13 @@ async def test_if_fires_on_position(
ent.entity_id, STATE_OPEN, attributes={"current_position": 50} ent.entity_id, STATE_OPEN, attributes={"current_position": 50}
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 3 assert len(service_calls) == 3
assert sorted( assert sorted(
[calls[0].data["some"], calls[1].data["some"], calls[2].data["some"]] [
service_calls[0].data["some"],
service_calls[1].data["some"],
service_calls[2].data["some"],
]
) == sorted( ) == sorted(
[ [
( (
@ -791,9 +788,9 @@ async def test_if_fires_on_position(
ent.entity_id, STATE_CLOSED, attributes={"current_position": 45} ent.entity_id, STATE_CLOSED, attributes={"current_position": 45}
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 4 assert len(service_calls) == 4
assert ( assert (
calls[3].data["some"] service_calls[3].data["some"]
== f"is_pos_lt_90 - device - {entry.entity_id} - closed - closed - None" == f"is_pos_lt_90 - device - {entry.entity_id} - closed - closed - None"
) )
@ -801,9 +798,9 @@ async def test_if_fires_on_position(
ent.entity_id, STATE_CLOSED, attributes={"current_position": 90} ent.entity_id, STATE_CLOSED, attributes={"current_position": 90}
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 5 assert len(service_calls) == 5
assert ( assert (
calls[4].data["some"] service_calls[4].data["some"]
== f"is_pos_gt_45 - device - {entry.entity_id} - closed - closed - None" == f"is_pos_gt_45 - device - {entry.entity_id} - closed - closed - None"
) )
@ -812,7 +809,7 @@ async def test_if_fires_on_tilt_position(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
mock_cover_entities: list[MockCover], mock_cover_entities: list[MockCover],
) -> None: ) -> None:
"""Test for tilt position triggers.""" """Test for tilt position triggers."""
@ -924,9 +921,13 @@ async def test_if_fires_on_tilt_position(
ent.entity_id, STATE_OPEN, attributes={"current_tilt_position": 50} ent.entity_id, STATE_OPEN, attributes={"current_tilt_position": 50}
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 3 assert len(service_calls) == 3
assert sorted( assert sorted(
[calls[0].data["some"], calls[1].data["some"], calls[2].data["some"]] [
service_calls[0].data["some"],
service_calls[1].data["some"],
service_calls[2].data["some"],
]
) == sorted( ) == sorted(
[ [
( (
@ -946,9 +947,9 @@ async def test_if_fires_on_tilt_position(
ent.entity_id, STATE_CLOSED, attributes={"current_tilt_position": 45} ent.entity_id, STATE_CLOSED, attributes={"current_tilt_position": 45}
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 4 assert len(service_calls) == 4
assert ( assert (
calls[3].data["some"] service_calls[3].data["some"]
== f"is_pos_lt_90 - device - {entry.entity_id} - closed - closed - None" == f"is_pos_lt_90 - device - {entry.entity_id} - closed - closed - None"
) )
@ -956,8 +957,8 @@ async def test_if_fires_on_tilt_position(
ent.entity_id, STATE_CLOSED, attributes={"current_tilt_position": 90} ent.entity_id, STATE_CLOSED, attributes={"current_tilt_position": 90}
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 5 assert len(service_calls) == 5
assert ( assert (
calls[4].data["some"] service_calls[4].data["some"]
== f"is_pos_gt_45 - device - {entry.entity_id} - closed - closed - None" == f"is_pos_gt_45 - device - {entry.entity_id} - closed - closed - None"
) )

32
tests/components/device_automation/test_init.py
View File

@ -23,13 +23,7 @@ from homeassistant.loader import IntegrationNotFound
from homeassistant.requirements import RequirementsNotFound from homeassistant.requirements import RequirementsNotFound
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from tests.common import ( from tests.common import MockConfigEntry, MockModule, mock_integration, mock_platform
MockConfigEntry,
MockModule,
async_mock_service,
mock_integration,
mock_platform,
)
from tests.typing import WebSocketGenerator from tests.typing import WebSocketGenerator
@ -1384,15 +1378,9 @@ async def test_automation_with_bad_condition(
assert expected_error.format(path="['condition'][0]") in caplog.text assert expected_error.format(path="['condition'][0]") in caplog.text
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_automation_with_sub_condition( async def test_automation_with_sub_condition(
hass: HomeAssistant, hass: HomeAssistant,
calls: list[ServiceCall], service_calls: list[ServiceCall],
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
) -> None: ) -> None:
@ -1492,29 +1480,29 @@ async def test_automation_with_sub_condition(
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get(entity_entry1.entity_id).state == STATE_ON assert hass.states.get(entity_entry1.entity_id).state == STATE_ON
assert hass.states.get(entity_entry2.entity_id).state == STATE_OFF assert hass.states.get(entity_entry2.entity_id).state == STATE_OFF
assert len(calls) == 0 assert len(service_calls) == 0
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "or event - test_event1" assert service_calls[0].data["some"] == "or event - test_event1"
hass.states.async_set(entity_entry1.entity_id, STATE_OFF) hass.states.async_set(entity_entry1.entity_id, STATE_OFF)
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
hass.states.async_set(entity_entry2.entity_id, STATE_ON) hass.states.async_set(entity_entry2.entity_id, STATE_ON)
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert calls[1].data["some"] == "or event - test_event1" assert service_calls[1].data["some"] == "or event - test_event1"
hass.states.async_set(entity_entry1.entity_id, STATE_ON) hass.states.async_set(entity_entry1.entity_id, STATE_ON)
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 4 assert len(service_calls) == 4
assert [calls[2].data["some"], calls[3].data["some"]] == unordered( assert [service_calls[2].data["some"], service_calls[3].data["some"]] == unordered(
["or event - test_event1", "and event - test_event1"] ["or event - test_event1", "and event - test_event1"]
) )

30
tests/components/device_automation/test_toggle_entity.py
View File

@ -11,7 +11,7 @@ from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util import homeassistant.util.dt as dt_util
from tests.common import MockConfigEntry, async_fire_time_changed, async_mock_service from tests.common import MockConfigEntry, async_fire_time_changed
@pytest.fixture(autouse=True, name="stub_blueprint_populate") @pytest.fixture(autouse=True, name="stub_blueprint_populate")
@ -19,17 +19,11 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder.""" """Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_if_fires_on_state_change( async def test_if_fires_on_state_change(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off triggers firing. """Test for turn_on and turn_off triggers firing.
@ -121,20 +115,20 @@ async def test_if_fires_on_state_change(
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get(entry.entity_id).state == STATE_ON assert hass.states.get(entry.entity_id).state == STATE_ON
assert len(calls) == 0 assert len(service_calls) == 0
hass.states.async_set(entry.entity_id, STATE_OFF) hass.states.async_set(entry.entity_id, STATE_OFF)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert {calls[0].data["some"], calls[1].data["some"]} == { assert {service_calls[0].data["some"], service_calls[1].data["some"]} == {
f"turn_off device - {entry.entity_id} - on - off - None", f"turn_off device - {entry.entity_id} - on - off - None",
f"turn_on_or_off device - {entry.entity_id} - on - off - None", f"turn_on_or_off device - {entry.entity_id} - on - off - None",
} }
hass.states.async_set(entry.entity_id, STATE_ON) hass.states.async_set(entry.entity_id, STATE_ON)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 4 assert len(service_calls) == 4
assert {calls[2].data["some"], calls[3].data["some"]} == { assert {service_calls[2].data["some"], service_calls[3].data["some"]} == {
f"turn_on device - {entry.entity_id} - off - on - None", f"turn_on device - {entry.entity_id} - off - on - None",
f"turn_on_or_off device - {entry.entity_id} - off - on - None", f"turn_on_or_off device - {entry.entity_id} - off - on - None",
} }
@ -145,7 +139,7 @@ async def test_if_fires_on_state_change_with_for(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
trigger: str, trigger: str,
) -> None: ) -> None:
"""Test for triggers firing with delay.""" """Test for triggers firing with delay."""
@ -193,16 +187,16 @@ async def test_if_fires_on_state_change_with_for(
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get(entry.entity_id).state == STATE_ON assert hass.states.get(entry.entity_id).state == STATE_ON
assert len(calls) == 0 assert len(service_calls) == 0
hass.states.async_set(entry.entity_id, STATE_OFF) hass.states.async_set(entry.entity_id, STATE_OFF)
await hass.async_block_till_done() 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)) async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
await hass.async_block_till_done() await hass.async_block_till_done()
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"turn_off device - {entry.entity_id} - on - off - 0:00:05" == f"turn_off device - {entry.entity_id} - on - off - 0:00:05"
) )

28
tests/components/device_tracker/test_device_condition.py
View File

@ -12,11 +12,7 @@ from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.entity_registry import RegistryEntryHider from homeassistant.helpers.entity_registry import RegistryEntryHider
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from tests.common import ( from tests.common import MockConfigEntry, async_get_device_automations
MockConfigEntry,
async_get_device_automations,
async_mock_service,
)
@pytest.fixture(autouse=True, name="stub_blueprint_populate") @pytest.fixture(autouse=True, name="stub_blueprint_populate")
@ -24,12 +20,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder.""" """Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_get_conditions( async def test_get_conditions(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
@ -114,7 +104,7 @@ async def test_if_state(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off conditions.""" """Test for turn_on and turn_off conditions."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -184,22 +174,22 @@ async def test_if_state(
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "is_home - event - test_event1" assert service_calls[0].data["some"] == "is_home - event - test_event1"
hass.states.async_set(entry.entity_id, "school") hass.states.async_set(entry.entity_id, "school")
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert calls[1].data["some"] == "is_not_home - event - test_event2" assert service_calls[1].data["some"] == "is_not_home - event - test_event2"
async def test_if_state_legacy( async def test_if_state_legacy(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off conditions.""" """Test for turn_on and turn_off conditions."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -247,5 +237,5 @@ async def test_if_state_legacy(
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "is_home - event - test_event1" assert service_calls[0].data["some"] == "is_home - event - test_event1"

28
tests/components/device_tracker/test_device_trigger.py
View File

@ -17,11 +17,7 @@ from homeassistant.helpers import (
from homeassistant.helpers.entity_registry import RegistryEntryHider from homeassistant.helpers.entity_registry import RegistryEntryHider
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from tests.common import ( from tests.common import MockConfigEntry, async_get_device_automations
MockConfigEntry,
async_get_device_automations,
async_mock_service,
)
@pytest.fixture(autouse=True, name="stub_blueprint_populate") @pytest.fixture(autouse=True, name="stub_blueprint_populate")
@ -36,12 +32,6 @@ HOME_LATITUDE = 32.880837
HOME_LONGITUDE = -117.237561 HOME_LONGITUDE = -117.237561
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def setup_zone(hass: HomeAssistant) -> None: def setup_zone(hass: HomeAssistant) -> None:
"""Create test zone.""" """Create test zone."""
@ -145,7 +135,7 @@ async def test_if_fires_on_zone_change(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for enter and leave triggers firing.""" """Test for enter and leave triggers firing."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -228,9 +218,9 @@ async def test_if_fires_on_zone_change(
{"latitude": HOME_LATITUDE, "longitude": HOME_LONGITUDE}, {"latitude": HOME_LATITUDE, "longitude": HOME_LONGITUDE},
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"enter - device - {entry.entity_id} - -117.235 - -117.238" == f"enter - device - {entry.entity_id} - -117.235 - -117.238"
) )
@ -241,9 +231,9 @@ async def test_if_fires_on_zone_change(
{"latitude": AWAY_LATITUDE, "longitude": AWAY_LONGITUDE}, {"latitude": AWAY_LATITUDE, "longitude": AWAY_LONGITUDE},
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert ( assert (
calls[1].data["some"] service_calls[1].data["some"]
== f"leave - device - {entry.entity_id} - -117.238 - -117.235" == f"leave - device - {entry.entity_id} - -117.238 - -117.235"
) )
@ -252,7 +242,7 @@ async def test_if_fires_on_zone_change_legacy(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for enter and leave triggers firing.""" """Test for enter and leave triggers firing."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -311,9 +301,9 @@ async def test_if_fires_on_zone_change_legacy(
{"latitude": HOME_LATITUDE, "longitude": HOME_LONGITUDE}, {"latitude": HOME_LATITUDE, "longitude": HOME_LONGITUDE},
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"enter - device - {entry.entity_id} - -117.235 - -117.238" == f"enter - device - {entry.entity_id} - -117.235 - -117.238"
) )

28
tests/components/fan/test_device_condition.py
View File

@ -12,11 +12,7 @@ from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.entity_registry import RegistryEntryHider from homeassistant.helpers.entity_registry import RegistryEntryHider
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from tests.common import ( from tests.common import MockConfigEntry, async_get_device_automations
MockConfigEntry,
async_get_device_automations,
async_mock_service,
)
@pytest.fixture(autouse=True, name="stub_blueprint_populate") @pytest.fixture(autouse=True, name="stub_blueprint_populate")
@ -24,12 +20,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder.""" """Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_get_conditions( async def test_get_conditions(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
@ -114,7 +104,7 @@ async def test_if_state(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off conditions.""" """Test for turn_on and turn_off conditions."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -184,22 +174,22 @@ async def test_if_state(
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "is_on - event - test_event1" assert service_calls[0].data["some"] == "is_on - event - test_event1"
hass.states.async_set(entry.entity_id, STATE_OFF) hass.states.async_set(entry.entity_id, STATE_OFF)
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert calls[1].data["some"] == "is_off - event - test_event2" assert service_calls[1].data["some"] == "is_off - event - test_event2"
async def test_if_state_legacy( async def test_if_state_legacy(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off conditions.""" """Test for turn_on and turn_off conditions."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -246,5 +236,5 @@ async def test_if_state_legacy(
) )
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "is_on - event - test_event1" assert service_calls[0].data["some"] == "is_on - event - test_event1"

33
tests/components/fan/test_device_trigger.py
View File

@ -20,7 +20,6 @@ from tests.common import (
async_fire_time_changed, async_fire_time_changed,
async_get_device_automation_capabilities, async_get_device_automation_capabilities,
async_get_device_automations, async_get_device_automations,
async_mock_service,
) )
@ -29,12 +28,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder.""" """Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_get_triggers( async def test_get_triggers(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
@ -180,7 +173,7 @@ async def test_if_fires_on_state_change(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off triggers firing.""" """Test for turn_on and turn_off triggers firing."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -273,8 +266,8 @@ async def test_if_fires_on_state_change(
# Fake that the entity is turning on. # Fake that the entity is turning on.
hass.states.async_set(entry.entity_id, STATE_ON) hass.states.async_set(entry.entity_id, STATE_ON)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert {calls[0].data["some"], calls[1].data["some"]} == { assert {service_calls[0].data["some"], service_calls[1].data["some"]} == {
f"turn_on - device - {entry.entity_id} - off - on - None", f"turn_on - device - {entry.entity_id} - off - on - None",
f"turn_on_or_off - device - {entry.entity_id} - off - on - None", f"turn_on_or_off - device - {entry.entity_id} - off - on - None",
} }
@ -282,8 +275,8 @@ async def test_if_fires_on_state_change(
# Fake that the entity is turning off. # Fake that the entity is turning off.
hass.states.async_set(entry.entity_id, STATE_OFF) hass.states.async_set(entry.entity_id, STATE_OFF)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 4 assert len(service_calls) == 4
assert {calls[2].data["some"], calls[3].data["some"]} == { assert {service_calls[2].data["some"], service_calls[3].data["some"]} == {
f"turn_off - device - {entry.entity_id} - on - off - None", f"turn_off - device - {entry.entity_id} - on - off - None",
f"turn_on_or_off - device - {entry.entity_id} - on - off - None", f"turn_on_or_off - device - {entry.entity_id} - on - off - None",
} }
@ -293,7 +286,7 @@ async def test_if_fires_on_state_change_legacy(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off triggers firing.""" """Test for turn_on and turn_off triggers firing."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -342,9 +335,9 @@ async def test_if_fires_on_state_change_legacy(
# Fake that the entity is turning on. # Fake that the entity is turning on.
hass.states.async_set(entry.entity_id, STATE_ON) hass.states.async_set(entry.entity_id, STATE_ON)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"turn_on - device - {entry.entity_id} - off - on - None" == f"turn_on - device - {entry.entity_id} - off - on - None"
) )
@ -353,7 +346,7 @@ async def test_if_fires_on_state_change_with_for(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for triggers firing with delay.""" """Test for triggers firing with delay."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -399,16 +392,16 @@ async def test_if_fires_on_state_change_with_for(
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
hass.states.async_set(entry.entity_id, STATE_OFF) hass.states.async_set(entry.entity_id, STATE_OFF)
await hass.async_block_till_done() 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)) async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
await hass.async_block_till_done() await hass.async_block_till_done()
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"turn_off device - {entry.entity_id} - on - off - 0:00:05" == f"turn_off device - {entry.entity_id} - on - off - 0:00:05"
) )

38
tests/components/humidifier/test_device_condition.py
View File

@ -17,11 +17,7 @@ from homeassistant.helpers import (
from homeassistant.helpers.entity_registry import RegistryEntryHider from homeassistant.helpers.entity_registry import RegistryEntryHider
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from tests.common import ( from tests.common import MockConfigEntry, async_get_device_automations
MockConfigEntry,
async_get_device_automations,
async_mock_service,
)
@pytest.fixture(autouse=True, name="stub_blueprint_populate") @pytest.fixture(autouse=True, name="stub_blueprint_populate")
@ -29,12 +25,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder.""" """Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
@pytest.mark.parametrize( @pytest.mark.parametrize(
("set_state", "features_reg", "features_state", "expected_condition_types"), ("set_state", "features_reg", "features_state", "expected_condition_types"),
[ [
@ -153,7 +143,7 @@ async def test_if_state(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off conditions.""" """Test for turn_on and turn_off conditions."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -238,42 +228,42 @@ async def test_if_state(
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "is_on event - test_event1" assert service_calls[0].data["some"] == "is_on event - test_event1"
hass.states.async_set(entry.entity_id, STATE_OFF) hass.states.async_set(entry.entity_id, STATE_OFF)
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert calls[1].data["some"] == "is_off event - test_event2" assert service_calls[1].data["some"] == "is_off event - test_event2"
hass.states.async_set(entry.entity_id, STATE_ON, {ATTR_MODE: const.MODE_AWAY}) hass.states.async_set(entry.entity_id, STATE_ON, {ATTR_MODE: const.MODE_AWAY})
hass.bus.async_fire("test_event3") hass.bus.async_fire("test_event3")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 3 assert len(service_calls) == 3
assert calls[2].data["some"] == "is_mode - event - test_event3" assert service_calls[2].data["some"] == "is_mode - event - test_event3"
hass.states.async_set(entry.entity_id, STATE_ON, {ATTR_MODE: const.MODE_HOME}) hass.states.async_set(entry.entity_id, STATE_ON, {ATTR_MODE: const.MODE_HOME})
# Should not fire # Should not fire
hass.bus.async_fire("test_event3") hass.bus.async_fire("test_event3")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 3 assert len(service_calls) == 3
async def test_if_state_legacy( async def test_if_state_legacy(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off conditions.""" """Test for turn_on and turn_off conditions."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -316,15 +306,15 @@ async def test_if_state_legacy(
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
hass.states.async_set(entry.entity_id, STATE_ON, {ATTR_MODE: const.MODE_AWAY}) hass.states.async_set(entry.entity_id, STATE_ON, {ATTR_MODE: const.MODE_AWAY})
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "is_mode - event - test_event1" assert service_calls[0].data["some"] == "is_mode - event - test_event1"
@pytest.mark.parametrize( @pytest.mark.parametrize(

49
tests/components/humidifier/test_device_trigger.py
View File

@ -30,7 +30,6 @@ from tests.common import (
MockConfigEntry, MockConfigEntry,
async_fire_time_changed, async_fire_time_changed,
async_get_device_automations, async_get_device_automations,
async_mock_service,
) )
@ -39,12 +38,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder.""" """Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_get_triggers( async def test_get_triggers(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
@ -166,7 +159,7 @@ async def test_if_fires_on_state_change(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off triggers firing.""" """Test for turn_on and turn_off triggers firing."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -356,8 +349,8 @@ async def test_if_fires_on_state_change(
{const.ATTR_HUMIDITY: 7, const.ATTR_CURRENT_HUMIDITY: 35}, {const.ATTR_HUMIDITY: 7, const.ATTR_CURRENT_HUMIDITY: 35},
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "target_humidity_changed_below" assert service_calls[0].data["some"] == "target_humidity_changed_below"
# Fake that the current humidity is changing # Fake that the current humidity is changing
hass.states.async_set( hass.states.async_set(
@ -366,8 +359,8 @@ async def test_if_fires_on_state_change(
{const.ATTR_HUMIDITY: 7, const.ATTR_CURRENT_HUMIDITY: 18}, {const.ATTR_HUMIDITY: 7, const.ATTR_CURRENT_HUMIDITY: 18},
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert calls[1].data["some"] == "current_humidity_changed_below" assert service_calls[1].data["some"] == "current_humidity_changed_below"
# Fake that the humidity target is changing # Fake that the humidity target is changing
hass.states.async_set( hass.states.async_set(
@ -376,8 +369,8 @@ async def test_if_fires_on_state_change(
{const.ATTR_HUMIDITY: 37, const.ATTR_CURRENT_HUMIDITY: 18}, {const.ATTR_HUMIDITY: 37, const.ATTR_CURRENT_HUMIDITY: 18},
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 3 assert len(service_calls) == 3
assert calls[2].data["some"] == "target_humidity_changed_above" assert service_calls[2].data["some"] == "target_humidity_changed_above"
# Fake that the current humidity is changing # Fake that the current humidity is changing
hass.states.async_set( hass.states.async_set(
@ -386,14 +379,14 @@ async def test_if_fires_on_state_change(
{const.ATTR_HUMIDITY: 37, const.ATTR_CURRENT_HUMIDITY: 41}, {const.ATTR_HUMIDITY: 37, const.ATTR_CURRENT_HUMIDITY: 41},
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 4 assert len(service_calls) == 4
assert calls[3].data["some"] == "current_humidity_changed_above" assert service_calls[3].data["some"] == "current_humidity_changed_above"
# Wait 6 minutes # Wait 6 minutes
async_fire_time_changed(hass, dt_util.utcnow() + datetime.timedelta(minutes=6)) async_fire_time_changed(hass, dt_util.utcnow() + datetime.timedelta(minutes=6))
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 6 assert len(service_calls) == 6
assert {calls[4].data["some"], calls[5].data["some"]} == { assert {service_calls[4].data["some"], service_calls[5].data["some"]} == {
"current_humidity_changed_above_for", "current_humidity_changed_above_for",
"target_humidity_changed_above_for", "target_humidity_changed_above_for",
} }
@ -405,8 +398,8 @@ async def test_if_fires_on_state_change(
{const.ATTR_HUMIDITY: 37, const.ATTR_CURRENT_HUMIDITY: 41}, {const.ATTR_HUMIDITY: 37, const.ATTR_CURRENT_HUMIDITY: 41},
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 8 assert len(service_calls) == 8
assert {calls[6].data["some"], calls[7].data["some"]} == { assert {service_calls[6].data["some"], service_calls[7].data["some"]} == {
"turn_off device - humidifier.test_5678 - on - off - None", "turn_off device - humidifier.test_5678 - on - off - None",
"turn_on_or_off device - humidifier.test_5678 - on - off - None", "turn_on_or_off device - humidifier.test_5678 - on - off - None",
} }
@ -418,8 +411,8 @@ async def test_if_fires_on_state_change(
{const.ATTR_HUMIDITY: 37, const.ATTR_CURRENT_HUMIDITY: 41}, {const.ATTR_HUMIDITY: 37, const.ATTR_CURRENT_HUMIDITY: 41},
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 10 assert len(service_calls) == 10
assert {calls[8].data["some"], calls[9].data["some"]} == { assert {service_calls[8].data["some"], service_calls[9].data["some"]} == {
"turn_on device - humidifier.test_5678 - off - on - None", "turn_on device - humidifier.test_5678 - off - on - None",
"turn_on_or_off device - humidifier.test_5678 - off - on - None", "turn_on_or_off device - humidifier.test_5678 - off - on - None",
} }
@ -429,7 +422,7 @@ async def test_if_fires_on_state_change_legacy(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off triggers firing.""" """Test for turn_on and turn_off triggers firing."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -479,12 +472,14 @@ async def test_if_fires_on_state_change_legacy(
# Fake that the humidity is changing # Fake that the humidity is changing
hass.states.async_set(entry.entity_id, STATE_ON, {const.ATTR_HUMIDITY: 7}) hass.states.async_set(entry.entity_id, STATE_ON, {const.ATTR_HUMIDITY: 7})
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "target_humidity_changed_below" assert service_calls[0].data["some"] == "target_humidity_changed_below"
async def test_invalid_config( async def test_invalid_config(
hass: HomeAssistant, entity_registry: er.EntityRegistry, calls: list[ServiceCall] hass: HomeAssistant,
entity_registry: er.EntityRegistry,
service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off triggers firing.""" """Test for turn_on and turn_off triggers firing."""
entry = entity_registry.async_get_or_create(DOMAIN, "test", "5678") entry = entity_registry.async_get_or_create(DOMAIN, "test", "5678")
@ -528,7 +523,7 @@ async def test_invalid_config(
hass.states.async_set(entry.entity_id, STATE_ON, {const.ATTR_HUMIDITY: 7}) hass.states.async_set(entry.entity_id, STATE_ON, {const.ATTR_HUMIDITY: 7})
await hass.async_block_till_done() await hass.async_block_till_done()
# Should not trigger for invalid config # Should not trigger for invalid config
assert len(calls) == 0 assert len(service_calls) == 0
async def test_get_trigger_capabilities_on(hass: HomeAssistant) -> None: async def test_get_trigger_capabilities_on(hass: HomeAssistant) -> None:

10
tests/components/light/test_device_action.py
View File

@ -14,7 +14,7 @@ from homeassistant.components.light import (
LightEntityFeature, LightEntityFeature,
) )
from homeassistant.const import EntityCategory from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant, ServiceCall from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.entity_registry import RegistryEntryHider from homeassistant.helpers.entity_registry import RegistryEntryHider
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
@ -32,12 +32,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder.""" """Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_get_actions( async def test_get_actions(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
@ -471,7 +465,6 @@ async def test_action(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off actions.""" """Test for turn_on and turn_off actions."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -636,7 +629,6 @@ async def test_action_legacy(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off actions.""" """Test for turn_on and turn_off actions."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})

41
tests/components/light/test_device_condition.py
View File

@ -22,7 +22,6 @@ from tests.common import (
MockConfigEntry, MockConfigEntry,
async_get_device_automation_capabilities, async_get_device_automation_capabilities,
async_get_device_automations, async_get_device_automations,
async_mock_service,
setup_test_component_platform, setup_test_component_platform,
) )
@ -32,12 +31,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder.""" """Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_get_conditions( async def test_get_conditions(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
@ -186,7 +179,7 @@ async def test_if_state(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off conditions.""" """Test for turn_on and turn_off conditions."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -252,20 +245,20 @@ async def test_if_state(
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "is_on event - test_event1" assert service_calls[0].data["some"] == "is_on event - test_event1"
hass.states.async_set(entry.entity_id, STATE_OFF) hass.states.async_set(entry.entity_id, STATE_OFF)
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert calls[1].data["some"] == "is_off event - test_event2" assert service_calls[1].data["some"] == "is_off event - test_event2"
@pytest.mark.usefixtures("enable_custom_integrations") @pytest.mark.usefixtures("enable_custom_integrations")
@ -273,7 +266,7 @@ async def test_if_state_legacy(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off conditions.""" """Test for turn_on and turn_off conditions."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -318,20 +311,20 @@ async def test_if_state_legacy(
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "is_on event - test_event1" assert service_calls[0].data["some"] == "is_on event - test_event1"
async def test_if_fires_on_for_condition( async def test_if_fires_on_for_condition(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
mock_light_entities: list[MockLight], mock_light_entities: list[MockLight],
) -> None: ) -> None:
"""Test for firing if condition is on with delay.""" """Test for firing if condition is on with delay."""
@ -385,26 +378,26 @@ async def test_if_fires_on_for_condition(
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
# Time travel 10 secs into the future # Time travel 10 secs into the future
freezer.move_to(point2) freezer.move_to(point2)
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
hass.states.async_set(entry.entity_id, STATE_OFF) hass.states.async_set(entry.entity_id, STATE_OFF)
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
# Time travel 20 secs into the future # Time travel 20 secs into the future
freezer.move_to(point3) freezer.move_to(point3)
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "is_off event - test_event1" assert service_calls[0].data["some"] == "is_off event - test_event1"

38
tests/components/light/test_device_trigger.py
View File

@ -20,7 +20,6 @@ from tests.common import (
async_fire_time_changed, async_fire_time_changed,
async_get_device_automation_capabilities, async_get_device_automation_capabilities,
async_get_device_automations, async_get_device_automations,
async_mock_service,
) )
DATA_TEMPLATE_ATTRIBUTES = ( DATA_TEMPLATE_ATTRIBUTES = (
@ -37,12 +36,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder.""" """Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_get_triggers( async def test_get_triggers(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
@ -189,7 +182,7 @@ async def test_if_fires_on_state_change(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off triggers firing.""" """Test for turn_on and turn_off triggers firing."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -258,20 +251,20 @@ async def test_if_fires_on_state_change(
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
hass.states.async_set(entry.entity_id, STATE_OFF) hass.states.async_set(entry.entity_id, STATE_OFF)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert {calls[0].data["some"], calls[1].data["some"]} == { assert {service_calls[0].data["some"], service_calls[1].data["some"]} == {
f"turn_off device - {entry.entity_id} - on - off - None", f"turn_off device - {entry.entity_id} - on - off - None",
f"turn_on_or_off device - {entry.entity_id} - on - off - None", f"turn_on_or_off device - {entry.entity_id} - on - off - None",
} }
hass.states.async_set(entry.entity_id, STATE_ON) hass.states.async_set(entry.entity_id, STATE_ON)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 4 assert len(service_calls) == 4
assert {calls[2].data["some"], calls[3].data["some"]} == { assert {service_calls[2].data["some"], service_calls[3].data["some"]} == {
f"turn_on device - {entry.entity_id} - off - on - None", f"turn_on device - {entry.entity_id} - off - on - None",
f"turn_on_or_off device - {entry.entity_id} - off - on - None", f"turn_on_or_off device - {entry.entity_id} - off - on - None",
} }
@ -282,7 +275,7 @@ async def test_if_fires_on_state_change_legacy(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off triggers firing.""" """Test for turn_on and turn_off triggers firing."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -321,13 +314,14 @@ async def test_if_fires_on_state_change_legacy(
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
hass.states.async_set(entry.entity_id, STATE_OFF) hass.states.async_set(entry.entity_id, STATE_OFF)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert ( assert (
calls[0].data["some"] == f"turn_on device - {entry.entity_id} - on - off - None" service_calls[0].data["some"]
== f"turn_on device - {entry.entity_id} - on - off - None"
) )
@ -336,7 +330,7 @@ async def test_if_fires_on_state_change_with_for(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for triggers firing with delay.""" """Test for triggers firing with delay."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -376,16 +370,16 @@ async def test_if_fires_on_state_change_with_for(
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
hass.states.async_set(entry.entity_id, STATE_OFF) hass.states.async_set(entry.entity_id, STATE_OFF)
await hass.async_block_till_done() 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)) async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
await hass.async_block_till_done() await hass.async_block_till_done()
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"turn_off device - {entry.entity_id} - on - off - 0:00:05" == f"turn_off device - {entry.entity_id} - on - off - 0:00:05"
) )

48
tests/components/lock/test_device_condition.py
View File

@ -21,11 +21,7 @@ from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.entity_registry import RegistryEntryHider from homeassistant.helpers.entity_registry import RegistryEntryHider
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from tests.common import ( from tests.common import MockConfigEntry, async_get_device_automations
MockConfigEntry,
async_get_device_automations,
async_mock_service,
)
@pytest.fixture(autouse=True, name="stub_blueprint_populate") @pytest.fixture(autouse=True, name="stub_blueprint_populate")
@ -33,12 +29,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder.""" """Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_get_conditions( async def test_get_conditions(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
@ -139,7 +129,7 @@ async def test_if_state(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off conditions.""" """Test for turn_on and turn_off conditions."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -291,52 +281,52 @@ async def test_if_state(
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "is_locked - event - test_event1" assert service_calls[0].data["some"] == "is_locked - event - test_event1"
hass.states.async_set(entry.entity_id, STATE_UNLOCKED) hass.states.async_set(entry.entity_id, STATE_UNLOCKED)
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert calls[1].data["some"] == "is_unlocked - event - test_event2" assert service_calls[1].data["some"] == "is_unlocked - event - test_event2"
hass.states.async_set(entry.entity_id, STATE_UNLOCKING) hass.states.async_set(entry.entity_id, STATE_UNLOCKING)
hass.bus.async_fire("test_event3") hass.bus.async_fire("test_event3")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 3 assert len(service_calls) == 3
assert calls[2].data["some"] == "is_unlocking - event - test_event3" assert service_calls[2].data["some"] == "is_unlocking - event - test_event3"
hass.states.async_set(entry.entity_id, STATE_LOCKING) hass.states.async_set(entry.entity_id, STATE_LOCKING)
hass.bus.async_fire("test_event4") hass.bus.async_fire("test_event4")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 4 assert len(service_calls) == 4
assert calls[3].data["some"] == "is_locking - event - test_event4" assert service_calls[3].data["some"] == "is_locking - event - test_event4"
hass.states.async_set(entry.entity_id, STATE_JAMMED) hass.states.async_set(entry.entity_id, STATE_JAMMED)
hass.bus.async_fire("test_event5") hass.bus.async_fire("test_event5")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 5 assert len(service_calls) == 5
assert calls[4].data["some"] == "is_jammed - event - test_event5" assert service_calls[4].data["some"] == "is_jammed - event - test_event5"
hass.states.async_set(entry.entity_id, STATE_OPENING) hass.states.async_set(entry.entity_id, STATE_OPENING)
hass.bus.async_fire("test_event6") hass.bus.async_fire("test_event6")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 6 assert len(service_calls) == 6
assert calls[5].data["some"] == "is_opening - event - test_event6" assert service_calls[5].data["some"] == "is_opening - event - test_event6"
hass.states.async_set(entry.entity_id, STATE_OPEN) hass.states.async_set(entry.entity_id, STATE_OPEN)
hass.bus.async_fire("test_event7") hass.bus.async_fire("test_event7")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 7 assert len(service_calls) == 7
assert calls[6].data["some"] == "is_open - event - test_event7" assert service_calls[6].data["some"] == "is_open - event - test_event7"
async def test_if_state_legacy( async def test_if_state_legacy(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off conditions.""" """Test for turn_on and turn_off conditions."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -380,5 +370,5 @@ async def test_if_state_legacy(
hass.bus.async_fire("test_event1") hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2") hass.bus.async_fire("test_event2")
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert calls[0].data["some"] == "is_locked - event - test_event1" assert service_calls[0].data["some"] == "is_locked - event - test_event1"

61
tests/components/lock/test_device_trigger.py
View File

@ -29,7 +29,6 @@ from tests.common import (
async_fire_time_changed, async_fire_time_changed,
async_get_device_automation_capabilities, async_get_device_automation_capabilities,
async_get_device_automations, async_get_device_automations,
async_mock_service,
) )
@ -38,12 +37,6 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder.""" """Stub copying the blueprints to the config folder."""
@pytest.fixture
def calls(hass: HomeAssistant) -> list[ServiceCall]:
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_get_triggers( async def test_get_triggers(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
@ -212,7 +205,7 @@ async def test_if_fires_on_state_change(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off triggers firing.""" """Test for turn_on and turn_off triggers firing."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -296,27 +289,27 @@ async def test_if_fires_on_state_change(
# Fake that the entity is turning on. # Fake that the entity is turning on.
hass.states.async_set(entry.entity_id, STATE_LOCKED) hass.states.async_set(entry.entity_id, STATE_LOCKED)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"locked - device - {entry.entity_id} - unlocked - locked - None" == f"locked - device - {entry.entity_id} - unlocked - locked - None"
) )
# Fake that the entity is turning off. # Fake that the entity is turning off.
hass.states.async_set(entry.entity_id, STATE_UNLOCKED) hass.states.async_set(entry.entity_id, STATE_UNLOCKED)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
assert ( assert (
calls[1].data["some"] service_calls[1].data["some"]
== f"unlocked - device - {entry.entity_id} - locked - unlocked - None" == f"unlocked - device - {entry.entity_id} - locked - unlocked - None"
) )
# Fake that the entity is opens. # Fake that the entity is opens.
hass.states.async_set(entry.entity_id, STATE_OPEN) hass.states.async_set(entry.entity_id, STATE_OPEN)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 3 assert len(service_calls) == 3
assert ( assert (
calls[2].data["some"] service_calls[2].data["some"]
== f"open - device - {entry.entity_id} - unlocked - open - None" == f"open - device - {entry.entity_id} - unlocked - open - None"
) )
@ -325,7 +318,7 @@ async def test_if_fires_on_state_change_legacy(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for turn_on and turn_off triggers firing.""" """Test for turn_on and turn_off triggers firing."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -371,9 +364,9 @@ async def test_if_fires_on_state_change_legacy(
# Fake that the entity is turning on. # Fake that the entity is turning on.
hass.states.async_set(entry.entity_id, STATE_LOCKED) hass.states.async_set(entry.entity_id, STATE_LOCKED)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"locked - device - {entry.entity_id} - unlocked - locked - None" == f"locked - device - {entry.entity_id} - unlocked - locked - None"
) )
@ -382,7 +375,7 @@ async def test_if_fires_on_state_change_with_for(
hass: HomeAssistant, hass: HomeAssistant,
device_registry: dr.DeviceRegistry, device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
calls: list[ServiceCall], service_calls: list[ServiceCall],
) -> None: ) -> None:
"""Test for triggers firing with delay.""" """Test for triggers firing with delay."""
config_entry = MockConfigEntry(domain="test", data={}) config_entry = MockConfigEntry(domain="test", data={})
@ -516,64 +509,64 @@ async def test_if_fires_on_state_change_with_for(
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(service_calls) == 0
hass.states.async_set(entry.entity_id, STATE_LOCKED) hass.states.async_set(entry.entity_id, STATE_LOCKED)
await hass.async_block_till_done() 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)) async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=10))
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
await hass.async_block_till_done() await hass.async_block_till_done()
assert ( assert (
calls[0].data["some"] service_calls[0].data["some"]
== f"turn_off device - {entry.entity_id} - unlocked - locked - 0:00:05" == f"turn_off device - {entry.entity_id} - unlocked - locked - 0:00:05"
) )
hass.states.async_set(entry.entity_id, STATE_UNLOCKING) hass.states.async_set(entry.entity_id, STATE_UNLOCKING)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(service_calls) == 1
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=16)) async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=16))
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
await hass.async_block_till_done() await hass.async_block_till_done()
assert ( assert (
calls[1].data["some"] service_calls[1].data["some"]
== f"turn_on device - {entry.entity_id} - locked - unlocking - 0:00:05" == f"turn_on device - {entry.entity_id} - locked - unlocking - 0:00:05"
) )
hass.states.async_set(entry.entity_id, STATE_JAMMED) hass.states.async_set(entry.entity_id, STATE_JAMMED)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 2 assert len(service_calls) == 2
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=21)) async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=21))
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 3 assert len(service_calls) == 3
await hass.async_block_till_done() await hass.async_block_till_done()
assert ( assert (
calls[2].data["some"] service_calls[2].data["some"]
== f"turn_off device - {entry.entity_id} - unlocking - jammed - 0:00:05" == f"turn_off device - {entry.entity_id} - unlocking - jammed - 0:00:05"
) )
hass.states.async_set(entry.entity_id, STATE_LOCKING) hass.states.async_set(entry.entity_id, STATE_LOCKING)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 3 assert len(service_calls) == 3
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=27)) async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=27))
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 4 assert len(service_calls) == 4
await hass.async_block_till_done() await hass.async_block_till_done()
assert ( assert (
calls[3].data["some"] service_calls[3].data["some"]
== f"turn_on device - {entry.entity_id} - jammed - locking - 0:00:05" == f"turn_on device - {entry.entity_id} - jammed - locking - 0:00:05"
) )
hass.states.async_set(entry.entity_id, STATE_OPENING) hass.states.async_set(entry.entity_id, STATE_OPENING)
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 4 assert len(service_calls) == 4
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=27)) async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=27))
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 5 assert len(service_calls) == 5
await hass.async_block_till_done() await hass.async_block_till_done()
assert ( assert (
calls[4].data["some"] service_calls[4].data["some"]
== f"turn_on device - {entry.entity_id} - locking - opening - 0:00:05" == f"turn_on device - {entry.entity_id} - locking - opening - 0:00:05"
) )