mirror of
https://github.com/home-assistant/core.git
synced 2026-03-07 06:24:56 +01:00
Compare commits
24 Commits
be370
...
add_trigge
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a0cb699e15 | ||
|
|
bee7b1da7d | ||
|
|
ac684b95dc | ||
|
|
f9baa1c7e2 | ||
|
|
a767ddcfc0 | ||
|
|
6cdd97075b | ||
|
|
4bda34dfdb | ||
|
|
f114717b97 | ||
|
|
743d841446 | ||
|
|
b976c5bd16 | ||
|
|
80724e507e | ||
|
|
79b4626a62 | ||
|
|
dab919bae3 | ||
|
|
707e8852fa | ||
|
|
90d7f8f779 | ||
|
|
b024c4811a | ||
|
|
9312791164 | ||
|
|
bd22e526ad | ||
|
|
ae8f9b8e3a | ||
|
|
45145e6cc0 | ||
|
|
ee4aac8747 | ||
|
|
2c836a782e | ||
|
|
3a58771bcc | ||
|
|
a855e0018b |
2
CODEOWNERS
generated
2
CODEOWNERS
generated
@@ -383,6 +383,8 @@ build.json @home-assistant/supervisor
|
||||
/tests/components/dlna_dms/ @chishm
|
||||
/homeassistant/components/dnsip/ @gjohansson-ST
|
||||
/tests/components/dnsip/ @gjohansson-ST
|
||||
/homeassistant/components/door/ @home-assistant/core
|
||||
/tests/components/door/ @home-assistant/core
|
||||
/homeassistant/components/doorbird/ @oblogic7 @bdraco @flacjacket
|
||||
/tests/components/doorbird/ @oblogic7 @bdraco @flacjacket
|
||||
/homeassistant/components/dormakaba_dkey/ @emontnemery
|
||||
|
||||
@@ -239,6 +239,9 @@ DEFAULT_INTEGRATIONS = {
|
||||
#
|
||||
# Base platforms:
|
||||
*BASE_PLATFORMS,
|
||||
#
|
||||
# Integrations providing triggers and conditions for base platforms:
|
||||
"door",
|
||||
}
|
||||
DEFAULT_INTEGRATIONS_RECOVERY_MODE = {
|
||||
# These integrations are set up if recovery mode is activated.
|
||||
|
||||
@@ -142,6 +142,7 @@ _EXPERIMENTAL_TRIGGER_PLATFORMS = {
|
||||
"climate",
|
||||
"cover",
|
||||
"device_tracker",
|
||||
"door",
|
||||
"fan",
|
||||
"humidifier",
|
||||
"lawn_mower",
|
||||
|
||||
15
homeassistant/components/door/__init__.py
Normal file
15
homeassistant/components/door/__init__.py
Normal file
@@ -0,0 +1,15 @@
|
||||
"""Integration for door triggers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
DOMAIN = "door"
|
||||
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
|
||||
|
||||
|
||||
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
"""Set up the component."""
|
||||
return True
|
||||
7
homeassistant/components/door/icons.json
Normal file
7
homeassistant/components/door/icons.json
Normal file
@@ -0,0 +1,7 @@
|
||||
{
|
||||
"triggers": {
|
||||
"opened": {
|
||||
"trigger": "mdi:door-open"
|
||||
}
|
||||
}
|
||||
}
|
||||
8
homeassistant/components/door/manifest.json
Normal file
8
homeassistant/components/door/manifest.json
Normal file
@@ -0,0 +1,8 @@
|
||||
{
|
||||
"domain": "door",
|
||||
"name": "Door",
|
||||
"codeowners": ["@home-assistant/core"],
|
||||
"documentation": "https://www.home-assistant.io/integrations/door",
|
||||
"integration_type": "system",
|
||||
"quality_scale": "internal"
|
||||
}
|
||||
28
homeassistant/components/door/strings.json
Normal file
28
homeassistant/components/door/strings.json
Normal file
@@ -0,0 +1,28 @@
|
||||
{
|
||||
"common": {
|
||||
"trigger_behavior_description": "The behavior of the targeted doors to trigger on.",
|
||||
"trigger_behavior_name": "Behavior"
|
||||
},
|
||||
"selector": {
|
||||
"trigger_behavior": {
|
||||
"options": {
|
||||
"any": "Any",
|
||||
"first": "First",
|
||||
"last": "Last"
|
||||
}
|
||||
}
|
||||
},
|
||||
"title": "Door",
|
||||
"triggers": {
|
||||
"opened": {
|
||||
"description": "Triggers after one or more doors open.",
|
||||
"fields": {
|
||||
"behavior": {
|
||||
"description": "[%key:component::door::common::trigger_behavior_description%]",
|
||||
"name": "[%key:component::door::common::trigger_behavior_name%]"
|
||||
}
|
||||
},
|
||||
"name": "Door opened"
|
||||
}
|
||||
}
|
||||
}
|
||||
75
homeassistant/components/door/trigger.py
Normal file
75
homeassistant/components/door/trigger.py
Normal file
@@ -0,0 +1,75 @@
|
||||
"""Provides triggers for doors."""
|
||||
|
||||
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
|
||||
from homeassistant.components.cover import ATTR_IS_CLOSED, DOMAIN as COVER_DOMAIN
|
||||
from homeassistant.const import STATE_ON, STATE_UNAVAILABLE, STATE_UNKNOWN
|
||||
from homeassistant.core import HomeAssistant, State, split_entity_id
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers.entity import get_device_class
|
||||
from homeassistant.helpers.trigger import EntityTriggerBase, Trigger
|
||||
from homeassistant.helpers.typing import UNDEFINED, UndefinedType
|
||||
|
||||
DEVICE_CLASS_DOOR = "door"
|
||||
|
||||
|
||||
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 DoorTriggerBase(EntityTriggerBase):
|
||||
"""Base trigger for door state changes."""
|
||||
|
||||
_domains = {BINARY_SENSOR_DOMAIN, COVER_DOMAIN}
|
||||
_binary_sensor_target_state: str
|
||||
_cover_is_closed_target_value: bool
|
||||
|
||||
def entity_filter(self, entities: set[str]) -> set[str]:
|
||||
"""Filter entities by door device class."""
|
||||
entities = super().entity_filter(entities)
|
||||
return {
|
||||
entity_id
|
||||
for entity_id in entities
|
||||
if get_device_class_or_undefined(self._hass, entity_id) == DEVICE_CLASS_DOOR
|
||||
}
|
||||
|
||||
def is_valid_state(self, state: State) -> bool:
|
||||
"""Check if the state matches the target door state."""
|
||||
if split_entity_id(state.entity_id)[0] == COVER_DOMAIN:
|
||||
return (
|
||||
state.attributes.get(ATTR_IS_CLOSED)
|
||||
== self._cover_is_closed_target_value
|
||||
)
|
||||
return state.state == self._binary_sensor_target_state
|
||||
|
||||
def is_valid_transition(self, from_state: State, to_state: State) -> bool:
|
||||
"""Check if the transition is valid for a door state change."""
|
||||
if from_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN):
|
||||
return False
|
||||
if split_entity_id(from_state.entity_id)[0] == COVER_DOMAIN:
|
||||
if (from_is_closed := from_state.attributes.get(ATTR_IS_CLOSED)) is None:
|
||||
return False
|
||||
return from_is_closed != to_state.attributes.get(ATTR_IS_CLOSED)
|
||||
return from_state.state != to_state.state
|
||||
|
||||
|
||||
class DoorOpenedTrigger(DoorTriggerBase):
|
||||
"""Trigger for door opened state changes."""
|
||||
|
||||
_binary_sensor_target_state = STATE_ON
|
||||
_cover_is_closed_target_value = False
|
||||
|
||||
|
||||
TRIGGERS: dict[str, type[Trigger]] = {
|
||||
"opened": DoorOpenedTrigger,
|
||||
}
|
||||
|
||||
|
||||
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
|
||||
"""Return the triggers for doors."""
|
||||
return TRIGGERS
|
||||
20
homeassistant/components/door/triggers.yaml
Normal file
20
homeassistant/components/door/triggers.yaml
Normal file
@@ -0,0 +1,20 @@
|
||||
.trigger_common_fields: &trigger_common_fields
|
||||
behavior:
|
||||
required: true
|
||||
default: any
|
||||
selector:
|
||||
select:
|
||||
translation_key: trigger_behavior
|
||||
options:
|
||||
- first
|
||||
- last
|
||||
- any
|
||||
|
||||
opened:
|
||||
fields: *trigger_common_fields
|
||||
target:
|
||||
entity:
|
||||
- domain: binary_sensor
|
||||
device_class: door
|
||||
- domain: cover
|
||||
device_class: door
|
||||
@@ -71,6 +71,7 @@ NO_IOT_CLASS = [
|
||||
"device_automation",
|
||||
"device_tracker",
|
||||
"diagnostics",
|
||||
"door",
|
||||
"downloader",
|
||||
"ffmpeg",
|
||||
"file_upload",
|
||||
|
||||
@@ -2110,6 +2110,7 @@ NO_QUALITY_SCALE = [
|
||||
"device_automation",
|
||||
"device_tracker",
|
||||
"diagnostics",
|
||||
"door",
|
||||
"ffmpeg",
|
||||
"file_upload",
|
||||
"frontend",
|
||||
|
||||
@@ -355,17 +355,20 @@ def parametrize_trigger_states(
|
||||
trigger_options: dict[str, Any] | None = None,
|
||||
target_states: list[str | None | tuple[str | None, dict]],
|
||||
other_states: list[str | None | tuple[str | None, dict]],
|
||||
extra_invalid_states: list[str | None | tuple[str | None, dict]] | None = None,
|
||||
additional_attributes: dict | None = None,
|
||||
trigger_from_none: bool = True,
|
||||
retrigger_on_target_state: bool = False,
|
||||
) -> list[tuple[str, dict[str, Any], list[TriggerStateDescription]]]:
|
||||
"""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.
|
||||
The target_states, other_states, and extra_invalid_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.
|
||||
when the initial state is None, this is relevant for triggers that limit
|
||||
entities to a certain device class because the device class can't be
|
||||
determined when the state is None.
|
||||
|
||||
Set `retrigger_on_target_state` to True if the trigger is expected to fire
|
||||
when the state changes to another target state.
|
||||
@@ -374,6 +377,8 @@ def parametrize_trigger_states(
|
||||
where states is a list of TriggerStateDescription dicts.
|
||||
"""
|
||||
|
||||
extra_invalid_states = extra_invalid_states or []
|
||||
invalid_states = [STATE_UNAVAILABLE, STATE_UNKNOWN, *extra_invalid_states]
|
||||
additional_attributes = additional_attributes or {}
|
||||
trigger_options = trigger_options or {}
|
||||
|
||||
@@ -463,34 +468,19 @@ def parametrize_trigger_states(
|
||||
)
|
||||
),
|
||||
),
|
||||
# Initial state unavailable / unknown
|
||||
# Initial state unavailable / unknown + extra invalid states
|
||||
(
|
||||
trigger,
|
||||
trigger_options,
|
||||
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,
|
||||
trigger_options,
|
||||
list(
|
||||
itertools.chain.from_iterable(
|
||||
(
|
||||
state_with_attributes(STATE_UNKNOWN, 0),
|
||||
state_with_attributes(invalid_state, 0),
|
||||
state_with_attributes(target_state, 0),
|
||||
state_with_attributes(other_state, 0),
|
||||
state_with_attributes(target_state, 1),
|
||||
)
|
||||
for invalid_state in invalid_states
|
||||
for target_state in target_states
|
||||
for other_state in other_states
|
||||
)
|
||||
|
||||
1
tests/components/door/__init__.py
Normal file
1
tests/components/door/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""Tests for the door integration."""
|
||||
572
tests/components/door/test_trigger.py
Normal file
572
tests/components/door/test_trigger.py
Normal file
@@ -0,0 +1,572 @@
|
||||
"""Test door trigger."""
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
|
||||
from homeassistant.components.cover import ATTR_IS_CLOSED, CoverDeviceClass, CoverState
|
||||
from homeassistant.components.door.trigger import DEVICE_CLASS_DOOR
|
||||
from homeassistant.const import (
|
||||
ATTR_DEVICE_CLASS,
|
||||
ATTR_LABEL_ID,
|
||||
CONF_ENTITY_ID,
|
||||
STATE_OFF,
|
||||
STATE_ON,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant, ServiceCall
|
||||
|
||||
from tests.components import (
|
||||
TriggerStateDescription,
|
||||
arm_trigger,
|
||||
parametrize_target_entities,
|
||||
parametrize_trigger_states,
|
||||
set_or_remove_state,
|
||||
target_entities,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def target_binary_sensors(hass: HomeAssistant) -> dict[str, list[str]]:
|
||||
"""Create multiple binary sensor entities associated with different targets."""
|
||||
return await target_entities(hass, "binary_sensor")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def target_covers(hass: HomeAssistant) -> dict[str, list[str]]:
|
||||
"""Create multiple cover entities associated with different targets."""
|
||||
return await target_entities(hass, "cover")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"trigger_key",
|
||||
[
|
||||
"door.opened",
|
||||
],
|
||||
)
|
||||
async def test_door_triggers_gated_by_labs_flag(
|
||||
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, trigger_key: str
|
||||
) -> None:
|
||||
"""Test the door triggers are gated by the labs flag."""
|
||||
await arm_trigger(hass, trigger_key, None, {ATTR_LABEL_ID: "test_label"})
|
||||
assert (
|
||||
"Unnamed automation failed to setup triggers and has been disabled: Trigger "
|
||||
f"'{trigger_key}' requires the experimental 'New triggers and conditions' "
|
||||
"feature to be enabled in Home Assistant Labs settings (feature flag: "
|
||||
"'new_triggers_conditions')"
|
||||
) in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("enable_labs_preview_features")
|
||||
@pytest.mark.parametrize(
|
||||
("trigger_target_config", "entity_id", "entities_in_target"),
|
||||
parametrize_target_entities("binary_sensor"),
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
("trigger", "trigger_options", "states"),
|
||||
[
|
||||
*parametrize_trigger_states(
|
||||
trigger="door.opened",
|
||||
target_states=[STATE_ON],
|
||||
other_states=[STATE_OFF],
|
||||
additional_attributes={ATTR_DEVICE_CLASS: "door"},
|
||||
trigger_from_none=False,
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_door_trigger_binary_sensor_behavior_any(
|
||||
hass: HomeAssistant,
|
||||
service_calls: list[ServiceCall],
|
||||
target_binary_sensors: dict[str, list[str]],
|
||||
trigger_target_config: dict,
|
||||
entity_id: str,
|
||||
entities_in_target: int,
|
||||
trigger: str,
|
||||
trigger_options: dict[str, Any],
|
||||
states: list[TriggerStateDescription],
|
||||
) -> None:
|
||||
"""Test door trigger fires for binary_sensor entities with device_class door."""
|
||||
other_entity_ids = set(target_binary_sensors["included"]) - {entity_id}
|
||||
excluded_entity_ids = set(target_binary_sensors["excluded"]) - {entity_id}
|
||||
|
||||
for eid in target_binary_sensors["included"]:
|
||||
set_or_remove_state(hass, eid, states[0]["included"])
|
||||
await hass.async_block_till_done()
|
||||
for eid in excluded_entity_ids:
|
||||
set_or_remove_state(hass, eid, states[0]["excluded"])
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await arm_trigger(hass, trigger, {}, trigger_target_config)
|
||||
|
||||
for state in states[1:]:
|
||||
excluded_state = state["excluded"]
|
||||
included_state = state["included"]
|
||||
set_or_remove_state(hass, entity_id, included_state)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == state["count"]
|
||||
for service_call in service_calls:
|
||||
assert service_call.data[CONF_ENTITY_ID] == entity_id
|
||||
service_calls.clear()
|
||||
|
||||
for other_entity_id in other_entity_ids:
|
||||
set_or_remove_state(hass, other_entity_id, included_state)
|
||||
await hass.async_block_till_done()
|
||||
for excluded_entity_id in excluded_entity_ids:
|
||||
set_or_remove_state(hass, excluded_entity_id, excluded_state)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == (entities_in_target - 1) * state["count"]
|
||||
service_calls.clear()
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("enable_labs_preview_features")
|
||||
@pytest.mark.parametrize(
|
||||
("trigger_target_config", "entity_id", "entities_in_target"),
|
||||
parametrize_target_entities("cover"),
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
("trigger", "trigger_options", "states"),
|
||||
[
|
||||
*parametrize_trigger_states(
|
||||
trigger="door.opened",
|
||||
target_states=[
|
||||
(CoverState.OPEN, {ATTR_IS_CLOSED: False}),
|
||||
(CoverState.OPENING, {ATTR_IS_CLOSED: False}),
|
||||
],
|
||||
other_states=[
|
||||
(CoverState.CLOSED, {ATTR_IS_CLOSED: True}),
|
||||
(CoverState.CLOSING, {ATTR_IS_CLOSED: True}),
|
||||
(CoverState.OPEN, {ATTR_IS_CLOSED: True}),
|
||||
],
|
||||
extra_invalid_states=[
|
||||
(CoverState.OPEN, {ATTR_IS_CLOSED: None}),
|
||||
(CoverState.OPEN, {}),
|
||||
],
|
||||
additional_attributes={ATTR_DEVICE_CLASS: "door"},
|
||||
trigger_from_none=False,
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_door_trigger_cover_behavior_any(
|
||||
hass: HomeAssistant,
|
||||
service_calls: list[ServiceCall],
|
||||
target_covers: dict[str, list[str]],
|
||||
trigger_target_config: dict,
|
||||
entity_id: str,
|
||||
entities_in_target: int,
|
||||
trigger: str,
|
||||
trigger_options: dict[str, Any],
|
||||
states: list[TriggerStateDescription],
|
||||
) -> None:
|
||||
"""Test door trigger fires for cover entities with device_class door."""
|
||||
other_entity_ids = set(target_covers["included"]) - {entity_id}
|
||||
excluded_entity_ids = set(target_covers["excluded"]) - {entity_id}
|
||||
|
||||
for eid in target_covers["included"]:
|
||||
set_or_remove_state(hass, eid, states[0]["included"])
|
||||
await hass.async_block_till_done()
|
||||
for eid in excluded_entity_ids:
|
||||
set_or_remove_state(hass, eid, states[0]["excluded"])
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await arm_trigger(hass, trigger, {}, trigger_target_config)
|
||||
|
||||
for state in states[1:]:
|
||||
excluded_state = state["excluded"]
|
||||
included_state = state["included"]
|
||||
set_or_remove_state(hass, entity_id, included_state)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == state["count"]
|
||||
for service_call in service_calls:
|
||||
assert service_call.data[CONF_ENTITY_ID] == entity_id
|
||||
service_calls.clear()
|
||||
|
||||
for other_entity_id in other_entity_ids:
|
||||
set_or_remove_state(hass, other_entity_id, included_state)
|
||||
await hass.async_block_till_done()
|
||||
for excluded_entity_id in excluded_entity_ids:
|
||||
set_or_remove_state(hass, excluded_entity_id, excluded_state)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == (entities_in_target - 1) * state["count"]
|
||||
service_calls.clear()
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("enable_labs_preview_features")
|
||||
@pytest.mark.parametrize(
|
||||
("trigger_target_config", "entity_id", "entities_in_target"),
|
||||
parametrize_target_entities("binary_sensor"),
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
("trigger", "trigger_options", "states"),
|
||||
[
|
||||
*parametrize_trigger_states(
|
||||
trigger="door.opened",
|
||||
target_states=[STATE_ON],
|
||||
other_states=[STATE_OFF],
|
||||
additional_attributes={ATTR_DEVICE_CLASS: "door"},
|
||||
trigger_from_none=False,
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_door_trigger_binary_sensor_behavior_first(
|
||||
hass: HomeAssistant,
|
||||
service_calls: list[ServiceCall],
|
||||
target_binary_sensors: dict[str, list[str]],
|
||||
trigger_target_config: dict,
|
||||
entity_id: str,
|
||||
entities_in_target: int,
|
||||
trigger: str,
|
||||
trigger_options: dict[str, Any],
|
||||
states: list[TriggerStateDescription],
|
||||
) -> None:
|
||||
"""Test door trigger fires on the first binary_sensor state change."""
|
||||
other_entity_ids = set(target_binary_sensors["included"]) - {entity_id}
|
||||
excluded_entity_ids = set(target_binary_sensors["excluded"]) - {entity_id}
|
||||
|
||||
for eid in target_binary_sensors["included"]:
|
||||
set_or_remove_state(hass, eid, states[0]["included"])
|
||||
await hass.async_block_till_done()
|
||||
for eid in excluded_entity_ids:
|
||||
set_or_remove_state(hass, eid, states[0]["excluded"])
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
|
||||
|
||||
for state in states[1:]:
|
||||
excluded_state = state["excluded"]
|
||||
included_state = state["included"]
|
||||
set_or_remove_state(hass, entity_id, included_state)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == state["count"]
|
||||
for service_call in service_calls:
|
||||
assert service_call.data[CONF_ENTITY_ID] == entity_id
|
||||
service_calls.clear()
|
||||
|
||||
for other_entity_id in other_entity_ids:
|
||||
set_or_remove_state(hass, other_entity_id, excluded_state)
|
||||
await hass.async_block_till_done()
|
||||
for excluded_entity_id in excluded_entity_ids:
|
||||
set_or_remove_state(hass, excluded_entity_id, excluded_state)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == 0
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("enable_labs_preview_features")
|
||||
@pytest.mark.parametrize(
|
||||
("trigger_target_config", "entity_id", "entities_in_target"),
|
||||
parametrize_target_entities("binary_sensor"),
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
("trigger", "trigger_options", "states"),
|
||||
[
|
||||
*parametrize_trigger_states(
|
||||
trigger="door.opened",
|
||||
target_states=[STATE_ON],
|
||||
other_states=[STATE_OFF],
|
||||
additional_attributes={ATTR_DEVICE_CLASS: "door"},
|
||||
trigger_from_none=False,
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_door_trigger_binary_sensor_behavior_last(
|
||||
hass: HomeAssistant,
|
||||
service_calls: list[ServiceCall],
|
||||
target_binary_sensors: dict[str, list[str]],
|
||||
trigger_target_config: dict,
|
||||
entity_id: str,
|
||||
entities_in_target: int,
|
||||
trigger: str,
|
||||
trigger_options: dict[str, Any],
|
||||
states: list[TriggerStateDescription],
|
||||
) -> None:
|
||||
"""Test door trigger fires when the last binary_sensor changes state."""
|
||||
other_entity_ids = set(target_binary_sensors["included"]) - {entity_id}
|
||||
excluded_entity_ids = set(target_binary_sensors["excluded"]) - {entity_id}
|
||||
|
||||
for eid in target_binary_sensors["included"]:
|
||||
set_or_remove_state(hass, eid, states[0]["included"])
|
||||
await hass.async_block_till_done()
|
||||
for eid in excluded_entity_ids:
|
||||
set_or_remove_state(hass, eid, states[0]["excluded"])
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
|
||||
|
||||
for state in states[1:]:
|
||||
excluded_state = state["excluded"]
|
||||
included_state = state["included"]
|
||||
for other_entity_id in other_entity_ids:
|
||||
set_or_remove_state(hass, other_entity_id, excluded_state)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == 0
|
||||
|
||||
set_or_remove_state(hass, entity_id, included_state)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == state["count"]
|
||||
for service_call in service_calls:
|
||||
assert service_call.data[CONF_ENTITY_ID] == entity_id
|
||||
service_calls.clear()
|
||||
|
||||
for excluded_entity_id in excluded_entity_ids:
|
||||
set_or_remove_state(hass, excluded_entity_id, excluded_state)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == 0
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("enable_labs_preview_features")
|
||||
@pytest.mark.parametrize(
|
||||
("trigger_target_config", "entity_id", "entities_in_target"),
|
||||
parametrize_target_entities("cover"),
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
("trigger", "trigger_options", "states"),
|
||||
[
|
||||
*parametrize_trigger_states(
|
||||
trigger="door.opened",
|
||||
target_states=[
|
||||
(CoverState.OPEN, {ATTR_IS_CLOSED: False}),
|
||||
(CoverState.OPENING, {ATTR_IS_CLOSED: False}),
|
||||
],
|
||||
other_states=[
|
||||
(CoverState.CLOSED, {ATTR_IS_CLOSED: True}),
|
||||
(CoverState.CLOSING, {ATTR_IS_CLOSED: True}),
|
||||
(CoverState.OPEN, {ATTR_IS_CLOSED: True}),
|
||||
],
|
||||
extra_invalid_states=[
|
||||
(CoverState.OPEN, {ATTR_IS_CLOSED: None}),
|
||||
(CoverState.OPEN, {}),
|
||||
],
|
||||
additional_attributes={ATTR_DEVICE_CLASS: "door"},
|
||||
trigger_from_none=False,
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_door_trigger_cover_behavior_first(
|
||||
hass: HomeAssistant,
|
||||
service_calls: list[ServiceCall],
|
||||
target_covers: dict[str, list[str]],
|
||||
trigger_target_config: dict,
|
||||
entity_id: str,
|
||||
entities_in_target: int,
|
||||
trigger: str,
|
||||
trigger_options: dict[str, Any],
|
||||
states: list[TriggerStateDescription],
|
||||
) -> None:
|
||||
"""Test door trigger fires on the first cover state change."""
|
||||
other_entity_ids = set(target_covers["included"]) - {entity_id}
|
||||
excluded_entity_ids = set(target_covers["excluded"]) - {entity_id}
|
||||
|
||||
for eid in target_covers["included"]:
|
||||
set_or_remove_state(hass, eid, states[0]["included"])
|
||||
await hass.async_block_till_done()
|
||||
for eid in excluded_entity_ids:
|
||||
set_or_remove_state(hass, eid, states[0]["excluded"])
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
|
||||
|
||||
for state in states[1:]:
|
||||
excluded_state = state["excluded"]
|
||||
included_state = state["included"]
|
||||
set_or_remove_state(hass, entity_id, included_state)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == state["count"]
|
||||
for service_call in service_calls:
|
||||
assert service_call.data[CONF_ENTITY_ID] == entity_id
|
||||
service_calls.clear()
|
||||
|
||||
for other_entity_id in other_entity_ids:
|
||||
set_or_remove_state(hass, other_entity_id, excluded_state)
|
||||
await hass.async_block_till_done()
|
||||
for excluded_entity_id in excluded_entity_ids:
|
||||
set_or_remove_state(hass, excluded_entity_id, excluded_state)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == 0
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("enable_labs_preview_features")
|
||||
@pytest.mark.parametrize(
|
||||
("trigger_target_config", "entity_id", "entities_in_target"),
|
||||
parametrize_target_entities("cover"),
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
("trigger", "trigger_options", "states"),
|
||||
[
|
||||
*parametrize_trigger_states(
|
||||
trigger="door.opened",
|
||||
target_states=[
|
||||
(CoverState.OPEN, {ATTR_IS_CLOSED: False}),
|
||||
(CoverState.OPENING, {ATTR_IS_CLOSED: False}),
|
||||
],
|
||||
other_states=[
|
||||
(CoverState.CLOSED, {ATTR_IS_CLOSED: True}),
|
||||
(CoverState.CLOSING, {ATTR_IS_CLOSED: True}),
|
||||
(CoverState.OPEN, {ATTR_IS_CLOSED: True}),
|
||||
],
|
||||
extra_invalid_states=[
|
||||
(CoverState.OPEN, {ATTR_IS_CLOSED: None}),
|
||||
(CoverState.OPEN, {}),
|
||||
],
|
||||
additional_attributes={ATTR_DEVICE_CLASS: "door"},
|
||||
trigger_from_none=False,
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_door_trigger_cover_behavior_last(
|
||||
hass: HomeAssistant,
|
||||
service_calls: list[ServiceCall],
|
||||
target_covers: dict[str, list[str]],
|
||||
trigger_target_config: dict,
|
||||
entity_id: str,
|
||||
entities_in_target: int,
|
||||
trigger: str,
|
||||
trigger_options: dict[str, Any],
|
||||
states: list[TriggerStateDescription],
|
||||
) -> None:
|
||||
"""Test door trigger fires when the last cover changes state."""
|
||||
other_entity_ids = set(target_covers["included"]) - {entity_id}
|
||||
excluded_entity_ids = set(target_covers["excluded"]) - {entity_id}
|
||||
|
||||
for eid in target_covers["included"]:
|
||||
set_or_remove_state(hass, eid, states[0]["included"])
|
||||
await hass.async_block_till_done()
|
||||
for eid in excluded_entity_ids:
|
||||
set_or_remove_state(hass, eid, states[0]["excluded"])
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
|
||||
|
||||
for state in states[1:]:
|
||||
excluded_state = state["excluded"]
|
||||
included_state = state["included"]
|
||||
for other_entity_id in other_entity_ids:
|
||||
set_or_remove_state(hass, other_entity_id, excluded_state)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == 0
|
||||
|
||||
set_or_remove_state(hass, entity_id, included_state)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == state["count"]
|
||||
for service_call in service_calls:
|
||||
assert service_call.data[CONF_ENTITY_ID] == entity_id
|
||||
service_calls.clear()
|
||||
|
||||
for excluded_entity_id in excluded_entity_ids:
|
||||
set_or_remove_state(hass, excluded_entity_id, excluded_state)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == 0
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("enable_labs_preview_features")
|
||||
@pytest.mark.parametrize(
|
||||
(
|
||||
"trigger_key",
|
||||
"binary_sensor_initial",
|
||||
"binary_sensor_target",
|
||||
"cover_initial",
|
||||
"cover_initial_is_closed",
|
||||
"cover_target",
|
||||
"cover_target_is_closed",
|
||||
),
|
||||
[
|
||||
(
|
||||
"door.opened",
|
||||
STATE_OFF,
|
||||
STATE_ON,
|
||||
CoverState.CLOSED,
|
||||
True,
|
||||
CoverState.OPEN,
|
||||
False,
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_door_trigger_excludes_non_door_device_class(
|
||||
hass: HomeAssistant,
|
||||
service_calls: list[ServiceCall],
|
||||
trigger_key: str,
|
||||
binary_sensor_initial: str,
|
||||
binary_sensor_target: str,
|
||||
cover_initial: str,
|
||||
cover_initial_is_closed: bool,
|
||||
cover_target: str,
|
||||
cover_target_is_closed: bool,
|
||||
) -> None:
|
||||
"""Test door trigger does not fire for entities without device_class door."""
|
||||
entity_id_door = "binary_sensor.test_door"
|
||||
entity_id_window = "binary_sensor.test_window"
|
||||
entity_id_cover_door = "cover.test_door"
|
||||
entity_id_cover_garage = "cover.test_garage"
|
||||
|
||||
# Set initial states
|
||||
hass.states.async_set(
|
||||
entity_id_door, binary_sensor_initial, {ATTR_DEVICE_CLASS: "door"}
|
||||
)
|
||||
hass.states.async_set(
|
||||
entity_id_window, binary_sensor_initial, {ATTR_DEVICE_CLASS: "window"}
|
||||
)
|
||||
hass.states.async_set(
|
||||
entity_id_cover_door,
|
||||
cover_initial,
|
||||
{ATTR_DEVICE_CLASS: "door", ATTR_IS_CLOSED: cover_initial_is_closed},
|
||||
)
|
||||
hass.states.async_set(
|
||||
entity_id_cover_garage,
|
||||
cover_initial,
|
||||
{ATTR_DEVICE_CLASS: "garage", ATTR_IS_CLOSED: cover_initial_is_closed},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await arm_trigger(
|
||||
hass,
|
||||
trigger_key,
|
||||
{},
|
||||
{
|
||||
CONF_ENTITY_ID: [
|
||||
entity_id_door,
|
||||
entity_id_window,
|
||||
entity_id_cover_door,
|
||||
entity_id_cover_garage,
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
# Door binary_sensor changes - should trigger
|
||||
hass.states.async_set(
|
||||
entity_id_door, binary_sensor_target, {ATTR_DEVICE_CLASS: "door"}
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == 1
|
||||
assert service_calls[0].data[CONF_ENTITY_ID] == entity_id_door
|
||||
service_calls.clear()
|
||||
|
||||
# Window binary_sensor changes - should NOT trigger (wrong device class)
|
||||
hass.states.async_set(
|
||||
entity_id_window, binary_sensor_target, {ATTR_DEVICE_CLASS: "window"}
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == 0
|
||||
|
||||
# Cover door changes - should trigger
|
||||
hass.states.async_set(
|
||||
entity_id_cover_door,
|
||||
cover_target,
|
||||
{ATTR_DEVICE_CLASS: "door", ATTR_IS_CLOSED: cover_target_is_closed},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == 1
|
||||
assert service_calls[0].data[CONF_ENTITY_ID] == entity_id_cover_door
|
||||
service_calls.clear()
|
||||
|
||||
# Garage cover changes - should NOT trigger (wrong device class)
|
||||
hass.states.async_set(
|
||||
entity_id_cover_garage,
|
||||
cover_target,
|
||||
{ATTR_DEVICE_CLASS: "garage", ATTR_IS_CLOSED: cover_target_is_closed},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert len(service_calls) == 0
|
||||
|
||||
|
||||
def test_door_device_class() -> None:
|
||||
"""Test the door trigger device class."""
|
||||
assert BinarySensorDeviceClass.DOOR == DEVICE_CLASS_DOOR
|
||||
assert CoverDeviceClass.DOOR == DEVICE_CLASS_DOOR
|
||||
@@ -30,6 +30,7 @@
|
||||
'device_automation',
|
||||
'device_tracker',
|
||||
'diagnostics',
|
||||
'door',
|
||||
'event',
|
||||
'fan',
|
||||
'ffmpeg',
|
||||
@@ -128,6 +129,7 @@
|
||||
'device_automation',
|
||||
'device_tracker',
|
||||
'diagnostics',
|
||||
'door',
|
||||
'event',
|
||||
'fan',
|
||||
'ffmpeg',
|
||||
|
||||
Reference in New Issue
Block a user