mirror of
https://github.com/home-assistant/core.git
synced 2025-11-13 21:10:25 +00:00
Compare commits
29 Commits
claude/tri
...
add_trigge
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
55eb01c894 | ||
|
|
5ccd2c20a4 | ||
|
|
333075246f | ||
|
|
3c2573e5b6 | ||
|
|
edf82db057 | ||
|
|
37644511f6 | ||
|
|
3685d0f7c2 | ||
|
|
3dabfeb329 | ||
|
|
8e7d2d7108 | ||
|
|
2fe4a1164b | ||
|
|
05175294f6 | ||
|
|
e2ddfb8782 | ||
|
|
f1cc133ff6 | ||
|
|
0cf97cf577 | ||
|
|
38cea2e5f0 | ||
|
|
71876d5b34 | ||
|
|
0f780254e1 | ||
|
|
9e40972b11 | ||
|
|
07ef61dd8d | ||
|
|
1bf6771a54 | ||
|
|
e7a7cb829e | ||
|
|
6f6b2f1ad3 | ||
|
|
1cc4890f75 | ||
|
|
d3dd9b26c9 | ||
|
|
a64d61df05 | ||
|
|
e7c6c5311d | ||
|
|
72a524c868 | ||
|
|
b437113f31 | ||
|
|
e0e263d3b5 |
@@ -96,5 +96,13 @@
|
||||
"turn_on": {
|
||||
"service": "mdi:power-on"
|
||||
}
|
||||
},
|
||||
"triggers": {
|
||||
"started_heating": {
|
||||
"trigger": "mdi:fire"
|
||||
},
|
||||
"turned_off": {
|
||||
"trigger": "mdi:power-off"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
23
homeassistant/components/climate/trigger.py
Normal file
23
homeassistant/components/climate/trigger.py
Normal 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
|
||||
19
homeassistant/components/climate/triggers.yaml
Normal file
19
homeassistant/components/climate/triggers.yaml
Normal 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
|
||||
@@ -108,5 +108,10 @@
|
||||
"toggle_cover_tilt": {
|
||||
"service": "mdi:arrow-top-right-bottom-left"
|
||||
}
|
||||
},
|
||||
"triggers": {
|
||||
"garage_opened": {
|
||||
"trigger": "mdi:garage-open"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
119
homeassistant/components/cover/trigger.py
Normal file
119
homeassistant/components/cover/trigger.py
Normal 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
|
||||
21
homeassistant/components/cover/triggers.yaml
Normal file
21
homeassistant/components/cover/triggers.yaml
Normal 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:
|
||||
@@ -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]),
|
||||
}
|
||||
)
|
||||
|
||||
131
homeassistant/components/light/condition.py
Normal file
131
homeassistant/components/light/condition.py
Normal 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
|
||||
28
homeassistant/components/light/conditions.yaml
Normal file
28
homeassistant/components/light/conditions.yaml
Normal 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
|
||||
@@ -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"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
17
homeassistant/components/light/trigger.py
Normal file
17
homeassistant/components/light/trigger.py
Normal 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
|
||||
31
homeassistant/components/light/triggers.yaml
Normal file
31
homeassistant/components/light/triggers.yaml
Normal 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
|
||||
@@ -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
|
||||
|
||||
@@ -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),
|
||||
],
|
||||
),
|
||||
]
|
||||
|
||||
373
tests/components/climate/test_trigger.py
Normal file
373
tests/components/climate/test_trigger.py
Normal 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()
|
||||
407
tests/components/cover/test_trigger.py
Normal file
407
tests/components/cover/test_trigger.py
Normal 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()
|
||||
203
tests/components/light/test_condition.py
Normal file
203
tests/components/light/test_condition.py
Normal 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)
|
||||
210
tests/components/light/test_trigger.py
Normal file
210
tests/components/light/test_trigger.py
Normal 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()
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user