diff --git a/homeassistant/components/deconz/device_trigger.py b/homeassistant/components/deconz/device_trigger.py index ae539ee5d48..c76aaf481bf 100644 --- a/homeassistant/components/deconz/device_trigger.py +++ b/homeassistant/components/deconz/device_trigger.py @@ -2,8 +2,14 @@ from __future__ import annotations +from typing import Any + import voluptuous as vol +from homeassistant.components.automation import ( + AutomationActionType, + AutomationTriggerInfo, +) from homeassistant.components.device_automation import DEVICE_TRIGGER_BASE_SCHEMA from homeassistant.components.device_automation.exceptions import ( InvalidDeviceAutomationConfig, @@ -17,11 +23,13 @@ from homeassistant.const import ( CONF_TYPE, CONF_UNIQUE_ID, ) -from homeassistant.core import HomeAssistant +from homeassistant.core import CALLBACK_TYPE, HomeAssistant from homeassistant.helpers import device_registry as dr +from homeassistant.helpers.typing import ConfigType from . import DOMAIN from .deconz_event import CONF_DECONZ_EVENT, CONF_GESTURE, DeconzAlarmEvent, DeconzEvent +from .gateway import DeconzGateway CONF_SUBTYPE = "subtype" @@ -622,7 +630,8 @@ def _get_deconz_event_from_device( device: dr.DeviceEntry, ) -> DeconzAlarmEvent | DeconzEvent: """Resolve deconz event from device.""" - for gateway in hass.data.get(DOMAIN, {}).values(): + gateways: dict[str, DeconzGateway] = hass.data.get(DOMAIN, {}) + for gateway in gateways.values(): for deconz_event in gateway.events: if device.id == deconz_event.device_id: return deconz_event @@ -632,7 +641,10 @@ def _get_deconz_event_from_device( ) -async def async_validate_trigger_config(hass, config): +async def async_validate_trigger_config( + hass: HomeAssistant, + config: dict[str, Any], +) -> vol.Schema: """Validate config.""" config = TRIGGER_SCHEMA(config) @@ -656,32 +668,42 @@ async def async_validate_trigger_config(hass, config): return config -async def async_attach_trigger(hass, config, action, automation_info): +async def async_attach_trigger( + hass: HomeAssistant, + config: ConfigType, + action: AutomationActionType, + automation_info: AutomationTriggerInfo, +) -> CALLBACK_TYPE: """Listen for state changes based on configuration.""" + event_data: dict[str, int | str] = {} + device_registry = dr.async_get(hass) - device = device_registry.async_get(config[CONF_DEVICE_ID]) - - trigger = (config[CONF_TYPE], config[CONF_SUBTYPE]) - - trigger = REMOTES[device.model][trigger] + device = device_registry.devices[config[CONF_DEVICE_ID]] deconz_event = _get_deconz_event_from_device(hass, device) + if event_id := deconz_event.serial: + event_data[CONF_UNIQUE_ID] = event_id - event_id = deconz_event.serial + if device_model := device.model: + config_trigger = (config[CONF_TYPE], config[CONF_SUBTYPE]) + event_data |= REMOTES[device_model][config_trigger] - event_config = { + raw_event_config = { event_trigger.CONF_PLATFORM: "event", event_trigger.CONF_EVENT_TYPE: CONF_DECONZ_EVENT, - event_trigger.CONF_EVENT_DATA: {CONF_UNIQUE_ID: event_id, **trigger}, + event_trigger.CONF_EVENT_DATA: event_data, } - event_config = event_trigger.TRIGGER_SCHEMA(event_config) + event_config = event_trigger.TRIGGER_SCHEMA(raw_event_config) return await event_trigger.async_attach_trigger( hass, event_config, action, automation_info, platform_type="device" ) -async def async_get_triggers(hass, device_id): +async def async_get_triggers( + hass: HomeAssistant, + device_id: str, +) -> list | None: """List device triggers. Make sure device is a supported remote model. @@ -689,10 +711,10 @@ async def async_get_triggers(hass, device_id): Generate device trigger list. """ device_registry = dr.async_get(hass) - device = device_registry.async_get(device_id) + device = device_registry.devices[device_id] if device.model not in REMOTES: - return + return None triggers = [] for trigger, subtype in REMOTES[device.model].keys(): diff --git a/tests/components/deconz/test_device_trigger.py b/tests/components/deconz/test_device_trigger.py index 4ae8fd32e45..91d8e0e1ea2 100644 --- a/tests/components/deconz/test_device_trigger.py +++ b/tests/components/deconz/test_device_trigger.py @@ -5,6 +5,13 @@ from unittest.mock import Mock, patch import pytest from homeassistant.components.automation import DOMAIN as AUTOMATION_DOMAIN +from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN +from homeassistant.components.binary_sensor.device_trigger import ( + CONF_BAT_LOW, + CONF_NOT_BAT_LOW, + CONF_NOT_TAMPERED, + CONF_TAMPERED, +) from homeassistant.components.deconz import device_trigger from homeassistant.components.deconz.const import DOMAIN as DECONZ_DOMAIN from homeassistant.components.deconz.device_trigger import CONF_SUBTYPE @@ -129,6 +136,91 @@ async def test_get_triggers(hass, aioclient_mock): assert_lists_same(triggers, expected_triggers) +async def test_get_triggers_for_alarm_event(hass, aioclient_mock): + """Test triggers work.""" + data = { + "sensors": { + "1": { + "config": { + "battery": 95, + "enrolled": 1, + "on": True, + "pending": [], + "reachable": True, + }, + "ep": 1, + "etag": "5aaa1c6bae8501f59929539c6e8f44d6", + "lastseen": "2021-07-25T18:07Z", + "manufacturername": "lk", + "modelid": "ZB-KeypadGeneric-D0002", + "name": "Keypad", + "state": { + "action": "armed_stay", + "lastupdated": "2021-07-25T18:02:51.172", + "lowbattery": False, + "panel": "exit_delay", + "seconds_remaining": 55, + "tampered": False, + }, + "swversion": "3.13", + "type": "ZHAAncillaryControl", + "uniqueid": "00:00:00:00:00:00:00:00-00", + } + } + } + with patch.dict(DECONZ_WEB_REQUEST, data): + await setup_deconz_integration(hass, aioclient_mock) + + device_registry = await hass.helpers.device_registry.async_get_registry() + device = device_registry.async_get_device( + identifiers={(DECONZ_DOMAIN, "00:00:00:00:00:00:00:00")} + ) + + triggers = await async_get_device_automations( + hass, DeviceAutomationType.TRIGGER, device.id + ) + + expected_triggers = [ + { + CONF_DEVICE_ID: device.id, + CONF_DOMAIN: BINARY_SENSOR_DOMAIN, + ATTR_ENTITY_ID: "binary_sensor.keypad_low_battery", + CONF_PLATFORM: "device", + CONF_TYPE: CONF_BAT_LOW, + }, + { + CONF_DEVICE_ID: device.id, + CONF_DOMAIN: BINARY_SENSOR_DOMAIN, + ATTR_ENTITY_ID: "binary_sensor.keypad_low_battery", + CONF_PLATFORM: "device", + CONF_TYPE: CONF_NOT_BAT_LOW, + }, + { + CONF_DEVICE_ID: device.id, + CONF_DOMAIN: BINARY_SENSOR_DOMAIN, + ATTR_ENTITY_ID: "binary_sensor.keypad_tampered", + CONF_PLATFORM: "device", + CONF_TYPE: CONF_TAMPERED, + }, + { + CONF_DEVICE_ID: device.id, + CONF_DOMAIN: BINARY_SENSOR_DOMAIN, + ATTR_ENTITY_ID: "binary_sensor.keypad_tampered", + CONF_PLATFORM: "device", + CONF_TYPE: CONF_NOT_TAMPERED, + }, + { + CONF_DEVICE_ID: device.id, + CONF_DOMAIN: SENSOR_DOMAIN, + ATTR_ENTITY_ID: "sensor.keypad_battery", + CONF_PLATFORM: "device", + CONF_TYPE: ATTR_BATTERY_LEVEL, + }, + ] + + assert_lists_same(triggers, expected_triggers) + + async def test_get_triggers_manage_unsupported_remotes(hass, aioclient_mock): """Verify no triggers for an unsupported remote.""" data = { @@ -244,9 +336,7 @@ async def test_functional_device_trigger( assert automation_calls[0].data["some"] == "test_trigger_button_press" -async def test_validate_trigger_unknown_device( - hass, aioclient_mock, mock_deconz_websocket -): +async def test_validate_trigger_unknown_device(hass, aioclient_mock): """Test unknown device does not return a trigger config.""" await setup_deconz_integration(hass, aioclient_mock) @@ -276,9 +366,7 @@ async def test_validate_trigger_unknown_device( assert len(hass.states.async_entity_ids(AUTOMATION_DOMAIN)) == 0 -async def test_validate_trigger_unsupported_device( - hass, aioclient_mock, mock_deconz_websocket -): +async def test_validate_trigger_unsupported_device(hass, aioclient_mock): """Test unsupported device doesn't return a trigger config.""" config_entry = await setup_deconz_integration(hass, aioclient_mock) @@ -315,9 +403,7 @@ async def test_validate_trigger_unsupported_device( assert len(hass.states.async_entity_ids(AUTOMATION_DOMAIN)) == 0 -async def test_validate_trigger_unsupported_trigger( - hass, aioclient_mock, mock_deconz_websocket -): +async def test_validate_trigger_unsupported_trigger(hass, aioclient_mock): """Test unsupported trigger does not return a trigger config.""" config_entry = await setup_deconz_integration(hass, aioclient_mock) @@ -356,9 +442,7 @@ async def test_validate_trigger_unsupported_trigger( assert len(hass.states.async_entity_ids(AUTOMATION_DOMAIN)) == 0 -async def test_attach_trigger_no_matching_event( - hass, aioclient_mock, mock_deconz_websocket -): +async def test_attach_trigger_no_matching_event(hass, aioclient_mock): """Test no matching event for device doesn't return a trigger config.""" config_entry = await setup_deconz_integration(hass, aioclient_mock)