jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-12-02 14:08:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve test

Browse Source
This commit is contained in:
Erik
2025-12-02 13:11:34 +01:00
parent 4213736eaf
commit 2bb71ec06b
12 changed files with 269 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
tests/components/__init__.py
View File

@@ -29,8 +29,15 @@ from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry, mock_device_registry
async def target_entities(hass: HomeAssistant, domain: str) -> list[str]:
"""Create multiple entities associated with different targets."""
async def target_entities(
hass: HomeAssistant, domain: str
) -> tuple[list[str], list[str]]:
"""Create multiple entities associated with different targets.
Returns a dict with the following keys:
- included: List of entity_ids meant to be targeted.
- excluded: List of entity_ids not meant to be targeted.
"""
await async_setup_component(hass, domain, {})
config_entry = MockConfigEntry(domain="test")
@@ -55,7 +62,7 @@ async def target_entities(hass: HomeAssistant, domain: str) -> list[str]:
mock_device_registry(hass, {device.id: device})
entity_reg = er.async_get(hass)
# Entity associated with area
# Entities associated with area
entity_area = entity_reg.async_get_or_create(
domain=domain,
platform="test",
@@ -63,8 +70,15 @@ async def target_entities(hass: HomeAssistant, domain: str) -> list[str]:
suggested_object_id=f"area_{domain}",
)
entity_reg.async_update_entity(entity_area.entity_id, area_id=area.id)
entity_area_excluded = entity_reg.async_get_or_create(
domain=domain,
platform="test",
unique_id=f"{domain}_area_excluded",
suggested_object_id=f"area_{domain}_excluded",
)
entity_reg.async_update_entity(entity_area_excluded.entity_id, area_id=area.id)
# Entity associated with device
# Entities associated with device
entity_reg.async_get_or_create(
domain=domain,
platform="test",
@@ -72,8 +86,15 @@ async def target_entities(hass: HomeAssistant, domain: str) -> list[str]:
suggested_object_id=f"device_{domain}",
device_id=device.id,
)
entity_reg.async_get_or_create(
domain=domain,
platform="test",
unique_id=f"{domain}_device_excluded",
suggested_object_id=f"device_{domain}_excluded",
device_id=device.id,
)
# Entity associated with label
# Entities associated with label
entity_label = entity_reg.async_get_or_create(
domain=domain,
platform="test",
@@ -81,14 +102,31 @@ async def target_entities(hass: HomeAssistant, domain: str) -> list[str]:
suggested_object_id=f"label_{domain}",
)
entity_reg.async_update_entity(entity_label.entity_id, labels={label.label_id})
entity_label_excluded = entity_reg.async_get_or_create(
domain=domain,
platform="test",
unique_id=f"{domain}_label_excluded",
suggested_object_id=f"label_{domain}_excluded",
)
entity_reg.async_update_entity(
entity_label_excluded.entity_id, labels={label.label_id}
)
# Return all available entities
return [
f"{domain}.standalone_{domain}",
f"{domain}.label_{domain}",
f"{domain}.area_{domain}",
f"{domain}.device_{domain}",
]
return {
"included": [
f"{domain}.standalone_{domain}",
f"{domain}.label_{domain}",
f"{domain}.area_{domain}",
f"{domain}.device_{domain}",
],
"excluded": [
f"{domain}.standalone_{domain}_excluded",
f"{domain}.label_{domain}_excluded",
f"{domain}.area_{domain}_excluded",
f"{domain}.device_{domain}_excluded",
],
}
def parametrize_target_entities(domain: str) -> list[tuple[dict, str, int]]:
@@ -112,11 +150,18 @@ def parametrize_target_entities(domain: str) -> list[tuple[dict, str, int]]:
]
class StateDescription(TypedDict):
class _StateDescription(TypedDict):
"""Test state and expected service call count."""
state: str | None
attributes: dict
class StateDescription(TypedDict):
"""Test state and expected service call count."""
included: _StateDescription
excluded: _StateDescription
count: int
@@ -147,10 +192,26 @@ def parametrize_trigger_states(
) -> dict:
"""Return (state, attributes) dict."""
if isinstance(state, str) or state is None:
return {"state": state, "attributes": additional_attributes, "count": count}
return {
"included": {
"state": state,
"attributes": additional_attributes,
},
"excluded": {
"state": state,
"attributes": {},
},
"count": count,
}
return {
"state": state[0],
"attributes": state[1] | additional_attributes,
"included": {
"state": state[0],
"attributes": state[1] | additional_attributes,
},
"excluded": {
"state": state[0],
"attributes": state[1],
},
"count": count,
}

23
tests/components/alarm_control_panel/test_trigger.py
View File

@@ -42,7 +42,7 @@ def enable_experimental_triggers_conditions() -> Generator[None]:
@pytest.fixture
async def target_alarm_control_panels(hass: HomeAssistant) -> list[str]:
"""Create multiple alarm control panel entities associated with different targets."""
return await target_entities(hass, "alarm_control_panel")
return (await target_entities(hass, "alarm_control_panel"))["included"]
@pytest.mark.parametrize(
@@ -160,13 +160,14 @@ async def test_alarm_control_panel_state_trigger_behavior_any(
# Set all alarm control panels, including the tested one, to the initial state
for eid in target_alarm_control_panels:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -175,7 +176,7 @@ async def test_alarm_control_panel_state_trigger_behavior_any(
# Check if changing other alarm control panels also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * state["count"]
service_calls.clear()
@@ -271,13 +272,14 @@ async def test_alarm_control_panel_state_trigger_behavior_first(
# Set all alarm control panels, including the tested one, to the initial state
for eid in target_alarm_control_panels:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -286,7 +288,7 @@ async def test_alarm_control_panel_state_trigger_behavior_first(
# Triggering other alarm control panels should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
@@ -381,18 +383,19 @@ async def test_alarm_control_panel_state_trigger_behavior_last(
# Set all alarm control panels, including the tested one, to the initial state
for eid in target_alarm_control_panels:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
included_state = state["included"]
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
set_or_remove_state(hass, entity_id, state)
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:

23
tests/components/assist_satellite/test_trigger.py
View File

@@ -39,7 +39,7 @@ def enable_experimental_triggers_conditions() -> Generator[None]:
@pytest.fixture
async def target_assist_satellites(hass: HomeAssistant) -> list[str]:
"""Create multiple assist satellite entities associated with different targets."""
return await target_entities(hass, "assist_satellite")
return (await target_entities(hass, "assist_satellite"))["included"]
@pytest.mark.parametrize(
@@ -111,13 +111,14 @@ async def test_assist_satellite_state_trigger_behavior_any(
# Set all assist satellites, including the tested one, to the initial state
for eid in target_assist_satellites:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -126,7 +127,7 @@ async def test_assist_satellite_state_trigger_behavior_any(
# Check if changing other assist satellites also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * state["count"]
service_calls.clear()
@@ -179,13 +180,14 @@ async def test_assist_satellite_state_trigger_behavior_first(
# Set all assist satellites, including the tested one, to the initial state
for eid in target_assist_satellites:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -194,7 +196,7 @@ async def test_assist_satellite_state_trigger_behavior_first(
# Triggering other assist satellites should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
@@ -246,18 +248,19 @@ async def test_assist_satellite_state_trigger_behavior_last(
# Set all assist satellites, including the tested one, to the initial state
for eid in target_assist_satellites:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
included_state = state["included"]
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
set_or_remove_state(hass, entity_id, state)
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:

63
tests/components/binary_sensor/test_trigger.py
View File

@@ -41,7 +41,7 @@ def enable_experimental_triggers_conditions() -> Generator[None]:
@pytest.fixture
async def target_binary_sensors(hass: HomeAssistant) -> list[str]:
async def target_binary_sensors(hass: HomeAssistant) -> tuple[list[str], list[str]]:
"""Create multiple binary sensor entities associated with different targets."""
return await target_entities(hass, "binary_sensor")
@@ -93,7 +93,7 @@ async def test_binary_sensor_triggers_gated_by_labs_flag(
async def test_binary_sensor_state_attribute_trigger_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_binary_sensors: list[str],
target_binary_sensors: dict[list[str], list[str]],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
@@ -103,17 +103,23 @@ async def test_binary_sensor_state_attribute_trigger_behavior_any(
"""Test that the binary sensor state trigger fires when any binary sensor state changes to a specific state."""
await async_setup_component(hass, "binary_sensor", {})
other_entity_ids = set(target_binary_sensors) - {entity_id}
other_entity_ids = set(target_binary_sensors["included"]) - {entity_id}
excluded_entity_ids = set(target_binary_sensors["excluded"]) - {entity_id}
# Set all binary sensors, including the tested binary sensor, to the initial state
for eid in target_binary_sensors:
set_or_remove_state(hass, eid, states[0])
for eid in target_binary_sensors["included"]:
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
for eid in excluded_entity_ids:
set_or_remove_state(hass, eid, states[0]["excluded"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
excluded_state = state["excluded"]
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -122,7 +128,10 @@ async def test_binary_sensor_state_attribute_trigger_behavior_any(
# Check if changing other binary sensors also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
for excluded_entity_id in excluded_entity_ids:
set_or_remove_state(hass, excluded_entity_id, excluded_state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * state["count"]
service_calls.clear()
@@ -165,17 +174,23 @@ async def test_binary_sensor_state_attribute_trigger_behavior_first(
"""Test that the binary sensor state trigger fires when the first binary sensor state changes to a specific state."""
await async_setup_component(hass, "binary_sensor", {})
other_entity_ids = set(target_binary_sensors) - {entity_id}
other_entity_ids = set(target_binary_sensors["included"]) - {entity_id}
excluded_entity_ids = set(target_binary_sensors["excluded"]) - {entity_id}
# Set all binary sensors, including the tested binary sensor, to the initial state
for eid in target_binary_sensors:
set_or_remove_state(hass, eid, states[0])
for eid in target_binary_sensors["included"]:
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
for eid in excluded_entity_ids:
set_or_remove_state(hass, eid, states[0]["excluded"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
excluded_state = state["excluded"]
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -184,7 +199,10 @@ async def test_binary_sensor_state_attribute_trigger_behavior_first(
# Triggering other binary sensors should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, excluded_state)
await hass.async_block_till_done()
for excluded_entity_id in excluded_entity_ids:
set_or_remove_state(hass, excluded_entity_id, excluded_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
@@ -226,24 +244,35 @@ async def test_binary_sensor_state_attribute_trigger_behavior_last(
"""Test that the binary sensor state trigger fires when the last binary sensor state changes to a specific state."""
await async_setup_component(hass, "binary_sensor", {})
other_entity_ids = set(target_binary_sensors) - {entity_id}
other_entity_ids = set(target_binary_sensors["included"]) - {entity_id}
excluded_entity_ids = set(target_binary_sensors["excluded"]) - {entity_id}
# Set all binary sensors, including the tested binary sensor, to the initial state
for eid in target_binary_sensors:
set_or_remove_state(hass, eid, states[0])
for eid in target_binary_sensors["included"]:
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
for eid in excluded_entity_ids:
set_or_remove_state(hass, eid, states[0]["excluded"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
excluded_state = state["excluded"]
included_state = state["included"]
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, excluded_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
set_or_remove_state(hass, entity_id, state)
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
for excluded_entity_id in excluded_entity_ids:
set_or_remove_state(hass, excluded_entity_id, excluded_state)
await hass.async_block_till_done()
assert len(service_calls) == 0

44
tests/components/climate/test_trigger.py
View File

@@ -43,7 +43,7 @@ def enable_experimental_triggers_conditions() -> Generator[None]:
@pytest.fixture
async def target_climates(hass: HomeAssistant) -> list[str]:
"""Create multiple climate entities associated with different targets."""
return await target_entities(hass, "climate")
return (await target_entities(hass, "climate"))["included"]
@pytest.mark.parametrize(
@@ -113,13 +113,14 @@ async def test_climate_state_trigger_behavior_any(
# Set all climates, including the tested climate, to the initial state
for eid in target_climates:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -128,7 +129,7 @@ async def test_climate_state_trigger_behavior_any(
# Check if changing other climates also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * state["count"]
service_calls.clear()
@@ -176,13 +177,14 @@ async def test_climate_state_attribute_trigger_behavior_any(
# Set all climates, including the tested climate, to the initial state
for eid in target_climates:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -191,7 +193,7 @@ async def test_climate_state_attribute_trigger_behavior_any(
# Check if changing other climates also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * state["count"]
service_calls.clear()
@@ -243,13 +245,14 @@ async def test_climate_state_trigger_behavior_first(
# Set all climates, including the tested climate, to the initial state
for eid in target_climates:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -258,7 +261,7 @@ async def test_climate_state_trigger_behavior_first(
# Triggering other climates should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
@@ -305,13 +308,14 @@ async def test_climate_state_attribute_trigger_behavior_first(
# Set all climates, including the tested climate, to the initial state
for eid in target_climates:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -320,7 +324,7 @@ async def test_climate_state_attribute_trigger_behavior_first(
# Triggering other climates should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
@@ -371,18 +375,19 @@ async def test_climate_state_trigger_behavior_last(
# Set all climates, including the tested climate, to the initial state
for eid in target_climates:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
included_state = state["included"]
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
set_or_remove_state(hass, entity_id, state)
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -432,18 +437,19 @@ async def test_climate_state_attribute_trigger_behavior_last(
# Set all climates, including the tested climate, to the initial state
for eid in target_climates:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
included_state = state["included"]
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
set_or_remove_state(hass, entity_id, state)
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:

23
tests/components/fan/test_trigger.py
View File

@@ -37,7 +37,7 @@ def enable_experimental_triggers_conditions() -> Generator[None]:
@pytest.fixture
async def target_fans(hass: HomeAssistant) -> list[str]:
"""Create multiple fan entities associated with different targets."""
return await target_entities(hass, "fan")
return (await target_entities(hass, "fan"))["included"]
@pytest.mark.parametrize(
@@ -97,13 +97,14 @@ async def test_fan_state_trigger_behavior_any(
# Set all fans, including the tested fan, to the initial state
for eid in target_fans:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -112,7 +113,7 @@ async def test_fan_state_trigger_behavior_any(
# Check if changing other fans also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * state["count"]
service_calls.clear()
@@ -155,13 +156,14 @@ async def test_fan_state_trigger_behavior_first(
# Set all fans, including the tested fan, to the initial state
for eid in target_fans:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -170,7 +172,7 @@ async def test_fan_state_trigger_behavior_first(
# Triggering other fans should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
@@ -212,18 +214,19 @@ async def test_fan_state_trigger_behavior_last(
# Set all fans, including the tested fan, to the initial state
for eid in target_fans:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
included_state = state["included"]
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
set_or_remove_state(hass, entity_id, state)
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:

23
tests/components/lawn_mower/test_trigger.py
View File

@@ -39,7 +39,7 @@ def enable_experimental_triggers_conditions() -> Generator[None]:
@pytest.fixture
async def target_lawn_mowers(hass: HomeAssistant) -> list[str]:
"""Create multiple lawn mower entities associated with different targets."""
return await target_entities(hass, "lawn_mower")
return (await target_entities(hass, "lawn_mower"))["included"]
@pytest.mark.parametrize(
@@ -111,13 +111,14 @@ async def test_lawn_mower_state_trigger_behavior_any(
# Set all lawn mowers, including the tested one, to the initial state
for eid in target_lawn_mowers:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -126,7 +127,7 @@ async def test_lawn_mower_state_trigger_behavior_any(
# Check if changing other lawn mowers also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * state["count"]
service_calls.clear()
@@ -179,13 +180,14 @@ async def test_lawn_mower_state_trigger_behavior_first(
# Set all lawn mowers, including the tested one, to the initial state
for eid in target_lawn_mowers:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -194,7 +196,7 @@ async def test_lawn_mower_state_trigger_behavior_first(
# Triggering other lawn mowers should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
@@ -246,18 +248,19 @@ async def test_lawn_mower_state_trigger_behavior_last(
# Set all lawn mowers, including the tested one, to the initial state
for eid in target_lawn_mowers:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
included_state = state["included"]
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
set_or_remove_state(hass, entity_id, state)
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:

4
tests/components/light/test_condition.py
View File

@@ -40,13 +40,13 @@ def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
@pytest.fixture
async def target_lights(hass: HomeAssistant) -> list[str]:
"""Create multiple light entities associated with different targets."""
return await target_entities(hass, "light")
return (await target_entities(hass, "light"))[0]
@pytest.fixture
async def target_switches(hass: HomeAssistant) -> list[str]:
"""Create multiple switch entities associated with different targets."""
return await target_entities(hass, "switch")
return (await target_entities(hass, "switch"))[0]
async def setup_automation_with_light_condition(

23
tests/components/light/test_trigger.py
View File

@@ -37,7 +37,7 @@ def enable_experimental_triggers_conditions() -> Generator[None]:
@pytest.fixture
async def target_lights(hass: HomeAssistant) -> list[str]:
"""Create multiple light entities associated with different targets."""
return await target_entities(hass, "light")
return (await target_entities(hass, "light"))["included"]
@pytest.mark.parametrize(
@@ -97,13 +97,14 @@ async def test_light_state_trigger_behavior_any(
# Set all lights, including the tested light, to the initial state
for eid in target_lights:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -112,7 +113,7 @@ async def test_light_state_trigger_behavior_any(
# Check if changing other lights also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * state["count"]
service_calls.clear()
@@ -155,13 +156,14 @@ async def test_light_state_trigger_behavior_first(
# Set all lights, including the tested light, to the initial state
for eid in target_lights:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -170,7 +172,7 @@ async def test_light_state_trigger_behavior_first(
# Triggering other lights should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
@@ -212,18 +214,19 @@ async def test_light_state_trigger_behavior_last(
# Set all lights, including the tested light, to the initial state
for eid in target_lights:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
included_state = state["included"]
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
set_or_remove_state(hass, entity_id, state)
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:

23
tests/components/media_player/test_trigger.py
View File

@@ -38,7 +38,7 @@ def enable_experimental_triggers_conditions() -> Generator[None]:
@pytest.fixture
async def target_media_players(hass: HomeAssistant) -> list[str]:
"""Create multiple media player entities associated with different targets."""
return await target_entities(hass, "media_player")
return (await target_entities(hass, "media_player"))["included"]
@pytest.mark.parametrize(
@@ -100,13 +100,14 @@ async def test_media_player_state_trigger_behavior_any(
# Set all media players, including the tested media player, to the initial state
for eid in target_media_players:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -115,7 +116,7 @@ async def test_media_player_state_trigger_behavior_any(
# Check if changing other media players also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * state["count"]
service_calls.clear()
@@ -161,13 +162,14 @@ async def test_media_player_state_trigger_behavior_first(
# Set all media players, including the tested media player, to the initial state
for eid in target_media_players:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -176,7 +178,7 @@ async def test_media_player_state_trigger_behavior_first(
# Triggering other media players should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
@@ -221,18 +223,19 @@ async def test_media_player_state_trigger_behavior_last(
# Set all media players, including the tested media player, to the initial state
for eid in target_media_players:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
included_state = state["included"]
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
set_or_remove_state(hass, entity_id, state)
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:

50
tests/components/text/test_trigger.py
View File

@@ -41,7 +41,7 @@ def enable_experimental_triggers_conditions() -> Generator[None]:
@pytest.fixture
async def target_texts(hass: HomeAssistant) -> list[str]:
"""Create multiple text entities associated with different targets."""
return await target_entities(hass, "text")
return (await target_entities(hass, "text"))["included"]
@pytest.mark.parametrize(
@@ -94,43 +94,50 @@ async def test_text_triggers_gated_by_labs_flag(
(
"text.changed",
[
{"state": None, "attributes": {}, "count": 0},
{"state": "bar", "attributes": {}, "count": 0},
{"state": "baz", "attributes": {}, "count": 1},
{"included": {"state": None, "attributes": {}}, "count": 0},
{"included": {"state": "bar", "attributes": {}}, "count": 0},
{"included": {"state": "baz", "attributes": {}}, "count": 1},
],
),
(
"text.changed",
[
{"state": "foo", "attributes": {}, "count": 0},
{"state": "bar", "attributes": {}, "count": 1},
{"state": "baz", "attributes": {}, "count": 1},
{"included": {"state": "foo", "attributes": {}}, "count": 0},
{"included": {"state": "bar", "attributes": {}}, "count": 1},
{"included": {"state": "baz", "attributes": {}}, "count": 1},
],
),
(
"text.changed",
[
{"state": "foo", "attributes": {}, "count": 0},
{"state": "", "attributes": {}, "count": 1}, # empty string
{"state": "baz", "attributes": {}, "count": 1},
{"included": {"state": "foo", "attributes": {}}, "count": 0},
# empty string
{"included": {"state": "", "attributes": {}}, "count": 1},
{"included": {"state": "baz", "attributes": {}}, "count": 1},
],
),
(
"text.changed",
[
{"state": STATE_UNAVAILABLE, "attributes": {}, "count": 0},
{"state": "bar", "attributes": {}, "count": 0},
{"state": "baz", "attributes": {}, "count": 1},
{"state": STATE_UNAVAILABLE, "attributes": {}, "count": 0},
{
"included": {"state": STATE_UNAVAILABLE, "attributes": {}},
"count": 0,
},
{"included": {"state": "bar", "attributes": {}}, "count": 0},
{"included": {"state": "baz", "attributes": {}}, "count": 1},
{
"included": {"state": STATE_UNAVAILABLE, "attributes": {}},
"count": 0,
},
],
),
(
"text.changed",
[
{"state": STATE_UNKNOWN, "attributes": {}, "count": 0},
{"state": "bar", "attributes": {}, "count": 0},
{"state": "baz", "attributes": {}, "count": 1},
{"state": STATE_UNKNOWN, "attributes": {}, "count": 0},
{"included": {"state": STATE_UNKNOWN, "attributes": {}}, "count": 0},
{"included": {"state": "bar", "attributes": {}}, "count": 0},
{"included": {"state": "baz", "attributes": {}}, "count": 1},
{"included": {"state": STATE_UNKNOWN, "attributes": {}}, "count": 0},
],
),
],
@@ -152,13 +159,14 @@ async def test_text_state_trigger_behavior_any(
# Set all texts, including the tested text, to the initial state
for eid in target_texts:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, None, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -167,7 +175,7 @@ async def test_text_state_trigger_behavior_any(
# Check if changing other texts also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * state["count"]
service_calls.clear()

23
tests/components/vacuum/test_trigger.py
View File

@@ -39,7 +39,7 @@ def enable_experimental_triggers_conditions() -> Generator[None]:
@pytest.fixture
async def target_vacuums(hass: HomeAssistant) -> list[str]:
"""Create multiple vacuum entities associated with different targets."""
return await target_entities(hass, "vacuum")
return (await target_entities(hass, "vacuum"))["included"]
@pytest.mark.parametrize(
@@ -111,13 +111,14 @@ async def test_vacuum_state_trigger_behavior_any(
# Set all vacuums, including the tested one, to the initial state
for eid in target_vacuums:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -126,7 +127,7 @@ async def test_vacuum_state_trigger_behavior_any(
# Check if changing other vacuums also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * state["count"]
service_calls.clear()
@@ -179,13 +180,14 @@ async def test_vacuum_state_trigger_behavior_first(
# Set all vacuums, including the tested one, to the initial state
for eid in target_vacuums:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
@@ -194,7 +196,7 @@ async def test_vacuum_state_trigger_behavior_first(
# Triggering other vacuums should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
@@ -246,18 +248,19 @@ async def test_vacuum_state_trigger_behavior_last(
# Set all vacuums, including the tested one, to the initial state
for eid in target_vacuums:
set_or_remove_state(hass, eid, states[0])
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
included_state = state["included"]
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == 0
set_or_remove_state(hass, entity_id, state)
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls: