Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-05-15 04:51:20 +01:00
Code Activity
Files
ef1a5ea2df2d2c405bc2c207cb6fa2df35f09b56
core/tests/helpers/test_trigger.py
T
Erik Montnemery a9becca321 Add duration to state based entity triggers (#167740)
2026-04-16 07:38:50 +02:00

3883 lines
125 KiB
Python
Raw Blame History

"""The tests for the trigger helper."""
from collections.abc import Mapping
from contextlib import AbstractContextManager, nullcontext as does_not_raise
import datetime
import io
import logging
from typing import Any
from unittest.mock import ANY, AsyncMock, MagicMock, Mock, call, patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from pytest_unordered import unordered
import voluptuous as vol
from homeassistant.components import automation
from homeassistant.components.sun import DOMAIN as SUN_DOMAIN
from homeassistant.components.system_health import DOMAIN as SYSTEM_HEALTH_DOMAIN
from homeassistant.components.tag import DOMAIN as TAG_DOMAIN
from homeassistant.components.text import DOMAIN as TEXT_DOMAIN
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_UNIT_OF_MEASUREMENT,
CONF_ENTITY_ID,
CONF_FOR,
CONF_OPTIONS,
CONF_PLATFORM,
CONF_TARGET,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
UnitOfTemperature,
)
from homeassistant.core import (
CALLBACK_TYPE,
Context,
HomeAssistant,
ServiceCall,
State,
callback,
)
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv, trigger
from homeassistant.helpers.automation import (
DomainSpec,
move_top_level_schema_fields_to_options,
)
from homeassistant.helpers.trigger import (
ATTR_BEHAVIOR,
BEHAVIOR_ANY,
BEHAVIOR_FIRST,
BEHAVIOR_LAST,
DATA_PLUGGABLE_ACTIONS,
TRIGGERS,
EntityNumericalStateChangedTriggerWithUnitBase,
EntityNumericalStateCrossedThresholdTriggerWithUnitBase,
EntityTriggerBase,
PluggableAction,
Trigger,
TriggerActionRunner,
TriggerConfig,
_async_get_trigger_platform,
async_initialize_triggers,
async_validate_trigger_config,
make_entity_numerical_state_changed_trigger,
make_entity_numerical_state_crossed_threshold_trigger,
make_entity_origin_state_trigger,
make_entity_target_state_trigger,
make_entity_transition_trigger,
)
from homeassistant.helpers.typing import ConfigType
from homeassistant.loader import Integration, async_get_integration
from homeassistant.setup import async_setup_component
from homeassistant.util.unit_conversion import TemperatureConverter
from homeassistant.util.yaml.loader import parse_yaml
from tests.common import (
MockModule,
MockPlatform,
async_fire_time_changed,
mock_integration,
mock_platform,
)
from tests.typing import WebSocketGenerator
async def test_bad_trigger_platform(hass: HomeAssistant) -> None:
"""Test bad trigger platform."""
with pytest.raises(vol.Invalid) as ex:
await async_validate_trigger_config(hass, [{"platform": "not_a_platform"}])
assert "Invalid trigger 'not_a_platform' specified" in str(ex)
async def test_trigger_subtype(hass: HomeAssistant) -> None:
"""Test trigger subtypes."""
with patch(
"homeassistant.helpers.trigger.async_get_integration",
return_value=MagicMock(async_get_platform=AsyncMock()),
) as integration_mock:
await _async_get_trigger_platform(hass, "test.subtype")
assert integration_mock.call_args == call(hass, "test")
async def test_trigger_variables(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test trigger variables."""
assert await async_setup_component(
hass,
"automation",
{
"automation": {
"trigger": {
"platform": "event",
"event_type": "test_event",
"variables": {
"name": "Paulus",
"via_event": "{{ trigger.event.event_type }}",
},
},
"action": {
"service": "test.automation",
"data_template": {"hello": "{{ name }} + {{ via_event }}"},
},
}
},
)
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(service_calls) == 1
assert service_calls[0].data["hello"] == "Paulus + test_event"
async def test_if_disabled_trigger_not_firing(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test disabled triggers don't fire."""
assert await async_setup_component(
hass,
"automation",
{
"automation": {
"trigger": [
{
"platform": "event",
"event_type": "enabled_trigger_event",
},
{
"enabled": False,
"platform": "event",
"event_type": "disabled_trigger_event",
},
],
"action": {
"service": "test.automation",
},
}
},
)
hass.bus.async_fire("disabled_trigger_event")
await hass.async_block_till_done()
assert not service_calls
hass.bus.async_fire("enabled_trigger_event")
await hass.async_block_till_done()
assert len(service_calls) == 1
async def test_trigger_enabled_templates(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test triggers enabled by template."""
assert await async_setup_component(
hass,
"automation",
{
"automation": {
"trigger": [
{
"enabled": "{{ 'some text' }}",
"platform": "event",
"event_type": "truthy_template_trigger_event",
},
{
"enabled": "{{ 3 == 4 }}",
"platform": "event",
"event_type": "falsy_template_trigger_event",
},
{
"enabled": False, # eg. from a blueprints input defaulting to `false`
"platform": "event",
"event_type": "falsy_trigger_event",
},
{
"enabled": "some text", # eg. from a blueprints input value
"platform": "event",
"event_type": "truthy_trigger_event",
},
],
"action": {
"service": "test.automation",
},
}
},
)
hass.bus.async_fire("falsy_template_trigger_event")
await hass.async_block_till_done()
assert not service_calls
hass.bus.async_fire("falsy_trigger_event")
await hass.async_block_till_done()
assert not service_calls
hass.bus.async_fire("truthy_template_trigger_event")
await hass.async_block_till_done()
assert len(service_calls) == 1
hass.bus.async_fire("truthy_trigger_event")
await hass.async_block_till_done()
assert len(service_calls) == 2
async def test_nested_trigger_list(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test triggers within nested list."""
assert await async_setup_component(
hass,
"automation",
{
"automation": {
"trigger": [
{
"triggers": {
"platform": "event",
"event_type": "trigger_1",
},
},
{
"platform": "event",
"event_type": "trigger_2",
},
{"triggers": []},
{"triggers": None},
{
"triggers": [
{
"platform": "event",
"event_type": "trigger_3",
},
{
"platform": "event",
"event_type": "trigger_4",
},
],
},
],
"action": {
"service": "test.automation",
},
}
},
)
hass.bus.async_fire("trigger_1")
await hass.async_block_till_done()
assert len(service_calls) == 1
hass.bus.async_fire("trigger_2")
await hass.async_block_till_done()
assert len(service_calls) == 2
hass.bus.async_fire("trigger_none")
await hass.async_block_till_done()
assert len(service_calls) == 2
hass.bus.async_fire("trigger_3")
await hass.async_block_till_done()
assert len(service_calls) == 3
hass.bus.async_fire("trigger_4")
await hass.async_block_till_done()
assert len(service_calls) == 4
async def test_trigger_enabled_template_limited(
hass: HomeAssistant,
service_calls: list[ServiceCall],
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test triggers enabled invalid template."""
assert await async_setup_component(
hass,
"automation",
{
"automation": {
"trigger": [
{
"enabled": "{{ states('sensor.limited') }}", # only limited template supported
"platform": "event",
"event_type": "test_event",
},
],
"action": {
"service": "test.automation",
},
}
},
)
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert not service_calls
assert "Error rendering enabled template" in caplog.text
async def test_trigger_alias(
hass: HomeAssistant,
service_calls: list[ServiceCall],
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test triggers support aliases."""
assert await async_setup_component(
hass,
"automation",
{
"automation": {
"trigger": [
{
"alias": "My event",
"platform": "event",
"event_type": "trigger_event",
}
],
"action": {
"service": "test.automation",
"data_template": {"alias": "{{ trigger.alias }}"},
},
}
},
)
hass.bus.async_fire("trigger_event")
await hass.async_block_till_done()
assert len(service_calls) == 1
assert service_calls[0].data["alias"] == "My event"
assert (
"Automation trigger 'My event' triggered by event 'trigger_event'"
in caplog.text
)
async def test_async_initialize_triggers(
hass: HomeAssistant,
service_calls: list[ServiceCall],
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test async_initialize_triggers with different action types."""
log_cb = MagicMock()
action_calls = []
trigger_config = await async_validate_trigger_config(
hass,
[
{
"platform": "event",
"event_type": ["trigger_event"],
"variables": {
"name": "Paulus",
"via_event": "{{ trigger.event.event_type }}",
},
}
],
)
async def async_action(*args):
action_calls.append([*args])
@callback
def cb_action(*args):
action_calls.append([*args])
def non_cb_action(*args):
action_calls.append([*args])
for action in (async_action, cb_action, non_cb_action):
action_calls = []
unsub = await async_initialize_triggers(
hass,
trigger_config,
action,
"test",
"",
log_cb,
)
await hass.async_block_till_done()
hass.bus.async_fire("trigger_event")
await hass.async_block_till_done()
await hass.async_block_till_done()
assert len(action_calls) == 1
assert action_calls[0][0]["name"] == "Paulus"
assert action_calls[0][0]["via_event"] == "trigger_event"
log_cb.assert_called_once_with(ANY, "Initialized trigger")
log_cb.reset_mock()
unsub()
async def test_pluggable_action(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test normal behavior of pluggable actions."""
update_1 = MagicMock()
update_2 = MagicMock()
action_1 = AsyncMock()
action_2 = AsyncMock()
trigger_1 = {"domain": "test", "device": "1"}
trigger_2 = {"domain": "test", "device": "2"}
variables_1 = {"source": "test 1"}
variables_2 = {"source": "test 2"}
context_1 = Context()
context_2 = Context()
plug_1 = PluggableAction(update_1)
plug_2 = PluggableAction(update_2)
# Verify plug is inactive without triggers
remove_plug_1 = plug_1.async_register(hass, trigger_1)
assert not plug_1
assert not plug_2
# Verify plug remain inactive with non matching trigger
remove_attach_2 = PluggableAction.async_attach_trigger(
hass, trigger_2, action_2, variables_2
)
assert not plug_1
assert not plug_2
update_1.assert_not_called()
update_2.assert_not_called()
# Verify plug is active, and update when matching trigger attaches
remove_attach_1 = PluggableAction.async_attach_trigger(
hass, trigger_1, action_1, variables_1
)
assert plug_1
assert not plug_2
update_1.assert_called()
update_1.reset_mock()
update_2.assert_not_called()
# Verify a non registered plug is inactive
remove_plug_1()
assert not plug_1
assert not plug_2
# Verify a plug registered to existing trigger is true
remove_plug_1 = plug_1.async_register(hass, trigger_1)
assert plug_1
assert not plug_2
remove_plug_2 = plug_2.async_register(hass, trigger_2)
assert plug_1
assert plug_2
# Verify no actions should have been triggered so far
action_1.assert_not_called()
action_2.assert_not_called()
# Verify action is triggered with correct data
await plug_1.async_run(hass, context_1)
await plug_2.async_run(hass, context_2)
action_1.assert_called_with(variables_1, context_1)
action_2.assert_called_with(variables_2, context_2)
# Verify plug goes inactive if trigger is removed
remove_attach_1()
assert not plug_1
# Verify registry is cleaned when no plugs nor triggers are attached
assert hass.data[DATA_PLUGGABLE_ACTIONS]
remove_plug_1()
remove_plug_2()
remove_attach_2()
assert not hass.data[DATA_PLUGGABLE_ACTIONS]
assert not plug_2
class TriggerActionFunctionTypeHelper:
"""Helper for testing different trigger action function types."""
def __init__(self) -> None:
"""Init helper."""
self.action_calls = []
@callback
def cb_action(self, *args):
"""Callback action."""
self.action_calls.append([*args])
def sync_action(self, *args):
"""Sync action."""
self.action_calls.append([*args])
async def async_action(self, *args):
"""Async action."""
self.action_calls.append([*args])
@pytest.mark.parametrize("action_method", ["cb_action", "sync_action", "async_action"])
async def test_platform_multiple_triggers(
hass: HomeAssistant, action_method: str
) -> None:
"""Test a trigger platform with multiple trigger."""
class MockTrigger(Trigger):
"""Mock trigger."""
@classmethod
async def async_validate_config(
cls, hass: HomeAssistant, config: ConfigType
) -> ConfigType:
"""Validate config."""
return config
class MockTrigger1(MockTrigger):
"""Mock trigger 1."""
async def async_attach_runner(
self, run_action: TriggerActionRunner
) -> CALLBACK_TYPE:
"""Attach a trigger."""
run_action({"extra": "test_trigger_1"}, "trigger 1 desc")
class MockTrigger2(MockTrigger):
"""Mock trigger 2."""
async def async_attach_runner(
self, run_action: TriggerActionRunner
) -> CALLBACK_TYPE:
"""Attach a trigger."""
run_action({"extra": "test_trigger_2"}, "trigger 2 desc")
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
return {
"_": MockTrigger1,
"trig_2": MockTrigger2,
}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
config_1 = [{"platform": "test"}]
config_2 = [{"platform": "test.trig_2", "options": {"x": 1}}]
config_3 = [{"platform": "test.unknown_trig"}]
assert await async_validate_trigger_config(hass, config_1) == config_1
assert await async_validate_trigger_config(hass, config_2) == config_2
with pytest.raises(
vol.Invalid, match="Invalid trigger 'test.unknown_trig' specified"
):
await async_validate_trigger_config(hass, config_3)
log_cb = MagicMock()
action_helper = TriggerActionFunctionTypeHelper()
action_method = getattr(action_helper, action_method)
await async_initialize_triggers(hass, config_1, action_method, "test", "", log_cb)
await hass.async_block_till_done()
assert len(action_helper.action_calls) == 1
assert action_helper.action_calls[0][0] == {
"trigger": {
"alias": None,
"description": "trigger 1 desc",
"extra": "test_trigger_1",
"id": "0",
"idx": "0",
"platform": "test",
}
}
action_helper.action_calls.clear()
await async_initialize_triggers(hass, config_2, action_method, "test", "", log_cb)
await hass.async_block_till_done()
assert len(action_helper.action_calls) == 1
assert action_helper.action_calls[0][0] == {
"trigger": {
"alias": None,
"description": "trigger 2 desc",
"extra": "test_trigger_2",
"id": "0",
"idx": "0",
"platform": "test.trig_2",
}
}
action_helper.action_calls.clear()
with pytest.raises(KeyError):
await async_initialize_triggers(
hass, config_3, action_method, "test", "", log_cb
)
async def test_platform_migrate_trigger(hass: HomeAssistant) -> None:
"""Test a trigger platform with a migration."""
OPTIONS_SCHEMA_DICT = {
vol.Required("option_1"): str,
vol.Optional("option_2"): int,
}
class MockTrigger(Trigger):
"""Mock trigger."""
@classmethod
async def async_validate_complete_config(
cls, hass: HomeAssistant, complete_config: ConfigType
) -> ConfigType:
"""Validate complete config."""
complete_config = move_top_level_schema_fields_to_options(
complete_config, OPTIONS_SCHEMA_DICT
)
return await super().async_validate_complete_config(hass, complete_config)
@classmethod
async def async_validate_config(
cls, hass: HomeAssistant, config: ConfigType
) -> ConfigType:
"""Validate config."""
return config
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
return {
"_": MockTrigger,
}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
config_1 = [{"platform": "test", "option_1": "value_1", "option_2": 2}]
config_2 = [{"platform": "test", "option_1": "value_1"}]
config_3 = [{"platform": "test", "options": {"option_1": "value_1", "option_2": 2}}]
config_4 = [{"platform": "test", "options": {"option_1": "value_1"}}]
assert await async_validate_trigger_config(hass, config_1) == config_3
assert await async_validate_trigger_config(hass, config_2) == config_4
assert await async_validate_trigger_config(hass, config_3) == config_3
assert await async_validate_trigger_config(hass, config_4) == config_4
async def test_platform_backwards_compatibility_for_new_style_configs(
hass: HomeAssistant,
) -> None:
"""Test backwards compatibility for old-style triggers with new-style configs."""
class MockTriggerPlatform:
"""Mock trigger platform."""
TRIGGER_SCHEMA = cv.TRIGGER_BASE_SCHEMA.extend(
{
vol.Required("option_1"): str,
vol.Optional("option_2"): int,
}
)
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", MockTriggerPlatform())
config_old_style = [{"platform": "test", "option_1": "value_1", "option_2": 2}]
result = await async_validate_trigger_config(hass, config_old_style)
assert result == config_old_style
config_new_style = [
{"platform": "test", "options": {"option_1": "value_1", "option_2": 2}}
]
result = await async_validate_trigger_config(hass, config_new_style)
assert result == config_old_style
async def test_get_trigger_platform_registers_triggers(
hass: HomeAssistant,
) -> None:
"""Test _async_get_trigger_platform registers triggers and notifies subscribers."""
class MockTrigger(Trigger):
"""Mock trigger."""
async def async_attach_runner(
self, run_action: TriggerActionRunner
) -> CALLBACK_TYPE:
return lambda: None
async def async_get_triggers(
hass: HomeAssistant,
) -> dict[str, type[Trigger]]:
return {"trig_a": MockTrigger, "trig_b": MockTrigger}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
subscriber_events: list[set[str]] = []
async def subscriber(new_triggers: set[str]) -> None:
subscriber_events.append(new_triggers)
trigger.async_subscribe_platform_events(hass, subscriber)
assert "test.trig_a" not in hass.data[TRIGGERS]
assert "test.trig_b" not in hass.data[TRIGGERS]
# First call registers all triggers from the platform and notifies subscribers
await _async_get_trigger_platform(hass, "test.trig_a")
assert hass.data[TRIGGERS]["test.trig_a"] == "test"
assert hass.data[TRIGGERS]["test.trig_b"] == "test"
assert len(subscriber_events) == 1
assert subscriber_events[0] == {"test.trig_a", "test.trig_b"}
# Subsequent calls are idempotent — no re-registration or re-notification
await _async_get_trigger_platform(hass, "test.trig_a")
await _async_get_trigger_platform(hass, "test.trig_b")
assert len(subscriber_events) == 1
@pytest.mark.parametrize(
"sun_trigger_descriptions",
[
"""
_:
fields:
event:
example: sunrise
selector:
select:
options:
- sunrise
- sunset
offset:
selector:
time: null
""",
"""
.anchor: &anchor
- sunrise
- sunset
_:
fields:
event:
example: sunrise
selector:
select:
options: *anchor
offset:
selector:
time: null
""",
],
)
async def test_async_get_all_descriptions(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
sun_trigger_descriptions: str,
) -> None:
"""Test async_get_all_descriptions."""
tag_trigger_descriptions = """
_:
target:
entity:
domain: alarm_control_panel
"""
text_trigger_descriptions = """
changed:
target:
entity:
domain: text
"""
ws_client = await hass_ws_client(hass)
assert await async_setup_component(hass, SUN_DOMAIN, {})
assert await async_setup_component(hass, SYSTEM_HEALTH_DOMAIN, {})
await hass.async_block_till_done()
def _load_yaml(fname, secrets=None):
if fname.endswith("sun/triggers.yaml"):
trigger_descriptions = sun_trigger_descriptions
elif fname.endswith("tag/triggers.yaml"):
trigger_descriptions = tag_trigger_descriptions
elif fname.endswith("text/triggers.yaml"):
trigger_descriptions = text_trigger_descriptions
with io.StringIO(trigger_descriptions) as file:
return parse_yaml(file)
with (
patch(
"homeassistant.helpers.trigger._load_triggers_files",
side_effect=trigger._load_triggers_files,
) as proxy_load_triggers_files,
patch(
"annotatedyaml.loader.load_yaml",
side_effect=_load_yaml,
),
patch.object(Integration, "has_triggers", return_value=True),
):
descriptions = await trigger.async_get_all_descriptions(hass)
# Test we only load triggers.yaml for integrations with triggers,
# system_health has no triggers
assert proxy_load_triggers_files.mock_calls[0][1][0] == unordered(
[
await async_get_integration(hass, SUN_DOMAIN),
]
)
# system_health does not have triggers and should not be in descriptions
expected_descriptions = {
"sun": {
"fields": {
"event": {
"example": "sunrise",
"selector": {
"select": {
"custom_value": False,
"multiple": False,
"options": ["sunrise", "sunset"],
"sort": False,
}
},
},
"offset": {"selector": {"time": {}}},
}
}
}
assert descriptions == expected_descriptions
# Verify the cache returns the same object
assert await trigger.async_get_all_descriptions(hass) is descriptions
# Load the tag integration and check a new cache object is created
assert await async_setup_component(hass, TAG_DOMAIN, {})
await hass.async_block_till_done()
with (
patch(
"annotatedyaml.loader.load_yaml",
side_effect=_load_yaml,
),
patch.object(Integration, "has_triggers", return_value=True),
):
new_descriptions = await trigger.async_get_all_descriptions(hass)
assert new_descriptions is not descriptions
# The tag trigger should now be present
expected_descriptions |= {
"tag": {
"target": {
"entity": [
{
"domain": ["alarm_control_panel"],
}
],
},
"fields": {},
},
}
assert new_descriptions == expected_descriptions
# Verify the cache returns the same object
assert await trigger.async_get_all_descriptions(hass) is new_descriptions
# Load the text integration and check a new cache object is created
assert await async_setup_component(hass, TEXT_DOMAIN, {})
await hass.async_block_till_done()
with (
patch(
"annotatedyaml.loader.load_yaml",
side_effect=_load_yaml,
),
patch.object(Integration, "has_triggers", return_value=True),
):
new_descriptions = await trigger.async_get_all_descriptions(hass)
assert new_descriptions is not descriptions
# No text triggers added, they are gated by the automation.new_triggers_conditions
# labs flag
assert new_descriptions == expected_descriptions
# Verify the cache returns the same object
assert await trigger.async_get_all_descriptions(hass) is new_descriptions
# Enable the new_triggers_conditions flag and verify text triggers are loaded
assert await async_setup_component(hass, "labs", {})
await ws_client.send_json_auto_id(
{
"type": "labs/update",
"domain": "automation",
"preview_feature": "new_triggers_conditions",
"enabled": True,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
await hass.async_block_till_done()
with (
patch(
"annotatedyaml.loader.load_yaml",
side_effect=_load_yaml,
),
patch.object(Integration, "has_triggers", return_value=True),
):
new_descriptions = await trigger.async_get_all_descriptions(hass)
assert new_descriptions is not descriptions
# The text triggers should now be present
assert new_descriptions == expected_descriptions | {
"text.changed": {
"fields": {},
"target": {
"entity": [
{
"domain": [
"text",
],
},
],
},
},
}
# Verify the cache returns the same object
assert await trigger.async_get_all_descriptions(hass) is new_descriptions
# Disable the new_triggers_conditions flag and verify text triggers are removed
assert await async_setup_component(hass, "labs", {})
await ws_client.send_json_auto_id(
{
"type": "labs/update",
"domain": "automation",
"preview_feature": "new_triggers_conditions",
"enabled": False,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
await hass.async_block_till_done()
with (
patch(
"annotatedyaml.loader.load_yaml",
side_effect=_load_yaml,
),
patch.object(Integration, "has_triggers", return_value=True),
):
new_descriptions = await trigger.async_get_all_descriptions(hass)
assert new_descriptions is not descriptions
# The text triggers should no longer be present
assert new_descriptions == expected_descriptions
# Verify the cache returns the same object
assert await trigger.async_get_all_descriptions(hass) is new_descriptions
await hass.data["entity_components"][SUN_DOMAIN]._async_reset()
@pytest.mark.parametrize(
("yaml_error", "expected_message"),
[
(
FileNotFoundError("Blah"),
"Unable to find triggers.yaml for the sun integration",
),
(
HomeAssistantError("Test error"),
"Unable to parse triggers.yaml for the sun integration: Test error",
),
],
)
async def test_async_get_all_descriptions_with_yaml_error(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
yaml_error: Exception,
expected_message: str,
) -> None:
"""Test async_get_all_descriptions."""
assert await async_setup_component(hass, SUN_DOMAIN, {})
await hass.async_block_till_done()
def _load_yaml_dict(fname, secrets=None):
raise yaml_error
with (
patch(
"homeassistant.helpers.trigger.load_yaml_dict",
side_effect=_load_yaml_dict,
),
patch.object(Integration, "has_triggers", return_value=True),
):
descriptions = await trigger.async_get_all_descriptions(hass)
assert descriptions == {SUN_DOMAIN: None}
assert expected_message in caplog.text
await hass.data["entity_components"][SUN_DOMAIN]._async_reset()
async def test_async_get_all_descriptions_with_bad_description(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test async_get_all_descriptions."""
sun_service_descriptions = """
_:
fields: not_a_dict
"""
assert await async_setup_component(hass, SUN_DOMAIN, {})
await hass.async_block_till_done()
def _load_yaml(fname, secrets=None):
with io.StringIO(sun_service_descriptions) as file:
return parse_yaml(file)
with (
patch(
"annotatedyaml.loader.load_yaml",
side_effect=_load_yaml,
),
patch.object(Integration, "has_triggers", return_value=True),
):
descriptions = await trigger.async_get_all_descriptions(hass)
assert descriptions == {SUN_DOMAIN: None}
assert (
"Unable to parse triggers.yaml for the sun integration: "
"expected a dictionary for dictionary value @ data['_']['fields']"
) in caplog.text
await hass.data["entity_components"][SUN_DOMAIN]._async_reset()
async def test_invalid_trigger_platform(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test invalid trigger platform."""
mock_integration(hass, MockModule("test", async_setup=AsyncMock(return_value=True)))
mock_platform(hass, "test.trigger", MockPlatform())
await async_setup_component(hass, "test", {})
assert "Integration test does not provide trigger support, skipping" in caplog.text
@patch("annotatedyaml.loader.load_yaml")
@patch.object(Integration, "has_triggers", return_value=True)
async def test_subscribe_triggers(
mock_has_triggers: Mock,
mock_load_yaml: Mock,
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test trigger.async_subscribe_platform_events."""
sun_trigger_descriptions = """
_: {}
"""
def _load_yaml(fname, secrets=None):
if fname.endswith("sun/triggers.yaml"):
trigger_descriptions = sun_trigger_descriptions
else:
raise FileNotFoundError
with io.StringIO(trigger_descriptions) as file:
return parse_yaml(file)
mock_load_yaml.side_effect = _load_yaml
async def broken_subscriber(_):
"""Simulate a broken subscriber."""
raise Exception("Boom!") # noqa: TRY002
trigger_events = []
async def good_subscriber(new_triggers: set[str]):
"""Simulate a working subscriber."""
trigger_events.append(new_triggers)
trigger.async_subscribe_platform_events(hass, broken_subscriber)
trigger.async_subscribe_platform_events(hass, good_subscriber)
assert await async_setup_component(hass, "sun", {})
assert trigger_events == [{"sun"}]
assert "Error while notifying trigger platform listener" in caplog.text
await hass.data["entity_components"][SUN_DOMAIN]._async_reset()
@patch("annotatedyaml.loader.load_yaml")
@patch.object(Integration, "has_triggers", return_value=True)
@pytest.mark.parametrize(
("new_triggers_conditions_enabled", "expected_events"),
[
(
True,
[
{
"light.brightness_changed",
"light.brightness_crossed_threshold",
"light.turned_off",
"light.turned_on",
}
],
),
(False, []),
],
)
async def test_subscribe_triggers_experimental_triggers(
mock_has_triggers: Mock,
mock_load_yaml: Mock,
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
caplog: pytest.LogCaptureFixture,
new_triggers_conditions_enabled: bool,
expected_events: list[set[str]],
) -> None:
"""Test trigger.async_subscribe_platform_events doesn't send events for disabled triggers."""
# Return empty triggers.yaml for light integration, the actual trigger descriptions
# are irrelevant for this test
light_trigger_descriptions = ""
def _load_yaml(fname, secrets=None):
if fname.endswith("light/triggers.yaml"):
trigger_descriptions = light_trigger_descriptions
else:
raise FileNotFoundError
with io.StringIO(trigger_descriptions) as file:
return parse_yaml(file)
mock_load_yaml.side_effect = _load_yaml
trigger_events = []
async def good_subscriber(new_triggers: set[str]):
"""Simulate a working subscriber."""
trigger_events.append(new_triggers)
ws_client = await hass_ws_client(hass)
assert await async_setup_component(hass, "labs", {})
await ws_client.send_json_auto_id(
{
"type": "labs/update",
"domain": "automation",
"preview_feature": "new_triggers_conditions",
"enabled": new_triggers_conditions_enabled,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
await hass.async_block_till_done()
trigger.async_subscribe_platform_events(hass, good_subscriber)
assert await async_setup_component(hass, "light", {})
await hass.async_block_till_done()
assert trigger_events == expected_events
@patch("annotatedyaml.loader.load_yaml")
@patch.object(Integration, "has_triggers", return_value=True)
@patch(
"homeassistant.components.light.trigger.async_get_triggers",
new=AsyncMock(return_value={}),
)
async def test_subscribe_triggers_no_triggers(
mock_has_triggers: Mock,
mock_load_yaml: Mock,
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test trigger.async_subscribe_platform_events doesn't send events for platforms without triggers."""
# Return empty triggers.yaml for light integration, the actual trigger descriptions
# are irrelevant for this test
light_trigger_descriptions = ""
def _load_yaml(fname, secrets=None):
if fname.endswith("light/triggers.yaml"):
trigger_descriptions = light_trigger_descriptions
else:
raise FileNotFoundError
with io.StringIO(trigger_descriptions) as file:
return parse_yaml(file)
mock_load_yaml.side_effect = _load_yaml
trigger_events = []
async def good_subscriber(new_triggers: set[str]):
"""Simulate a working subscriber."""
trigger_events.append(new_triggers)
ws_client = await hass_ws_client(hass)
assert await async_setup_component(hass, "labs", {})
await ws_client.send_json_auto_id(
{
"type": "labs/update",
"domain": "automation",
"preview_feature": "new_triggers_conditions",
"enabled": True,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
await hass.async_block_till_done()
trigger.async_subscribe_platform_events(hass, good_subscriber)
assert await async_setup_component(hass, "light", {})
await hass.async_block_till_done()
assert trigger_events == []
@pytest.mark.parametrize(
("trigger_options", "expected_result"),
[
# Test validating climate.target_temperature_changed
# Valid: no limits at all
(
{"threshold": {"type": "any"}},
does_not_raise(),
),
# Valid: numerical limits
(
{"threshold": {"type": "above", "value": {"number": 10}}},
does_not_raise(),
),
(
{"threshold": {"type": "below", "value": {"number": 90}}},
does_not_raise(),
),
(
{
"threshold": {
"type": "between",
"value_min": {"number": 10},
"value_max": {"number": 90},
}
},
does_not_raise(),
),
# Valid: entity references
(
{"threshold": {"type": "above", "value": {"entity": "sensor.test"}}},
does_not_raise(),
),
(
{"threshold": {"type": "below", "value": {"entity": "sensor.test"}}},
does_not_raise(),
),
(
{
"threshold": {
"type": "between",
"value_min": {"entity": "sensor.test"},
"value_max": {"entity": "sensor.test"},
}
},
does_not_raise(),
),
# Valid: Mix of numerical limits and entity references
(
{
"threshold": {
"type": "between",
"value_min": {"entity": "sensor.test"},
"value_max": {"number": 90},
}
},
does_not_raise(),
),
(
{
"threshold": {
"type": "between",
"value_min": {"number": 10},
"value_max": {"entity": "sensor.test"},
}
},
does_not_raise(),
),
# Test invalid configurations
(
# Missing threshold type
{},
pytest.raises(vol.Invalid),
),
(
# Invalid threshold type
{"threshold": {"type": "invalid_type"}},
pytest.raises(vol.Invalid),
),
(
# Must be valid entity id
{
"threshold": {
"type": "between",
"value_min": {"entity": "cat"},
"value_max": {"entity": "dog"},
}
},
pytest.raises(vol.Invalid),
),
(
# Above must be smaller than below
{
"threshold": {
"type": "between",
"value_min": {"number": 90},
"value_max": {"number": 10},
}
},
pytest.raises(vol.Invalid),
),
],
)
async def test_numerical_state_attribute_changed_trigger_config_validation(
hass: HomeAssistant,
trigger_options: dict[str, Any],
expected_result: AbstractContextManager,
) -> None:
"""Test numerical state attribute change trigger config validation."""
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
return {
"test_trigger": make_entity_numerical_state_changed_trigger(
{"test": DomainSpec(value_source="test_attribute")}
),
}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
with expected_result:
await async_validate_trigger_config(
hass,
[
{
"platform": "test.test_trigger",
CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
CONF_OPTIONS: trigger_options,
}
],
)
def _make_with_unit_changed_trigger_class() -> type[
EntityNumericalStateChangedTriggerWithUnitBase
]:
"""Create a concrete WithUnit changed trigger class for testing."""
class _TestChangedTrigger(
EntityNumericalStateChangedTriggerWithUnitBase,
):
_base_unit = UnitOfTemperature.CELSIUS
_domain_specs = {"test": DomainSpec(value_source="test_attribute")}
_unit_converter = TemperatureConverter
return _TestChangedTrigger
@pytest.mark.parametrize(
("trigger_options", "expected_result"),
[
# Valid: no limits at all
(
{"threshold": {"type": "any"}},
does_not_raise(),
),
# Valid: unit provided with numerical limits
(
{
"threshold": {
"type": "above",
"value": {"number": 10, "unit_of_measurement": "°C"},
}
},
does_not_raise(),
),
(
{
"threshold": {
"type": "below",
"value": {"number": 90, "unit_of_measurement": "°F"},
}
},
does_not_raise(),
),
(
{
"threshold": {
"type": "between",
"value_min": {"number": 10, "unit_of_measurement": "°C"},
"value_max": {"number": 90, "unit_of_measurement": "°F"},
}
},
does_not_raise(),
),
# Valid: no unit needed when using entity references
(
{"threshold": {"type": "above", "value": {"entity": "sensor.test"}}},
does_not_raise(),
),
(
{"threshold": {"type": "below", "value": {"entity": "sensor.test"}}},
does_not_raise(),
),
(
{
"threshold": {
"type": "between",
"value_min": {"entity": "sensor.test"},
"value_max": {"entity": "sensor.test"},
}
},
does_not_raise(),
),
# Valid: unit only needed for numerical limits, not entity references
(
{
"threshold": {
"type": "between",
"value_min": {"entity": "sensor.test"},
"value_max": {"number": 90, "unit_of_measurement": "°C"},
}
},
does_not_raise(),
),
(
{
"threshold": {
"type": "between",
"value_min": {"number": 10, "unit_of_measurement": "°C"},
"value_max": {"entity": "sensor.test"},
}
},
does_not_raise(),
),
# Invalid: missing threshold type
(
{},
pytest.raises(vol.Invalid),
),
# Invalid: invalid threshold type
(
{"threshold": {"type": "invalid_type"}},
pytest.raises(vol.Invalid),
),
# Invalid: numerical limit without unit
(
{"threshold": {"type": "above", "value": {"number": 10}}},
pytest.raises(vol.Invalid),
),
(
{"threshold": {"type": "below", "value": {"number": 90}}},
pytest.raises(vol.Invalid),
),
(
{
"threshold": {
"type": "between",
"value_min": {"number": 90},
"value_max": {"number": 90},
}
},
pytest.raises(vol.Invalid),
),
# Invalid: one numerical limit without unit (other is entity)
(
{
"threshold": {
"type": "between",
"value_min": {"number": 10},
"value_max": {"entity": "sensor.test"},
}
},
pytest.raises(vol.Invalid),
),
(
{
"threshold": {
"type": "between",
"value_min": {"entity": "sensor.test"},
"value_max": {"number": 90},
}
},
pytest.raises(vol.Invalid),
),
# Invalid: invalid unit value
(
{
"threshold": {
"type": "above",
"value": {"number": 10, "unit_of_measurement": "invalid_unit"},
}
},
pytest.raises(vol.Invalid),
),
# Invalid: Must use valid entity id
(
{
"threshold": {
"type": "between",
"value_min": {"entity": "cat"},
"value_max": {"entity": "dog"},
}
},
pytest.raises(vol.Invalid),
),
# Invalid: above must be smaller than below
(
{
"threshold": {
"type": "between",
"value_min": {"number": 90, "unit_of_measurement": "°C"},
"value_max": {"number": 10, "unit_of_measurement": "°F"},
}
},
pytest.raises(vol.Invalid),
),
],
)
async def test_numerical_state_attribute_changed_with_unit_trigger_config_validation(
hass: HomeAssistant,
trigger_options: dict[str, Any],
expected_result: AbstractContextManager,
) -> None:
"""Test numerical state attribute change with unit trigger config validation."""
trigger_cls = _make_with_unit_changed_trigger_class()
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
return {"test_trigger": trigger_cls}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
with expected_result:
await async_validate_trigger_config(
hass,
[
{
"platform": "test.test_trigger",
CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
CONF_OPTIONS: trigger_options,
}
],
)
async def test_numerical_state_attribute_changed_error_handling(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test numerical state attribute change error handling."""
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
return {
"attribute_changed": make_entity_numerical_state_changed_trigger(
{"test": DomainSpec(value_source="test_attribute")}
),
}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
hass.states.async_set("test.test_entity", "on", {"test_attribute": 20})
options = {
CONF_OPTIONS: {
"threshold": {
"type": "between",
"value_min": {"entity": "sensor.above"},
"value_max": {"entity": "sensor.below"},
}
}
}
await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {
CONF_PLATFORM: "test.attribute_changed",
CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
}
| options,
"action": {
"service": "test.automation",
"data_template": {CONF_ENTITY_ID: "{{ trigger.entity_id }}"},
},
}
},
)
assert len(service_calls) == 0
# Test the trigger works
hass.states.async_set("sensor.above", "10")
hass.states.async_set("sensor.below", "90")
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 1
service_calls.clear()
# Test the trigger fires again when still within limits
hass.states.async_set("test.test_entity", "on", {"test_attribute": 51})
await hass.async_block_till_done()
assert len(service_calls) == 1
service_calls.clear()
# Test the trigger does not fire when the from-state is unknown or unavailable
for from_state in (STATE_UNKNOWN, STATE_UNAVAILABLE):
hass.states.async_set("test.test_entity", from_state)
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the attribute value is outside the limits
for value in (5, 95):
hass.states.async_set("test.test_entity", "on", {"test_attribute": value})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the attribute value is missing
hass.states.async_set("test.test_entity", "on", {})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the attribute value is invalid
for value in ("cat", None):
hass.states.async_set("test.test_entity", "on", {"test_attribute": value})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the above sensor does not exist
hass.states.async_remove("sensor.above")
hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the above sensor state is not numeric
for invalid_value in ("cat", None):
hass.states.async_set("sensor.above", invalid_value)
hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Reset the above sensor state to a valid numeric value
hass.states.async_set("sensor.above", "10")
# Test the trigger does not fire when the below sensor does not exist
hass.states.async_remove("sensor.below")
hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the below sensor state is not numeric
for invalid_value in ("cat", None):
hass.states.async_set("sensor.below", invalid_value)
hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
async def test_numerical_state_attribute_changed_entity_limit_unit_validation(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test that entity limits with wrong unit are rejected."""
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
return {
"attribute_changed": make_entity_numerical_state_changed_trigger(
{"test": DomainSpec(value_source="test_attribute")},
valid_unit="%",
),
}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
hass.states.async_set("test.test_entity", "on", {"test_attribute": 20})
options = {
CONF_OPTIONS: {
"threshold": {
"type": "between",
"value_min": {"entity": "sensor.above"},
"value_max": {"entity": "sensor.below"},
}
}
}
await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {
CONF_PLATFORM: "test.attribute_changed",
CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
}
| options,
"action": {
"service": "test.automation",
"data_template": {CONF_ENTITY_ID: "{{ trigger.entity_id }}"},
},
}
},
)
assert len(service_calls) == 0
# Test the trigger works with correct unit on limit entities
hass.states.async_set("sensor.above", "10", {ATTR_UNIT_OF_MEASUREMENT: "%"})
hass.states.async_set("sensor.below", "90", {ATTR_UNIT_OF_MEASUREMENT: "%"})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 1
service_calls.clear()
# Test the trigger does not fire when the above sensor has wrong unit
hass.states.async_set("sensor.above", "10", {ATTR_UNIT_OF_MEASUREMENT: "°C"})
hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the above sensor has no unit
hass.states.async_set("sensor.above", "10")
hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Reset the above sensor to correct unit
hass.states.async_set("sensor.above", "10", {ATTR_UNIT_OF_MEASUREMENT: "%"})
# Test the trigger does not fire when the below sensor has wrong unit
hass.states.async_set("sensor.below", "90", {ATTR_UNIT_OF_MEASUREMENT: "°C"})
hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the below sensor has no unit
hass.states.async_set("sensor.below", "90")
hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
async def test_numerical_state_attribute_changed_with_unit_error_handling(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test numerical state attribute change with unit conversion error handling."""
trigger_cls = _make_with_unit_changed_trigger_class()
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
return {"attribute_changed": trigger_cls}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
# Entity reports in °F, trigger configured in °C with above 20°C, below 30°C
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 68, # 68°F = 20°C
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
CONF_PLATFORM: "test.attribute_changed",
CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
CONF_OPTIONS: {
"threshold": {
"type": "between",
"value_min": {
"number": 20,
"unit_of_measurement": "°C",
},
"value_max": {
"number": 30,
"unit_of_measurement": "°C",
},
}
},
},
"action": {
"service": "test.numerical_automation",
"data_template": {CONF_ENTITY_ID: "{{ trigger.entity_id }}"},
},
},
{
"trigger": {
CONF_PLATFORM: "test.attribute_changed",
CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
CONF_OPTIONS: {
"threshold": {
"type": "between",
"value_min": {"entity": "sensor.above"},
"value_max": {"entity": "sensor.below"},
}
},
},
"action": {
"service": "test.entity_automation",
"data_template": {CONF_ENTITY_ID: "{{ trigger.entity_id }}"},
},
},
]
},
)
assert len(service_calls) == 0
# 77°F = 25°C, within range (above 20, below 30) - should trigger numerical
# Entity automation won't trigger because sensor.above/below don't exist yet
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 77,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 1
assert service_calls[0].service == "numerical_automation"
service_calls.clear()
# 59°F = 15°C, below 20°C - should NOT trigger
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 59,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# 95°F = 35°C, above 30°C - should NOT trigger
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 95,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# Set up entity limits referencing sensors that report in °F
hass.states.async_set(
"sensor.above",
"68", # 68°F = 20°C
{ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT},
)
hass.states.async_set(
"sensor.below",
"86", # 86°F = 30°C
{ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT},
)
# 77°F = 25°C, between 20°C and 30°C - should trigger both automations
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 77,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 2
assert {call.service for call in service_calls} == {
"numerical_automation",
"entity_automation",
}
service_calls.clear()
# Test the trigger does not fire when the attribute value is missing
hass.states.async_set("test.test_entity", "on", {})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the attribute value is invalid
for value in ("cat", None):
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": value,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the unit is incompatible
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 50,
ATTR_UNIT_OF_MEASUREMENT: "invalid_unit",
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the above sensor does not exist
hass.states.async_remove("sensor.above")
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": None,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
hass.states.async_set(
"test.test_entity",
"on",
{"test_attribute": 50, ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the above sensor state is not numeric
for invalid_value in ("cat", None):
hass.states.async_set(
"sensor.above",
invalid_value,
{ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT},
)
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": None,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 50,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the above sensor's unit is incompatible
hass.states.async_set(
"sensor.above",
"68", # 68°F = 20°C
{ATTR_UNIT_OF_MEASUREMENT: "invalid_unit"},
)
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": None,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
hass.states.async_set(
"test.test_entity",
"on",
{"test_attribute": 50, ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# Reset the above sensor state to a valid numeric value
hass.states.async_set(
"sensor.above",
"68", # 68°F = 20°C
{ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT},
)
# Test the trigger does not fire when the below sensor does not exist
hass.states.async_remove("sensor.below")
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": None,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
hass.states.async_set(
"test.test_entity",
"on",
{"test_attribute": 50, ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the below sensor state is not numeric
for invalid_value in ("cat", None):
hass.states.async_set("sensor.below", invalid_value)
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": None,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 50,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the below sensor's unit is incompatible
hass.states.async_set(
"sensor.below",
"68", # 68°F = 20°C
{ATTR_UNIT_OF_MEASUREMENT: "invalid_unit"},
)
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": None,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
hass.states.async_set(
"test.test_entity",
"on",
{"test_attribute": 50, ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
@pytest.mark.parametrize(
("trigger_options", "expected_result"),
[
# Valid configurations
# Don't use the enum in tests to allow testing validation of strings when the source is JSON or YAML
(
{"threshold": {"type": "above", "value": {"number": 10}}},
does_not_raise(),
),
(
{"threshold": {"type": "above", "value": {"entity": "sensor.test"}}},
does_not_raise(),
),
(
{"threshold": {"type": "below", "value": {"number": 90}}},
does_not_raise(),
),
(
{"threshold": {"type": "below", "value": {"entity": "sensor.test"}}},
does_not_raise(),
),
(
{
"threshold": {
"type": "between",
"value_min": {"number": 10},
"value_max": {"number": 90},
}
},
does_not_raise(),
),
(
{
"threshold": {
"type": "between",
"value_min": {"number": 10},
"value_max": {"entity": "sensor.test"},
}
},
does_not_raise(),
),
(
{
"threshold": {
"type": "between",
"value_min": {"entity": "sensor.test"},
"value_max": {"number": 90},
}
},
does_not_raise(),
),
(
{
"threshold": {
"type": "between",
"value_min": {"entity": "sensor.test"},
"value_max": {"entity": "sensor.test"},
}
},
does_not_raise(),
),
(
{
"threshold": {
"type": "outside",
"value_min": {"number": 10},
"value_max": {"number": 90},
}
},
does_not_raise(),
),
(
{
"threshold": {
"type": "outside",
"value_min": {"number": 10},
"value_max": {"entity": "sensor.test"},
}
},
does_not_raise(),
),
(
{
"threshold": {
"type": "outside",
"value_min": {"entity": "sensor.test"},
"value_max": {"number": 90},
}
},
does_not_raise(),
),
(
{
"threshold": {
"type": "outside",
"value_min": {"entity": "sensor.test"},
"value_max": {"entity": "sensor.test"},
}
},
does_not_raise(),
),
# Test verbose choose selector options
# Test invalid configurations
(
# Missing threshold type
{},
pytest.raises(vol.Invalid),
),
(
# Missing threshold type
{"threshold": {}},
pytest.raises(vol.Invalid),
),
(
# Invalid threshold type
{"threshold": {"type": "cat"}},
pytest.raises(vol.Invalid),
),
(
# Must provide lower limit for ABOVE
{"threshold": {"type": "above"}},
pytest.raises(vol.Invalid),
),
(
# Must provide lower limit for ABOVE
{"threshold": {"type": "above", "value_min": {"number": 10}}},
pytest.raises(vol.Invalid),
),
(
# Must provide lower limit for ABOVE
{"threshold": {"type": "above", "value_max": {"number": 90}}},
pytest.raises(vol.Invalid),
),
(
# Must provide upper limit for BELOW
{"threshold": {"type": "below"}},
pytest.raises(vol.Invalid),
),
(
# Must provide upper limit for BELOW
{"threshold": {"type": "below", "value_min": {"number": 10}}},
pytest.raises(vol.Invalid),
),
(
# Must provide upper limit for BELOW
{"threshold": {"type": "below", "value_max": {"number": 10}}},
pytest.raises(vol.Invalid),
),
(
# Must provide upper and lower limits for BETWEEN
{"threshold": {"type": "between"}},
pytest.raises(vol.Invalid),
),
(
# Must provide upper and lower limits for BETWEEN
{"threshold": {"type": "between", "value_min": {"number": 10}}},
pytest.raises(vol.Invalid),
),
(
# Must provide upper and lower limits for BETWEEN
{"threshold": {"type": "between", "value_max": {"number": 90}}},
pytest.raises(vol.Invalid),
),
(
# Must provide upper and lower limits for OUTSIDE
{"threshold": {"type": "outside"}},
pytest.raises(vol.Invalid),
),
(
# Must provide upper and lower limits for OUTSIDE
{"threshold": {"type": "outside", "value_min": {"number": 10}}},
pytest.raises(vol.Invalid),
),
(
# Must provide upper and lower limits for OUTSIDE
{"threshold": {"type": "outside", "value_max": {"number": 90}}},
pytest.raises(vol.Invalid),
),
(
# Must be valid entity id
{
"threshold": {
"type": "between",
"value_min": {"entity": "cat"},
"value_max": {"entity": "dog"},
}
},
pytest.raises(vol.Invalid),
),
(
# Min must be smaller than max
{
"threshold": {
"type": "between",
"value_min": {"number": 90},
"value_max": {"number": 10},
}
},
pytest.raises(vol.Invalid),
),
],
)
async def test_numerical_state_attribute_crossed_threshold_trigger_config_validation(
hass: HomeAssistant,
trigger_options: dict[str, Any],
expected_result: AbstractContextManager,
) -> None:
"""Test numerical state attribute change trigger config validation."""
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
return {
"test_trigger": make_entity_numerical_state_crossed_threshold_trigger(
{"test": DomainSpec(value_source="test_attribute")}
),
}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
with expected_result:
await async_validate_trigger_config(
hass,
[
{
"platform": "test.test_trigger",
CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
CONF_OPTIONS: trigger_options,
}
],
)
def _make_with_unit_crossed_threshold_trigger_class() -> type[
EntityNumericalStateCrossedThresholdTriggerWithUnitBase
]:
"""Create a concrete WithUnit crossed threshold trigger class for testing."""
class _TestCrossedThresholdTrigger(
EntityNumericalStateCrossedThresholdTriggerWithUnitBase,
):
_base_unit = UnitOfTemperature.CELSIUS
_domain_specs = {"test": DomainSpec(value_source="test_attribute")}
_unit_converter = TemperatureConverter
return _TestCrossedThresholdTrigger
@pytest.mark.parametrize(
("trigger_options", "expected_result"),
[
# Valid: unit provided with numerical limits
(
{
"threshold": {
"type": "above",
"value": {
"number": 10,
"unit_of_measurement": UnitOfTemperature.CELSIUS,
},
}
},
does_not_raise(),
),
(
{
"threshold": {
"type": "below",
"value": {
"number": 90,
"unit_of_measurement": UnitOfTemperature.FAHRENHEIT,
},
}
},
does_not_raise(),
),
(
{
"threshold": {
"type": "between",
"value_min": {
"number": 10,
"unit_of_measurement": UnitOfTemperature.CELSIUS,
},
"value_max": {
"number": 90,
"unit_of_measurement": UnitOfTemperature.FAHRENHEIT,
},
}
},
does_not_raise(),
),
# Valid: no unit needed when using entity references
(
{
"threshold": {
"type": "above",
"value": {"entity": "sensor.test"},
}
},
does_not_raise(),
),
(
{
"threshold": {
"type": "between",
"value_min": {"entity": "sensor.test"},
"value_max": {"entity": "sensor.test"},
}
},
does_not_raise(),
),
# Invalid: numerical limit without unit
(
{"threshold": {"type": "above", "value": {"number": 10}}},
pytest.raises(vol.Invalid),
),
(
{
"threshold": {
"type": "between",
"value_min": {"number": 10},
"value_max": {"number": 90},
}
},
pytest.raises(vol.Invalid),
),
# Invalid: one numerical limit without unit (other is entity)
(
{
"threshold": {
"type": "between",
"value_min": {"number": 10},
"value_max": {"entity": "sensor.test"},
}
},
pytest.raises(vol.Invalid),
),
# Invalid: invalid unit value
(
{
"threshold": {
"type": "above",
"value": {"number": 10, "unit_of_measurement": "invalid_unit"},
}
},
pytest.raises(vol.Invalid),
),
# Invalid: missing threshold type
(
{},
pytest.raises(vol.Invalid),
),
# Invalid: missing threshold type
(
{"threshold": {}},
pytest.raises(vol.Invalid),
),
],
)
async def test_numerical_state_attribute_crossed_threshold_with_unit_trigger_config_validation(
hass: HomeAssistant,
trigger_options: dict[str, Any],
expected_result: AbstractContextManager,
) -> None:
"""Test numerical state attribute crossed threshold with unit trigger config validation."""
trigger_cls = _make_with_unit_crossed_threshold_trigger_class()
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
return {"test_trigger": trigger_cls}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
with expected_result:
await async_validate_trigger_config(
hass,
[
{
"platform": "test.test_trigger",
CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
CONF_OPTIONS: trigger_options,
}
],
)
async def test_numerical_state_attribute_crossed_threshold_error_handling(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test numerical state attribute crossed threshold error handling."""
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
return {
"crossed_threshold": make_entity_numerical_state_crossed_threshold_trigger(
{"test": DomainSpec(value_source="test_attribute")}
),
}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
hass.states.async_set("test.test_entity", "on", {"test_attribute": 0})
options = {
CONF_OPTIONS: {
"threshold": {
"type": "between",
"value_min": {"entity": "sensor.lower"},
"value_max": {"entity": "sensor.upper"},
}
},
}
await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {
CONF_PLATFORM: "test.crossed_threshold",
CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
}
| options,
"action": {
"service": "test.automation",
"data_template": {CONF_ENTITY_ID: "{{ trigger.entity_id }}"},
},
}
},
)
assert len(service_calls) == 0
# Test the trigger works
hass.states.async_set("sensor.lower", "10")
hass.states.async_set("sensor.upper", "90")
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 1
service_calls.clear()
# Test the trigger does not fire again when still within limits
hass.states.async_set("test.test_entity", "on", {"test_attribute": 51})
await hass.async_block_till_done()
assert len(service_calls) == 0
service_calls.clear()
# Test the trigger does not fire when the from-state is unknown or unavailable
for from_state in (STATE_UNKNOWN, STATE_UNAVAILABLE):
hass.states.async_set("test.test_entity", from_state)
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does fire when the attribute value is changing from None
hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 1
service_calls.clear()
# Test the trigger does not fire when the attribute value is outside the limits
for value in (5, 95):
hass.states.async_set("test.test_entity", "on", {"test_attribute": value})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the attribute value is missing
hass.states.async_set("test.test_entity", "on", {})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the attribute value is invalid
for value in ("cat", None):
hass.states.async_set("test.test_entity", "on", {"test_attribute": value})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the lower sensor does not exist
hass.states.async_remove("sensor.lower")
hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the lower sensor state is not numeric
for invalid_value in ("cat", None):
hass.states.async_set("sensor.lower", invalid_value)
hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Reset the lower sensor state to a valid numeric value
hass.states.async_set("sensor.lower", "10")
# Test the trigger does not fire when the upper sensor does not exist
hass.states.async_remove("sensor.upper")
hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the upper sensor state is not numeric
for invalid_value in ("cat", None):
hass.states.async_set("sensor.upper", invalid_value)
hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
async def test_numerical_state_attribute_crossed_threshold_entity_limit_unit_validation(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test that entity limits with wrong unit are rejected for crossed threshold."""
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
return {
"crossed_threshold": make_entity_numerical_state_crossed_threshold_trigger(
{"test": DomainSpec(value_source="test_attribute")},
valid_unit="%",
),
}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
hass.states.async_set("test.test_entity", "on", {"test_attribute": 0})
options = {
CONF_OPTIONS: {
"threshold": {
"type": "between",
"value_min": {"entity": "sensor.lower"},
"value_max": {"entity": "sensor.upper"},
}
},
}
await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {
CONF_PLATFORM: "test.crossed_threshold",
CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
}
| options,
"action": {
"service": "test.automation",
"data_template": {CONF_ENTITY_ID: "{{ trigger.entity_id }}"},
},
}
},
)
assert len(service_calls) == 0
# Test the trigger works with correct unit on limit entities
hass.states.async_set("sensor.lower", "10", {ATTR_UNIT_OF_MEASUREMENT: "%"})
hass.states.async_set("sensor.upper", "90", {ATTR_UNIT_OF_MEASUREMENT: "%"})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 1
service_calls.clear()
# Test the trigger does not fire when the lower sensor has wrong unit
hass.states.async_set("sensor.lower", "10", {ATTR_UNIT_OF_MEASUREMENT: "°C"})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 0})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the lower sensor has no unit
hass.states.async_set("sensor.lower", "10")
hass.states.async_set("test.test_entity", "on", {"test_attribute": 0})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Reset the lower sensor to correct unit
hass.states.async_set("sensor.lower", "10", {ATTR_UNIT_OF_MEASUREMENT: "%"})
# Test the trigger does not fire when the upper sensor has wrong unit
hass.states.async_set("sensor.upper", "90", {ATTR_UNIT_OF_MEASUREMENT: "°C"})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 0})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
# Test the trigger does not fire when the upper sensor has no unit
hass.states.async_set("sensor.upper", "90")
hass.states.async_set("test.test_entity", "on", {"test_attribute": 0})
hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
await hass.async_block_till_done()
assert len(service_calls) == 0
async def test_numerical_state_attribute_crossed_threshold_with_unit_error_handling(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
"""Test numerical state attribute crossed threshold with unit conversion."""
trigger_cls = _make_with_unit_crossed_threshold_trigger_class()
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
return {"crossed_threshold": trigger_cls}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
# Entity reports in °F, trigger configured in °C: above 25°C
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 68, # 68°F = 20°C, below threshold
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
options = {
CONF_OPTIONS: {
"threshold": {
"type": "above",
"value": {
"number": 25,
"unit_of_measurement": UnitOfTemperature.CELSIUS,
},
}
},
}
await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {
CONF_PLATFORM: "test.crossed_threshold",
CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
}
| options,
"action": {
"service": "test.automation",
"data_template": {CONF_ENTITY_ID: "{{ trigger.entity_id }}"},
},
}
},
)
assert len(service_calls) == 0
# 80.6°F = 27°C, above 25°C threshold - should trigger
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 80.6,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 1
service_calls.clear()
# Still above threshold - should NOT trigger (already crossed)
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 82,
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
# Drop below threshold and cross again
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 68, # 20°C, below 25°C
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 80.6, # 27°C, above 25°C again
ATTR_UNIT_OF_MEASUREMENT: UnitOfTemperature.FAHRENHEIT,
},
)
await hass.async_block_till_done()
assert len(service_calls) == 1
service_calls.clear()
# Test with incompatible unit - should NOT trigger
hass.states.async_set(
"test.test_entity",
"on",
{
"test_attribute": 50,
ATTR_UNIT_OF_MEASUREMENT: "invalid_unit",
},
)
await hass.async_block_till_done()
assert len(service_calls) == 0
def _make_trigger(
hass: HomeAssistant, domain_specs: Mapping[str, DomainSpec]
) -> EntityTriggerBase:
"""Create a minimal EntityTriggerBase subclass with the given domain specs."""
class _SimpleTrigger(EntityTriggerBase):
"""Minimal concrete trigger for testing entity_filter."""
_domain_specs = domain_specs
def is_valid_transition(self, from_state: State, to_state: State) -> bool:
"""Accept any transition."""
return True
def is_valid_state(self, state: State) -> bool:
"""Accept any state."""
return True
config = TriggerConfig(key="test.test_trigger", target={CONF_ENTITY_ID: []})
return _SimpleTrigger(hass, config)
async def test_entity_filter_by_domain_only(hass: HomeAssistant) -> None:
"""Test entity_filter includes entities matching domain, excludes others."""
trig = _make_trigger(hass, {"sensor": DomainSpec(), "switch": DomainSpec()})
entities = {
"sensor.temp",
"sensor.humidity",
"switch.light",
"light.bedroom",
"cover.garage",
}
result = trig.entity_filter(entities)
assert result == {"sensor.temp", "sensor.humidity", "switch.light"}
async def test_entity_filter_by_device_class(hass: HomeAssistant) -> None:
"""Test entity_filter filters by device_class when specified."""
trig = _make_trigger(hass, {"sensor": DomainSpec(device_class="humidity")})
# Set states with device_class attributes
hass.states.async_set("sensor.humidity_1", "50", {ATTR_DEVICE_CLASS: "humidity"})
hass.states.async_set(
"sensor.temperature_1", "22", {ATTR_DEVICE_CLASS: "temperature"}
)
hass.states.async_set("sensor.no_class", "10", {})
entities = {"sensor.humidity_1", "sensor.temperature_1", "sensor.no_class"}
result = trig.entity_filter(entities)
assert result == {"sensor.humidity_1"}
async def test_entity_filter_device_class_unknown_entity(
hass: HomeAssistant,
) -> None:
"""Test entity_filter excludes entities not in state machine or registry."""
trig = _make_trigger(hass, {"sensor": DomainSpec(device_class="humidity")})
# Entity not in state machine and not in entity registry -> UNDEFINED
entities = {"sensor.nonexistent"}
result = trig.entity_filter(entities)
assert result == set()
async def test_entity_filter_multiple_domains_with_device_class(
hass: HomeAssistant,
) -> None:
"""Test entity_filter with multiple domains, some with device_class filtering."""
trig = _make_trigger(
hass,
{
"climate": DomainSpec(value_source="current_humidity"),
"sensor": DomainSpec(device_class="humidity"),
"weather": DomainSpec(value_source="humidity"),
},
)
hass.states.async_set("sensor.humidity", "60", {ATTR_DEVICE_CLASS: "humidity"})
hass.states.async_set(
"sensor.temperature", "20", {ATTR_DEVICE_CLASS: "temperature"}
)
hass.states.async_set("climate.hvac", "heat", {})
hass.states.async_set("weather.home", "sunny", {})
hass.states.async_set("light.bedroom", "on", {})
entities = {
"sensor.humidity",
"sensor.temperature",
"climate.hvac",
"weather.home",
"light.bedroom",
}
result = trig.entity_filter(entities)
# sensor.temperature excluded (wrong device_class)
# light.bedroom excluded (no matching domain)
assert result == {"sensor.humidity", "climate.hvac", "weather.home"}
async def test_entity_filter_no_device_class_means_match_all_in_domain(
hass: HomeAssistant,
) -> None:
"""Test that DomainSpec without device_class matches all entities in the domain."""
trig = _make_trigger(hass, {"cover": DomainSpec()})
hass.states.async_set("cover.door", "open", {ATTR_DEVICE_CLASS: "door"})
hass.states.async_set("cover.garage", "closed", {ATTR_DEVICE_CLASS: "garage"})
hass.states.async_set("cover.plain", "open", {})
entities = {"cover.door", "cover.garage", "cover.plain"}
result = trig.entity_filter(entities)
assert result == entities
@pytest.mark.parametrize(
("domain_specs", "to_states", "from_state", "to_state", "wrong_value_state"),
[
pytest.param(
{"light": DomainSpec()},
{"on"},
State("light.bed", "off"),
State("light.bed", "on"),
State("light.bed", "off"),
id="state_based",
),
pytest.param(
{"light": DomainSpec(value_source="color_mode")},
{"hs"},
State("light.bed", "on", {"color_mode": "color_temp"}),
State("light.bed", "on", {"color_mode": "hs"}),
State("light.bed", "on", {"color_mode": "rgb"}),
id="attribute_based",
),
pytest.param(
"light",
{"on"},
State("light.bed", "off"),
State("light.bed", "on"),
State("light.bed", "off"),
id="state_based_domain_string",
),
],
)
async def test_make_entity_target_state_trigger(
hass: HomeAssistant,
domain_specs: Mapping[str, DomainSpec] | str,
to_states: set[str],
from_state: State,
to_state: State,
wrong_value_state: State,
) -> None:
"""Test make_entity_target_state_trigger with state and attribute-based DomainSpec."""
trigger_cls = make_entity_target_state_trigger(domain_specs, to_states=to_states)
config = TriggerConfig(key="light.turned_on", target={"entity_id": "light.bed"})
trig = trigger_cls(hass, config)
# Value changed to target — valid
assert trig.is_valid_transition(from_state, to_state)
assert trig.is_valid_state(to_state)
# Value did not change — not a valid transition
assert not trig.is_valid_transition(from_state, from_state)
# From unavailable — not valid
unavailable = State("light.bed", STATE_UNAVAILABLE, {})
assert not trig.is_valid_transition(unavailable, to_state)
# Value not in to_states — not valid
assert not trig.is_valid_state(wrong_value_state)
@pytest.mark.parametrize(
(
"domain_specs",
"from_states",
"to_states",
"from_state",
"to_state",
"wrong_from",
"wrong_to",
),
[
pytest.param(
{"climate": DomainSpec()},
{"off"},
{"heat"},
State("climate.living", "off"),
State("climate.living", "heat"),
State("climate.living", "cool"),
State("climate.living", "cool"),
id="state_based",
),
pytest.param(
{"climate": DomainSpec(value_source="hvac_action")},
{"idle"},
{"heating"},
State("climate.living", "heat", {"hvac_action": "idle"}),
State("climate.living", "heat", {"hvac_action": "heating"}),
State("climate.living", "heat", {"hvac_action": "heating"}),
State("climate.living", "heat", {"hvac_action": "idle"}),
id="attribute_based",
),
],
)
async def test_make_entity_transition_trigger(
hass: HomeAssistant,
domain_specs: Mapping[str, DomainSpec],
from_states: set[str],
to_states: set[str],
from_state: State,
to_state: State,
wrong_from: State,
wrong_to: State,
) -> None:
"""Test make_entity_transition_trigger with state and attribute-based DomainSpec."""
trigger_cls = make_entity_transition_trigger(
domain_specs, from_states=from_states, to_states=to_states
)
config = TriggerConfig(
key="climate.hvac_action", target={"entity_id": "climate.living"}
)
trig = trigger_cls(hass, config)
# Valid transition
assert trig.is_valid_transition(from_state, to_state)
assert trig.is_valid_state(to_state)
# Wrong origin (not in from_states)
assert not trig.is_valid_transition(wrong_from, to_state)
# Wrong target (not in to_states)
assert not trig.is_valid_state(wrong_to)
# No change in tracked value — not a valid transition
assert not trig.is_valid_transition(from_state, from_state)
# From unavailable — not valid
unavailable = State("climate.living", STATE_UNAVAILABLE, {})
assert not trig.is_valid_transition(unavailable, to_state)
@pytest.mark.parametrize(
("domain_specs", "origin", "from_state", "to_state", "wrong_from"),
[
pytest.param(
{"climate": DomainSpec()},
"off",
State("climate.living", "off"),
State("climate.living", "heat"),
State("climate.living", "cool"),
id="state_based",
),
pytest.param(
{"climate": DomainSpec(value_source="hvac_action")},
"idle",
State("climate.living", "heat", {"hvac_action": "idle"}),
State("climate.living", "heat", {"hvac_action": "heating"}),
State("climate.living", "heat", {"hvac_action": "heating"}),
id="attribute_based",
),
],
)
async def test_make_entity_origin_state_trigger(
hass: HomeAssistant,
domain_specs: Mapping[str, DomainSpec],
origin: str,
from_state: State,
to_state: State,
wrong_from: State,
) -> None:
"""Test make_entity_origin_state_trigger with state and attribute-based DomainSpec."""
trigger_cls = make_entity_origin_state_trigger(domain_specs, from_state=origin)
config = TriggerConfig(
key="climate.started_heating", target={"entity_id": "climate.living"}
)
trig = trigger_cls(hass, config)
# Valid: changed from expected origin to something else
assert trig.is_valid_transition(from_state, to_state)
assert trig.is_valid_state(to_state)
# Wrong origin (not the expected from_state)
assert not trig.is_valid_transition(wrong_from, to_state)
# No change in tracked value — not a valid transition
assert not trig.is_valid_transition(from_state, from_state)
# To-state still matches from_state — not valid
assert not trig.is_valid_state(from_state)
class _OffToOnTrigger(EntityTriggerBase):
"""Test trigger that fires when state becomes 'on'."""
_domain_specs = {"test": DomainSpec()}
def is_valid_transition(self, from_state: State, to_state: State) -> bool:
"""Valid if transitioning from a non-'on' state."""
if from_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN):
return False
return from_state.state != STATE_ON
def is_valid_state(self, state: State) -> bool:
"""Valid if the state is 'on'."""
return state.state == STATE_ON
async def _arm_off_to_on_trigger(
hass: HomeAssistant,
entity_ids: list[str],
behavior: str,
calls: list[dict[str, Any]],
duration: dict[str, int] | None,
) -> CALLBACK_TYPE:
"""Set up _OffToOnTrigger via async_initialize_triggers."""
async def async_get_triggers(
hass: HomeAssistant,
) -> dict[str, type[Trigger]]:
return {"off_to_on": _OffToOnTrigger}
mock_integration(hass, MockModule("test"))
mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))
options: dict[str, Any] = {ATTR_BEHAVIOR: behavior}
if duration is not None:
options[CONF_FOR] = duration
trigger_config = {
CONF_PLATFORM: "test.off_to_on",
CONF_TARGET: {CONF_ENTITY_ID: entity_ids},
CONF_OPTIONS: options,
}
log = logging.getLogger(__name__)
@callback
def action(run_variables: dict[str, Any], context: Context | None = None) -> None:
calls.append(run_variables["trigger"])
validated_config = await async_validate_trigger_config(hass, [trigger_config])
return await async_initialize_triggers(
hass,
validated_config,
action,
domain="test",
name="test_off_to_on",
log_cb=log.log,
)
def _set_or_remove_state(
hass: HomeAssistant, entity_id: str, state: str | None
) -> None:
"""Set or remove state based on whether state is None."""
if state is None:
hass.states.async_remove(entity_id)
else:
hass.states.async_set(entity_id, state)
@pytest.mark.parametrize("behavior", [BEHAVIOR_ANY, BEHAVIOR_FIRST, BEHAVIOR_LAST])
async def test_entity_trigger_fires_on_valid_transition(
hass: HomeAssistant, behavior: str
) -> None:
"""Test EntityTriggerBase fires immediately on a valid off→on transition without duration."""
entity_id = "test.entity_1"
hass.states.async_set(entity_id, STATE_OFF)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_id], behavior, calls, duration=None
)
hass.states.async_set(entity_id, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0]["entity_id"] == entity_id
assert calls[0]["from_state"].state == STATE_OFF
assert calls[0]["to_state"].state == STATE_ON
assert calls[0]["for"] is None
# Transition back and trigger again
calls.clear()
hass.states.async_set(entity_id, STATE_OFF)
await hass.async_block_till_done()
hass.states.async_set(entity_id, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 1
unsub()
@pytest.mark.parametrize("behavior", [BEHAVIOR_ANY, BEHAVIOR_FIRST, BEHAVIOR_LAST])
@pytest.mark.parametrize(
"initial_state",
[STATE_UNAVAILABLE, STATE_UNKNOWN, None],
ids=["unavailable", "unknown", "no_state"],
)
async def test_entity_trigger_from_invalid_initial_state(
hass: HomeAssistant, behavior: str, initial_state: str | None
) -> None:
"""Test that the trigger does not fire when transitioning from unavailable, unknown, or no state."""
entity_id = "test.entity_1"
_set_or_remove_state(hass, entity_id, initial_state)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_id], behavior, calls, duration=None
)
# Transition to "on" from the invalid initial state
_set_or_remove_state(hass, entity_id, STATE_ON)
await hass.async_block_till_done()
# Should NOT fire — transition from invalid state is rejected
assert len(calls) == 0
# Now transition back to off and then to on — should fire
_set_or_remove_state(hass, entity_id, STATE_OFF)
await hass.async_block_till_done()
_set_or_remove_state(hass, entity_id, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 1
unsub()
async def test_entity_trigger_last_requires_all(
hass: HomeAssistant,
) -> None:
"""Test behavior last: trigger fires only when ALL entities are on."""
entity_a = "test.entity_a"
entity_b = "test.entity_b"
hass.states.async_set(entity_a, STATE_OFF)
hass.states.async_set(entity_b, STATE_OFF)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_a, entity_b], BEHAVIOR_LAST, calls, duration=None
)
# Turn only A on — not all match, should not fire
hass.states.async_set(entity_a, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 0
# Turn B on — now all match, should fire
hass.states.async_set(entity_b, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 1
unsub()
async def test_entity_trigger_first_requires_exactly_one(
hass: HomeAssistant,
) -> None:
"""Test behavior first: trigger fires only when exactly one entity matches."""
entity_a = "test.entity_a"
entity_b = "test.entity_b"
hass.states.async_set(entity_a, STATE_OFF)
hass.states.async_set(entity_b, STATE_OFF)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_a, entity_b], BEHAVIOR_FIRST, calls, duration=None
)
# Turn A on — exactly one matches, should fire
hass.states.async_set(entity_a, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 1
# Turn B on — now two match, B's transition should NOT fire
hass.states.async_set(entity_b, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 1
unsub()
@pytest.mark.parametrize(
"invalid_state",
[STATE_UNAVAILABLE, STATE_UNKNOWN],
ids=["unavailable", "unknown"],
)
async def test_entity_trigger_last_ignores_unavailable_and_unknown_entity(
hass: HomeAssistant, invalid_state: str
) -> None:
"""Test behavior last: unavailable/unknown entities are excluded from check_all_match.
With three entities (A=off, B=unavailable, C=off), turning A on should
not fire because C is still off, so the available entities do not all
match. Turning C on then fires because all *available* entities (A and C)
match. Without the exclusion, B would fail the "all match" check.
"""
entity_a = "test.entity_a"
entity_b = "test.entity_b"
entity_c = "test.entity_c"
hass.states.async_set(entity_a, STATE_OFF)
hass.states.async_set(entity_b, invalid_state)
hass.states.async_set(entity_c, STATE_OFF)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_a, entity_b, entity_c], BEHAVIOR_LAST, calls, duration=None
)
# Turn A on — B is unavailable and skipped, only A is on → all doesn't match
hass.states.async_set(entity_a, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 0
# Turn C on — B is unavailable and skipped, A and C are both on → all match
hass.states.async_set(entity_c, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0]["entity_id"] == entity_c
# B recovers to off — now not all available entities match, so
# turning A off→on should NOT fire
calls.clear()
hass.states.async_set(entity_b, STATE_OFF)
await hass.async_block_till_done()
hass.states.async_set(entity_a, STATE_OFF)
await hass.async_block_till_done()
hass.states.async_set(entity_a, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 0
unsub()
@pytest.mark.parametrize(
"invalid_state",
[STATE_UNAVAILABLE, STATE_UNKNOWN],
ids=["unavailable", "unknown"],
)
async def test_entity_trigger_first_ignores_unavailable_and_unknown_entity(
hass: HomeAssistant, invalid_state: str
) -> None:
"""Test behavior first: unavailable/unknown entities are excluded from check_one_match.
With three entities (A=off, B=unavailable, C=off), turning A on should
fire because exactly one *available* entity matches. B is skipped.
Then turning C on should NOT fire because now two available entities match.
"""
entity_a = "test.entity_a"
entity_b = "test.entity_b"
entity_c = "test.entity_c"
hass.states.async_set(entity_a, STATE_OFF)
hass.states.async_set(entity_b, invalid_state)
hass.states.async_set(entity_c, STATE_OFF)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_a, entity_b, entity_c], BEHAVIOR_FIRST, calls, duration=None
)
# Turn A on — B is unavailable and skipped, only A matches → exactly one
hass.states.async_set(entity_a, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0]["entity_id"] == entity_a
# Turn C on — now two available entities match (A and C), should NOT fire
hass.states.async_set(entity_c, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 1
unsub()
@pytest.mark.parametrize("behavior", [BEHAVIOR_ANY, BEHAVIOR_FIRST, BEHAVIOR_LAST])
async def test_entity_trigger_with_duration(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, behavior: str
) -> None:
"""Test EntityTriggerBase waits for duration before firing."""
entity_id = "test.entity_1"
hass.states.async_set(entity_id, STATE_OFF)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_id], behavior, calls, duration={"seconds": 5}
)
# Turn on — should NOT fire immediately
hass.states.async_set(entity_id, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 0
# Advance time past duration — should fire
freezer.tick(datetime.timedelta(seconds=6))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0]["entity_id"] == entity_id
assert calls[0]["from_state"].state == STATE_OFF
assert calls[0]["to_state"].state == STATE_ON
assert calls[0]["for"] == datetime.timedelta(seconds=5)
unsub()
@pytest.mark.parametrize("behavior", [BEHAVIOR_ANY, BEHAVIOR_FIRST, BEHAVIOR_LAST])
async def test_entity_trigger_duration_cancelled_on_state_change(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, behavior: str
) -> None:
"""Test that the duration timer is cancelled if state changes back."""
entity_id = "test.entity_1"
hass.states.async_set(entity_id, STATE_OFF)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_id], behavior, calls, duration={"seconds": 5}
)
# Turn on, then back off before duration expires
hass.states.async_set(entity_id, STATE_ON)
await hass.async_block_till_done()
freezer.tick(datetime.timedelta(seconds=2))
async_fire_time_changed(hass)
await hass.async_block_till_done()
hass.states.async_set(entity_id, STATE_OFF)
await hass.async_block_till_done()
# Advance past the original duration — should NOT fire
freezer.tick(datetime.timedelta(seconds=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 0
unsub()
async def test_entity_trigger_duration_any_independent(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test behavior any tracks per-entity durations independently."""
entity_a = "test.entity_a"
entity_b = "test.entity_b"
hass.states.async_set(entity_a, STATE_OFF)
hass.states.async_set(entity_b, STATE_OFF)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_a, entity_b], BEHAVIOR_ANY, calls, duration={"seconds": 5}
)
# Turn A on
hass.states.async_set(entity_a, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 0
# Turn B on 2 seconds later
freezer.tick(datetime.timedelta(seconds=2))
async_fire_time_changed(hass)
hass.states.async_set(entity_b, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 0
# After 5s from A's turn-on, A should fire
freezer.tick(datetime.timedelta(seconds=3))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0]["entity_id"] == entity_a
# After 5s from B's turn-on (2 more seconds), B should fire
freezer.tick(datetime.timedelta(seconds=2))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 2
assert calls[1]["entity_id"] == entity_b
unsub()
async def test_entity_trigger_duration_any_entity_off_cancels_only_that_entity(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test behavior any: turning off one entity doesn't cancel the other's timer."""
entity_a = "test.entity_a"
entity_b = "test.entity_b"
hass.states.async_set(entity_a, STATE_OFF)
hass.states.async_set(entity_b, STATE_OFF)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_a, entity_b], BEHAVIOR_ANY, calls, duration={"seconds": 5}
)
# Turn both on
hass.states.async_set(entity_a, STATE_ON)
hass.states.async_set(entity_b, STATE_ON)
await hass.async_block_till_done()
# Turn A off after 2 seconds — cancels A's timer but not B's
freezer.tick(datetime.timedelta(seconds=2))
async_fire_time_changed(hass)
hass.states.async_set(entity_a, STATE_OFF)
await hass.async_block_till_done()
# After 5s total, B should fire but A should not
freezer.tick(datetime.timedelta(seconds=3))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0]["entity_id"] == entity_b
unsub()
async def test_entity_trigger_duration_last_requires_all(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test behavior last: trigger fires only when ALL entities are on for duration."""
entity_a = "test.entity_a"
entity_b = "test.entity_b"
hass.states.async_set(entity_a, STATE_OFF)
hass.states.async_set(entity_b, STATE_OFF)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_a, entity_b], BEHAVIOR_LAST, calls, duration={"seconds": 5}
)
# Turn only A on — should not start timer (not all match)
hass.states.async_set(entity_a, STATE_ON)
await hass.async_block_till_done()
freezer.tick(datetime.timedelta(seconds=6))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 0
# Turn B on — now all match, timer starts
hass.states.async_set(entity_b, STATE_ON)
await hass.async_block_till_done()
freezer.tick(datetime.timedelta(seconds=6))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
unsub()
async def test_entity_trigger_duration_last_cancelled_when_one_turns_off(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test behavior last: timer is cancelled when one entity turns off."""
entity_a = "test.entity_a"
entity_b = "test.entity_b"
hass.states.async_set(entity_a, STATE_OFF)
hass.states.async_set(entity_b, STATE_OFF)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_a, entity_b], BEHAVIOR_LAST, calls, duration={"seconds": 5}
)
# Turn both on
hass.states.async_set(entity_a, STATE_ON)
hass.states.async_set(entity_b, STATE_ON)
await hass.async_block_till_done()
# Turn A off after 2 seconds
freezer.tick(datetime.timedelta(seconds=2))
async_fire_time_changed(hass)
hass.states.async_set(entity_a, STATE_OFF)
await hass.async_block_till_done()
# Advance past original duration — should NOT fire
freezer.tick(datetime.timedelta(seconds=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 0
unsub()
async def test_entity_trigger_duration_last_timer_reset(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test behavior last: timer resets when combined state goes off and back on."""
entity_a = "test.entity_a"
entity_b = "test.entity_b"
hass.states.async_set(entity_a, STATE_OFF)
hass.states.async_set(entity_b, STATE_OFF)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_a, entity_b], BEHAVIOR_LAST, calls, duration={"seconds": 5}
)
# Turn both on — combined state "all on", timer starts
hass.states.async_set(entity_a, STATE_ON)
await hass.async_block_till_done()
hass.states.async_set(entity_b, STATE_ON)
await hass.async_block_till_done()
# After 2 seconds, B turns off — combined state breaks, timer cancelled
freezer.tick(datetime.timedelta(seconds=2))
async_fire_time_changed(hass)
hass.states.async_set(entity_b, STATE_OFF)
await hass.async_block_till_done()
# B turns back on — combined state restored, timer restarts
freezer.tick(datetime.timedelta(seconds=1))
async_fire_time_changed(hass)
hass.states.async_set(entity_b, STATE_ON)
await hass.async_block_till_done()
# 4 seconds after restart (not enough) — should NOT fire
freezer.tick(datetime.timedelta(seconds=4))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 0
# 1 more second (5 total from restart) — should fire
freezer.tick(datetime.timedelta(seconds=1))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
unsub()
async def test_entity_trigger_duration_first_fires_when_any_on(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test behavior first: trigger fires when first entity turns on for duration."""
entity_a = "test.entity_a"
entity_b = "test.entity_b"
hass.states.async_set(entity_a, STATE_OFF)
hass.states.async_set(entity_b, STATE_OFF)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_a, entity_b], BEHAVIOR_FIRST, calls, duration={"seconds": 5}
)
# Turn A on — combined state goes to "at least one on", timer starts
hass.states.async_set(entity_a, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 0
# Advance past duration — should fire
freezer.tick(datetime.timedelta(seconds=6))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
unsub()
async def test_entity_trigger_duration_first_not_cancelled_by_second(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test behavior first: second entity turning on doesn't restart timer."""
entity_a = "test.entity_a"
entity_b = "test.entity_b"
hass.states.async_set(entity_a, STATE_OFF)
hass.states.async_set(entity_b, STATE_OFF)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_a, entity_b], BEHAVIOR_FIRST, calls, duration={"seconds": 5}
)
# Turn A on
hass.states.async_set(entity_a, STATE_ON)
await hass.async_block_till_done()
# Turn B on 3 seconds later — combined state was already "any on",
# so this should NOT restart the timer
freezer.tick(datetime.timedelta(seconds=3))
async_fire_time_changed(hass)
hass.states.async_set(entity_b, STATE_ON)
await hass.async_block_till_done()
assert len(calls) == 0
# 2 more seconds (5 total from A) — should fire
freezer.tick(datetime.timedelta(seconds=2))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
unsub()
async def test_entity_trigger_duration_first_not_cancelled_by_partial_off(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test behavior first: one entity off doesn't cancel if another is still on."""
entity_a = "test.entity_a"
entity_b = "test.entity_b"
hass.states.async_set(entity_a, STATE_OFF)
hass.states.async_set(entity_b, STATE_OFF)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_a, entity_b], BEHAVIOR_FIRST, calls, duration={"seconds": 5}
)
# Turn both on
hass.states.async_set(entity_a, STATE_ON)
await hass.async_block_till_done()
hass.states.async_set(entity_b, STATE_ON)
await hass.async_block_till_done()
# Turn A off after 2 seconds — combined state still "any on" (B is on)
freezer.tick(datetime.timedelta(seconds=2))
async_fire_time_changed(hass)
hass.states.async_set(entity_a, STATE_OFF)
await hass.async_block_till_done()
# Advance past duration — should still fire (combined state never went to "none on")
freezer.tick(datetime.timedelta(seconds=4))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
unsub()
async def test_entity_trigger_duration_first_cancelled_when_all_off(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test behavior first: timer cancelled when ALL entities turn off."""
entity_a = "test.entity_a"
entity_b = "test.entity_b"
hass.states.async_set(entity_a, STATE_OFF)
hass.states.async_set(entity_b, STATE_OFF)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_a, entity_b], BEHAVIOR_FIRST, calls, duration={"seconds": 5}
)
# Turn both on
hass.states.async_set(entity_a, STATE_ON)
hass.states.async_set(entity_b, STATE_ON)
await hass.async_block_till_done()
# Turn both off after 2 seconds — combined state goes to "none on"
freezer.tick(datetime.timedelta(seconds=2))
async_fire_time_changed(hass)
hass.states.async_set(entity_a, STATE_OFF)
hass.states.async_set(entity_b, STATE_OFF)
await hass.async_block_till_done()
# Advance past original duration — should NOT fire
freezer.tick(datetime.timedelta(seconds=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 0
unsub()
async def test_entity_trigger_duration_any_retrigger_resets_timer(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test behavior any: turning an entity off and on resets its timer."""
entity_id = "test.entity_1"
hass.states.async_set(entity_id, STATE_OFF)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_id], BEHAVIOR_ANY, calls, duration={"seconds": 5}
)
# Turn on
hass.states.async_set(entity_id, STATE_ON)
await hass.async_block_till_done()
# After 3 seconds, turn off and on again — resets the timer
freezer.tick(datetime.timedelta(seconds=3))
async_fire_time_changed(hass)
hass.states.async_set(entity_id, STATE_OFF)
await hass.async_block_till_done()
hass.states.async_set(entity_id, STATE_ON)
await hass.async_block_till_done()
# 3 more seconds (6 from start, but only 3 from retrigger) — should NOT fire
freezer.tick(datetime.timedelta(seconds=3))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 0
# 2 more seconds (5 from retrigger) — should fire
freezer.tick(datetime.timedelta(seconds=2))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == 1
unsub()
@pytest.mark.parametrize(
("behavior", "expected_calls"),
[(BEHAVIOR_ANY, 0), (BEHAVIOR_FIRST, 0), (BEHAVIOR_LAST, 1)],
)
@pytest.mark.parametrize(
"invalid_state",
[STATE_UNAVAILABLE, STATE_UNKNOWN, None],
ids=["unavailable", "unknown", "removed"],
)
async def test_entity_trigger_duration_cancelled_on_invalid_state(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
behavior: str,
expected_calls: int,
invalid_state: str | None,
) -> None:
"""Test if the duration timer is cancelled if entity becomes unavailable, unknown, or is removed.
This is expected to happen in first and any modes, but not in last mode.
"""
entity_a = "test.entity_a"
entity_b = "test.entity_b"
_set_or_remove_state(hass, entity_a, STATE_OFF)
_set_or_remove_state(hass, entity_b, STATE_OFF)
await hass.async_block_till_done()
calls: list[dict[str, Any]] = []
unsub = await _arm_off_to_on_trigger(
hass, [entity_a, entity_b], behavior, calls, duration={"seconds": 5}
)
# Turn on the entities needed to start the timer
_set_or_remove_state(hass, entity_a, STATE_ON)
await hass.async_block_till_done()
if behavior == BEHAVIOR_LAST:
_set_or_remove_state(hass, entity_b, STATE_ON)
await hass.async_block_till_done()
# Entity A becomes invalid during the wait
freezer.tick(datetime.timedelta(seconds=2))
async_fire_time_changed(hass)
_set_or_remove_state(hass, entity_a, invalid_state)
await hass.async_block_till_done()
# Advance past the original duration — should NOT fire
freezer.tick(datetime.timedelta(seconds=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert len(calls) == expected_calls
unsub()
View Git Blame Copy Permalink