Compare commits

...

5 Commits

Author SHA1 Message Date
abmantis
196a42a44d Fix stale docstrings 2025-12-11 20:05:02 +00:00
abmantis
4cc46ecea4 Fix module docstring 2025-12-11 20:04:49 +00:00
abmantis
1a0791b59a Simplify condition 2025-12-11 19:59:11 +00:00
abmantis
3d23f43461 Add button pressed trigger 2025-12-11 19:52:12 +00:00
abmantis
2b3de7fa2b Move Trigger from_state validation to is_valid_transition 2025-12-11 19:50:12 +00:00
8 changed files with 270 additions and 10 deletions

View File

@@ -125,6 +125,7 @@ _EXPERIMENTAL_TRIGGER_PLATFORMS = {
"alarm_control_panel",
"assist_satellite",
"binary_sensor",
"button",
"climate",
"cover",
"fan",

View File

@@ -17,5 +17,10 @@
"press": {
"service": "mdi:gesture-tap-button"
}
},
"triggers": {
"pressed": {
"trigger": "mdi:gesture-tap-button"
}
}
}

View File

@@ -27,5 +27,11 @@
"name": "Press"
}
},
"title": "Button"
"title": "Button",
"triggers": {
"pressed": {
"description": "Triggers after one or several buttons were pressed.",
"name": "Button pressed"
}
}
}

View File

@@ -0,0 +1,42 @@
"""Provides triggers for buttons."""
from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers.trigger import (
ENTITY_STATE_TRIGGER_SCHEMA,
EntityTriggerBase,
Trigger,
)
from . import DOMAIN
class ButtonPressedTrigger(EntityTriggerBase):
"""Trigger for button entity presses."""
_domain = DOMAIN
_schema = ENTITY_STATE_TRIGGER_SCHEMA
def is_valid_transition(self, from_state: State, to_state: State) -> bool:
"""Check if the origin state is not an expected target states."""
# UNKNOWN is a valid from_state, otherwise the first time the button is pressed
# would not trigger
if from_state.state == STATE_UNAVAILABLE:
return False
return from_state.state != to_state.state
def is_valid_state(self, state: State) -> bool:
"""Check if the new state is not invalid."""
return state.state not in (STATE_UNAVAILABLE, STATE_UNKNOWN)
TRIGGERS: dict[str, type[Trigger]] = {
"pressed": ButtonPressedTrigger,
}
async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
"""Return the triggers for buttons."""
return TRIGGERS

View File

@@ -0,0 +1,4 @@
pressed:
target:
entity:
domain: button

View File

@@ -17,10 +17,6 @@ class TextChangedTrigger(EntityTriggerBase):
_domain = DOMAIN
_schema = ENTITY_STATE_TRIGGER_SCHEMA
def is_valid_transition(self, from_state: State, to_state: State) -> bool:
"""Check if the old and new states are different."""
return from_state.state != to_state.state
def is_valid_state(self, state: State) -> bool:
"""Check if the new state is not invalid."""
return state.state not in (STATE_UNAVAILABLE, STATE_UNKNOWN)

View File

@@ -338,8 +338,11 @@ class EntityTriggerBase(Trigger):
self._target = config.target
def is_valid_transition(self, from_state: State, to_state: State) -> bool:
"""Check if the origin state is not an expected target states."""
return not self.is_valid_state(from_state)
"""Check if the origin state is valid and the state has changed."""
if from_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN):
return False
return from_state.state != to_state.state
@abc.abstractmethod
def is_valid_state(self, state: State) -> bool:
@@ -390,12 +393,11 @@ class EntityTriggerBase(Trigger):
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):
if not from_state or not to_state:
return
# The trigger should never fire if the new state is not valid
if not to_state or not self.is_valid_state(to_state):
if not self.is_valid_state(to_state):
return
# The trigger should never fire if the transition is not valid
@@ -446,6 +448,9 @@ class ConditionalEntityStateTriggerBase(EntityTriggerBase):
def is_valid_transition(self, from_state: State, to_state: State) -> bool:
"""Check if the origin state matches the expected ones."""
if not super().is_valid_transition(from_state, to_state):
return False
return from_state.state in self._from_states
def is_valid_state(self, state: State) -> bool:
@@ -459,6 +464,15 @@ class EntityStateAttributeTriggerBase(EntityTriggerBase):
_attribute: str
_attribute_to_state: str
def is_valid_transition(self, from_state: State, to_state: State) -> bool:
"""Check if the origin state is valid and the state has changed."""
if from_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN):
return False
return from_state.attributes.get(self._attribute) != to_state.attributes.get(
self._attribute
)
def is_valid_state(self, state: State) -> bool:
"""Check if the new state attribute matches the expected one."""
return state.attributes.get(self._attribute) == self._attribute_to_state

View File

@@ -0,0 +1,192 @@
"""Test button trigger."""
from collections.abc import Generator
from unittest.mock import patch
import pytest
from homeassistant.const import (
ATTR_LABEL_ID,
CONF_ENTITY_ID,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant, ServiceCall
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(name="enable_experimental_triggers_conditions")
def enable_experimental_triggers_conditions() -> Generator[None]:
"""Enable experimental triggers and conditions."""
with patch(
"homeassistant.components.labs.async_is_preview_feature_enabled",
return_value=True,
):
yield
@pytest.fixture
async def target_buttons(hass: HomeAssistant) -> list[str]:
"""Create multiple button entities associated with different targets."""
return (await target_entities(hass, "button"))["included"]
@pytest.mark.parametrize("trigger_key", ["button.pressed"])
async def test_button_triggers_gated_by_labs_flag(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, trigger_key: str
) -> None:
"""Test the button 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_experimental_triggers_conditions")
@pytest.mark.parametrize(
("trigger_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("button"),
)
@pytest.mark.parametrize(
("trigger", "states"),
[
(
"button.pressed",
[
{"included": {"state": None, "attributes": {}}, "count": 0},
{
"included": {
"state": "2021-01-01T23:59:59+00:00",
"attributes": {},
},
"count": 0,
},
{
"included": {
"state": "2022-01-01T23:59:59+00:00",
"attributes": {},
},
"count": 1,
},
],
),
(
"button.pressed",
[
{"included": {"state": "foo", "attributes": {}}, "count": 0},
{
"included": {
"state": "2021-01-01T23:59:59+00:00",
"attributes": {},
},
"count": 1,
},
{
"included": {
"state": "2022-01-01T23:59:59+00:00",
"attributes": {},
},
"count": 1,
},
],
),
(
"button.pressed",
[
{
"included": {"state": STATE_UNAVAILABLE, "attributes": {}},
"count": 0,
},
{
"included": {
"state": "2021-01-01T23:59:59+00:00",
"attributes": {},
},
"count": 0,
},
{
"included": {
"state": "2022-01-01T23:59:59+00:00",
"attributes": {},
},
"count": 1,
},
{
"included": {"state": STATE_UNAVAILABLE, "attributes": {}},
"count": 0,
},
],
),
(
"button.pressed",
[
{"included": {"state": STATE_UNKNOWN, "attributes": {}}, "count": 0},
{
"included": {
"state": "2021-01-01T23:59:59+00:00",
"attributes": {},
},
"count": 1,
},
{
"included": {
"state": "2022-01-01T23:59:59+00:00",
"attributes": {},
},
"count": 1,
},
{"included": {"state": STATE_UNKNOWN, "attributes": {}}, "count": 0},
],
),
],
)
async def test_button_state_trigger_behavior_any(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_buttons: list[str],
trigger_target_config: dict,
entity_id: str,
entities_in_target: int,
trigger: str,
states: list[StateDescription],
) -> None:
"""Test that the button state trigger fires when any button state changes to a specific state."""
other_entity_ids = set(target_buttons) - {entity_id}
# Set all buttons, including the tested button, to the initial state
for eid in target_buttons:
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
await arm_trigger(hass, trigger, None, trigger_target_config)
for state in states[1:]:
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()
# Check if changing other buttons also triggers
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert len(service_calls) == (entities_in_target - 1) * state["count"]
service_calls.clear()