Compare commits

...

29 Commits

Author SHA1 Message Date
Erik
55eb01c894 Address review comments 2025-11-13 20:48:59 +01:00
Erik
5ccd2c20a4 Improve test parametrization 2025-11-13 18:21:53 +01:00
Erik
333075246f Remove debug print 2025-11-13 18:04:35 +01:00
Erik
3c2573e5b6 Add trigger cover.garage_opened 2025-11-13 18:00:58 +01:00
Erik Montnemery
edf82db057 Small improvement of entity trigger tests (#156498) 2025-11-13 11:00:11 +01:00
Erik Montnemery
37644511f6 Simplify entity state trigger (#156495) 2025-11-13 09:19:40 +01:00
Erik Montnemery
3685d0f7c2 Add trigger climate.started_heating (#156473) 2025-11-13 07:44:02 +01:00
Erik Montnemery
3dabfeb329 Add more test helpers for trigger tests (#156464) 2025-11-12 19:33:59 +01:00
Erik Montnemery
8e7d2d7108 Add test helper for trigger tests (#156462) 2025-11-12 17:12:32 +01:00
Erik Montnemery
2fe4a1164b Add climate turned off trigger (#156453) 2025-11-12 16:48:10 +01:00
Erik Montnemery
05175294f6 Check from state in light triggers (#156443) 2025-11-12 13:15:30 +01:00
Erik Montnemery
e2ddfb8782 Add light trigger tests where initial state is none, unknown or unavailable (#156430) 2025-11-12 11:54:15 +01:00
Erik Montnemery
f1cc133ff6 Improve light trigger tests (#156424) 2025-11-12 09:12:41 +01:00
Erik Montnemery
0cf97cf577 Improve light trigger tests (#156378) 2025-11-11 21:15:26 +01:00
Franck Nijhof
38cea2e5f0 Clean up CONF_OPTIONS constant from light triggers (#156375) 2025-11-11 17:49:45 +01:00
Erik Montnemery
71876d5b34 Fix light trigger descriptions (#156380) 2025-11-11 17:47:48 +01:00
Abílio Costa
0f780254e1 Add light state condition (#149830)
Co-authored-by: Artur Pragacz <49985303+arturpragacz@users.noreply.github.com>
2025-11-11 13:46:43 +01:00
Abílio Costa
9e40972b11 Split light state trigger (#156272)
Co-authored-by: Erik <erik@montnemery.com>
2025-11-11 13:42:40 +01:00
abmantis
07ef61dd8d Merge branch 'dev' of github.com:home-assistant/core into dev_target_triggers_conditions 2025-11-10 14:19:18 +00:00
abmantis
1bf6771a54 Merge branch 'dev' of github.com:home-assistant/core into dev_target_triggers_conditions 2025-11-06 19:57:40 +00:00
abmantis
e7a7cb829e Merge branch 'dev' of github.com:home-assistant/core into dev_target_triggers_conditions 2025-11-04 12:28:39 +00:00
abmantis
6f6b2f1ad3 Merge branch 'dev_target_triggers_conditions' of github.com:home-assistant/core into dev_target_triggers_conditions 2025-10-15 17:03:28 +01:00
abmantis
1cc4890f75 Merge branch 'dev' of github.com:home-assistant/core into dev_target_triggers_conditions 2025-10-15 17:03:18 +01:00
Bram Kragten
d3dd9b26c9 Fixes for triggers.yaml descriptions (#153841)
Co-authored-by: Artur Pragacz <49985303+arturpragacz@users.noreply.github.com>
2025-10-09 18:00:56 +01:00
Abílio Costa
a64d61df05 Fix light trigger with new Trigger class changes (#154087) 2025-10-09 18:14:55 +02:00
abmantis
e7c6c5311d Merge branch 'dev' of github.com:home-assistant/core into dev_target_triggers_conditions 2025-10-09 15:55:39 +01:00
abmantis
72a524c868 Merge branch 'dev' of github.com:home-assistant/core into dev_target_triggers_conditions 2025-09-29 16:56:23 +01:00
abmantis
b437113f31 Merge branch 'dev' of github.com:home-assistant/core into dev_target_triggers_conditions 2025-09-29 11:18:39 +01:00
Abílio Costa
e0e263d3b5 Add state trigger to light component (#148416)
Co-authored-by: Artur Pragacz <49985303+arturpragacz@users.noreply.github.com>
2025-09-18 19:53:26 +01:00
22 changed files with 2158 additions and 7 deletions

View File

@@ -96,5 +96,13 @@
"turn_on": {
"service": "mdi:power-on"
}
},
"triggers": {
"started_heating": {
"trigger": "mdi:fire"
},
"turned_off": {
"trigger": "mdi:power-off"
}
}
}

View File

@@ -187,6 +187,13 @@
"heat_cool": "Heat/cool",
"off": "[%key:common::state::off%]"
}
},
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
}
}
},
"services": {
@@ -285,5 +292,29 @@
"name": "[%key:common::action::turn_on%]"
}
},
"title": "Climate"
"title": "Climate",
"triggers": {
"started_heating": {
"description": "Triggers when a climate starts to heat.",
"description_configured": "Triggers when a climate starts to heat",
"fields": {
"behavior": {
"description": "[%key:component::climate::triggers::turned_off::fields::behavior::description%]",
"name": "[%key:component::climate::triggers::turned_off::fields::behavior::name%]"
}
},
"name": "When a climate starts to heat"
},
"turned_off": {
"description": "Triggers when a climate is turned off.",
"description_configured": "Triggers when a climate is turned off",
"fields": {
"behavior": {
"description": "The behavior of the targeted climates to trigger on.",
"name": "Behavior"
}
},
"name": "When a climate is turned off"
}
}
}

View File

@@ -0,0 +1,23 @@
"""Provides triggers for climates."""
from homeassistant.const import STATE_OFF
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import (
Trigger,
make_entity_state_attribute_trigger,
make_entity_state_trigger,
)
from .const import ATTR_HVAC_ACTION, DOMAIN, HVACAction
TRIGGERS: dict[str, type[Trigger]] = {
"turned_off": make_entity_state_trigger(DOMAIN, STATE_OFF),
"started_heating": make_entity_state_attribute_trigger(
DOMAIN, ATTR_HVAC_ACTION, HVACAction.HEATING
),
}
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
"""Return the triggers for climates."""
return TRIGGERS

View File

@@ -0,0 +1,19 @@
.trigger_common: &trigger_common
target:
entity:
domain: climate
fields:
behavior:
required: true
default: any
selector:
select:
translation_key: trigger_behavior
options:
- first
- last
- any
started_heating: *trigger_common
turned_off: *trigger_common

View File

@@ -108,5 +108,10 @@
"toggle_cover_tilt": {
"service": "mdi:arrow-top-right-bottom-left"
}
},
"triggers": {
"garage_opened": {
"trigger": "mdi:garage-open"
}
}
}

View File

@@ -82,6 +82,15 @@
"name": "Window"
}
},
"selector": {
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
}
}
},
"services": {
"close_cover": {
"description": "Closes a cover.",
@@ -136,5 +145,22 @@
"name": "Toggle tilt"
}
},
"title": "Cover"
"title": "Cover",
"triggers": {
"garage_opened": {
"description": "Triggers when a garage door opens.",
"description_configured": "[%key:component::cover::triggers::garage_opened::description%]",
"fields": {
"behavior": {
"description": "The behavior of the targeted garage doors to trigger on.",
"name": "Behavior"
},
"fully_opened": {
"description": "Require the garage doors to be fully opened before triggering.",
"name": "Fully opened"
}
},
"name": "When a garage door opens"
}
}
}

View File

@@ -0,0 +1,119 @@
"""Provides triggers for covers."""
from typing import Final
import voluptuous as vol
from homeassistant.const import CONF_OPTIONS
from homeassistant.core import HomeAssistant, State
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity import get_device_class
from homeassistant.helpers.trigger import (
ENTITY_STATE_TRIGGER_SCHEMA,
EntityTriggerBase,
Trigger,
TriggerConfig,
)
from homeassistant.helpers.typing import UNDEFINED, UndefinedType
from . import ATTR_CURRENT_POSITION, CoverDeviceClass, CoverState
from .const import DOMAIN
ATTR_FULLY_OPENED: Final = "fully_opened"
COVER_OPENED_TRIGGER_SCHEMA = ENTITY_STATE_TRIGGER_SCHEMA.extend(
{
vol.Required(CONF_OPTIONS): {
vol.Required(ATTR_FULLY_OPENED, default=False): bool,
},
}
)
def get_device_class_or_undefined(
hass: HomeAssistant, entity_id: str
) -> str | None | UndefinedType:
"""Get the device class of an entity or UNDEFINED if not found."""
try:
return get_device_class(hass, entity_id)
except HomeAssistantError:
return UNDEFINED
class CoverOpenedClosedTrigger(EntityTriggerBase):
"""Class for cover opened and closed triggers."""
_attribute: str = ATTR_CURRENT_POSITION
_attribute_value: int | None = None
_device_class: CoverDeviceClass | None
_domain: str = DOMAIN
_to_states: set[str]
def is_state_same(self, from_state: State, to_state: State) -> bool:
"""Check if the old and new states are considered the same."""
if from_state.state != to_state.state:
return False
if self._attribute_value is not None:
from_value = from_state.attributes.get(self._attribute)
to_value = to_state.attributes.get(self._attribute)
if from_value != to_value:
return False
return True
def is_state_to_state(self, state: State) -> bool:
"""Check if the state matches the target state."""
if state.state not in self._to_states:
return False
if (
self._attribute_value is not None
and (value := state.attributes.get(self._attribute)) is not None
and value != self._attribute_value
):
return False
return True
def entity_filter(self, entities: set[str]) -> set[str]:
"""Filter entities of this domain."""
entities = super().entity_filter(entities)
return {
entity_id
for entity_id in entities
if get_device_class_or_undefined(self._hass, entity_id)
== self._device_class
}
class CoverOpenedTrigger(CoverOpenedClosedTrigger):
"""Class for cover opened triggers."""
_schema = COVER_OPENED_TRIGGER_SCHEMA
_to_states = {CoverState.OPEN, CoverState.OPENING}
def __init__(self, hass: HomeAssistant, config: TriggerConfig) -> None:
"""Initialize the state trigger."""
super().__init__(hass, config)
if self._options.get(ATTR_FULLY_OPENED):
self._attribute_value = 100
def make_cover_opened_trigger(
device_class: CoverDeviceClass | None,
) -> type[CoverOpenedTrigger]:
"""Create an entity state attribute trigger class."""
class CustomTrigger(CoverOpenedTrigger):
"""Trigger for entity state changes."""
_device_class = device_class
return CustomTrigger
TRIGGERS: dict[str, type[Trigger]] = {
"garage_opened": make_cover_opened_trigger(CoverDeviceClass.GARAGE),
}
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
"""Return the triggers for covers."""
return TRIGGERS

View File

@@ -0,0 +1,21 @@
garage_opened:
target:
entity:
domain: climate
device_class: garage
fields:
behavior:
required: true
default: any
selector:
select:
translation_key: trigger_behavior
options:
- first
- last
- any
fully_opened:
required: true
default: false
selector:
boolean:

View File

@@ -770,7 +770,9 @@ class ManifestJSONView(HomeAssistantView):
@websocket_api.websocket_command(
{
"type": "frontend/get_icons",
vol.Required("category"): vol.In({"entity", "entity_component", "services"}),
vol.Required("category"): vol.In(
{"entity", "entity_component", "services", "triggers"}
),
vol.Optional("integration"): vol.All(cv.ensure_list, [str]),
}
)

View File

@@ -0,0 +1,131 @@
"""Provides conditions for lights."""
from collections.abc import Callable
from typing import TYPE_CHECKING, Any, Final, override
import voluptuous as vol
from homeassistant.const import CONF_OPTIONS, CONF_TARGET, STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant, split_entity_id
from homeassistant.helpers import config_validation as cv, target
from homeassistant.helpers.condition import (
Condition,
ConditionCheckerType,
ConditionConfig,
trace_condition_function,
)
from homeassistant.helpers.typing import ConfigType, TemplateVarsType
from .const import DOMAIN
ATTR_BEHAVIOR: Final = "behavior"
BEHAVIOR_ANY: Final = "any"
BEHAVIOR_ALL: Final = "all"
STATE_CONDITION_VALID_STATES: Final = [STATE_ON, STATE_OFF]
STATE_CONDITION_OPTIONS_SCHEMA: dict[vol.Marker, Any] = {
vol.Required(ATTR_BEHAVIOR, default=BEHAVIOR_ANY): vol.In(
[BEHAVIOR_ANY, BEHAVIOR_ALL]
),
}
STATE_CONDITION_SCHEMA = vol.Schema(
{
vol.Required(CONF_TARGET): cv.TARGET_FIELDS,
vol.Required(CONF_OPTIONS): STATE_CONDITION_OPTIONS_SCHEMA,
}
)
class StateConditionBase(Condition):
"""State condition."""
@override
@classmethod
async def async_validate_config(
cls, hass: HomeAssistant, config: ConfigType
) -> ConfigType:
"""Validate config."""
return STATE_CONDITION_SCHEMA(config) # type: ignore[no-any-return]
def __init__(
self, hass: HomeAssistant, config: ConditionConfig, state: str
) -> None:
"""Initialize condition."""
self._hass = hass
if TYPE_CHECKING:
assert config.target
assert config.options
self._target = config.target
self._behavior = config.options[ATTR_BEHAVIOR]
self._state = state
@override
async def async_get_checker(self) -> ConditionCheckerType:
"""Get the condition checker."""
def check_any_match_state(states: list[str]) -> bool:
"""Test if any entity match the state."""
return any(state == self._state for state in states)
def check_all_match_state(states: list[str]) -> bool:
"""Test if all entities match the state."""
return all(state == self._state for state in states)
matcher: Callable[[list[str]], bool]
if self._behavior == BEHAVIOR_ANY:
matcher = check_any_match_state
elif self._behavior == BEHAVIOR_ALL:
matcher = check_all_match_state
@trace_condition_function
def test_state(hass: HomeAssistant, variables: TemplateVarsType = None) -> bool:
"""Test state condition."""
selector_data = target.TargetSelectorData(self._target)
targeted_entities = target.async_extract_referenced_entity_ids(
hass, selector_data, expand_group=False
)
referenced_entity_ids = targeted_entities.referenced.union(
targeted_entities.indirectly_referenced
)
light_entity_ids = {
entity_id
for entity_id in referenced_entity_ids
if split_entity_id(entity_id)[0] == DOMAIN
}
light_entity_states = [
state.state
for entity_id in light_entity_ids
if (state := hass.states.get(entity_id))
and state.state in STATE_CONDITION_VALID_STATES
]
return matcher(light_entity_states)
return test_state
class IsOnCondition(StateConditionBase):
"""Is on condition."""
def __init__(self, hass: HomeAssistant, config: ConditionConfig) -> None:
"""Initialize condition."""
super().__init__(hass, config, STATE_ON)
class IsOffCondition(StateConditionBase):
"""Is off condition."""
def __init__(self, hass: HomeAssistant, config: ConditionConfig) -> None:
"""Initialize condition."""
super().__init__(hass, config, STATE_OFF)
CONDITIONS: dict[str, type[Condition]] = {
"is_off": IsOffCondition,
"is_on": IsOnCondition,
}
async def async_get_conditions(hass: HomeAssistant) -> dict[str, type[Condition]]:
"""Return the light conditions."""
return CONDITIONS

View File

@@ -0,0 +1,28 @@
is_off:
target:
entity:
domain: light
fields:
behavior:
required: true
default: any
selector:
select:
translation_key: condition_behavior
options:
- all
- any
is_on:
target:
entity:
domain: light
fields:
behavior:
required: true
default: any
selector:
select:
translation_key: condition_behavior
options:
- all
- any

View File

@@ -1,4 +1,12 @@
{
"conditions": {
"is_off": {
"condition": "mdi:lightbulb-off"
},
"is_on": {
"condition": "mdi:lightbulb-on"
}
},
"entity_component": {
"_": {
"default": "mdi:lightbulb",
@@ -25,5 +33,13 @@
"turn_on": {
"service": "mdi:lightbulb-on"
}
},
"triggers": {
"turned_off": {
"trigger": "mdi:lightbulb-off"
},
"turned_on": {
"trigger": "mdi:lightbulb-on"
}
}
}

View File

@@ -36,6 +36,30 @@
"field_xy_color_name": "XY-color",
"section_advanced_fields_name": "Advanced options"
},
"conditions": {
"is_off": {
"description": "Test if a light is off.",
"description_configured": "Test if a light is off",
"fields": {
"behavior": {
"description": "How the state should match on the targeted lights.",
"name": "Behavior"
}
},
"name": "If a light is off"
},
"is_on": {
"description": "Test if a light is on.",
"description_configured": "Test if a light is on",
"fields": {
"behavior": {
"description": "How the state should match on the targeted lights.",
"name": "Behavior"
}
},
"name": "If a light is on"
}
},
"device_automation": {
"action_type": {
"brightness_decrease": "Decrease {entity_name} brightness",
@@ -284,11 +308,30 @@
"yellowgreen": "Yellow green"
}
},
"condition_behavior": {
"options": {
"all": "All",
"any": "Any"
}
},
"flash": {
"options": {
"long": "Long",
"short": "Short"
}
},
"state": {
"options": {
"off": "[%key:common::state::off%]",
"on": "[%key:common::state::on%]"
}
},
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
}
}
},
"services": {
@@ -462,5 +505,29 @@
}
}
},
"title": "Light"
"title": "Light",
"triggers": {
"turned_off": {
"description": "Triggers when a light is turned off.",
"description_configured": "Triggers when a light is turned off",
"fields": {
"behavior": {
"description": "The behavior of the targeted lights to trigger on.",
"name": "Behavior"
}
},
"name": "When a light is turned off"
},
"turned_on": {
"description": "Triggers when a light is turned on.",
"description_configured": "Triggers when a light is turned on",
"fields": {
"behavior": {
"description": "[%key:component::light::triggers::turned_off::fields::behavior::description%]",
"name": "[%key:component::light::triggers::turned_off::fields::behavior::name%]"
}
},
"name": "When a light is turned on"
}
}
}

View File

@@ -0,0 +1,17 @@
"""Provides triggers for lights."""
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import Trigger, make_entity_state_trigger
from .const import DOMAIN
TRIGGERS: dict[str, type[Trigger]] = {
"turned_off": make_entity_state_trigger(DOMAIN, STATE_OFF),
"turned_on": make_entity_state_trigger(DOMAIN, STATE_ON),
}
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
"""Return the triggers for lights."""
return TRIGGERS

View File

@@ -0,0 +1,31 @@
turned_on:
target:
entity:
domain: light
fields:
behavior:
required: true
default: any
selector:
select:
options:
- first
- last
- any
translation_key: trigger_behavior
turned_off:
target:
entity:
domain: light
fields:
behavior:
required: true
default: any
selector:
select:
translation_key: trigger_behavior
options:
- first
- last
- any

View File

@@ -10,11 +10,12 @@ from dataclasses import dataclass, field
import functools
import inspect
import logging
from typing import TYPE_CHECKING, Any, Protocol, TypedDict, cast
from typing import TYPE_CHECKING, Any, Final, Protocol, TypedDict, cast, override
import voluptuous as vol
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_ALIAS,
CONF_ENABLED,
CONF_ID,
@@ -23,6 +24,8 @@ from homeassistant.const import (
CONF_SELECTOR,
CONF_TARGET,
CONF_VARIABLES,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import (
CALLBACK_TYPE,
@@ -30,9 +33,11 @@ from homeassistant.core import (
HassJob,
HassJobType,
HomeAssistant,
State,
callback,
get_hassjob_callable_job_type,
is_callback,
split_entity_id,
)
from homeassistant.exceptions import HomeAssistantError, TemplateError
from homeassistant.loader import (
@@ -49,6 +54,10 @@ from . import config_validation as cv, selector
from .automation import get_absolute_description_key, get_relative_description_key
from .integration_platform import async_process_integration_platforms
from .selector import TargetSelector
from .target import (
TargetStateChangedData,
async_track_target_selector_state_change_event,
)
from .template import Template
from .typing import ConfigType, TemplateVarsType
@@ -108,6 +117,22 @@ _TRIGGERS_DESCRIPTION_SCHEMA = vol.Schema(
}
)
ATTR_BEHAVIOR: Final = "behavior"
BEHAVIOR_FIRST: Final = "first"
BEHAVIOR_LAST: Final = "last"
BEHAVIOR_ANY: Final = "any"
ENTITY_STATE_TRIGGER_SCHEMA = vol.Schema(
{
vol.Required(CONF_OPTIONS): {
vol.Required(ATTR_BEHAVIOR, default=BEHAVIOR_ANY): vol.In(
[BEHAVIOR_FIRST, BEHAVIOR_LAST, BEHAVIOR_ANY]
),
},
vol.Required(CONF_TARGET): cv.TARGET_FIELDS,
}
)
async def async_setup(hass: HomeAssistant) -> None:
"""Set up the trigger helper."""
@@ -245,6 +270,178 @@ class Trigger(abc.ABC):
"""Attach the trigger to an action runner."""
class EntityTriggerBase(Trigger):
"""Trigger for entity state changes."""
_domain: str
_schema: vol.Schema = ENTITY_STATE_TRIGGER_SCHEMA
_to_state: str
@override
@classmethod
async def async_validate_config(
cls, hass: HomeAssistant, config: ConfigType
) -> ConfigType:
"""Validate config."""
return cast(ConfigType, cls._schema(config))
def __init__(self, hass: HomeAssistant, config: TriggerConfig) -> None:
"""Initialize the state trigger."""
super().__init__(hass, config)
if TYPE_CHECKING:
assert config.options is not None
assert config.target is not None
self._options = config.options
self._target = config.target
@abc.abstractmethod
def is_state_same(self, from_state: State, to_state: State) -> bool:
"""Check if the old and new states are considered the same."""
@abc.abstractmethod
def is_state_to_state(self, state: State) -> bool:
"""Check if the state matches the target state."""
def check_all_match(self, entity_ids: set[str]) -> bool:
"""Check if all entity states match."""
return all(
self.is_state_to_state(state)
for entity_id in entity_ids
if (state := self._hass.states.get(entity_id)) is not None
)
def check_one_match(self, entity_ids: set[str]) -> bool:
"""Check that only one entity state matches."""
return (
sum(
self.is_state_to_state(state)
for entity_id in entity_ids
if (state := self._hass.states.get(entity_id)) is not None
)
== 1
)
def entity_filter(self, entities: set[str]) -> set[str]:
"""Filter entities of this domain."""
return {
entity_id
for entity_id in entities
if split_entity_id(entity_id)[0] == self._domain
}
@override
async def async_attach_runner(
self, run_action: TriggerActionRunner
) -> CALLBACK_TYPE:
"""Attach the trigger to an action runner."""
behavior = self._options.get(ATTR_BEHAVIOR)
@callback
def state_change_listener(
target_state_change_data: TargetStateChangedData,
) -> None:
"""Listen for state changes and call action."""
event = target_state_change_data.state_change_event
entity_id = event.data["entity_id"]
from_state = event.data["old_state"]
to_state = event.data["new_state"]
# The trigger should never fire if the previous state was not a valid state
if not from_state or from_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN):
return
# The trigger should never fire if the new state is not the to state
if not to_state or not self.is_state_to_state(to_state):
return
# The trigger should never fire if the previous and new states are the same
if self.is_state_same(from_state, to_state):
return
if behavior == BEHAVIOR_LAST:
if not self.check_all_match(
target_state_change_data.targeted_entity_ids
):
return
elif behavior == BEHAVIOR_FIRST:
if not self.check_one_match(
target_state_change_data.targeted_entity_ids
):
return
run_action(
{
ATTR_ENTITY_ID: entity_id,
"from_state": from_state,
"to_state": to_state,
},
f"state of {entity_id}",
event.context,
)
return async_track_target_selector_state_change_event(
self._hass, self._target, state_change_listener, self.entity_filter
)
class EntityStateTriggerBase(EntityTriggerBase):
"""Trigger for entity state changes."""
def is_state_same(self, from_state: State, to_state: State) -> bool:
"""Check if the old and new states are considered the same."""
return from_state.state == to_state.state
def is_state_to_state(self, state: State) -> bool:
"""Check if the state matches the target state."""
return state.state == self._to_state
class EntityStateAttributeTriggerBase(EntityTriggerBase):
"""Trigger for entity state attribute changes."""
_attribute: str
def is_state_same(self, from_state: State, to_state: State) -> bool:
"""Check if the old and new states are considered the same."""
return from_state.attributes.get(self._attribute) == to_state.attributes.get(
self._attribute
)
def is_state_to_state(self, state: State) -> bool:
"""Check if the state matches the target state."""
return state.attributes.get(self._attribute) == self._to_state
def make_entity_state_trigger(
domain: str, to_state: str
) -> type[EntityStateTriggerBase]:
"""Create an entity state trigger class."""
class CustomTrigger(EntityStateTriggerBase):
"""Trigger for entity state changes."""
_domain = domain
_to_state = to_state
return CustomTrigger
def make_entity_state_attribute_trigger(
domain: str, attribute: str, to_state: str
) -> type[EntityStateAttributeTriggerBase]:
"""Create an entity state attribute trigger class."""
class CustomTrigger(EntityStateAttributeTriggerBase):
"""Trigger for entity state changes."""
_domain = domain
_to_state = to_state
_attribute = attribute
return CustomTrigger
class TriggerProtocol(Protocol):
"""Define the format of trigger modules.
@@ -806,6 +1003,9 @@ async def async_get_all_descriptions(
description = {"fields": yaml_description.get("fields", {})}
if (target := yaml_description.get("target")) is not None:
description["target"] = target
new_descriptions_cache[missing_trigger] = description
hass.data[TRIGGER_DESCRIPTION_CACHE] = new_descriptions_cache

View File

@@ -1 +1,215 @@
"""The tests for components."""
from homeassistant.const import (
ATTR_AREA_ID,
ATTR_DEVICE_ID,
ATTR_FLOOR_ID,
ATTR_LABEL_ID,
CONF_ENTITY_ID,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import (
area_registry as ar,
device_registry as dr,
entity_registry as er,
floor_registry as fr,
label_registry as lr,
)
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry, mock_device_registry
async def target_entities(hass: HomeAssistant, domain: str) -> None:
"""Create multiple entities associated with different targets."""
await async_setup_component(hass, domain, {})
config_entry = MockConfigEntry(domain="test")
config_entry.add_to_hass(hass)
floor_reg = fr.async_get(hass)
floor = floor_reg.async_create("Test Floor")
area_reg = ar.async_get(hass)
area = area_reg.async_create("Test Area", floor_id=floor.floor_id)
label_reg = lr.async_get(hass)
label = label_reg.async_create("Test Label")
device = dr.DeviceEntry(id="test_device", area_id=area.id, labels={label.label_id})
mock_device_registry(hass, {device.id: device})
entity_reg = er.async_get(hass)
# Entity associated with area
entity_area = entity_reg.async_get_or_create(
domain=domain,
platform="test",
unique_id=f"{domain}_area",
suggested_object_id=f"area_{domain}",
)
entity_reg.async_update_entity(entity_area.entity_id, area_id=area.id)
# Entity associated with device
entity_reg.async_get_or_create(
domain=domain,
platform="test",
unique_id=f"{domain}_device",
suggested_object_id=f"device_{domain}",
device_id=device.id,
)
# Entity associated with label
entity_label = entity_reg.async_get_or_create(
domain=domain,
platform="test",
unique_id=f"{domain}_label",
suggested_object_id=f"label_{domain}",
)
entity_reg.async_update_entity(entity_label.entity_id, labels={label.label_id})
# Return all available entities
return [
f"{domain}.standalone_{domain}",
f"{domain}.label_{domain}",
f"{domain}.area_{domain}",
f"{domain}.device_{domain}",
]
def parametrize_target_entities(domain: str) -> list[tuple[dict, str, int]]:
"""Parametrize target entities for different target types.
Meant to be used with target_entities.
"""
return [
(
{CONF_ENTITY_ID: f"{domain}.standalone_{domain}"},
f"{domain}.standalone_{domain}",
1,
),
({ATTR_LABEL_ID: "test_label"}, f"{domain}.label_{domain}", 2),
({ATTR_AREA_ID: "test_area"}, f"{domain}.area_{domain}", 2),
({ATTR_FLOOR_ID: "test_floor"}, f"{domain}.area_{domain}", 2),
({ATTR_LABEL_ID: "test_label"}, f"{domain}.device_{domain}", 2),
({ATTR_AREA_ID: "test_area"}, f"{domain}.device_{domain}", 2),
({ATTR_FLOOR_ID: "test_floor"}, f"{domain}.device_{domain}", 2),
({ATTR_DEVICE_ID: "test_device"}, f"{domain}.device_{domain}", 1),
]
def parametrize_trigger_states(
trigger: str, target_state: str, other_state: str
) -> list[tuple[str, str | None, list[tuple[str, int]]]]:
"""Parametrize states and expected service call counts.
Returns a list of tuples with (trigger, initial_state, list of states),
where states is a list of tuples (state to set, expected service call count).
"""
return [
# Initial state None
(
trigger,
None,
[(target_state, 0), (other_state, 0), (target_state, 1)],
),
# Initial state different from target state
(
trigger,
other_state,
[(target_state, 1), (other_state, 0), (target_state, 1)],
),
# Initial state same as target state
(
trigger,
target_state,
[(target_state, 0), (other_state, 0), (target_state, 1)],
),
# Initial state unavailable / unknown
(
trigger,
STATE_UNAVAILABLE,
[(target_state, 0), (other_state, 0), (target_state, 1)],
),
(
trigger,
STATE_UNKNOWN,
[(target_state, 0), (other_state, 0), (target_state, 1)],
),
]
def parametrize_attribute_trigger_states(
trigger: str, state: str, attribute: str, target_state: str, other_state: str
) -> list[
tuple[str, tuple[str | None, dict], list[tuple[tuple[str | None, dict], int]]]
]:
"""Parametrize states and expected service call counts.
Returns a list of tuples with (trigger, initial_state, list of states),
where states is a list of tuples (state to set, expected service call count).
The initial_state and state to set are tuples of (state, {attribute: value}).
"""
return [
# Initial state None
(
trigger,
(None, {}),
[
((state, {attribute: target_state}), 0),
((state, {}), 0),
((state, {attribute: target_state}), 1),
],
),
# No initial state attribute
(
trigger,
(state, {}),
[
((state, {attribute: target_state}), 1),
((state, {}), 0),
((state, {attribute: target_state}), 1),
],
),
# Initial state attribute different from target state
(
trigger,
(state, {attribute: other_state}),
[
((state, {attribute: target_state}), 1),
((state, {}), 0),
((state, {attribute: target_state}), 1),
],
),
# Initial state attribute same as target state
(
trigger,
(state, {attribute: target_state}),
[
((state, {attribute: target_state}), 0),
((state, {}), 0),
((state, {attribute: target_state}), 1),
],
),
# Initial state unavailable / unknown
(
trigger,
(STATE_UNAVAILABLE, {}),
[
((state, {attribute: target_state}), 0),
((state, {}), 0),
((state, {attribute: target_state}), 1),
],
),
(
trigger,
(STATE_UNKNOWN, {}),
[
((state, {attribute: target_state}), 0),
((state, {}), 0),
((state, {attribute: target_state}), 1),
],
),
]

View File

@@ -0,0 +1,373 @@
"""Test climate trigger."""
import pytest
from homeassistant.components import automation
from homeassistant.components.climate.const import (
ATTR_HVAC_ACTION,
HVACAction,
HVACMode,
)
from homeassistant.const import CONF_ENTITY_ID, CONF_OPTIONS, CONF_PLATFORM, CONF_TARGET
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.setup import async_setup_component
from tests.components import (
parametrize_attribute_trigger_states,
parametrize_target_entities,
parametrize_trigger_states,
target_entities,
)
@pytest.fixture(autouse=True, name="stub_blueprint_populate")
def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder."""
@pytest.fixture
async def target_climates(hass: HomeAssistant) -> None:
"""Create multiple climate entities associated with different targets."""
return await target_entities(hass, "climate")
def set_or_remove_state(
hass: HomeAssistant,
entity_id: str,
state: str | None,
attributes: dict | None = None,
) -> None:
"""Set or clear the state of an entity."""
if state is None:
hass.states.async_remove(entity_id)
else:
hass.states.async_set(entity_id, state, attributes, force_update=True)
async def setup_automation(
hass: HomeAssistant, trigger: str, trigger_options: dict, trigger_target: dict
) -> None:
"""Set up automation component with given config."""
await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {
CONF_PLATFORM: trigger,
CONF_OPTIONS: {**trigger_options},
CONF_TARGET: {**trigger_target},
},
"action": {
"service": "test.automation",
"data_template": {CONF_ENTITY_ID: "{{ trigger.entity_id }}"},
},
}
},
)
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("climate"),
)
@pytest.mark.parametrize(
("trigger", "initial_state", "states"),
[
*parametrize_trigger_states("climate.turned_off", HVACMode.OFF, HVACMode.HEAT),
],
)
async def test_climate_state_trigger_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_climates: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
initial_state: str | None,
states: list[tuple[str, int]],
) -> None:
"""Test that the climate state trigger fires when any climate state changes to a specific state."""
await async_setup_component(hass, "climate", {})
other_entity_ids = set(target_climates) - {entity_id}
# Set all climates, including the tested climate, to the initial state
for eid in target_climates:
set_or_remove_state(hass, eid, initial_state)
await hass.async_block_till_done()
await setup_automation(hass, trigger, {}, trigger_target_config)
for state, expected_calls in states:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == expected_calls
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Check if changing other climates also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * expected_calls
service_calls.clear()
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("climate"),
)
@pytest.mark.parametrize(
("trigger", "initial_state", "states"),
[
*parametrize_attribute_trigger_states(
"climate.started_heating",
HVACMode.OFF,
ATTR_HVAC_ACTION,
HVACAction.HEATING,
HVACAction.IDLE,
)
],
)
async def test_climate_state_attribute_trigger_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_climates: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
initial_state: tuple[str | None, dict],
states: list[tuple[tuple[str, dict], int]],
) -> None:
"""Test that the climate state trigger fires when any climate state changes to a specific state."""
await async_setup_component(hass, "climate", {})
other_entity_ids = set(target_climates) - {entity_id}
# Set all climates, including the tested climate, to the initial state
for eid in target_climates:
set_or_remove_state(hass, eid, *initial_state)
await hass.async_block_till_done()
await setup_automation(hass, trigger, {}, trigger_target_config)
for state, expected_calls in states:
set_or_remove_state(hass, entity_id, *state)
await hass.async_block_till_done()
assert len(service_calls) == expected_calls
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Check if changing other climates also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, *state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * expected_calls
service_calls.clear()
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("climate"),
)
@pytest.mark.parametrize(
("trigger", "initial_state", "states"),
[
*parametrize_trigger_states("climate.turned_off", HVACMode.OFF, HVACMode.HEAT),
],
)
async def test_climate_state_trigger_behavior_first(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_climates: list[str],
trigger_target_config: dict,
entities_in_target: int,
entity_id: str,
trigger: str,
initial_state: str | None,
states: list[tuple[str, int, list[str]]],
) -> None:
"""Test that the climate state trigger fires when the first climate changes to a specific state."""
await async_setup_component(hass, "climate", {})
other_entity_ids = set(target_climates) - {entity_id}
# Set all climates, including the tested climate, to the initial state
for eid in target_climates:
set_or_remove_state(hass, eid, initial_state)
await hass.async_block_till_done()
await setup_automation(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state, expected_calls in states:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == expected_calls
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Triggering other climates should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == 0
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("climate"),
)
@pytest.mark.parametrize(
("trigger", "initial_state", "states"),
[
*parametrize_attribute_trigger_states(
"climate.started_heating",
HVACMode.OFF,
ATTR_HVAC_ACTION,
HVACAction.HEATING,
HVACAction.IDLE,
)
],
)
async def test_climate_state_attribute_trigger_behavior_first(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_climates: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
initial_state: tuple[str | None, dict],
states: list[tuple[tuple[str, dict], int]],
) -> None:
"""Test that the climate state trigger fires when any climate state changes to a specific state."""
await async_setup_component(hass, "climate", {})
other_entity_ids = set(target_climates) - {entity_id}
# Set all climates, including the tested climate, to the initial state
for eid in target_climates:
set_or_remove_state(hass, eid, *initial_state)
await hass.async_block_till_done()
await setup_automation(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state, expected_calls in states:
set_or_remove_state(hass, entity_id, *state)
await hass.async_block_till_done()
assert len(service_calls) == expected_calls
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Triggering other climates should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, *state)
await hass.async_block_till_done()
assert len(service_calls) == 0
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("climate"),
)
@pytest.mark.parametrize(
("trigger", "initial_state", "states"),
[
*parametrize_trigger_states("climate.turned_off", HVACMode.OFF, HVACMode.HEAT),
],
)
async def test_climate_state_trigger_behavior_last(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_climates: list[str],
trigger_target_config: dict,
entities_in_target: int,
entity_id: str,
trigger: str,
initial_state: str | None,
states: list[tuple[str, int]],
) -> None:
"""Test that the climate state trigger fires when the last climate changes to a specific state."""
await async_setup_component(hass, "climate", {})
other_entity_ids = set(target_climates) - {entity_id}
# Set all climates, including the tested climate, to the initial state
for eid in target_climates:
set_or_remove_state(hass, eid, initial_state)
await hass.async_block_till_done()
await setup_automation(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state, expected_calls in states:
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == 0
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == expected_calls
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("climate"),
)
@pytest.mark.parametrize(
("trigger", "initial_state", "states"),
[
*parametrize_attribute_trigger_states(
"climate.started_heating",
HVACMode.OFF,
ATTR_HVAC_ACTION,
HVACAction.HEATING,
HVACAction.IDLE,
)
],
)
async def test_climate_state_attribute_trigger_behavior_last(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_climates: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
initial_state: tuple[str | None, dict],
states: list[tuple[tuple[str, dict], int]],
) -> None:
"""Test that the climate state trigger fires when any climate state changes to a specific state."""
await async_setup_component(hass, "climate", {})
other_entity_ids = set(target_climates) - {entity_id}
# Set all climates, including the tested climate, to the initial state
for eid in target_climates:
set_or_remove_state(hass, eid, *initial_state)
await hass.async_block_till_done()
await setup_automation(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state, expected_calls in states:
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, *state)
await hass.async_block_till_done()
assert len(service_calls) == 0
set_or_remove_state(hass, entity_id, *state)
await hass.async_block_till_done()
assert len(service_calls) == expected_calls
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()

View File

@@ -0,0 +1,407 @@
"""Test cover trigger."""
import pytest
from homeassistant.components import automation
from homeassistant.components.cover import ATTR_CURRENT_POSITION, CoverState
from homeassistant.const import (
CONF_ENTITY_ID,
CONF_OPTIONS,
CONF_PLATFORM,
CONF_TARGET,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.setup import async_setup_component
from tests.components import parametrize_target_entities, target_entities
@pytest.fixture(autouse=True, name="stub_blueprint_populate")
def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder."""
@pytest.fixture
async def target_covers(hass: HomeAssistant) -> None:
"""Create multiple cover entities associated with different targets."""
return await target_entities(hass, "cover")
def set_or_remove_state(
hass: HomeAssistant,
entity_id: str,
state: str | None,
attributes: dict | None = None,
) -> None:
"""Set or clear the state of an entity."""
if state is None:
hass.states.async_remove(entity_id)
else:
hass.states.async_set(entity_id, state, attributes, force_update=True)
async def setup_automation(
hass: HomeAssistant, trigger: str, trigger_options: dict, trigger_target: dict
) -> None:
"""Set up automation component with given config."""
await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {
CONF_PLATFORM: trigger,
CONF_OPTIONS: {**trigger_options},
CONF_TARGET: {**trigger_target},
},
"action": {
"service": "test.automation",
"data_template": {CONF_ENTITY_ID: "{{ trigger.entity_id }}"},
},
}
},
)
def parametrize_opened_closed_trigger_states(
trigger: str,
trigger_options: dict,
device_class: str,
target_state: tuple[str, dict],
other_state: tuple[str, dict],
) -> list[
tuple[str, tuple[str | None, dict], list[tuple[tuple[str | None, dict], int]]]
]:
"""Parametrize states and expected service call counts.
Returns a list of tuples with (trigger, trigger_options, device_class,
initial_state, list of states), where states is a list of tuples
(state to set, expected service call count).
"""
return [
## TODO: Check what happens if attribute is missing
# Initial state None
(
trigger,
trigger_options,
device_class,
(None, {}),
[
(target_state, 0),
(other_state, 0),
# This doesn't trigger because the device class is not set
# when the trigger is created. We need to teach TargetStateChangeTracker
# about device classes.
(target_state, 0),
],
),
# Initial state different from target state
(
trigger,
trigger_options,
device_class,
other_state,
[
(target_state, 1),
(other_state, 0),
(target_state, 1),
],
),
# Initial state same as target state
(
trigger,
trigger_options,
device_class,
target_state,
[
(target_state, 0),
(other_state, 0),
(target_state, 1),
],
),
# Initial state unavailable / unknown
(
trigger,
trigger_options,
device_class,
(STATE_UNAVAILABLE, {}),
[
(target_state, 0),
(other_state, 0),
(target_state, 1),
],
),
(
trigger,
trigger_options,
device_class,
(STATE_UNKNOWN, {}),
[
(target_state, 0),
(other_state, 0),
(target_state, 1),
],
),
]
def parametrize_opened_trigger_states(
trigger: str, device_class: str
) -> list[
tuple[str, tuple[str | None, dict], list[tuple[tuple[str | None, dict], int]]]
]:
"""Parametrize states and expected service call counts.
Returns a list of tuples with (trigger, trigger_options, device_class,
initial_state, list of states), where states is a list of tuples
(state to set, expected service call count).
"""
return [
# States without current position attribute
*parametrize_opened_closed_trigger_states(
trigger,
{"fully_opened": True},
device_class,
(CoverState.OPEN, {}),
(CoverState.CLOSED, {}),
),
*parametrize_opened_closed_trigger_states(
trigger,
{"fully_opened": True},
device_class,
(CoverState.OPENING, {}),
(CoverState.CLOSED, {}),
),
*parametrize_opened_closed_trigger_states(
trigger,
{},
device_class,
(CoverState.OPEN, {}),
(CoverState.CLOSED, {}),
),
*parametrize_opened_closed_trigger_states(
trigger,
{},
device_class,
(CoverState.OPENING, {}),
(CoverState.CLOSED, {}),
),
# States with current position attribute
*parametrize_opened_closed_trigger_states(
trigger,
{"fully_opened": True},
device_class,
(CoverState.OPEN, {ATTR_CURRENT_POSITION: 100}),
(CoverState.OPEN, {ATTR_CURRENT_POSITION: 0}),
),
*parametrize_opened_closed_trigger_states(
trigger,
{"fully_opened": True},
device_class,
(CoverState.OPENING, {ATTR_CURRENT_POSITION: 100}),
(CoverState.OPENING, {ATTR_CURRENT_POSITION: 0}),
),
*parametrize_opened_closed_trigger_states(
trigger,
{},
device_class,
(CoverState.OPEN, {ATTR_CURRENT_POSITION: 1}),
(CoverState.CLOSED, {ATTR_CURRENT_POSITION: 0}),
),
*parametrize_opened_closed_trigger_states(
trigger,
{},
device_class,
(CoverState.OPENING, {ATTR_CURRENT_POSITION: 1}),
(CoverState.CLOSED, {ATTR_CURRENT_POSITION: 0}),
),
]
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("cover"),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "device_class", "initial_state", "states"),
[
*parametrize_opened_trigger_states("cover.garage_opened", "garage"),
# No initial state attribute.
(
"cover.garage_opened",
{"fully_opened": True},
"garage",
(CoverState.OPEN, {}),
[
((CoverState.OPEN, {ATTR_CURRENT_POSITION: 100}), 1),
((CoverState.OPEN, {ATTR_CURRENT_POSITION: 0}), 0),
((CoverState.OPEN, {ATTR_CURRENT_POSITION: 100}), 1),
],
),
],
)
async def test_cover_state_attribute_trigger_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_covers: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
trigger_options: dict,
device_class: str,
initial_state: tuple[str | None, dict],
states: list[tuple[tuple[str, dict], int]],
) -> None:
"""Test that the cover state trigger fires when any cover state changes to a specific state."""
await async_setup_component(hass, "cover", {})
other_entity_ids = set(target_covers) - {entity_id}
# Set all covers, including the tested cover, to the initial state
for eid in target_covers:
set_or_remove_state(
hass, eid, initial_state[0], initial_state[1] | {"device_class": "garage"}
)
await hass.async_block_till_done()
await setup_automation(hass, trigger, trigger_options, trigger_target_config)
for state, expected_calls in states:
set_or_remove_state(
hass, entity_id, state[0], state[1] | {"device_class": "garage"}
)
await hass.async_block_till_done()
assert len(service_calls) == expected_calls
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Check if changing other covers also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(
hass, other_entity_id, state[0], state[1] | {"device_class": "garage"}
)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * expected_calls
service_calls.clear()
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("cover"),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "device_class", "initial_state", "states"),
[
*parametrize_opened_trigger_states("cover.garage_opened", "garage"),
],
)
async def test_cover_state_attribute_trigger_behavior_first(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_covers: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
trigger_options: dict,
device_class: str,
initial_state: tuple[str | None, dict],
states: list[tuple[tuple[str, dict], int]],
) -> None:
"""Test that the cover state trigger fires when the first cover state changes to a specific state."""
await async_setup_component(hass, "cover", {})
other_entity_ids = set(target_covers) - {entity_id}
# Set all covers, including the tested cover, to the initial state
for eid in target_covers:
set_or_remove_state(
hass, eid, initial_state[0], initial_state[1] | {"device_class": "garage"}
)
await hass.async_block_till_done()
await setup_automation(
hass,
trigger,
{"behavior": "first"} | trigger_options,
trigger_target_config,
)
for state, expected_calls in states:
set_or_remove_state(
hass, entity_id, state[0], state[1] | {"device_class": "garage"}
)
await hass.async_block_till_done()
assert len(service_calls) == expected_calls
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Triggering other covers should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(
hass, other_entity_id, state[0], state[1] | {"device_class": "garage"}
)
await hass.async_block_till_done()
assert len(service_calls) == 0
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("cover"),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "device_class", "initial_state", "states"),
[
*parametrize_opened_trigger_states("cover.garage_opened", "garage"),
],
)
async def test_cover_state_attribute_trigger_behavior_last(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_covers: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
trigger_options: dict,
device_class: str,
initial_state: tuple[str | None, dict],
states: list[tuple[tuple[str, dict], int]],
) -> None:
"""Test that the cover state trigger fires when the last cover state changes to a specific state."""
await async_setup_component(hass, "cover", {})
other_entity_ids = set(target_covers) - {entity_id}
# Set all covers, including the tested cover, to the initial state
for eid in target_covers:
set_or_remove_state(
hass, eid, initial_state[0], initial_state[1] | {"device_class": "garage"}
)
await hass.async_block_till_done()
await setup_automation(
hass, trigger, {"behavior": "last"} | trigger_options, trigger_target_config
)
for state, expected_calls in states:
for other_entity_id in other_entity_ids:
set_or_remove_state(
hass, other_entity_id, state[0], state[1] | {"device_class": "garage"}
)
await hass.async_block_till_done()
assert len(service_calls) == 0
set_or_remove_state(
hass, entity_id, state[0], state[1] | {"device_class": "garage"}
)
await hass.async_block_till_done()
assert len(service_calls) == expected_calls
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()

View File

@@ -0,0 +1,203 @@
"""Test light conditions."""
import pytest
from homeassistant.components import automation
from homeassistant.const import (
ATTR_LABEL_ID,
CONF_CONDITION,
CONF_OPTIONS,
CONF_TARGET,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.helpers import entity_registry as er, label_registry as lr
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
@pytest.fixture(autouse=True, name="stub_blueprint_populate")
def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder."""
@pytest.fixture
async def label_entities(hass: HomeAssistant) -> list[str]:
"""Create multiple entities associated with labels."""
await async_setup_component(hass, "light", {})
config_entry = MockConfigEntry(domain="test_labels")
config_entry.add_to_hass(hass)
label_reg = lr.async_get(hass)
label = label_reg.async_create("Test Label")
entity_reg = er.async_get(hass)
for i in range(3):
light_entity = entity_reg.async_get_or_create(
domain="light",
platform="test",
unique_id=f"label_light_{i}",
suggested_object_id=f"label_light_{i}",
)
entity_reg.async_update_entity(light_entity.entity_id, labels={label.label_id})
# Also create switches to test that they don't impact the conditions
for i in range(2):
switch_entity = entity_reg.async_get_or_create(
domain="switch",
platform="test",
unique_id=f"label_switch_{i}",
suggested_object_id=f"label_switch_{i}",
)
entity_reg.async_update_entity(switch_entity.entity_id, labels={label.label_id})
return [
"light.label_light_0",
"light.label_light_1",
"light.label_light_2",
]
async def setup_automation_with_light_condition(
hass: HomeAssistant,
*,
condition: str,
target: dict,
behavior: str,
) -> None:
"""Set up automation with light state condition."""
await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {"platform": "event", "event_type": "test_event"},
"condition": {
CONF_CONDITION: condition,
CONF_TARGET: target,
CONF_OPTIONS: {"behavior": behavior},
},
"action": {
"service": "test.automation",
},
}
},
)
async def has_calls_after_trigger(
hass: HomeAssistant, service_calls: list[ServiceCall]
) -> bool:
"""Check if there are service calls after the trigger event."""
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
has_calls = len(service_calls) == 1
service_calls.clear()
return has_calls
@pytest.mark.parametrize(
("condition", "state", "reverse_state"),
[("light.is_on", STATE_ON, STATE_OFF), ("light.is_off", STATE_OFF, STATE_ON)],
)
async def test_light_state_condition_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
label_entities: list[str],
condition: str,
state: str,
reverse_state: str,
) -> None:
"""Test the light state condition with the 'any' behavior."""
await async_setup_component(hass, "light", {})
for entity_id in label_entities:
hass.states.async_set(entity_id, reverse_state)
await setup_automation_with_light_condition(
hass,
condition=condition,
target={ATTR_LABEL_ID: "test_label", "entity_id": "light.nonexistent"},
behavior="any",
)
# Set state for two switches to ensure that they don't impact the condition
hass.states.async_set("switch.label_switch_1", STATE_OFF)
hass.states.async_set("switch.label_switch_2", STATE_ON)
# No lights on the condition state
assert not await has_calls_after_trigger(hass, service_calls)
# Set one light to the condition state -> condition pass
hass.states.async_set(label_entities[0], state)
assert await has_calls_after_trigger(hass, service_calls)
# Set all lights to the condition state -> condition pass
for entity_id in label_entities:
hass.states.async_set(entity_id, state)
assert await has_calls_after_trigger(hass, service_calls)
# Set one light to unavailable -> condition pass
hass.states.async_set(label_entities[0], STATE_UNAVAILABLE)
assert await has_calls_after_trigger(hass, service_calls)
# Set all lights to unavailable -> condition fail
for entity_id in label_entities:
hass.states.async_set(entity_id, STATE_UNAVAILABLE)
assert not await has_calls_after_trigger(hass, service_calls)
@pytest.mark.parametrize(
("condition", "state", "reverse_state"),
[("light.is_on", STATE_ON, STATE_OFF), ("light.is_off", STATE_OFF, STATE_ON)],
)
async def test_light_state_condition_behavior_all(
hass: HomeAssistant,
service_calls: list[ServiceCall],
label_entities: list[str],
condition: str,
state: str,
reverse_state: str,
) -> None:
"""Test the light state condition with the 'all' behavior."""
await async_setup_component(hass, "light", {})
# Set state for two switches to ensure that they don't impact the condition
hass.states.async_set("switch.label_switch_1", STATE_OFF)
hass.states.async_set("switch.label_switch_2", STATE_ON)
for entity_id in label_entities:
hass.states.async_set(entity_id, reverse_state)
await setup_automation_with_light_condition(
hass,
condition=condition,
target={ATTR_LABEL_ID: "test_label", "entity_id": "light.nonexistent"},
behavior="all",
)
# No lights on the condition state
assert not await has_calls_after_trigger(hass, service_calls)
# Set one light to the condition state -> condition fail
hass.states.async_set(label_entities[0], state)
assert not await has_calls_after_trigger(hass, service_calls)
# Set all lights to the condition state -> condition pass
for entity_id in label_entities:
hass.states.async_set(entity_id, state)
assert await has_calls_after_trigger(hass, service_calls)
# Set one light to unavailable -> condition still pass
hass.states.async_set(label_entities[0], STATE_UNAVAILABLE)
assert await has_calls_after_trigger(hass, service_calls)
# Set all lights to unavailable -> condition passes
for entity_id in label_entities:
hass.states.async_set(entity_id, STATE_UNAVAILABLE)
assert await has_calls_after_trigger(hass, service_calls)

View File

@@ -0,0 +1,210 @@
"""Test light trigger."""
import pytest
from homeassistant.components import automation
from homeassistant.const import (
CONF_ENTITY_ID,
CONF_OPTIONS,
CONF_PLATFORM,
CONF_TARGET,
STATE_OFF,
STATE_ON,
)
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.setup import async_setup_component
from tests.components import (
parametrize_target_entities,
parametrize_trigger_states,
target_entities,
)
@pytest.fixture(autouse=True, name="stub_blueprint_populate")
def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder."""
@pytest.fixture
async def target_lights(hass: HomeAssistant) -> None:
"""Create multiple light entities associated with different targets."""
return await target_entities(hass, "light")
def set_or_remove_state(hass: HomeAssistant, entity_id: str, state: str | None) -> None:
"""Set or clear the state of an entity."""
if state is None:
hass.states.async_remove(entity_id)
else:
hass.states.async_set(entity_id, state, force_update=True)
async def setup_automation(
hass: HomeAssistant, trigger: str, trigger_options: dict, trigger_target: dict
) -> None:
"""Set up automation component with given config."""
await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {
CONF_PLATFORM: trigger,
CONF_OPTIONS: {**trigger_options},
CONF_TARGET: {**trigger_target},
},
"action": {
"service": "test.automation",
"data_template": {CONF_ENTITY_ID: "{{ trigger.entity_id }}"},
},
}
},
)
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("light"),
)
@pytest.mark.parametrize(
("trigger", "initial_state", "states"),
[
*parametrize_trigger_states("light.turned_on", STATE_ON, STATE_OFF),
*parametrize_trigger_states("light.turned_off", STATE_OFF, STATE_ON),
],
)
async def test_light_state_trigger_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_lights: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
initial_state: str | None,
states: list[tuple[str, int]],
) -> None:
"""Test that the light state trigger fires when any light state changes to a specific state."""
await async_setup_component(hass, "light", {})
other_entity_ids = set(target_lights) - {entity_id}
# Set all lights, including the tested light, to the initial state
for eid in target_lights:
set_or_remove_state(hass, eid, initial_state)
await hass.async_block_till_done()
await setup_automation(hass, trigger, {}, trigger_target_config)
for state, expected_calls in states:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == expected_calls
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Check if changing other lights also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * expected_calls
service_calls.clear()
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("light"),
)
@pytest.mark.parametrize(
("trigger", "initial_state", "states"),
[
*parametrize_trigger_states("light.turned_on", STATE_ON, STATE_OFF),
*parametrize_trigger_states("light.turned_off", STATE_OFF, STATE_ON),
],
)
async def test_light_state_trigger_behavior_first(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_lights: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
initial_state: str | None,
states: list[tuple[str, int, list[str]]],
) -> None:
"""Test that the light state trigger fires when the first light changes to a specific state."""
await async_setup_component(hass, "light", {})
other_entity_ids = set(target_lights) - {entity_id}
# Set all lights, including the tested light, to the initial state
for eid in target_lights:
set_or_remove_state(hass, eid, initial_state)
await hass.async_block_till_done()
await setup_automation(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state, expected_calls in states:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == expected_calls
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Triggering other lights should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == 0
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("light"),
)
@pytest.mark.parametrize(
("trigger", "initial_state", "states"),
[
*parametrize_trigger_states("light.turned_on", STATE_ON, STATE_OFF),
*parametrize_trigger_states("light.turned_off", STATE_OFF, STATE_ON),
],
)
async def test_light_state_trigger_behavior_last(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_lights: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
initial_state: str | None,
states: list[tuple[str, int]],
) -> None:
"""Test that the light state trigger fires when the last light changes to a specific state."""
await async_setup_component(hass, "light", {})
other_entity_ids = set(target_lights) - {entity_id}
# Set all lights, including the tested light, to the initial state
for eid in target_lights:
set_or_remove_state(hass, eid, initial_state)
await hass.async_block_till_done()
await setup_automation(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state, expected_calls in states:
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == 0
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == expected_calls
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()

View File

@@ -450,10 +450,10 @@ async def test_caching(hass: HomeAssistant) -> None:
side_effect=translation.build_resources,
) as mock_build_resources:
load1 = await translation.async_get_translations(hass, "en", "entity_component")
assert len(mock_build_resources.mock_calls) == 7
assert len(mock_build_resources.mock_calls) == 9
load2 = await translation.async_get_translations(hass, "en", "entity_component")
assert len(mock_build_resources.mock_calls) == 7
assert len(mock_build_resources.mock_calls) == 9
assert load1 == load2