Compare commits

..

3 Commits

Author SHA1 Message Date
Erik Montnemery
f0981c00e5 Revert "Return target in trigger description command (#156766)"
This reverts commit 343ea1b82d.
2025-11-19 07:58:17 +01:00
Andre Lengwenus
82d3190016 Bump pypck to 0.9.5 (#156847) 2025-11-19 06:52:29 +01:00
omrishiv
d8cbcc1977 Bump pylutron-caseta to 0.26.0 (#156825)
Signed-off-by: omrishiv <327609+omrishiv@users.noreply.github.com>
2025-11-18 23:01:34 +01:00
63 changed files with 46 additions and 4700 deletions

View File

@@ -36,28 +36,5 @@
"alarm_trigger": {
"service": "mdi:bell-ring"
}
},
"triggers": {
"armed": {
"trigger": "mdi:shield"
},
"armed_away": {
"trigger": "mdi:shield-lock"
},
"armed_home": {
"trigger": "mdi:shield-home"
},
"armed_night": {
"trigger": "mdi:shield-moon"
},
"armed_vacation": {
"trigger": "mdi:shield-airplane"
},
"disarmed": {
"trigger": "mdi:shield-off"
},
"triggered": {
"trigger": "mdi:bell-ring"
}
}
}

View File

@@ -1,8 +1,4 @@
{
"common": {
"trigger_behavior_description": "The behavior of the targeted alarms to trigger on.",
"trigger_behavior_name": "Behavior"
},
"device_automation": {
"action_type": {
"arm_away": "Arm {entity_name} away",
@@ -75,15 +71,6 @@
"message": "Arming requires a code but none was given for {entity_id}."
}
},
"selector": {
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
}
}
},
"services": {
"alarm_arm_away": {
"description": "Arms the alarm in the away mode.",
@@ -156,84 +143,5 @@
"name": "Trigger"
}
},
"title": "Alarm control panel",
"triggers": {
"armed": {
"description": "Triggers when an alarm is armed.",
"description_configured": "[%key:component::alarm_control_panel::triggers::armed::description%]",
"fields": {
"behavior": {
"description": "[%key:component::alarm_control_panel::common::trigger_behavior_description%]",
"name": "[%key:component::alarm_control_panel::common::trigger_behavior_name%]"
}
},
"name": "When an alarm is armed"
},
"armed_away": {
"description": "Triggers when an alarm is armed away.",
"description_configured": "[%key:component::alarm_control_panel::triggers::armed_away::description%]",
"fields": {
"behavior": {
"description": "[%key:component::alarm_control_panel::common::trigger_behavior_description%]",
"name": "[%key:component::alarm_control_panel::common::trigger_behavior_name%]"
}
},
"name": "When an alarm is armed away"
},
"armed_home": {
"description": "Triggers when an alarm is armed home.",
"description_configured": "[%key:component::alarm_control_panel::triggers::armed_home::description%]",
"fields": {
"behavior": {
"description": "[%key:component::alarm_control_panel::common::trigger_behavior_description%]",
"name": "[%key:component::alarm_control_panel::common::trigger_behavior_name%]"
}
},
"name": "When an alarm is armed home"
},
"armed_night": {
"description": "Triggers when an alarm is armed night.",
"description_configured": "[%key:component::alarm_control_panel::triggers::armed_night::description%]",
"fields": {
"behavior": {
"description": "[%key:component::alarm_control_panel::common::trigger_behavior_description%]",
"name": "[%key:component::alarm_control_panel::common::trigger_behavior_name%]"
}
},
"name": "When an alarm is armed night"
},
"armed_vacation": {
"description": "Triggers when an alarm is armed vacation.",
"description_configured": "[%key:component::alarm_control_panel::triggers::armed_vacation::description%]",
"fields": {
"behavior": {
"description": "[%key:component::alarm_control_panel::common::trigger_behavior_description%]",
"name": "[%key:component::alarm_control_panel::common::trigger_behavior_name%]"
}
},
"name": "When an alarm is armed vacation"
},
"disarmed": {
"description": "Triggers when an alarm is disarmed.",
"description_configured": "[%key:component::alarm_control_panel::triggers::disarmed::description%]",
"fields": {
"behavior": {
"description": "[%key:component::alarm_control_panel::common::trigger_behavior_description%]",
"name": "[%key:component::alarm_control_panel::common::trigger_behavior_name%]"
}
},
"name": "When an alarm is disarmed"
},
"triggered": {
"description": "Triggers when an alarm is triggered.",
"description_configured": "[%key:component::alarm_control_panel::triggers::triggered::description%]",
"fields": {
"behavior": {
"description": "[%key:component::alarm_control_panel::common::trigger_behavior_description%]",
"name": "[%key:component::alarm_control_panel::common::trigger_behavior_name%]"
}
},
"name": "When an alarm is triggered"
}
}
"title": "Alarm control panel"
}

View File

@@ -1,99 +0,0 @@
"""Provides triggers for alarm control panels."""
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity import get_supported_features
from homeassistant.helpers.trigger import (
EntityStateTriggerBase,
Trigger,
make_conditional_entity_state_trigger,
make_entity_state_trigger,
)
from .const import DOMAIN, AlarmControlPanelEntityFeature, AlarmControlPanelState
def supports_feature(hass: HomeAssistant, entity_id: str, features: int) -> bool:
"""Get the device class of an entity or UNDEFINED if not found."""
try:
return bool(get_supported_features(hass, entity_id) & features)
except HomeAssistantError:
return False
class EntityStateTriggerRequiredFeatures(EntityStateTriggerBase):
"""Trigger for entity state changes."""
_required_features: int
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 supports_feature(self._hass, entity_id, self._required_features)
}
def make_entity_state_trigger_required_features(
domain: str, to_state: str, required_features: int
) -> type[EntityStateTriggerBase]:
"""Create an entity state trigger class."""
class CustomTrigger(EntityStateTriggerRequiredFeatures):
"""Trigger for entity state changes."""
_domain = domain
_to_state = to_state
_required_features = required_features
return CustomTrigger
TRIGGERS: dict[str, type[Trigger]] = {
"armed": make_conditional_entity_state_trigger(
DOMAIN,
from_states={
AlarmControlPanelState.ARMING,
AlarmControlPanelState.DISARMED,
AlarmControlPanelState.DISARMING,
AlarmControlPanelState.PENDING,
AlarmControlPanelState.TRIGGERED,
},
to_states={
AlarmControlPanelState.ARMED_AWAY,
AlarmControlPanelState.ARMED_CUSTOM_BYPASS,
AlarmControlPanelState.ARMED_HOME,
AlarmControlPanelState.ARMED_NIGHT,
AlarmControlPanelState.ARMED_VACATION,
},
),
"armed_away": make_entity_state_trigger_required_features(
DOMAIN,
AlarmControlPanelState.ARMED_AWAY,
AlarmControlPanelEntityFeature.ARM_AWAY,
),
"armed_home": make_entity_state_trigger_required_features(
DOMAIN,
AlarmControlPanelState.ARMED_HOME,
AlarmControlPanelEntityFeature.ARM_HOME,
),
"armed_night": make_entity_state_trigger_required_features(
DOMAIN,
AlarmControlPanelState.ARMED_NIGHT,
AlarmControlPanelEntityFeature.ARM_NIGHT,
),
"armed_vacation": make_entity_state_trigger_required_features(
DOMAIN,
AlarmControlPanelState.ARMED_VACATION,
AlarmControlPanelEntityFeature.ARM_VACATION,
),
"disarmed": make_entity_state_trigger(DOMAIN, AlarmControlPanelState.DISARMED),
"triggered": make_entity_state_trigger(DOMAIN, AlarmControlPanelState.TRIGGERED),
}
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
"""Return the triggers for alarm control panels."""
return TRIGGERS

View File

@@ -1,53 +0,0 @@
.trigger_common: &trigger_common
target:
entity:
domain: alarm_control_panel
fields: &trigger_common_fields
behavior:
required: true
default: any
selector:
select:
options:
- first
- last
- any
translation_key: trigger_behavior
armed: *trigger_common
armed_away:
fields: *trigger_common_fields
target:
entity:
domain: alarm_control_panel
supported_features:
- alarm_control_panel.AlarmControlPanelEntityFeature.ARM_AWAY
armed_home:
fields: *trigger_common_fields
target:
entity:
domain: alarm_control_panel
supported_features:
- alarm_control_panel.AlarmControlPanelEntityFeature.ARM_HOME
armed_night:
fields: *trigger_common_fields
target:
entity:
domain: alarm_control_panel
supported_features:
- alarm_control_panel.AlarmControlPanelEntityFeature.ARM_NIGHT
armed_vacation:
fields: *trigger_common_fields
target:
entity:
domain: alarm_control_panel
supported_features:
- alarm_control_panel.AlarmControlPanelEntityFeature.ARM_VACATION
disarmed: *trigger_common
triggered: *trigger_common

View File

@@ -14,19 +14,5 @@
"start_conversation": {
"service": "mdi:forum"
}
},
"triggers": {
"idle": {
"trigger": "mdi:chat-sleep"
},
"listening": {
"trigger": "mdi:chat-question"
},
"processing": {
"trigger": "mdi:chat-processing"
},
"responding": {
"trigger": "mdi:chat-alert"
}
}
}

View File

@@ -1,8 +1,4 @@
{
"common": {
"trigger_behavior_description": "The behavior of the targeted assist satellites to trigger on.",
"trigger_behavior_name": "Behavior"
},
"entity_component": {
"_": {
"name": "Assist satellite",
@@ -20,13 +16,6 @@
"id": "Answer ID",
"sentences": "Sentences"
}
},
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
}
}
},
"services": {
@@ -109,51 +98,5 @@
"name": "Start conversation"
}
},
"title": "Assist satellite",
"triggers": {
"idle": {
"description": "Triggers when an assist satellite becomes idle.",
"description_configured": "[%key:component::assist_satellite::triggers::idle::description%]",
"fields": {
"behavior": {
"description": "[%key:component::assist_satellite::common::trigger_behavior_description%]",
"name": "[%key:component::assist_satellite::common::trigger_behavior_name%]"
}
},
"name": "When an assist satellite becomes idle"
},
"listening": {
"description": "Triggers when an assist satellite starts listening.",
"description_configured": "[%key:component::assist_satellite::triggers::listening::description%]",
"fields": {
"behavior": {
"description": "[%key:component::assist_satellite::common::trigger_behavior_description%]",
"name": "[%key:component::assist_satellite::common::trigger_behavior_name%]"
}
},
"name": "When an assist satellite starts listening"
},
"processing": {
"description": "Triggers when an assist satellite is processing.",
"description_configured": "[%key:component::assist_satellite::triggers::processing::description%]",
"fields": {
"behavior": {
"description": "[%key:component::assist_satellite::common::trigger_behavior_description%]",
"name": "[%key:component::assist_satellite::common::trigger_behavior_name%]"
}
},
"name": "When an assist satellite is processing"
},
"responding": {
"description": "Triggers when an assist satellite is responding.",
"description_configured": "[%key:component::assist_satellite::triggers::responding::description%]",
"fields": {
"behavior": {
"description": "[%key:component::assist_satellite::common::trigger_behavior_description%]",
"name": "[%key:component::assist_satellite::common::trigger_behavior_name%]"
}
},
"name": "When an assist satellite is responding"
}
}
"title": "Assist satellite"
}

View File

@@ -1,19 +0,0 @@
"""Provides triggers for assist satellites."""
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import Trigger, make_entity_state_trigger
from .const import DOMAIN
from .entity import AssistSatelliteState
TRIGGERS: dict[str, type[Trigger]] = {
"idle": make_entity_state_trigger(DOMAIN, AssistSatelliteState.IDLE),
"listening": make_entity_state_trigger(DOMAIN, AssistSatelliteState.LISTENING),
"processing": make_entity_state_trigger(DOMAIN, AssistSatelliteState.PROCESSING),
"responding": make_entity_state_trigger(DOMAIN, AssistSatelliteState.RESPONDING),
}
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
"""Return the triggers for assist satellites."""
return TRIGGERS

View File

@@ -1,20 +0,0 @@
.trigger_common: &trigger_common
target:
entity:
domain: assist_satellite
fields:
behavior:
required: true
default: any
selector:
select:
options:
- first
- last
- any
translation_key: trigger_behavior
idle: *trigger_common
listening: *trigger_common
processing: *trigger_common
responding: *trigger_common

View File

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

View File

@@ -1,8 +1,4 @@
{
"common": {
"trigger_behavior_description": "The behavior of the targeted climates to trigger on.",
"trigger_behavior_name": "Behavior"
},
"device_automation": {
"action_type": {
"set_hvac_mode": "Change HVAC mode on {entity_name}",
@@ -191,13 +187,6 @@
"heat_cool": "Heat/cool",
"off": "[%key:common::state::off%]"
}
},
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
}
}
},
"services": {
@@ -296,40 +285,5 @@
"name": "[%key:common::action::turn_on%]"
}
},
"title": "Climate",
"triggers": {
"started_heating": {
"description": "Triggers when a climate starts to heat.",
"description_configured": "[%key:component::climate::triggers::started_heating::description%]",
"fields": {
"behavior": {
"description": "[%key:component::climate::common::trigger_behavior_description%]",
"name": "[%key:component::climate::common::trigger_behavior_name%]"
}
},
"name": "When a climate starts to heat"
},
"turned_off": {
"description": "Triggers when a climate is turned off.",
"description_configured": "[%key:component::climate::triggers::turned_off::description%]",
"fields": {
"behavior": {
"description": "[%key:component::climate::common::trigger_behavior_description%]",
"name": "[%key:component::climate::common::trigger_behavior_name%]"
}
},
"name": "When a climate is turned off"
},
"turned_on": {
"description": "Triggers when a climate is turned on.",
"description_configured": "[%key:component::climate::triggers::turned_on::description%]",
"fields": {
"behavior": {
"description": "[%key:component::climate::common::trigger_behavior_description%]",
"name": "[%key:component::climate::common::trigger_behavior_name%]"
}
},
"name": "When a climate is turned on"
}
}
"title": "Climate"
}

View File

@@ -1,37 +0,0 @@
"""Provides triggers for climates."""
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import (
Trigger,
make_conditional_entity_state_trigger,
make_entity_state_attribute_trigger,
make_entity_state_trigger,
)
from .const import ATTR_HVAC_ACTION, DOMAIN, HVACAction, HVACMode
TRIGGERS: dict[str, type[Trigger]] = {
"turned_off": make_entity_state_trigger(DOMAIN, HVACMode.OFF),
"turned_on": make_conditional_entity_state_trigger(
DOMAIN,
from_states={
HVACMode.OFF,
},
to_states={
HVACMode.AUTO,
HVACMode.COOL,
HVACMode.DRY,
HVACMode.FAN_ONLY,
HVACMode.HEAT,
HVACMode.HEAT_COOL,
},
),
"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

@@ -1,19 +0,0 @@
.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
turned_on: *trigger_common

View File

@@ -108,34 +108,5 @@
"toggle_cover_tilt": {
"service": "mdi:arrow-top-right-bottom-left"
}
},
"triggers": {
"awning_opened": {
"trigger": "mdi:awning-outline"
},
"blind_opened": {
"trigger": "mdi:blinds-horizontal"
},
"curtain_opened": {
"trigger": "mdi:curtains"
},
"door_opened": {
"trigger": "mdi:door-open"
},
"garage_opened": {
"trigger": "mdi:garage-open"
},
"gate_opened": {
"trigger": "mdi:gate-open"
},
"shade_opened": {
"trigger": "mdi:roller-shade"
},
"shutter_opened": {
"trigger": "mdi:window-shutter-open"
},
"window_opened": {
"trigger": "mdi:window-open"
}
}
}

View File

@@ -1,16 +1,4 @@
{
"common": {
"trigger_behavior_description_awning": "The behavior of the targeted awnings to trigger on.",
"trigger_behavior_description_blind": "The behavior of the targeted blinds to trigger on.",
"trigger_behavior_description_curtain": "The behavior of the targeted curtains to trigger on.",
"trigger_behavior_description_door": "The behavior of the targeted doors to trigger on.",
"trigger_behavior_description_garage": "The behavior of the targeted garage doors to trigger on.",
"trigger_behavior_description_gate": "The behavior of the targeted gates to trigger on.",
"trigger_behavior_description_shade": "The behavior of the targeted shades to trigger on.",
"trigger_behavior_description_shutter": "The behavior of the targeted shutters to trigger on.",
"trigger_behavior_description_window": "The behavior of the targeted windows to trigger on.",
"trigger_behavior_name": "Behavior"
},
"device_automation": {
"action_type": {
"close": "Close {entity_name}",
@@ -94,15 +82,6 @@
"name": "Window"
}
},
"selector": {
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
}
}
},
"services": {
"close_cover": {
"description": "Closes a cover.",
@@ -157,142 +136,5 @@
"name": "Toggle tilt"
}
},
"title": "Cover",
"triggers": {
"awning_opened": {
"description": "Triggers when an awning opens.",
"description_configured": "[%key:component::cover::triggers::awning_opened::description%]",
"fields": {
"behavior": {
"description": "[%key:component::cover::common::trigger_behavior_description_awning%]",
"name": "[%key:component::cover::common::trigger_behavior_name%]"
},
"fully_opened": {
"description": "Require the awnings to be fully opened before triggering.",
"name": "Fully opened"
}
},
"name": "When an awning opens"
},
"blind_opened": {
"description": "Triggers when a blind opens.",
"description_configured": "[%key:component::cover::triggers::blind_opened::description%]",
"fields": {
"behavior": {
"description": "[%key:component::cover::common::trigger_behavior_description_blind%]",
"name": "[%key:component::cover::common::trigger_behavior_name%]"
},
"fully_opened": {
"description": "Require the blinds to be fully opened before triggering.",
"name": "Fully opened"
}
},
"name": "When a blind opens"
},
"curtain_opened": {
"description": "Triggers when a curtain opens.",
"description_configured": "[%key:component::cover::triggers::curtain_opened::description%]",
"fields": {
"behavior": {
"description": "[%key:component::cover::common::trigger_behavior_description_curtain%]",
"name": "[%key:component::cover::common::trigger_behavior_name%]"
},
"fully_opened": {
"description": "Require the curtains to be fully opened before triggering.",
"name": "Fully opened"
}
},
"name": "When a curtain opens"
},
"door_opened": {
"description": "Triggers when a door opens.",
"description_configured": "[%key:component::cover::triggers::door_opened::description%]",
"fields": {
"behavior": {
"description": "[%key:component::cover::common::trigger_behavior_description_door%]",
"name": "[%key:component::cover::common::trigger_behavior_name%]"
},
"fully_opened": {
"description": "Require the doors to be fully opened before triggering.",
"name": "Fully opened"
}
},
"name": "When a door opens"
},
"garage_opened": {
"description": "Triggers when a garage door opens.",
"description_configured": "[%key:component::cover::triggers::garage_opened::description%]",
"fields": {
"behavior": {
"description": "[%key:component::cover::common::trigger_behavior_description_garage%]",
"name": "[%key:component::cover::common::trigger_behavior_name%]"
},
"fully_opened": {
"description": "Require the garage doors to be fully opened before triggering.",
"name": "Fully opened"
}
},
"name": "When a garage door opens"
},
"gate_opened": {
"description": "Triggers when a gate opens.",
"description_configured": "[%key:component::cover::triggers::gate_opened::description%]",
"fields": {
"behavior": {
"description": "[%key:component::cover::common::trigger_behavior_description_gate%]",
"name": "[%key:component::cover::common::trigger_behavior_name%]"
},
"fully_opened": {
"description": "Require the gates to be fully opened before triggering.",
"name": "Fully opened"
}
},
"name": "When a gate opens"
},
"shade_opened": {
"description": "Triggers when a shade opens.",
"description_configured": "[%key:component::cover::triggers::shade_opened::description%]",
"fields": {
"behavior": {
"description": "[%key:component::cover::common::trigger_behavior_description_shade%]",
"name": "[%key:component::cover::common::trigger_behavior_name%]"
},
"fully_opened": {
"description": "Require the shades to be fully opened before triggering.",
"name": "Fully opened"
}
},
"name": "When a shade opens"
},
"shutter_opened": {
"description": "Triggers when a shutter opens.",
"description_configured": "[%key:component::cover::triggers::shutter_opened::description%]",
"fields": {
"behavior": {
"description": "[%key:component::cover::common::trigger_behavior_description_shutter%]",
"name": "[%key:component::cover::common::trigger_behavior_name%]"
},
"fully_opened": {
"description": "Require the shutters to be fully opened before triggering.",
"name": "Fully opened"
}
},
"name": "When a shutter opens"
},
"window_opened": {
"description": "Triggers when a window opens.",
"description_configured": "[%key:component::cover::triggers::window_opened::description%]",
"fields": {
"behavior": {
"description": "[%key:component::cover::common::trigger_behavior_description_window%]",
"name": "[%key:component::cover::common::trigger_behavior_name%]"
},
"fully_opened": {
"description": "Require the windows to be fully opened before triggering.",
"name": "Fully opened"
}
},
"name": "When a window opens"
}
}
"title": "Cover"
}

View File

@@ -1,116 +0,0 @@
"""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_FIRST_LAST,
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_FIRST_LAST.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_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]] = {
"awning_opened": make_cover_opened_trigger(CoverDeviceClass.AWNING),
"blind_opened": make_cover_opened_trigger(CoverDeviceClass.BLIND),
"curtain_opened": make_cover_opened_trigger(CoverDeviceClass.CURTAIN),
"door_opened": make_cover_opened_trigger(CoverDeviceClass.DOOR),
"garage_opened": make_cover_opened_trigger(CoverDeviceClass.GARAGE),
"gate_opened": make_cover_opened_trigger(CoverDeviceClass.GATE),
"shade_opened": make_cover_opened_trigger(CoverDeviceClass.SHADE),
"shutter_opened": make_cover_opened_trigger(CoverDeviceClass.SHUTTER),
"window_opened": make_cover_opened_trigger(CoverDeviceClass.WINDOW),
}
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
"""Return the triggers for covers."""
return TRIGGERS

View File

@@ -1,79 +0,0 @@
.trigger_common_fields: &trigger_common_fields
behavior:
required: true
default: any
selector:
select:
translation_key: trigger_behavior
options:
- first
- last
- any
fully_opened:
required: true
default: false
selector:
boolean:
awning_opened:
fields: *trigger_common_fields
target:
entity:
domain: cover
device_class: awning
blind_opened:
fields: *trigger_common_fields
target:
entity:
domain: cover
device_class: blind
curtain_opened:
fields: *trigger_common_fields
target:
entity:
domain: cover
device_class: curtain
door_opened:
fields: *trigger_common_fields
target:
entity:
domain: cover
device_class: door
garage_opened:
fields: *trigger_common_fields
target:
entity:
domain: cover
device_class: garage
gate_opened:
fields: *trigger_common_fields
target:
entity:
domain: cover
device_class: gate
shade_opened:
fields: *trigger_common_fields
target:
entity:
domain: cover
device_class: shade
shutter_opened:
fields: *trigger_common_fields
target:
entity:
domain: cover
device_class: shutter
window_opened:
fields: *trigger_common_fields
target:
entity:
domain: cover
device_class: window

View File

@@ -47,13 +47,5 @@
"turn_on": {
"service": "mdi:fan"
}
},
"triggers": {
"turned_off": {
"trigger": "mdi:fan-off"
},
"turned_on": {
"trigger": "mdi:fan"
}
}
}

View File

@@ -1,8 +1,4 @@
{
"common": {
"trigger_behavior_description": "The behavior of the targeted fans to trigger on.",
"trigger_behavior_name": "Behavior"
},
"device_automation": {
"action_type": {
"toggle": "[%key:common::device_automation::action_type::toggle%]",
@@ -70,13 +66,6 @@
"forward": "Forward",
"reverse": "Reverse"
}
},
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
}
}
},
"services": {
@@ -163,29 +152,5 @@
"name": "[%key:common::action::turn_on%]"
}
},
"title": "Fan",
"triggers": {
"turned_off": {
"description": "Triggers when a fan is turned off.",
"description_configured": "[%key:component::fan::triggers::turned_off::description%]",
"fields": {
"behavior": {
"description": "[%key:component::fan::common::trigger_behavior_description%]",
"name": "[%key:component::fan::common::trigger_behavior_name%]"
}
},
"name": "When a fan is turned off"
},
"turned_on": {
"description": "Triggers when a fan is turned on.",
"description_configured": "[%key:component::fan::triggers::turned_on::description%]",
"fields": {
"behavior": {
"description": "[%key:component::fan::common::trigger_behavior_description%]",
"name": "[%key:component::fan::common::trigger_behavior_name%]"
}
},
"name": "When a fan is turned on"
}
}
"title": "Fan"
}

View File

@@ -1,17 +0,0 @@
"""Provides triggers for fans."""
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import Trigger, make_entity_state_trigger
from . 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 fans."""
return TRIGGERS

View File

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

View File

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

View File

@@ -14,19 +14,5 @@
"start_mowing": {
"service": "mdi:play"
}
},
"triggers": {
"docked": {
"trigger": "mdi:home-import-outline"
},
"errored": {
"trigger": "mdi:alert-circle-outline"
},
"paused_mowing": {
"trigger": "mdi:pause"
},
"started_mowing": {
"trigger": "mdi:play"
}
}
}

View File

@@ -1,8 +1,4 @@
{
"common": {
"trigger_behavior_description": "The behavior of the targeted lawn mowers to trigger on.",
"trigger_behavior_name": "Behavior"
},
"entity_component": {
"_": {
"name": "[%key:component::lawn_mower::title%]",
@@ -15,15 +11,6 @@
}
}
},
"selector": {
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
}
}
},
"services": {
"dock": {
"description": "Stops the mowing task and returns to the dock.",
@@ -38,51 +25,5 @@
"name": "Start mowing"
}
},
"title": "Lawn mower",
"triggers": {
"docked": {
"description": "Triggers when a lawn mower has docked.",
"description_configured": "[%key:component::lawn_mower::triggers::docked::description%]",
"fields": {
"behavior": {
"description": "[%key:component::lawn_mower::common::trigger_behavior_description%]",
"name": "[%key:component::lawn_mower::common::trigger_behavior_name%]"
}
},
"name": "When a lawn mower has docked"
},
"errored": {
"description": "Triggers when a lawn mower has errored.",
"description_configured": "[%key:component::lawn_mower::triggers::errored::description%]",
"fields": {
"behavior": {
"description": "[%key:component::lawn_mower::common::trigger_behavior_description%]",
"name": "[%key:component::lawn_mower::common::trigger_behavior_name%]"
}
},
"name": "When a lawn mower has errored"
},
"paused_mowing": {
"description": "Triggers when a lawn mower has paused mowing.",
"description_configured": "[%key:component::lawn_mower::triggers::paused_mowing::description%]",
"fields": {
"behavior": {
"description": "[%key:component::lawn_mower::common::trigger_behavior_description%]",
"name": "[%key:component::lawn_mower::common::trigger_behavior_name%]"
}
},
"name": "When a lawn mower has paused mowing"
},
"started_mowing": {
"description": "Triggers when a lawn mower has started mowing.",
"description_configured": "[%key:component::lawn_mower::triggers::started_mowing::description%]",
"fields": {
"behavior": {
"description": "[%key:component::lawn_mower::common::trigger_behavior_description%]",
"name": "[%key:component::lawn_mower::common::trigger_behavior_name%]"
}
},
"name": "When a lawn mower has started mowing"
}
}
"title": "Lawn mower"
}

View File

@@ -1,18 +0,0 @@
"""Provides triggers for lawn mowers."""
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import Trigger, make_entity_state_trigger
from .const import DOMAIN, LawnMowerActivity
TRIGGERS: dict[str, type[Trigger]] = {
"docked": make_entity_state_trigger(DOMAIN, LawnMowerActivity.DOCKED),
"errored": make_entity_state_trigger(DOMAIN, LawnMowerActivity.ERROR),
"paused_mowing": make_entity_state_trigger(DOMAIN, LawnMowerActivity.PAUSED),
"started_mowing": make_entity_state_trigger(DOMAIN, LawnMowerActivity.MOWING),
}
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
"""Return the triggers for lawn mowers."""
return TRIGGERS

View File

@@ -1,20 +0,0 @@
.trigger_common: &trigger_common
target:
entity:
domain: lawn_mower
fields:
behavior:
required: true
default: any
selector:
select:
options:
- first
- last
- any
translation_key: trigger_behavior
docked: *trigger_common
errored: *trigger_common
paused_mowing: *trigger_common
started_mowing: *trigger_common

View File

@@ -9,5 +9,5 @@
"iot_class": "local_polling",
"loggers": ["pypck"],
"quality_scale": "bronze",
"requirements": ["pypck==0.9.4", "lcn-frontend==0.2.7"]
"requirements": ["pypck==0.9.5", "lcn-frontend==0.2.7"]
}

View File

@@ -1,131 +0,0 @@
"""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

@@ -1,28 +0,0 @@
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,12 +1,4 @@
{
"conditions": {
"is_off": {
"condition": "mdi:lightbulb-off"
},
"is_on": {
"condition": "mdi:lightbulb-on"
}
},
"entity_component": {
"_": {
"default": "mdi:lightbulb",
@@ -33,13 +25,5 @@
"turn_on": {
"service": "mdi:lightbulb-on"
}
},
"triggers": {
"turned_off": {
"trigger": "mdi:lightbulb-off"
},
"turned_on": {
"trigger": "mdi:lightbulb-on"
}
}
}

View File

@@ -1,7 +1,5 @@
{
"common": {
"condition_behavior_description": "How the state should match on the targeted lights.",
"condition_behavior_name": "Behavior",
"field_brightness_description": "Number indicating brightness, where 0 turns the light off, 1 is the minimum brightness, and 255 is the maximum brightness.",
"field_brightness_name": "Brightness value",
"field_brightness_pct_description": "Number indicating the percentage of full brightness, where 0 turns the light off, 1 is the minimum brightness, and 100 is the maximum brightness.",
@@ -36,33 +34,7 @@
"field_white_name": "White",
"field_xy_color_description": "Color in XY-format. A list of two decimal numbers between 0 and 1.",
"field_xy_color_name": "XY-color",
"section_advanced_fields_name": "Advanced options",
"trigger_behavior_description": "The behavior of the targeted lights to trigger on.",
"trigger_behavior_name": "Behavior"
},
"conditions": {
"is_off": {
"description": "Test if a light is off.",
"description_configured": "[%key:component::light::conditions::is_off::description%]",
"fields": {
"behavior": {
"description": "[%key:component::light::common::condition_behavior_description%]",
"name": "[%key:component::light::common::condition_behavior_name%]"
}
},
"name": "If a light is off"
},
"is_on": {
"description": "Test if a light is on.",
"description_configured": "[%key:component::light::conditions::is_on::description%]",
"fields": {
"behavior": {
"description": "[%key:component::light::common::condition_behavior_description%]",
"name": "[%key:component::light::common::condition_behavior_name%]"
}
},
"name": "If a light is on"
}
"section_advanced_fields_name": "Advanced options"
},
"device_automation": {
"action_type": {
@@ -312,30 +284,11 @@
"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": {
@@ -509,29 +462,5 @@
}
}
},
"title": "Light",
"triggers": {
"turned_off": {
"description": "Triggers when a light is turned off.",
"description_configured": "[%key:component::light::triggers::turned_off::description%]",
"fields": {
"behavior": {
"description": "[%key:component::light::common::trigger_behavior_description%]",
"name": "[%key:component::light::common::trigger_behavior_name%]"
}
},
"name": "When a light is turned off"
},
"turned_on": {
"description": "Triggers when a light is turned on.",
"description_configured": "[%key:component::light::triggers::turned_on::description%]",
"fields": {
"behavior": {
"description": "[%key:component::light::common::trigger_behavior_description%]",
"name": "[%key:component::light::common::trigger_behavior_name%]"
}
},
"name": "When a light is turned on"
}
}
"title": "Light"
}

View File

@@ -1,17 +0,0 @@
"""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

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

View File

@@ -9,7 +9,7 @@
},
"iot_class": "local_push",
"loggers": ["pylutron_caseta"],
"requirements": ["pylutron-caseta==0.25.0"],
"requirements": ["pylutron-caseta==0.26.0"],
"zeroconf": [
{
"properties": {

View File

@@ -104,10 +104,5 @@
"volume_up": {
"service": "mdi:volume-plus"
}
},
"triggers": {
"stopped_playing": {
"trigger": "mdi:stop"
}
}
}

View File

@@ -1,8 +1,4 @@
{
"common": {
"trigger_behavior_description": "The behavior of the targeted media players to trigger on.",
"trigger_behavior_name": "Behavior"
},
"device_automation": {
"condition_type": {
"is_buffering": "{entity_name} is buffering",
@@ -181,13 +177,6 @@
"off": "[%key:common::state::off%]",
"one": "Repeat one"
}
},
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
}
}
},
"services": {
@@ -378,18 +367,5 @@
"name": "Turn up volume"
}
},
"title": "Media player",
"triggers": {
"stopped_playing": {
"description": "Triggers when a media player stops playing.",
"description_configured": "[%key:component::media_player::triggers::stopped_playing::description%]",
"fields": {
"behavior": {
"description": "[%key:component::media_player::common::trigger_behavior_description%]",
"name": "[%key:component::media_player::common::trigger_behavior_name%]"
}
},
"name": "When a media player stops playing"
}
}
"title": "Media player"
}

View File

@@ -1,28 +0,0 @@
"""Provides triggers for media players."""
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import Trigger, make_conditional_entity_state_trigger
from . import MediaPlayerState
from .const import DOMAIN
TRIGGERS: dict[str, type[Trigger]] = {
"stopped_playing": make_conditional_entity_state_trigger(
DOMAIN,
from_states={
MediaPlayerState.BUFFERING,
MediaPlayerState.PAUSED,
MediaPlayerState.PLAYING,
},
to_states={
MediaPlayerState.IDLE,
MediaPlayerState.OFF,
MediaPlayerState.ON,
},
),
}
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
"""Return the triggers for media players."""
return TRIGGERS

View File

@@ -1,15 +0,0 @@
stopped_playing:
target:
entity:
domain: media_player
fields:
behavior:
required: true
default: any
selector:
select:
translation_key: trigger_behavior
options:
- first
- last
- any

View File

@@ -8,10 +8,5 @@
"set_value": {
"service": "mdi:form-textbox"
}
},
"triggers": {
"changed": {
"trigger": "mdi:form-textbox"
}
}
}

View File

@@ -42,12 +42,5 @@
"name": "Set value"
}
},
"title": "Text",
"triggers": {
"changed": {
"description": "Triggers when the text changes.",
"description_configured": "[%key:component::text::triggers::changed::description%]",
"name": "When the text changes"
}
}
"title": "Text"
}

View File

@@ -1,35 +0,0 @@
"""Provides triggers for lights."""
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers.trigger import (
ENTITY_STATE_TRIGGER_SCHEMA,
EntityTriggerBase,
Trigger,
)
from .const import DOMAIN
class TextChangedTrigger(EntityTriggerBase):
"""Trigger for text entity when its content changes."""
_domain = DOMAIN
_schema = ENTITY_STATE_TRIGGER_SCHEMA
def is_from_state(self, from_state: State, to_state: State) -> bool:
"""Check if the state matches the origin state."""
return from_state.state != to_state.state
def is_to_state(self, state: State) -> bool:
"""Check if the state matches the target state."""
return True
TRIGGERS: dict[str, type[Trigger]] = {
"changed": TextChangedTrigger,
}
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
"""Return the triggers for lights."""
return TRIGGERS

View File

@@ -1,4 +0,0 @@
changed:
target:
entity:
domain: text

View File

@@ -41,19 +41,5 @@
"turn_on": {
"service": "mdi:play"
}
},
"triggers": {
"docked": {
"trigger": "mdi:home-import-outline"
},
"errored": {
"trigger": "mdi:alert-circle-outline"
},
"paused_cleaning": {
"trigger": "mdi:pause"
},
"started_cleaning": {
"trigger": "mdi:play"
}
}
}

View File

@@ -1,8 +1,4 @@
{
"common": {
"trigger_behavior_description": "The behavior of the targeted vacuum cleaners to trigger on.",
"trigger_behavior_name": "Behavior"
},
"device_automation": {
"action_type": {
"clean": "Let {entity_name} clean",
@@ -35,15 +31,6 @@
}
}
},
"selector": {
"trigger_behavior": {
"options": {
"any": "Any",
"first": "First",
"last": "Last"
}
}
},
"services": {
"clean_spot": {
"description": "Tells the vacuum cleaner to do a spot clean-up.",
@@ -110,51 +97,5 @@
"name": "[%key:common::action::turn_on%]"
}
},
"title": "Vacuum",
"triggers": {
"docked": {
"description": "Triggers when a vacuum cleaner has docked.",
"description_configured": "[%key:component::vacuum::triggers::docked::description%]",
"fields": {
"behavior": {
"description": "[%key:component::vacuum::common::trigger_behavior_description%]",
"name": "[%key:component::vacuum::common::trigger_behavior_name%]"
}
},
"name": "When a vacuum cleaner has docked"
},
"errored": {
"description": "Triggers when a vacuum cleaner has errored.",
"description_configured": "[%key:component::vacuum::triggers::errored::description%]",
"fields": {
"behavior": {
"description": "[%key:component::vacuum::common::trigger_behavior_description%]",
"name": "[%key:component::vacuum::common::trigger_behavior_name%]"
}
},
"name": "When a vacuum cleaner has errored"
},
"paused_cleaning": {
"description": "Triggers when a vacuum cleaner has paused cleaning.",
"description_configured": "[%key:component::vacuum::triggers::paused_cleaning::description%]",
"fields": {
"behavior": {
"description": "[%key:component::vacuum::common::trigger_behavior_description%]",
"name": "[%key:component::vacuum::common::trigger_behavior_name%]"
}
},
"name": "When a vacuum cleaner has paused cleaning"
},
"started_cleaning": {
"description": "Triggers when a vacuum cleaner has started cleaning.",
"description_configured": "[%key:component::vacuum::triggers::started_cleaning::description%]",
"fields": {
"behavior": {
"description": "[%key:component::vacuum::common::trigger_behavior_description%]",
"name": "[%key:component::vacuum::common::trigger_behavior_name%]"
}
},
"name": "When a vacuum cleaner has started cleaning"
}
}
"title": "Vacuum"
}

View File

@@ -1,18 +0,0 @@
"""Provides triggers for vacuum cleaners."""
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import Trigger, make_entity_state_trigger
from .const import DOMAIN, VacuumActivity
TRIGGERS: dict[str, type[Trigger]] = {
"docked": make_entity_state_trigger(DOMAIN, VacuumActivity.DOCKED),
"errored": make_entity_state_trigger(DOMAIN, VacuumActivity.ERROR),
"paused_cleaning": make_entity_state_trigger(DOMAIN, VacuumActivity.PAUSED),
"started_cleaning": make_entity_state_trigger(DOMAIN, VacuumActivity.CLEANING),
}
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
"""Return the triggers for vacuum cleaners."""
return TRIGGERS

View File

@@ -1,20 +0,0 @@
.trigger_common: &trigger_common
target:
entity:
domain: vacuum
fields:
behavior:
required: true
default: any
selector:
select:
options:
- first
- last
- any
translation_key: trigger_behavior
docked: *trigger_common
errored: *trigger_common
paused_cleaning: *trigger_common
started_cleaning: *trigger_common

View File

@@ -289,7 +289,7 @@ class TargetStateChangeTracker:
)
tracked_entities = self._entity_filter(
selected.referenced | selected.indirectly_referenced
selected.referenced.union(selected.indirectly_referenced)
)
@callback

View File

@@ -10,12 +10,11 @@ from dataclasses import dataclass, field
import functools
import inspect
import logging
from typing import TYPE_CHECKING, Any, Final, Protocol, TypedDict, cast, override
from typing import TYPE_CHECKING, Any, Protocol, TypedDict, cast
import voluptuous as vol
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_ALIAS,
CONF_ENABLED,
CONF_ID,
@@ -24,8 +23,6 @@ from homeassistant.const import (
CONF_SELECTOR,
CONF_TARGET,
CONF_VARIABLES,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import (
CALLBACK_TYPE,
@@ -33,11 +30,9 @@ 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 (
@@ -54,10 +49,6 @@ 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
@@ -254,222 +245,6 @@ class Trigger(abc.ABC):
"""Attach the trigger to an action runner."""
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_TARGET): cv.TARGET_FIELDS,
}
)
ENTITY_STATE_TRIGGER_SCHEMA_FIRST_LAST = ENTITY_STATE_TRIGGER_SCHEMA.extend(
{
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,
}
)
class EntityTriggerBase(Trigger):
"""Trigger for entity state changes."""
_domain: str
_schema: vol.Schema = ENTITY_STATE_TRIGGER_SCHEMA_FIRST_LAST
@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.target is not None
self._options = config.options or {}
self._target = config.target
def is_from_state(self, from_state: State, to_state: State) -> bool:
"""Check if the state matches the origin state."""
return not self.is_to_state(from_state)
@abc.abstractmethod
def is_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_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_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 previous state was not the from state
if not to_state or not self.is_from_state(from_state, to_state):
return
# The trigger should never fire if the new state is not the to state
if not self.is_to_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."""
_to_state: str
def is_to_state(self, state: State) -> bool:
"""Check if the state matches the target state."""
return state.state == self._to_state
class ConditionalEntityStateTriggerBase(EntityTriggerBase):
"""Class for entity state changes where the from state is restricted."""
_from_states: set[str]
_to_states: set[str]
def is_from_state(self, from_state: State, to_state: State) -> bool:
"""Check if the state matches the origin state."""
return from_state.state in self._from_states
def is_to_state(self, state: State) -> bool:
"""Check if the state matches the target state."""
return state.state in self._to_states
class EntityStateAttributeTriggerBase(EntityTriggerBase):
"""Trigger for entity state attribute changes."""
_attribute: str
_attribute_to_state: str
def is_to_state(self, state: State) -> bool:
"""Check if the state matches the target state."""
return state.attributes.get(self._attribute) == self._attribute_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_conditional_entity_state_trigger(
domain: str, *, from_states: set[str], to_states: set[str]
) -> type[ConditionalEntityStateTriggerBase]:
"""Create a conditional entity state trigger class."""
class CustomTrigger(ConditionalEntityStateTriggerBase):
"""Trigger for conditional entity state changes."""
_domain = domain
_from_states = from_states
_to_states = to_states
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
_attribute = attribute
_attribute_to_state = to_state
return CustomTrigger
class TriggerProtocol(Protocol):
"""Define the format of trigger modules.
@@ -1030,8 +805,6 @@ async def async_get_all_descriptions(
continue
description = {"fields": yaml_description.get("fields", {})}
if (target := yaml_description.get("target")) is not None:
description["target"] = target
new_descriptions_cache[missing_trigger] = description

4
requirements_all.txt generated
View File

@@ -2150,7 +2150,7 @@ pylitejet==0.6.3
pylitterbot==2025.0.0
# homeassistant.components.lutron_caseta
pylutron-caseta==0.25.0
pylutron-caseta==0.26.0
# homeassistant.components.lutron
pylutron==0.2.18
@@ -2269,7 +2269,7 @@ pypaperless==4.1.1
pypca==0.0.7
# homeassistant.components.lcn
pypck==0.9.4
pypck==0.9.5
# homeassistant.components.pglab
pypglab==0.0.5

View File

@@ -1794,7 +1794,7 @@ pylitejet==0.6.3
pylitterbot==2025.0.0
# homeassistant.components.lutron_caseta
pylutron-caseta==0.25.0
pylutron-caseta==0.26.0
# homeassistant.components.lutron
pylutron==0.2.18
@@ -1892,7 +1892,7 @@ pypalazzetti==0.1.20
pypaperless==4.1.1
# homeassistant.components.lcn
pypck==0.9.4
pypck==0.9.5
# homeassistant.components.pglab
pypglab==0.0.5

View File

@@ -1,280 +1 @@
"""The tests for components."""
from enum import StrEnum
import itertools
from typing import TypedDict
from homeassistant.const import (
ATTR_AREA_ID,
ATTR_DEVICE_ID,
ATTR_FLOOR_ID,
ATTR_LABEL_ID,
CONF_ENTITY_ID,
CONF_OPTIONS,
CONF_PLATFORM,
CONF_TARGET,
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),
]
class StateDescription(TypedDict):
"""Test state and expected service call count."""
state: str | None
attributes: dict
count: int
def parametrize_trigger_states(
*,
trigger: str,
target_states: list[str | None | tuple[str | None, dict]],
other_states: list[str | None | tuple[str | None, dict]],
additional_attributes: dict | None = None,
trigger_from_none: bool = True,
) -> list[tuple[str, list[StateDescription]]]:
"""Parametrize states and expected service call counts.
The target_states and other_states iterables are either iterables of
states or iterables of (state, attributes) tuples.
Set `trigger_from_none` to False if the trigger is not expected to fire
when the initial state is None.
Returns a list of tuples with (trigger, list of states),
where states is a list of StateDescription dicts.
"""
additional_attributes = additional_attributes or {}
def state_with_attributes(
state: str | None | tuple[str | None, dict], count: int
) -> dict:
"""Return (state, attributes) dict."""
if isinstance(state, str) or state is None:
return {"state": state, "attributes": additional_attributes, "count": count}
return {
"state": state[0],
"attributes": state[1] | additional_attributes,
"count": count,
}
return [
# Initial state None
(
trigger,
list(
itertools.chain.from_iterable(
(
state_with_attributes(None, 0),
state_with_attributes(target_state, 0),
state_with_attributes(other_state, 0),
state_with_attributes(
target_state, 1 if trigger_from_none else 0
),
)
for target_state in target_states
for other_state in other_states
)
),
),
# Initial state different from target state
(
trigger,
# other_state,
list(
itertools.chain.from_iterable(
(
state_with_attributes(other_state, 0),
state_with_attributes(target_state, 1),
state_with_attributes(other_state, 0),
state_with_attributes(target_state, 1),
)
for target_state in target_states
for other_state in other_states
)
),
),
# Initial state same as target state
(
trigger,
list(
itertools.chain.from_iterable(
(
state_with_attributes(target_state, 0),
state_with_attributes(target_state, 0),
state_with_attributes(other_state, 0),
state_with_attributes(target_state, 1),
)
for target_state in target_states
for other_state in other_states
)
),
),
# Initial state unavailable / unknown
(
trigger,
list(
itertools.chain.from_iterable(
(
state_with_attributes(STATE_UNAVAILABLE, 0),
state_with_attributes(target_state, 0),
state_with_attributes(other_state, 0),
state_with_attributes(target_state, 1),
)
for target_state in target_states
for other_state in other_states
)
),
),
(
trigger,
list(
itertools.chain.from_iterable(
(
state_with_attributes(STATE_UNKNOWN, 0),
state_with_attributes(target_state, 0),
state_with_attributes(other_state, 0),
state_with_attributes(target_state, 1),
)
for target_state in target_states
for other_state in other_states
)
),
),
]
async def arm_trigger(
hass: HomeAssistant, trigger: str, trigger_options: dict, trigger_target: dict
) -> None:
"""Arm the specified trigger, call service test.automation when it triggers."""
# Local include to avoid importing the automation component unnecessarily
from homeassistant.components import automation # noqa: PLC0415
await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {
CONF_PLATFORM: trigger,
CONF_TARGET: {**trigger_target},
}
| ({CONF_OPTIONS: {**trigger_options}} if trigger_options else {}),
"action": {
"service": "test.automation",
"data_template": {CONF_ENTITY_ID: "{{ trigger.entity_id }}"},
},
}
},
)
def set_or_remove_state(
hass: HomeAssistant,
entity_id: str,
state: StateDescription,
) -> None:
"""Set or remove the state of an entity."""
if state["state"] is None:
hass.states.async_remove(entity_id)
else:
hass.states.async_set(
entity_id, state["state"], state["attributes"], force_update=True
)
def other_states(state: StrEnum) -> list[str]:
"""Return a sorted list with all states except the specified one."""
return sorted({s.value for s in state.__class__} - {state.value})

View File

@@ -1,359 +0,0 @@
"""Test alarm control panel triggers."""
import pytest
from homeassistant.components.alarm_control_panel import (
AlarmControlPanelEntityFeature,
AlarmControlPanelState,
)
from homeassistant.const import ATTR_SUPPORTED_FEATURES, CONF_ENTITY_ID
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.setup import async_setup_component
from tests.components import (
StateDescription,
arm_trigger,
other_states,
parametrize_target_entities,
parametrize_trigger_states,
set_or_remove_state,
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_alarm_control_panels(hass: HomeAssistant) -> None:
"""Create multiple alarm control panel entities associated with different targets."""
return await target_entities(hass, "alarm_control_panel")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("alarm_control_panel"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="alarm_control_panel.armed",
target_states=[
AlarmControlPanelState.ARMED_AWAY,
AlarmControlPanelState.ARMED_CUSTOM_BYPASS,
AlarmControlPanelState.ARMED_HOME,
AlarmControlPanelState.ARMED_NIGHT,
AlarmControlPanelState.ARMED_VACATION,
],
other_states=[
AlarmControlPanelState.ARMING,
AlarmControlPanelState.DISARMED,
AlarmControlPanelState.DISARMING,
AlarmControlPanelState.PENDING,
AlarmControlPanelState.TRIGGERED,
],
),
*parametrize_trigger_states(
trigger="alarm_control_panel.armed_away",
target_states=[AlarmControlPanelState.ARMED_AWAY],
other_states=other_states(AlarmControlPanelState.ARMED_AWAY),
additional_attributes={
ATTR_SUPPORTED_FEATURES: AlarmControlPanelEntityFeature.ARM_AWAY
},
trigger_from_none=False,
),
*parametrize_trigger_states(
trigger="alarm_control_panel.armed_home",
target_states=[AlarmControlPanelState.ARMED_HOME],
other_states=other_states(AlarmControlPanelState.ARMED_HOME),
additional_attributes={
ATTR_SUPPORTED_FEATURES: AlarmControlPanelEntityFeature.ARM_HOME
},
trigger_from_none=False,
),
*parametrize_trigger_states(
trigger="alarm_control_panel.armed_night",
target_states=[AlarmControlPanelState.ARMED_NIGHT],
other_states=other_states(AlarmControlPanelState.ARMED_NIGHT),
additional_attributes={
ATTR_SUPPORTED_FEATURES: AlarmControlPanelEntityFeature.ARM_NIGHT
},
trigger_from_none=False,
),
*parametrize_trigger_states(
trigger="alarm_control_panel.armed_vacation",
target_states=[AlarmControlPanelState.ARMED_VACATION],
other_states=other_states(AlarmControlPanelState.ARMED_VACATION),
additional_attributes={
ATTR_SUPPORTED_FEATURES: AlarmControlPanelEntityFeature.ARM_VACATION
},
trigger_from_none=False,
),
*parametrize_trigger_states(
trigger="alarm_control_panel.disarmed",
target_states=[AlarmControlPanelState.DISARMED],
other_states=other_states(AlarmControlPanelState.DISARMED),
),
*parametrize_trigger_states(
trigger="alarm_control_panel.triggered",
target_states=[AlarmControlPanelState.TRIGGERED],
other_states=other_states(AlarmControlPanelState.TRIGGERED),
),
],
)
async def test_alarm_control_panel_state_trigger_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_alarm_control_panels: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the alarm control panel state trigger fires when any alarm control panel state changes to a specific state."""
await async_setup_component(hass, "alarm_control_panel", {})
other_entity_ids = set(target_alarm_control_panels) - {entity_id}
# Set all alarm control panels, including the tested one, to the initial state
for eid in target_alarm_control_panels:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Check if changing other alarm control panels also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * state["count"]
service_calls.clear()
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("alarm_control_panel"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="alarm_control_panel.armed",
target_states=[
AlarmControlPanelState.ARMED_AWAY,
AlarmControlPanelState.ARMED_CUSTOM_BYPASS,
AlarmControlPanelState.ARMED_HOME,
AlarmControlPanelState.ARMED_NIGHT,
AlarmControlPanelState.ARMED_VACATION,
],
other_states=[
AlarmControlPanelState.ARMING,
AlarmControlPanelState.DISARMED,
AlarmControlPanelState.DISARMING,
AlarmControlPanelState.PENDING,
AlarmControlPanelState.TRIGGERED,
],
),
*parametrize_trigger_states(
trigger="alarm_control_panel.armed_away",
target_states=[AlarmControlPanelState.ARMED_AWAY],
other_states=other_states(AlarmControlPanelState.ARMED_AWAY),
additional_attributes={
ATTR_SUPPORTED_FEATURES: AlarmControlPanelEntityFeature.ARM_AWAY
},
trigger_from_none=False,
),
*parametrize_trigger_states(
trigger="alarm_control_panel.armed_home",
target_states=[AlarmControlPanelState.ARMED_HOME],
other_states=other_states(AlarmControlPanelState.ARMED_HOME),
additional_attributes={
ATTR_SUPPORTED_FEATURES: AlarmControlPanelEntityFeature.ARM_HOME
},
trigger_from_none=False,
),
*parametrize_trigger_states(
trigger="alarm_control_panel.armed_night",
target_states=[AlarmControlPanelState.ARMED_NIGHT],
other_states=other_states(AlarmControlPanelState.ARMED_NIGHT),
additional_attributes={
ATTR_SUPPORTED_FEATURES: AlarmControlPanelEntityFeature.ARM_NIGHT
},
trigger_from_none=False,
),
*parametrize_trigger_states(
trigger="alarm_control_panel.armed_vacation",
target_states=[AlarmControlPanelState.ARMED_VACATION],
other_states=other_states(AlarmControlPanelState.ARMED_VACATION),
additional_attributes={
ATTR_SUPPORTED_FEATURES: AlarmControlPanelEntityFeature.ARM_VACATION
},
trigger_from_none=False,
),
*parametrize_trigger_states(
trigger="alarm_control_panel.disarmed",
target_states=[AlarmControlPanelState.DISARMED],
other_states=other_states(AlarmControlPanelState.DISARMED),
),
*parametrize_trigger_states(
trigger="alarm_control_panel.triggered",
target_states=[AlarmControlPanelState.TRIGGERED],
other_states=other_states(AlarmControlPanelState.TRIGGERED),
),
],
)
async def test_alarm_control_panel_state_trigger_behavior_first(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_alarm_control_panels: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the alarm control panel state trigger fires when the first alarm control panel changes to a specific state."""
await async_setup_component(hass, "alarm_control_panel", {})
other_entity_ids = set(target_alarm_control_panels) - {entity_id}
# Set all alarm control panels, including the tested one, to the initial state
for eid in target_alarm_control_panels:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Triggering other alarm control panels should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
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("alarm_control_panel"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="alarm_control_panel.armed",
target_states=[
AlarmControlPanelState.ARMED_AWAY,
AlarmControlPanelState.ARMED_CUSTOM_BYPASS,
AlarmControlPanelState.ARMED_HOME,
AlarmControlPanelState.ARMED_NIGHT,
AlarmControlPanelState.ARMED_VACATION,
],
other_states=[
AlarmControlPanelState.ARMING,
AlarmControlPanelState.DISARMED,
AlarmControlPanelState.DISARMING,
AlarmControlPanelState.PENDING,
AlarmControlPanelState.TRIGGERED,
],
),
*parametrize_trigger_states(
trigger="alarm_control_panel.armed_away",
target_states=[AlarmControlPanelState.ARMED_AWAY],
other_states=other_states(AlarmControlPanelState.ARMED_AWAY),
additional_attributes={
ATTR_SUPPORTED_FEATURES: AlarmControlPanelEntityFeature.ARM_AWAY
},
trigger_from_none=False,
),
*parametrize_trigger_states(
trigger="alarm_control_panel.armed_home",
target_states=[AlarmControlPanelState.ARMED_HOME],
other_states=other_states(AlarmControlPanelState.ARMED_HOME),
additional_attributes={
ATTR_SUPPORTED_FEATURES: AlarmControlPanelEntityFeature.ARM_HOME
},
trigger_from_none=False,
),
*parametrize_trigger_states(
trigger="alarm_control_panel.armed_night",
target_states=[AlarmControlPanelState.ARMED_NIGHT],
other_states=other_states(AlarmControlPanelState.ARMED_NIGHT),
additional_attributes={
ATTR_SUPPORTED_FEATURES: AlarmControlPanelEntityFeature.ARM_NIGHT
},
trigger_from_none=False,
),
*parametrize_trigger_states(
trigger="alarm_control_panel.armed_vacation",
target_states=[AlarmControlPanelState.ARMED_VACATION],
other_states=other_states(AlarmControlPanelState.ARMED_VACATION),
additional_attributes={
ATTR_SUPPORTED_FEATURES: AlarmControlPanelEntityFeature.ARM_VACATION
},
trigger_from_none=False,
),
*parametrize_trigger_states(
trigger="alarm_control_panel.disarmed",
target_states=[AlarmControlPanelState.DISARMED],
other_states=other_states(AlarmControlPanelState.DISARMED),
),
*parametrize_trigger_states(
trigger="alarm_control_panel.triggered",
target_states=[AlarmControlPanelState.TRIGGERED],
other_states=other_states(AlarmControlPanelState.TRIGGERED),
),
],
)
async def test_alarm_control_panel_state_trigger_behavior_last(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_alarm_control_panels: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the alarm_control_panel state trigger fires when the last alarm_control_panel changes to a specific state."""
await async_setup_component(hass, "alarm_control_panel", {})
other_entity_ids = set(target_alarm_control_panels) - {entity_id}
# Set all alarm control panels, including the tested one, to the initial state
for eid in target_alarm_control_panels:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
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) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()

View File

@@ -1,227 +0,0 @@
"""Test assist satellite triggers."""
import pytest
from homeassistant.components.assist_satellite.entity import AssistSatelliteState
from homeassistant.const import CONF_ENTITY_ID
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.setup import async_setup_component
from tests.components import (
StateDescription,
arm_trigger,
other_states,
parametrize_target_entities,
parametrize_trigger_states,
set_or_remove_state,
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_assist_satellites(hass: HomeAssistant) -> None:
"""Create multiple assist satellite entities associated with different targets."""
return await target_entities(hass, "assist_satellite")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("assist_satellite"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="assist_satellite.idle",
target_states=[AssistSatelliteState.IDLE],
other_states=other_states(AssistSatelliteState.IDLE),
),
*parametrize_trigger_states(
trigger="assist_satellite.listening",
target_states=[AssistSatelliteState.LISTENING],
other_states=other_states(AssistSatelliteState.LISTENING),
),
*parametrize_trigger_states(
trigger="assist_satellite.processing",
target_states=[AssistSatelliteState.PROCESSING],
other_states=other_states(AssistSatelliteState.PROCESSING),
),
*parametrize_trigger_states(
trigger="assist_satellite.responding",
target_states=[AssistSatelliteState.RESPONDING],
other_states=other_states(AssistSatelliteState.RESPONDING),
),
],
)
async def test_assist_satellite_state_trigger_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_assist_satellites: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the assist satellite state trigger fires when any assist satellite state changes to a specific state."""
await async_setup_component(hass, "assist_satellite", {})
other_entity_ids = set(target_assist_satellites) - {entity_id}
# Set all assist satellites, including the tested one, to the initial state
for eid in target_assist_satellites:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Check if changing other assist satellites also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * state["count"]
service_calls.clear()
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("assist_satellite"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="assist_satellite.idle",
target_states=[AssistSatelliteState.IDLE],
other_states=other_states(AssistSatelliteState.IDLE),
),
*parametrize_trigger_states(
trigger="assist_satellite.listening",
target_states=[AssistSatelliteState.LISTENING],
other_states=other_states(AssistSatelliteState.LISTENING),
),
*parametrize_trigger_states(
trigger="assist_satellite.processing",
target_states=[AssistSatelliteState.PROCESSING],
other_states=other_states(AssistSatelliteState.PROCESSING),
),
*parametrize_trigger_states(
trigger="assist_satellite.responding",
target_states=[AssistSatelliteState.RESPONDING],
other_states=other_states(AssistSatelliteState.RESPONDING),
),
],
)
async def test_assist_satellite_state_trigger_behavior_first(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_assist_satellites: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the assist satellite state trigger fires when the first assist satellite changes to a specific state."""
await async_setup_component(hass, "assist_satellite", {})
other_entity_ids = set(target_assist_satellites) - {entity_id}
# Set all assist satellites, including the tested one, to the initial state
for eid in target_assist_satellites:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Triggering other assist satellites should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
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("assist_satellite"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="assist_satellite.idle",
target_states=[AssistSatelliteState.IDLE],
other_states=other_states(AssistSatelliteState.IDLE),
),
*parametrize_trigger_states(
trigger="assist_satellite.listening",
target_states=[AssistSatelliteState.LISTENING],
other_states=other_states(AssistSatelliteState.LISTENING),
),
*parametrize_trigger_states(
trigger="assist_satellite.processing",
target_states=[AssistSatelliteState.PROCESSING],
other_states=other_states(AssistSatelliteState.PROCESSING),
),
*parametrize_trigger_states(
trigger="assist_satellite.responding",
target_states=[AssistSatelliteState.RESPONDING],
other_states=other_states(AssistSatelliteState.RESPONDING),
),
],
)
async def test_assist_satellite_state_trigger_behavior_last(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_assist_satellites: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the assist_satellite state trigger fires when the last assist_satellite changes to a specific state."""
await async_setup_component(hass, "assist_satellite", {})
other_entity_ids = set(target_assist_satellites) - {entity_id}
# Set all assist satellites, including the tested one, to the initial state
for eid in target_assist_satellites:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
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) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()

View File

@@ -1,381 +0,0 @@
"""Test climate trigger."""
import pytest
from homeassistant.components.climate.const import (
ATTR_HVAC_ACTION,
HVACAction,
HVACMode,
)
from homeassistant.const import CONF_ENTITY_ID
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.setup import async_setup_component
from tests.components import (
StateDescription,
arm_trigger,
other_states,
parametrize_target_entities,
parametrize_trigger_states,
set_or_remove_state,
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")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("climate"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="climate.turned_off",
target_states=[HVACMode.OFF],
other_states=other_states(HVACMode.OFF),
),
*parametrize_trigger_states(
trigger="climate.turned_on",
target_states=[
HVACMode.AUTO,
HVACMode.COOL,
HVACMode.DRY,
HVACMode.FAN_ONLY,
HVACMode.HEAT,
HVACMode.HEAT_COOL,
],
other_states=[
HVACMode.OFF,
],
),
],
)
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,
states: list[StateDescription],
) -> 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, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# 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) * state["count"]
service_calls.clear()
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("climate"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="climate.started_heating",
target_states=[(HVACMode.OFF, {ATTR_HVAC_ACTION: HVACAction.HEATING})],
other_states=[(HVACMode.OFF, {ATTR_HVAC_ACTION: 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,
states: list[StateDescription],
) -> 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, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# 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) * state["count"]
service_calls.clear()
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("climate"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="climate.turned_off",
target_states=[HVACMode.OFF],
other_states=other_states(HVACMode.OFF),
),
*parametrize_trigger_states(
trigger="climate.turned_on",
target_states=[
HVACMode.AUTO,
HVACMode.COOL,
HVACMode.DRY,
HVACMode.FAN_ONLY,
HVACMode.HEAT,
HVACMode.HEAT_COOL,
],
other_states=[
HVACMode.OFF,
],
),
],
)
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,
states: list[StateDescription],
) -> 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, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# 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", "states"),
[
*parametrize_trigger_states(
trigger="climate.started_heating",
target_states=[(HVACMode.OFF, {ATTR_HVAC_ACTION: HVACAction.HEATING})],
other_states=[(HVACMode.OFF, {ATTR_HVAC_ACTION: 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,
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, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# 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", "states"),
[
*parametrize_trigger_states(
trigger="climate.turned_off",
target_states=[HVACMode.OFF],
other_states=other_states(HVACMode.OFF),
),
*parametrize_trigger_states(
trigger="climate.turned_on",
target_states=[
HVACMode.AUTO,
HVACMode.COOL,
HVACMode.DRY,
HVACMode.FAN_ONLY,
HVACMode.HEAT,
HVACMode.HEAT_COOL,
],
other_states=[
HVACMode.OFF,
],
),
],
)
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,
states: list[StateDescription],
) -> 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, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
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) == state["count"]
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", "states"),
[
*parametrize_trigger_states(
trigger="climate.started_heating",
target_states=[(HVACMode.OFF, {ATTR_HVAC_ACTION: HVACAction.HEATING})],
other_states=[(HVACMode.OFF, {ATTR_HVAC_ACTION: 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,
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, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
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) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()

View File

@@ -1,254 +0,0 @@
"""Test cover trigger."""
import pytest
from homeassistant.components.cover import ATTR_CURRENT_POSITION, CoverState
from homeassistant.const import ATTR_DEVICE_CLASS, CONF_ENTITY_ID
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.setup import async_setup_component
from tests.components import (
StateDescription,
arm_trigger,
parametrize_target_entities,
parametrize_trigger_states,
set_or_remove_state,
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 parametrize_opened_trigger_states(
trigger: str, device_class: str
) -> list[tuple[str, dict, str, list[StateDescription]]]:
"""Parametrize states and expected service call counts.
Returns a list of tuples with (trigger, trigger_options,
list of StateDescription).
"""
additional_attributes = {ATTR_DEVICE_CLASS: device_class}
return [
# Test fully_opened = True
*(
(s[0], {"fully_opened": True}, *s[1:])
for s in parametrize_trigger_states(
trigger=trigger,
target_states=[
(CoverState.OPEN, {}),
(CoverState.OPENING, {}),
(CoverState.OPEN, {ATTR_CURRENT_POSITION: 100}),
(CoverState.OPENING, {ATTR_CURRENT_POSITION: 100}),
],
other_states=[
(CoverState.CLOSED, {}),
(CoverState.OPEN, {ATTR_CURRENT_POSITION: 0}),
],
additional_attributes=additional_attributes,
trigger_from_none=False,
)
),
# Test fully_opened = False
*(
(s[0], {}, *s[1:])
for s in parametrize_trigger_states(
trigger=trigger,
target_states=[
(CoverState.OPEN, {}),
(CoverState.OPENING, {}),
(CoverState.OPEN, {ATTR_CURRENT_POSITION: 1}),
(CoverState.OPENING, {ATTR_CURRENT_POSITION: 1}),
],
other_states=[
(CoverState.CLOSED, {}),
(CoverState.CLOSED, {ATTR_CURRENT_POSITION: 0}),
],
additional_attributes=additional_attributes,
trigger_from_none=False,
)
),
]
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("cover"),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_opened_trigger_states("cover.awning_opened", "awning"),
*parametrize_opened_trigger_states("cover.blind_opened", "blind"),
*parametrize_opened_trigger_states("cover.curtain_opened", "curtain"),
*parametrize_opened_trigger_states("cover.door_opened", "door"),
*parametrize_opened_trigger_states("cover.garage_opened", "garage"),
*parametrize_opened_trigger_states("cover.gate_opened", "gate"),
*parametrize_opened_trigger_states("cover.shade_opened", "shade"),
*parametrize_opened_trigger_states("cover.shutter_opened", "shutter"),
*parametrize_opened_trigger_states("cover.window_opened", "window"),
],
)
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,
states: list[StateDescription],
) -> 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, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, trigger_options, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Check if changing other covers 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) * state["count"]
service_calls.clear()
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("cover"),
)
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_opened_trigger_states("cover.awning_opened", "awning"),
*parametrize_opened_trigger_states("cover.blind_opened", "blind"),
*parametrize_opened_trigger_states("cover.curtain_opened", "curtain"),
*parametrize_opened_trigger_states("cover.door_opened", "door"),
*parametrize_opened_trigger_states("cover.garage_opened", "garage"),
*parametrize_opened_trigger_states("cover.gate_opened", "gate"),
*parametrize_opened_trigger_states("cover.shade_opened", "shade"),
*parametrize_opened_trigger_states("cover.shutter_opened", "shutter"),
*parametrize_opened_trigger_states("cover.window_opened", "window"),
],
)
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,
states: list[StateDescription],
) -> 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, states[0])
await hass.async_block_till_done()
await arm_trigger(
hass,
trigger,
{"behavior": "first"} | trigger_options,
trigger_target_config,
)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# 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)
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", "states"),
[
*parametrize_opened_trigger_states("cover.awning_opened", "awning"),
*parametrize_opened_trigger_states("cover.blind_opened", "blind"),
*parametrize_opened_trigger_states("cover.curtain_opened", "curtain"),
*parametrize_opened_trigger_states("cover.door_opened", "door"),
*parametrize_opened_trigger_states("cover.garage_opened", "garage"),
*parametrize_opened_trigger_states("cover.gate_opened", "gate"),
*parametrize_opened_trigger_states("cover.shade_opened", "shade"),
*parametrize_opened_trigger_states("cover.shutter_opened", "shutter"),
*parametrize_opened_trigger_states("cover.window_opened", "window"),
],
)
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,
states: list[StateDescription],
) -> 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, states[0])
await hass.async_block_till_done()
await arm_trigger(
hass, trigger, {"behavior": "last"} | trigger_options, trigger_target_config
)
for state in states[1:]:
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) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()

View File

@@ -1,195 +0,0 @@
"""Test fan trigger."""
import pytest
from homeassistant.const import CONF_ENTITY_ID, STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.setup import async_setup_component
from tests.components import (
StateDescription,
arm_trigger,
parametrize_target_entities,
parametrize_trigger_states,
set_or_remove_state,
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_fans(hass: HomeAssistant) -> None:
"""Create multiple fan entities associated with different targets."""
return await target_entities(hass, "fan")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("fan"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="fan.turned_on",
target_states=[STATE_ON],
other_states=[STATE_OFF],
),
*parametrize_trigger_states(
trigger="fan.turned_off",
target_states=[STATE_OFF],
other_states=[STATE_ON],
),
],
)
async def test_fan_state_trigger_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_fans: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the fan state trigger fires when any fan state changes to a specific state."""
await async_setup_component(hass, "fan", {})
other_entity_ids = set(target_fans) - {entity_id}
# Set all fans, including the tested fan, to the initial state
for eid in target_fans:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Check if changing other fans 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) * state["count"]
service_calls.clear()
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("fan"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="fan.turned_on",
target_states=[STATE_ON],
other_states=[STATE_OFF],
),
*parametrize_trigger_states(
trigger="fan.turned_off",
target_states=[STATE_OFF],
other_states=[STATE_ON],
),
],
)
async def test_fan_state_trigger_behavior_first(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_fans: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the fan state trigger fires when the first fan changes to a specific state."""
await async_setup_component(hass, "fan", {})
other_entity_ids = set(target_fans) - {entity_id}
# Set all fans, including the tested fan, to the initial state
for eid in target_fans:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Triggering other fans should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
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("fan"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="fan.turned_on",
target_states=[STATE_ON],
other_states=[STATE_OFF],
),
*parametrize_trigger_states(
trigger="fan.turned_off",
target_states=[STATE_OFF],
other_states=[STATE_ON],
),
],
)
async def test_fan_state_trigger_behavior_last(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_fans: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the fan state trigger fires when the last fan changes to a specific state."""
await async_setup_component(hass, "fan", {})
other_entity_ids = set(target_fans) - {entity_id}
# Set all fans, including the tested fan, to the initial state
for eid in target_fans:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
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) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()

View File

@@ -1,227 +0,0 @@
"""Test lawn mower triggers."""
import pytest
from homeassistant.components.lawn_mower import LawnMowerActivity
from homeassistant.const import CONF_ENTITY_ID
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.setup import async_setup_component
from tests.components import (
StateDescription,
arm_trigger,
other_states,
parametrize_target_entities,
parametrize_trigger_states,
set_or_remove_state,
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_lawn_mowers(hass: HomeAssistant) -> None:
"""Create multiple lawn mower entities associated with different targets."""
return await target_entities(hass, "lawn_mower")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("lawn_mower"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="lawn_mower.docked",
target_states=[LawnMowerActivity.DOCKED],
other_states=other_states(LawnMowerActivity.DOCKED),
),
*parametrize_trigger_states(
trigger="lawn_mower.errored",
target_states=[LawnMowerActivity.ERROR],
other_states=other_states(LawnMowerActivity.ERROR),
),
*parametrize_trigger_states(
trigger="lawn_mower.paused_mowing",
target_states=[LawnMowerActivity.PAUSED],
other_states=other_states(LawnMowerActivity.PAUSED),
),
*parametrize_trigger_states(
trigger="lawn_mower.started_mowing",
target_states=[LawnMowerActivity.MOWING],
other_states=other_states(LawnMowerActivity.MOWING),
),
],
)
async def test_lawn_mower_state_trigger_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_lawn_mowers: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the lawn mower state trigger fires when any lawn mower state changes to a specific state."""
await async_setup_component(hass, "lawn_mower", {})
other_entity_ids = set(target_lawn_mowers) - {entity_id}
# Set all lawn mowers, including the tested one, to the initial state
for eid in target_lawn_mowers:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Check if changing other lawn mowers also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * state["count"]
service_calls.clear()
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("lawn_mower"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="lawn_mower.docked",
target_states=[LawnMowerActivity.DOCKED],
other_states=other_states(LawnMowerActivity.DOCKED),
),
*parametrize_trigger_states(
trigger="lawn_mower.errored",
target_states=[LawnMowerActivity.ERROR],
other_states=other_states(LawnMowerActivity.ERROR),
),
*parametrize_trigger_states(
trigger="lawn_mower.paused_mowing",
target_states=[LawnMowerActivity.PAUSED],
other_states=other_states(LawnMowerActivity.PAUSED),
),
*parametrize_trigger_states(
trigger="lawn_mower.started_mowing",
target_states=[LawnMowerActivity.MOWING],
other_states=other_states(LawnMowerActivity.MOWING),
),
],
)
async def test_lawn_mower_state_trigger_behavior_first(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_lawn_mowers: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the lawn mower state trigger fires when the first lawn mower changes to a specific state."""
await async_setup_component(hass, "lawn_mower", {})
other_entity_ids = set(target_lawn_mowers) - {entity_id}
# Set all lawn mowers, including the tested one, to the initial state
for eid in target_lawn_mowers:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Triggering other lawn mowers should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
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("lawn_mower"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="lawn_mower.docked",
target_states=[LawnMowerActivity.DOCKED],
other_states=other_states(LawnMowerActivity.DOCKED),
),
*parametrize_trigger_states(
trigger="lawn_mower.errored",
target_states=[LawnMowerActivity.ERROR],
other_states=other_states(LawnMowerActivity.ERROR),
),
*parametrize_trigger_states(
trigger="lawn_mower.paused_mowing",
target_states=[LawnMowerActivity.PAUSED],
other_states=other_states(LawnMowerActivity.PAUSED),
),
*parametrize_trigger_states(
trigger="lawn_mower.started_mowing",
target_states=[LawnMowerActivity.MOWING],
other_states=other_states(LawnMowerActivity.MOWING),
),
],
)
async def test_lawn_mower_state_trigger_behavior_last(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_lawn_mowers: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the lawn_mower state trigger fires when the last lawn_mower changes to a specific state."""
await async_setup_component(hass, "lawn_mower", {})
other_entity_ids = set(target_lawn_mowers) - {entity_id}
# Set all lawn mowers, including the tested one, to the initial state
for eid in target_lawn_mowers:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
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) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()

View File

@@ -1,203 +0,0 @@
"""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

@@ -1,195 +0,0 @@
"""Test light trigger."""
import pytest
from homeassistant.const import CONF_ENTITY_ID, STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.setup import async_setup_component
from tests.components import (
StateDescription,
arm_trigger,
parametrize_target_entities,
parametrize_trigger_states,
set_or_remove_state,
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")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("light"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="light.turned_on",
target_states=[STATE_ON],
other_states=[STATE_OFF],
),
*parametrize_trigger_states(
trigger="light.turned_off",
target_states=[STATE_OFF],
other_states=[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,
states: list[StateDescription],
) -> 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, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# 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) * state["count"]
service_calls.clear()
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("light"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="light.turned_on",
target_states=[STATE_ON],
other_states=[STATE_OFF],
),
*parametrize_trigger_states(
trigger="light.turned_off",
target_states=[STATE_OFF],
other_states=[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,
states: list[StateDescription],
) -> 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, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# 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", "states"),
[
*parametrize_trigger_states(
trigger="light.turned_on",
target_states=[STATE_ON],
other_states=[STATE_OFF],
),
*parametrize_trigger_states(
trigger="light.turned_off",
target_states=[STATE_OFF],
other_states=[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,
states: list[StateDescription],
) -> 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, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
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) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()

View File

@@ -1,205 +0,0 @@
"""Test media player trigger."""
import pytest
from homeassistant.components.media_player import MediaPlayerState
from homeassistant.const import CONF_ENTITY_ID
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.setup import async_setup_component
from tests.components import (
StateDescription,
arm_trigger,
parametrize_target_entities,
parametrize_trigger_states,
set_or_remove_state,
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_media_players(hass: HomeAssistant) -> None:
"""Create multiple media player entities associated with different targets."""
return await target_entities(hass, "media_player")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("media_player"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="media_player.stopped_playing",
target_states=[
MediaPlayerState.IDLE,
MediaPlayerState.OFF,
MediaPlayerState.ON,
],
other_states=[
MediaPlayerState.BUFFERING,
MediaPlayerState.PAUSED,
MediaPlayerState.PLAYING,
],
),
],
)
async def test_media_player_state_trigger_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_media_players: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the media player state trigger fires when any media player state changes to a specific state."""
await async_setup_component(hass, "media_player", {})
other_entity_ids = set(target_media_players) - {entity_id}
# Set all media players, including the tested media player, to the initial state
for eid in target_media_players:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Check if changing other media players also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * state["count"]
service_calls.clear()
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("media_player"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="media_player.stopped_playing",
target_states=[
MediaPlayerState.IDLE,
MediaPlayerState.OFF,
MediaPlayerState.ON,
],
other_states=[
MediaPlayerState.BUFFERING,
MediaPlayerState.PAUSED,
MediaPlayerState.PLAYING,
],
),
],
)
async def test_media_player_state_trigger_behavior_first(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_media_players: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the media player state trigger fires when the first media player changes to a specific state."""
await async_setup_component(hass, "media_player", {})
other_entity_ids = set(target_media_players) - {entity_id}
# Set all media players, including the tested media player, to the initial state
for eid in target_media_players:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Triggering other media players should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
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("media_player"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="media_player.stopped_playing",
target_states=[
MediaPlayerState.IDLE,
MediaPlayerState.OFF,
MediaPlayerState.ON,
],
other_states=[
MediaPlayerState.BUFFERING,
MediaPlayerState.PAUSED,
MediaPlayerState.PLAYING,
],
),
],
)
async def test_media_player_state_trigger_behavior_last(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_media_players: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the media player state trigger fires when the last media player changes to a specific state."""
await async_setup_component(hass, "media_player", {})
other_entity_ids = set(target_media_players) - {entity_id}
# Set all media players, including the tested media player, to the initial state
for eid in target_media_players:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
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) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()

View File

@@ -1,113 +0,0 @@
"""Test text trigger."""
import pytest
from homeassistant.const import CONF_ENTITY_ID, STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.setup import async_setup_component
from tests.components import (
StateDescription,
arm_trigger,
parametrize_target_entities,
set_or_remove_state,
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_texts(hass: HomeAssistant) -> None:
"""Create multiple text entities associated with different targets."""
return await target_entities(hass, "text")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("text"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
(
"text.changed",
[
{"state": None, "attributes": {}, "count": 0},
{"state": "bar", "attributes": {}, "count": 0},
{"state": "baz", "attributes": {}, "count": 1},
],
),
(
"text.changed",
[
{"state": "foo", "attributes": {}, "count": 0},
{"state": "bar", "attributes": {}, "count": 1},
{"state": "baz", "attributes": {}, "count": 1},
],
),
(
"text.changed",
[
{"state": "foo", "attributes": {}, "count": 0},
{"state": "", "attributes": {}, "count": 1}, # empty string
{"state": "baz", "attributes": {}, "count": 1},
],
),
(
"text.changed",
[
{"state": STATE_UNAVAILABLE, "attributes": {}, "count": 0},
{"state": "bar", "attributes": {}, "count": 0},
{"state": "baz", "attributes": {}, "count": 1},
],
),
(
"text.changed",
[
{"state": STATE_UNKNOWN, "attributes": {}, "count": 0},
{"state": "bar", "attributes": {}, "count": 0},
{"state": "baz", "attributes": {}, "count": 1},
],
),
],
)
async def test_text_state_trigger_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_texts: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the text state trigger fires when any text state changes to a specific state."""
await async_setup_component(hass, "text", {})
other_entity_ids = set(target_texts) - {entity_id}
# Set all texts, including the tested text, to the initial state
for eid in target_texts:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Check if changing other texts 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) * state["count"]
service_calls.clear()

View File

@@ -1,227 +0,0 @@
"""Test vacuum triggers."""
import pytest
from homeassistant.components.vacuum import VacuumActivity
from homeassistant.const import CONF_ENTITY_ID
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.setup import async_setup_component
from tests.components import (
StateDescription,
arm_trigger,
other_states,
parametrize_target_entities,
parametrize_trigger_states,
set_or_remove_state,
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_vacuums(hass: HomeAssistant) -> None:
"""Create multiple vacuum entities associated with different targets."""
return await target_entities(hass, "vacuum")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("vacuum"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="vacuum.docked",
target_states=[VacuumActivity.DOCKED],
other_states=other_states(VacuumActivity.DOCKED),
),
*parametrize_trigger_states(
trigger="vacuum.errored",
target_states=[VacuumActivity.ERROR],
other_states=other_states(VacuumActivity.ERROR),
),
*parametrize_trigger_states(
trigger="vacuum.paused_cleaning",
target_states=[VacuumActivity.PAUSED],
other_states=other_states(VacuumActivity.PAUSED),
),
*parametrize_trigger_states(
trigger="vacuum.started_cleaning",
target_states=[VacuumActivity.CLEANING],
other_states=other_states(VacuumActivity.CLEANING),
),
],
)
async def test_vacuum_state_trigger_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_vacuums: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the vacuum state trigger fires when any vacuum state changes to a specific state."""
await async_setup_component(hass, "vacuum", {})
other_entity_ids = set(target_vacuums) - {entity_id}
# Set all vacuums, including the tested one, to the initial state
for eid in target_vacuums:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Check if changing other vacuums 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) * state["count"]
service_calls.clear()
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("vacuum"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="vacuum.docked",
target_states=[VacuumActivity.DOCKED],
other_states=other_states(VacuumActivity.DOCKED),
),
*parametrize_trigger_states(
trigger="vacuum.errored",
target_states=[VacuumActivity.ERROR],
other_states=other_states(VacuumActivity.ERROR),
),
*parametrize_trigger_states(
trigger="vacuum.paused_cleaning",
target_states=[VacuumActivity.PAUSED],
other_states=other_states(VacuumActivity.PAUSED),
),
*parametrize_trigger_states(
trigger="vacuum.started_cleaning",
target_states=[VacuumActivity.CLEANING],
other_states=other_states(VacuumActivity.CLEANING),
),
],
)
async def test_vacuum_state_trigger_behavior_first(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_vacuums: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the vacuum state trigger fires when the first vacuum changes to a specific state."""
await async_setup_component(hass, "vacuum", {})
other_entity_ids = set(target_vacuums) - {entity_id}
# Set all vacuums, including the tested one, to the initial state
for eid in target_vacuums:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
for state in states[1:]:
set_or_remove_state(hass, entity_id, state)
await hass.async_block_till_done()
assert len(service_calls) == state["count"]
for service_call in service_calls:
assert service_call.data[CONF_ENTITY_ID] == entity_id
service_calls.clear()
# Triggering other vacuums should not cause the trigger to fire again
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, state)
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("vacuum"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
*parametrize_trigger_states(
trigger="vacuum.docked",
target_states=[VacuumActivity.DOCKED],
other_states=other_states(VacuumActivity.DOCKED),
),
*parametrize_trigger_states(
trigger="vacuum.errored",
target_states=[VacuumActivity.ERROR],
other_states=other_states(VacuumActivity.ERROR),
),
*parametrize_trigger_states(
trigger="vacuum.paused_cleaning",
target_states=[VacuumActivity.PAUSED],
other_states=other_states(VacuumActivity.PAUSED),
),
*parametrize_trigger_states(
trigger="vacuum.started_cleaning",
target_states=[VacuumActivity.CLEANING],
other_states=other_states(VacuumActivity.CLEANING),
),
],
)
async def test_vacuum_state_trigger_behavior_last(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_vacuums: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the vacuum state trigger fires when the last vacuum changes to a specific state."""
await async_setup_component(hass, "vacuum", {})
other_entity_ids = set(target_vacuums) - {entity_id}
# Set all vacuums, including the tested one, to the initial state
for eid in target_vacuums:
set_or_remove_state(hass, eid, states[0])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
for state in states[1:]:
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) == state["count"]
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) == 9
assert len(mock_build_resources.mock_calls) == 7
load2 = await translation.async_get_translations(hass, "en", "entity_component")
assert len(mock_build_resources.mock_calls) == 9
assert len(mock_build_resources.mock_calls) == 7
assert load1 == load2

View File

@@ -647,9 +647,14 @@ async def test_async_get_all_descriptions(
"""Test async_get_all_descriptions."""
tag_trigger_descriptions = """
_:
target:
fields:
entity:
domain: alarm_control_panel
selector:
entity:
filter:
domain: alarm_control_panel
supported_features:
- alarm_control_panel.AlarmControlPanelEntityFeature.ARM_HOME
"""
assert await async_setup_component(hass, DOMAIN_SUN, {})
@@ -739,14 +744,22 @@ async def test_async_get_all_descriptions(
}
},
"tag": {
"target": {
"entity": [
{
"domain": ["alarm_control_panel"],
}
],
},
"fields": {},
"fields": {
"entity": {
"selector": {
"entity": {
"filter": [
{
"domain": ["alarm_control_panel"],
"supported_features": [1],
}
],
"multiple": False,
"reorder": False,
},
},
},
}
},
}
@@ -878,5 +891,6 @@ async def test_subscribe_triggers(
trigger.async_subscribe_platform_events(hass, good_subscriber)
assert await async_setup_component(hass, "sun", {})
assert trigger_events == [{"sun"}]
assert "Error while notifying trigger platform listener" in caplog.text