From 0d8527997b6a667175bf933db02615c1b630d3f7 Mon Sep 17 00:00:00 2001 From: Joakim Plate Date: Wed, 30 Sep 2026 14:02:25 +0200 Subject: [PATCH] Add basic support for events driven from zha (#183109) --- homeassistant/components/zha/__init__.py | 1 + homeassistant/components/zha/event.py | 70 ++++++ tests/components/zha/test_event.py | 307 +++++++++++++++++++++++ 3 files changed, 378 insertions(+) create mode 100644 homeassistant/components/zha/event.py create mode 100644 tests/components/zha/test_event.py diff --git a/homeassistant/components/zha/__init__.py b/homeassistant/components/zha/__init__.py index 64944e9f6ce5..8cba532bb4cb 100644 --- a/homeassistant/components/zha/__init__.py +++ b/homeassistant/components/zha/__init__.py @@ -97,6 +97,7 @@ PLATFORMS = ( Platform.CLIMATE, Platform.COVER, Platform.DEVICE_TRACKER, + Platform.EVENT, Platform.FAN, Platform.LIGHT, Platform.LOCK, diff --git a/homeassistant/components/zha/event.py b/homeassistant/components/zha/event.py new file mode 100644 index 000000000000..5fb1cab2bf9e --- /dev/null +++ b/homeassistant/components/zha/event.py @@ -0,0 +1,70 @@ +"""Events on Zigbee Home Automation networks.""" + +import functools +from typing import override + +from zha.application.platforms.event import EntityEventTriggeredEvent + +from homeassistant.components.event import EventDeviceClass, EventEntity +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import Platform +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers.dispatcher import async_dispatcher_connect +from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback + +from .entity import ZHAEntity +from .helpers import ( + SIGNAL_ADD_ENTITIES, + async_add_entities as zha_async_add_entities, + get_zha_data, +) + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + """Set up the Zigbee Home Automation event from config entry.""" + zha_data = get_zha_data(hass) + entities_to_create = zha_data.platforms[Platform.EVENT] + + unsub = async_dispatcher_connect( + hass, + SIGNAL_ADD_ENTITIES, + functools.partial( + zha_async_add_entities, async_add_entities, ZHAEvent, entities_to_create + ), + ) + config_entry.async_on_unload(unsub) + + +class ZHAEvent(ZHAEntity, EventEntity): + """ZHA event.""" + + @override + def _update_capability_attrs(self) -> None: + """Re-derive capability attributes from the cached state.""" + super()._update_capability_attrs() + + device_class = self._zha_state.device_class + self._attr_device_class = ( + EventDeviceClass(device_class) if device_class is not None else None + ) + self._attr_event_types = self._zha_state.event_types + + @override + async def async_added_to_hass(self) -> None: + await super().async_added_to_hass() + self._unsubs.append( + self.entity_data.entity.on_event( + EntityEventTriggeredEvent.event, self._handle_entity_events + ) + ) + + @callback + def _handle_entity_events(self, data: EntityEventTriggeredEvent) -> None: + """Handle an event triggered by the ZHA entity.""" + self.debug("Handling event from entity: %s", data.triggered.event_type) + self._trigger_event(data.triggered.event_type, data.triggered.event_attributes) + self.async_write_ha_state() diff --git a/tests/components/zha/test_event.py b/tests/components/zha/test_event.py new file mode 100644 index 000000000000..f9df3910de54 --- /dev/null +++ b/tests/components/zha/test_event.py @@ -0,0 +1,307 @@ +"""Test ZHA event.""" + +from collections.abc import Callable, Coroutine, Generator +from typing import Any +from unittest.mock import patch + +import pytest +from zha.application.platforms.event import BaseEvent, EntityEventTriggeredEvent +from zha.application.platforms.event.const import ( + ATTR_MULTI_PRESS_COUNT, + ButtonEventType, + EventDeviceClass as ZHAEventDeviceClass, +) +from zha.zigbee.device import DeviceEntityAddedEvent +from zigpy.const import SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_PROFILE, SIG_EP_TYPE +from zigpy.device import Device +from zigpy.profiles import zha +from zigpy.zcl.clusters import general + +from homeassistant.components.event import ( + ATTR_EVENT_TYPE, + ATTR_EVENT_TYPES, + EventDeviceClass, +) +from homeassistant.components.zha.const import DOMAIN +from homeassistant.components.zha.helpers import ( + ZHADeviceProxy, + ZHAGatewayProxy, + get_zha_gateway, + get_zha_gateway_proxy, +) +from homeassistant.const import ATTR_DEVICE_CLASS, STATE_UNKNOWN, Platform +from homeassistant.core import HomeAssistant, State +from homeassistant.helpers import entity_registry as er + +from tests.common import mock_restore_cache_with_extra_data + +ENTITY_ID = "event.fakemanufacturer_fakemodel_button" + + +class FakeEvent(BaseEvent): + """Event entity that can be triggered from tests.""" + + _attr_fallback_name = "Fake" + + def trigger( + self, event_type: str, event_attributes: dict[str, Any] | None = None + ) -> None: + """Trigger an event, as a concrete subclass would.""" + self._trigger_event(event_type, event_attributes) + + +class FakeButtonEvent(FakeEvent): + """Button event entity.""" + + _unique_id_suffix = "fake_button" + _attr_device_class = ZHAEventDeviceClass.BUTTON + _attr_event_types = [ButtonEventType.PRESS_END, ButtonEventType.MULTI_PRESS_END] + + +class FakeMotionEvent(FakeEvent): + """Motion event entity.""" + + _unique_id_suffix = "fake_motion" + _attr_device_class = ZHAEventDeviceClass.MOTION + _attr_event_types = ["motion"] + + +class FakeNoDeviceClassEvent(FakeEvent): + """Event entity without a device class.""" + + _unique_id_suffix = "fake_no_device_class" + _attr_device_class = None + _attr_event_types = ["triggered"] + + +@pytest.fixture(autouse=True) +def event_platform_only() -> Generator[None]: + """Only set up the event and required base platforms to speed up tests.""" + with patch( + "homeassistant.components.zha.PLATFORMS", + (Platform.BINARY_SENSOR, Platform.EVENT, Platform.SENSOR), + ): + yield + + +@pytest.fixture +def speed_up_radio_mgr() -> Generator[None]: + """Speed up the radio manager connection time by removing delays. + + This fixture replaces the fixture in conftest.py by patching the connect + and shutdown delays to 0 to allow waiting for the patched delays when + running tests with time frozen, which otherwise blocks forever. + """ + with ( + patch("homeassistant.components.zha.radio_manager.CONNECT_DELAY_S", 0), + patch("zha.application.gateway.SHUT_DOWN_DELAY_S", 0), + ): + yield + + +async def _setup_device( + hass: HomeAssistant, + setup_zha: Callable[..., Coroutine[None]], + zigpy_device_mock: Callable[..., Device], +) -> ZHADeviceProxy: + """Join a device that event entities can be added to.""" + await setup_zha() + + gateway = get_zha_gateway(hass) + gateway_proxy: ZHAGatewayProxy = get_zha_gateway_proxy(hass) + + zigpy_device = zigpy_device_mock( + { + 1: { + SIG_EP_PROFILE: zha.PROFILE_ID, + SIG_EP_TYPE: zha.DeviceType.ON_OFF_SWITCH, + SIG_EP_INPUT: [general.Basic.cluster_id], + SIG_EP_OUTPUT: [general.OnOff.cluster_id], + } + }, + ieee="01:2d:6f:00:0a:90:69:e8", + ) + + gateway.get_or_create_device(zigpy_device) + await gateway.async_device_initialized(zigpy_device) + await hass.async_block_till_done(wait_background_tasks=True) + + return gateway_proxy.get_device_proxy(zigpy_device.ieee) + + +async def _add_event( + hass: HomeAssistant, + zha_device_proxy: ZHADeviceProxy, + event_class: type[FakeEvent], +) -> tuple[str, FakeEvent]: + """Add a fake event entity to a device.""" + zha_device = zha_device_proxy.device + endpoint = zha_device.endpoints[1] + + zha_entity = event_class( + endpoint=endpoint, + device=zha_device, + cluster=endpoint.zigpy_endpoint.out_clusters[general.OnOff.cluster_id], + ) + zha_device.platform_entities[(zha_entity.PLATFORM, zha_entity.unique_id)] = ( + zha_entity + ) + zha_device.emit( + DeviceEntityAddedEvent.event_type, + DeviceEntityAddedEvent( + platform=zha_entity.PLATFORM, unique_id=zha_entity.unique_id + ), + ) + await hass.async_block_till_done() + + entity_id = er.async_get(hass).async_get_entity_id( + Platform.EVENT, DOMAIN, zha_entity.unique_id + ) + assert entity_id is not None + return entity_id, zha_entity + + +@pytest.mark.parametrize( + ("event_class", "device_class", "event_types"), + [ + pytest.param( + FakeButtonEvent, + EventDeviceClass.BUTTON, + ["press_end", "multi_press_end"], + id="button", + ), + pytest.param(FakeMotionEvent, EventDeviceClass.MOTION, ["motion"], id="motion"), + pytest.param(FakeNoDeviceClassEvent, None, ["triggered"], id="no_device_class"), + ], +) +async def test_event_entity( + hass: HomeAssistant, + setup_zha: Callable[..., Coroutine[None]], + zigpy_device_mock: Callable[..., Device], + event_class: type[FakeEvent], + device_class: EventDeviceClass | None, + event_types: list[str], +) -> None: + """Test ZHA event entity is created with the capabilities of the zha entity.""" + zha_device_proxy = await _setup_device(hass, setup_zha, zigpy_device_mock) + entity_id, _ = await _add_event(hass, zha_device_proxy, event_class) + + state = hass.states.get(entity_id) + assert state + assert state.state == STATE_UNKNOWN + assert state.attributes.get(ATTR_DEVICE_CLASS) == device_class + assert state.attributes[ATTR_EVENT_TYPES] == event_types + assert state.attributes[ATTR_EVENT_TYPE] is None + + +@pytest.mark.freeze_time("2026-09-25 12:00:00+00:00") +async def test_event_triggered( + hass: HomeAssistant, + setup_zha: Callable[..., Coroutine[None]], + zigpy_device_mock: Callable[..., Device], +) -> None: + """Test events triggered by the zha entity are fired in Home Assistant.""" + zha_device_proxy = await _setup_device(hass, setup_zha, zigpy_device_mock) + entity_id, zha_entity = await _add_event(hass, zha_device_proxy, FakeButtonEvent) + + zha_entity.trigger(ButtonEventType.MULTI_PRESS_END, {ATTR_MULTI_PRESS_COUNT: 2}) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert state + assert state.state == "2026-09-25T12:00:00.000+00:00" + assert state.attributes[ATTR_EVENT_TYPE] == "multi_press_end" + assert state.attributes[ATTR_MULTI_PRESS_COUNT] == 2 + + zha_entity.trigger(ButtonEventType.PRESS_END) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert state + assert state.attributes[ATTR_EVENT_TYPE] == "press_end" + assert ATTR_MULTI_PRESS_COUNT not in state.attributes + + +@pytest.mark.freeze_time("2026-09-25 12:00:00+00:00") +async def test_multiple_events( + hass: HomeAssistant, + setup_zha: Callable[..., Coroutine[None]], + zigpy_device_mock: Callable[..., Device], +) -> None: + """Test event entities on the same device are triggered independently.""" + zha_device_proxy = await _setup_device(hass, setup_zha, zigpy_device_mock) + button_id, button = await _add_event(hass, zha_device_proxy, FakeButtonEvent) + motion_id, motion = await _add_event(hass, zha_device_proxy, FakeMotionEvent) + assert button_id != motion_id + + button.trigger(ButtonEventType.PRESS_END) + await hass.async_block_till_done() + + state = hass.states.get(button_id) + assert state + assert state.state == "2026-09-25T12:00:00.000+00:00" + state = hass.states.get(motion_id) + assert state + assert state.state == STATE_UNKNOWN + + motion.trigger("motion") + await hass.async_block_till_done() + + state = hass.states.get(button_id) + assert state + assert state.attributes[ATTR_EVENT_TYPE] == "press_end" + state = hass.states.get(motion_id) + assert state + assert state.attributes[ATTR_EVENT_TYPE] == "motion" + + +async def test_event_restored( + hass: HomeAssistant, + setup_zha: Callable[..., Coroutine[None]], + zigpy_device_mock: Callable[..., Device], +) -> None: + """Test the last event is restored when the entity is set up again.""" + mock_restore_cache_with_extra_data( + hass, + [ + ( + State(ENTITY_ID, "2026-09-25T12:00:00.000+00:00"), + { + "last_event_type": "multi_press_end", + "last_event_attributes": {ATTR_MULTI_PRESS_COUNT: 2}, + }, + ) + ], + ) + + zha_device_proxy = await _setup_device(hass, setup_zha, zigpy_device_mock) + entity_id, _ = await _add_event(hass, zha_device_proxy, FakeButtonEvent) + assert entity_id == ENTITY_ID + + state = hass.states.get(entity_id) + assert state + assert state.state == "2026-09-25T12:00:00.000+00:00" + assert state.attributes[ATTR_EVENT_TYPE] == "multi_press_end" + assert state.attributes[ATTR_MULTI_PRESS_COUNT] == 2 + + +async def test_event_removed( + hass: HomeAssistant, + entity_registry: er.EntityRegistry, + setup_zha: Callable[..., Coroutine[None]], + zigpy_device_mock: Callable[..., Device], +) -> None: + """Test removing the entity stops listening to the zha entity.""" + zha_device_proxy = await _setup_device(hass, setup_zha, zigpy_device_mock) + entity_id, zha_entity = await _add_event(hass, zha_device_proxy, FakeButtonEvent) + assert zha_entity._listeners[EntityEventTriggeredEvent.event] + + entity_registry.async_remove(entity_id) + await hass.async_block_till_done() + assert hass.states.get(entity_id) is None + assert not zha_entity._listeners[EntityEventTriggeredEvent.event] + + # Must not write state for the removed entity + zha_entity.trigger(ButtonEventType.PRESS_END) + await hass.async_block_till_done() + assert hass.states.get(entity_id) is None