evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add typing of deCONZ device_trigger (#67496)

Browse source
This commit is contained in:
Robert Svensson 2022-03-14 17:37:48 +01:00 • committed by GitHub
parent 38306417ad
commit 362191a0e6
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 134 additions and 28 deletions
Download patch file Download diff file Expand all files Collapse all files

54
homeassistant/components/deconz/device_trigger.py
View file

@ -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():

108
tests/components/deconz/test_device_trigger.py
View file

@ -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)