From 74d5cbd3a9471824223f446e59fcb0cf9201657a Mon Sep 17 00:00:00 2001 From: Andre Lengwenus Date: Sat, 11 Dec 2021 06:02:17 +0100 Subject: [PATCH] Add events and device triggers to LCN (#58745) --- homeassistant/components/lcn/__init__.py | 86 +++- homeassistant/components/lcn/const.py | 29 ++ .../components/lcn/device_trigger.py | 105 +++++ homeassistant/components/lcn/helpers.py | 2 +- homeassistant/components/lcn/manifest.json | 2 +- homeassistant/components/lcn/strings.json | 10 + .../components/lcn/translations/en.json | 10 + requirements_all.txt | 2 +- requirements_test_all.txt | 2 +- tests/components/lcn/conftest.py | 26 +- tests/components/lcn/test_device_trigger.py | 409 ++++++++++++++++++ tests/components/lcn/test_events.py | 153 +++++++ tests/components/lcn/test_init.py | 1 + 13 files changed, 831 insertions(+), 6 deletions(-) create mode 100644 homeassistant/components/lcn/device_trigger.py create mode 100644 homeassistant/components/lcn/strings.json create mode 100644 homeassistant/components/lcn/translations/en.json create mode 100644 tests/components/lcn/test_device_trigger.py create mode 100644 tests/components/lcn/test_events.py diff --git a/homeassistant/components/lcn/__init__.py b/homeassistant/components/lcn/__init__.py index d019c156f37..b0b231cb9e9 100644 --- a/homeassistant/components/lcn/__init__.py +++ b/homeassistant/components/lcn/__init__.py @@ -2,6 +2,7 @@ from __future__ import annotations from collections.abc import Callable +from functools import partial import logging import pypck @@ -9,6 +10,7 @@ import pypck from homeassistant import config_entries from homeassistant.const import ( CONF_ADDRESS, + CONF_DEVICE_ID, CONF_DOMAIN, CONF_IP_ADDRESS, CONF_NAME, @@ -18,6 +20,7 @@ from homeassistant.const import ( CONF_USERNAME, ) from homeassistant.core import HomeAssistant +from homeassistant.helpers import device_registry as dr from homeassistant.helpers.entity import DeviceInfo, Entity from homeassistant.helpers.typing import ConfigType @@ -119,6 +122,13 @@ async def async_setup_entry( # forward config_entry to components hass.config_entries.async_setup_platforms(config_entry, PLATFORMS) + # register for LCN bus messages + device_registry = dr.async_get(hass) + input_received = partial( + async_host_input_received, hass, config_entry, device_registry + ) + lcn_connection.register_for_inputs(input_received) + # register service calls for service_name, service in SERVICES: if not hass.services.has_service(DOMAIN, service_name): @@ -150,6 +160,80 @@ async def async_unload_entry( return unload_ok +def async_host_input_received( + hass: HomeAssistant, + config_entry: config_entries.ConfigEntry, + device_registry: dr.DeviceRegistry, + inp: pypck.inputs.Input, +) -> None: + """Process received input object (command) from LCN bus.""" + if not isinstance(inp, pypck.inputs.ModInput): + return + + lcn_connection = hass.data[DOMAIN][config_entry.entry_id][CONNECTION] + logical_address = lcn_connection.physical_to_logical(inp.physical_source_addr) + address = ( + logical_address.seg_id, + logical_address.addr_id, + logical_address.is_group, + ) + identifiers = {(DOMAIN, generate_unique_id(config_entry.entry_id, address))} + device = device_registry.async_get_device(identifiers, set()) + if device is None: + return + + if isinstance(inp, pypck.inputs.ModStatusAccessControl): + _async_fire_access_control_event(hass, device, address, inp) + elif isinstance(inp, pypck.inputs.ModSendKeysHost): + _async_fire_send_keys_event(hass, device, address, inp) + + +def _async_fire_access_control_event( + hass: HomeAssistant, device: dr.DeviceEntry, address: AddressType, inp: InputType +) -> None: + """Fire access control event (transponder, transmitter, fingerprint).""" + event_data = { + "segment_id": address[0], + "module_id": address[1], + "code": inp.code, + } + + if device is not None: + event_data.update({CONF_DEVICE_ID: device.id}) + + if inp.periphery == pypck.lcn_defs.AccessControlPeriphery.TRANSMITTER: + event_data.update( + {"level": inp.level, "key": inp.key, "action": inp.action.value} + ) + + event_name = f"lcn_{inp.periphery.value.lower()}" + hass.bus.async_fire(event_name, event_data) + + +def _async_fire_send_keys_event( + hass: HomeAssistant, device: dr.DeviceEntry, address: AddressType, inp: InputType +) -> None: + """Fire send_keys event.""" + for table, action in enumerate(inp.actions): + if action == pypck.lcn_defs.SendKeyCommand.DONTSEND: + continue + + for key, selected in enumerate(inp.keys): + if not selected: + continue + event_data = { + "segment_id": address[0], + "module_id": address[1], + "key": pypck.lcn_defs.Key(table * 8 + key).name.lower(), + "action": action.name.lower(), + } + + if device is not None: + event_data.update({CONF_DEVICE_ID: device.id}) + + hass.bus.async_fire("lcn_send_keys", event_data) + + class LcnEntity(Entity): """Parent class for all entities associated with the LCN component.""" @@ -183,7 +267,7 @@ class LcnEntity(Entity): def device_info(self) -> DeviceInfo | None: """Return device specific attributes.""" address = f"{'g' if self.address[2] else 'm'}{self.address[0]:03d}{self.address[1]:03d}" - model = f"LCN {get_device_model(self.config[CONF_DOMAIN], self.config[CONF_DOMAIN_DATA])}" + model = f"LCN resource ({get_device_model(self.config[CONF_DOMAIN], self.config[CONF_DOMAIN_DATA])})" return { "identifiers": {(DOMAIN, self.unique_id)}, diff --git a/homeassistant/components/lcn/const.py b/homeassistant/components/lcn/const.py index f2059827e84..ef0ad6481f4 100644 --- a/homeassistant/components/lcn/const.py +++ b/homeassistant/components/lcn/const.py @@ -192,6 +192,35 @@ RELVARREF = ["CURRENT", "PROG"] SENDKEYCOMMANDS = ["HIT", "MAKE", "BREAK", "DONTSEND"] +SENDKEYS = [ + "A1", + "A2", + "A3", + "A4", + "A5", + "A6", + "A7", + "A8", + "B1", + "B2", + "B3", + "B4", + "B5", + "B6", + "B7", + "B8", + "C1", + "C2", + "C3", + "C4", + "C5", + "C6", + "C7", + "C8", +] + +KEY_ACTIONS = ["HIT", "MAKE", "BREAK"] + TIME_UNITS = [ "SECONDS", "SECOND", diff --git a/homeassistant/components/lcn/device_trigger.py b/homeassistant/components/lcn/device_trigger.py new file mode 100644 index 00000000000..b21c3b820af --- /dev/null +++ b/homeassistant/components/lcn/device_trigger.py @@ -0,0 +1,105 @@ +"""Provides device triggers for LCN.""" +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.homeassistant.triggers import event +from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_PLATFORM, CONF_TYPE +from homeassistant.core import CALLBACK_TYPE, HomeAssistant +from homeassistant.helpers import config_validation as cv, device_registry as dr +from homeassistant.helpers.typing import ConfigType + +from . import DOMAIN +from .const import KEY_ACTIONS, SENDKEYS + +TRIGGER_TYPES = {"transmitter", "transponder", "fingerprint", "send_keys"} + +LCN_DEVICE_TRIGGER_BASE_SCHEMA = DEVICE_TRIGGER_BASE_SCHEMA.extend( + {vol.Required(CONF_TYPE): vol.In(TRIGGER_TYPES)} +) + +ACCESS_CONTROL_SCHEMA = {vol.Optional("code"): vol.All(vol.Lower, cv.string)} +TRANSMITTER_SCHEMA = { + **ACCESS_CONTROL_SCHEMA, + vol.Optional("level"): cv.positive_int, + vol.Optional("key"): cv.positive_int, + vol.Optional("action"): vol.In([action.lower() for action in KEY_ACTIONS]), +} + +SENDKEYS_SCHEMA = { + vol.Optional("key"): vol.In([key.lower() for key in SENDKEYS]), + vol.Optional("action"): vol.In([action.lower() for action in KEY_ACTIONS]), +} + +TRIGGER_SCHEMA = vol.Any( + LCN_DEVICE_TRIGGER_BASE_SCHEMA.extend(ACCESS_CONTROL_SCHEMA), + LCN_DEVICE_TRIGGER_BASE_SCHEMA.extend(TRANSMITTER_SCHEMA), + LCN_DEVICE_TRIGGER_BASE_SCHEMA.extend(SENDKEYS_SCHEMA), +) + +TYPE_SCHEMAS = { + "transmitter": {"extra_fields": vol.Schema(TRANSMITTER_SCHEMA)}, + "transponder": {"extra_fields": vol.Schema(ACCESS_CONTROL_SCHEMA)}, + "fingerprint": {"extra_fields": vol.Schema(ACCESS_CONTROL_SCHEMA)}, + "send_keys": {"extra_fields": vol.Schema(SENDKEYS_SCHEMA)}, +} + + +async def async_get_triggers( + hass: HomeAssistant, device_id: str +) -> list[dict[str, Any]]: + """List device triggers for LCN devices.""" + device_registry = dr.async_get(hass) + device = device_registry.async_get(device_id) + + if device.model.startswith(("LCN host", "LCN group", "LCN resource")): # type: ignore[union-attr] + return [] + + base_trigger = { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_DEVICE_ID: device_id, + } + + return [{**base_trigger, CONF_TYPE: type_} for type_ in TRIGGER_TYPES] + + +async def async_attach_trigger( + hass: HomeAssistant, + config: ConfigType, + action: AutomationActionType, + automation_info: AutomationTriggerInfo, +) -> CALLBACK_TYPE: + """Attach a trigger.""" + event_data = { + CONF_DEVICE_ID: config[CONF_DEVICE_ID], + **{ + key: config[key] + for key in ("code", "level", "key", "action") + if key in config + }, + } + + event_config = event.TRIGGER_SCHEMA( + { + event.CONF_PLATFORM: "event", + event.CONF_EVENT_TYPE: f"lcn_{config[CONF_TYPE]}", + event.CONF_EVENT_DATA: event_data, + } + ) + + return await event.async_attach_trigger( + hass, event_config, action, automation_info, platform_type="device" + ) + + +async def async_get_trigger_capabilities(hass: HomeAssistant, config: dict) -> dict: + """List trigger capabilities.""" + return TYPE_SCHEMAS.get(config[CONF_TYPE], {}) diff --git a/homeassistant/components/lcn/helpers.py b/homeassistant/components/lcn/helpers.py index b879c2d3f72..74f135c4d1e 100644 --- a/homeassistant/components/lcn/helpers.py +++ b/homeassistant/components/lcn/helpers.py @@ -256,7 +256,7 @@ def register_lcn_host_device(hass: HomeAssistant, config_entry: ConfigEntry) -> identifiers={(DOMAIN, config_entry.entry_id)}, manufacturer="Issendorff", name=config_entry.title, - model="PCHK", + model=f"LCN host ({config_entry.data[CONF_IP_ADDRESS]}:{config_entry.data[CONF_PORT]})", ) diff --git a/homeassistant/components/lcn/manifest.json b/homeassistant/components/lcn/manifest.json index 1adc407d692..4b06f7075f6 100644 --- a/homeassistant/components/lcn/manifest.json +++ b/homeassistant/components/lcn/manifest.json @@ -3,7 +3,7 @@ "name": "LCN", "config_flow": false, "documentation": "https://www.home-assistant.io/integrations/lcn", - "requirements": ["pypck==0.7.10"], + "requirements": ["pypck==0.7.11"], "codeowners": ["@alengwenus"], "iot_class": "local_push" } diff --git a/homeassistant/components/lcn/strings.json b/homeassistant/components/lcn/strings.json new file mode 100644 index 00000000000..078172ff34a --- /dev/null +++ b/homeassistant/components/lcn/strings.json @@ -0,0 +1,10 @@ +{ + "device_automation": { + "trigger_type": { + "transmitter": "transmitter code received", + "transponder": "transpoder code received", + "fingerprint": "fingerprint code received", + "send_keys": "send keys received" + } + } +} diff --git a/homeassistant/components/lcn/translations/en.json b/homeassistant/components/lcn/translations/en.json new file mode 100644 index 00000000000..9a27b35a4d4 --- /dev/null +++ b/homeassistant/components/lcn/translations/en.json @@ -0,0 +1,10 @@ +{ + "device_automation": { + "trigger_type": { + "fingerprint": "fingerprint code received", + "send_keys": "send keys received", + "transmitter": "transmitter code received", + "transponder": "transpoder code received" + } + } +} \ No newline at end of file diff --git a/requirements_all.txt b/requirements_all.txt index 721bfa3a0ae..aa8200cf48d 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -1726,7 +1726,7 @@ pyownet==0.10.0.post1 pypca==0.0.7 # homeassistant.components.lcn -pypck==0.7.10 +pypck==0.7.11 # homeassistant.components.pjlink pypjlink2==1.2.1 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index c543bba974e..f70f8f281d1 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -1072,7 +1072,7 @@ pyowm==3.2.0 pyownet==0.10.0.post1 # homeassistant.components.lcn -pypck==0.7.10 +pypck==0.7.11 # homeassistant.components.plaato pyplaato==0.0.15 diff --git a/tests/components/lcn/conftest.py b/tests/components/lcn/conftest.py index 81c2fdc68e4..aebae09547a 100644 --- a/tests/components/lcn/conftest.py +++ b/tests/components/lcn/conftest.py @@ -9,10 +9,12 @@ from pypck.module import GroupConnection, ModuleConnection import pytest from homeassistant.components.lcn.const import DOMAIN +from homeassistant.components.lcn.helpers import generate_unique_id from homeassistant.const import CONF_HOST +from homeassistant.helpers import device_registry as dr from homeassistant.setup import async_setup_component -from tests.common import MockConfigEntry, load_fixture +from tests.common import MockConfigEntry, async_mock_service, load_fixture class MockModuleConnection(ModuleConnection): @@ -39,6 +41,13 @@ class MockGroupConnection(GroupConnection): class MockPchkConnectionManager(PchkConnectionManager): """Fake connection handler.""" + return_value = None + + def __init__(self, *args, **kwargs): + """Initialize MockPchkCOnnectionManager.""" + super().__init__(*args, **kwargs) + self.__class__.return_value = self + async def async_connect(self, timeout=30): """Mock establishing a connection to PCHK.""" self.authentication_completed_future.set_result(True) @@ -75,6 +84,12 @@ def create_config_entry(name): return entry +@pytest.fixture +def calls(hass): + """Track calls to a mock service.""" + return async_mock_service(hass, "test", "automation") + + @pytest.fixture(name="entry") def create_config_entry_pchk(): """Return one specific config entry.""" @@ -101,3 +116,12 @@ async def setup_component(hass): await async_setup_component(hass, DOMAIN, config_data) await hass.async_block_till_done() + + +def get_device(hass, entry, address): + """Get LCN device for specified address.""" + device_registry = dr.async_get(hass) + identifiers = {(DOMAIN, generate_unique_id(entry.entry_id, address))} + device = device_registry.async_get_device(identifiers) + assert device + return device diff --git a/tests/components/lcn/test_device_trigger.py b/tests/components/lcn/test_device_trigger.py new file mode 100644 index 00000000000..00138594a5c --- /dev/null +++ b/tests/components/lcn/test_device_trigger.py @@ -0,0 +1,409 @@ +"""Tests for LCN device triggers.""" +from unittest.mock import patch + +from pypck.inputs import ModSendKeysHost, ModStatusAccessControl +from pypck.lcn_addr import LcnAddr +from pypck.lcn_defs import AccessControlPeriphery, KeyAction, SendKeyCommand +import voluptuous_serialize + +from homeassistant.components import automation +from homeassistant.components.lcn import device_trigger +from homeassistant.components.lcn.const import DOMAIN, KEY_ACTIONS, SENDKEYS +from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_PLATFORM, CONF_TYPE +from homeassistant.helpers import config_validation as cv, device_registry as dr +from homeassistant.setup import async_setup_component + +from .conftest import MockPchkConnectionManager, get_device, init_integration + +from tests.common import assert_lists_same, async_get_device_automations + + +@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager) +async def test_get_triggers_module_device(hass, entry): + """Test we get the expected triggers from a LCN module device.""" + await init_integration(hass, entry) + device = get_device(hass, entry, (0, 7, False)) + + expected_triggers = [ + { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_TYPE: "transmitter", + CONF_DEVICE_ID: device.id, + }, + { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_TYPE: "transponder", + CONF_DEVICE_ID: device.id, + }, + { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_TYPE: "fingerprint", + CONF_DEVICE_ID: device.id, + }, + { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_TYPE: "send_keys", + CONF_DEVICE_ID: device.id, + }, + ] + + triggers = await async_get_device_automations(hass, "trigger", device.id) + assert_lists_same(triggers, expected_triggers) + + +@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager) +async def test_get_triggers_non_module_device(hass, entry): + """Test we get the expected triggers from a LCN non-module device.""" + not_included_types = ("transmitter", "transponder", "fingerprint", "send_keys") + + await init_integration(hass, entry) + device_registry = dr.async_get(hass) + for device_id in device_registry.devices: + device = device_registry.async_get(device_id) + if device.model.startswith(("LCN host", "LCN group", "LCN resource")): + triggers = await async_get_device_automations(hass, "trigger", device_id) + for trigger in triggers: + assert trigger[CONF_TYPE] not in not_included_types + + +@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager) +async def test_if_fires_on_transponder_event(hass, calls, entry): + """Test for transponder event triggers firing.""" + await init_integration(hass, entry) + address = (0, 7, False) + device = get_device(hass, entry, address) + + assert await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + { + "trigger": { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_DEVICE_ID: device.id, + CONF_TYPE: "transponder", + }, + "action": { + "service": "test.automation", + "data_template": { + "test": "test_trigger_transponder", + "code": "{{ trigger.event.data.code }}", + }, + }, + }, + ] + }, + ) + + inp = ModStatusAccessControl( + LcnAddr(*address), + periphery=AccessControlPeriphery.TRANSPONDER, + code="aabbcc", + ) + + lcn_connection = MockPchkConnectionManager.return_value + await lcn_connection.async_process_input(inp) + await hass.async_block_till_done() + + assert len(calls) == 1 + assert calls[0].data == { + "test": "test_trigger_transponder", + "code": "aabbcc", + } + + +@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager) +async def test_if_fires_on_fingerprint_event(hass, calls, entry): + """Test for fingerprint event triggers firing.""" + await init_integration(hass, entry) + address = (0, 7, False) + device = get_device(hass, entry, address) + + assert await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + { + "trigger": { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_DEVICE_ID: device.id, + CONF_TYPE: "fingerprint", + }, + "action": { + "service": "test.automation", + "data_template": { + "test": "test_trigger_fingerprint", + "code": "{{ trigger.event.data.code }}", + }, + }, + }, + ] + }, + ) + + inp = ModStatusAccessControl( + LcnAddr(*address), + periphery=AccessControlPeriphery.FINGERPRINT, + code="aabbcc", + ) + + lcn_connection = MockPchkConnectionManager.return_value + await lcn_connection.async_process_input(inp) + await hass.async_block_till_done() + + assert len(calls) == 1 + assert calls[0].data == { + "test": "test_trigger_fingerprint", + "code": "aabbcc", + } + + +@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager) +async def test_if_fires_on_transmitter_event(hass, calls, entry): + """Test for transmitter event triggers firing.""" + await init_integration(hass, entry) + address = (0, 7, False) + device = get_device(hass, entry, address) + + assert await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + { + "trigger": { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_DEVICE_ID: device.id, + CONF_TYPE: "transmitter", + }, + "action": { + "service": "test.automation", + "data_template": { + "test": "test_trigger_transmitter", + "code": "{{ trigger.event.data.code }}", + "level": "{{ trigger.event.data.level }}", + "key": "{{ trigger.event.data.key }}", + "action": "{{ trigger.event.data.action }}", + }, + }, + }, + ] + }, + ) + + inp = ModStatusAccessControl( + LcnAddr(*address), + periphery=AccessControlPeriphery.TRANSMITTER, + code="aabbcc", + level=0, + key=0, + action=KeyAction.HIT, + ) + + lcn_connection = MockPchkConnectionManager.return_value + await lcn_connection.async_process_input(inp) + await hass.async_block_till_done() + + assert len(calls) == 1 + assert calls[0].data == { + "test": "test_trigger_transmitter", + "code": "aabbcc", + "level": 0, + "key": 0, + "action": "hit", + } + + +@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager) +async def test_if_fires_on_send_keys_event(hass, calls, entry): + """Test for send_keys event triggers firing.""" + await init_integration(hass, entry) + address = (0, 7, False) + device = get_device(hass, entry, address) + + assert await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + { + "trigger": { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_DEVICE_ID: device.id, + CONF_TYPE: "send_keys", + }, + "action": { + "service": "test.automation", + "data_template": { + "test": "test_trigger_send_keys", + "key": "{{ trigger.event.data.key }}", + "action": "{{ trigger.event.data.action }}", + }, + }, + }, + ] + }, + ) + + inp = ModSendKeysHost( + LcnAddr(*address), + actions=[SendKeyCommand.HIT, SendKeyCommand.DONTSEND, SendKeyCommand.DONTSEND], + keys=[True, False, False, False, False, False, False, False], + ) + + lcn_connection = MockPchkConnectionManager.return_value + await lcn_connection.async_process_input(inp) + await hass.async_block_till_done() + + assert len(calls) == 1 + assert calls[0].data == { + "test": "test_trigger_send_keys", + "key": "a1", + "action": "hit", + } + + +@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager) +async def test_get_transponder_trigger_capabilities(hass, entry): + """Test we get the expected capabilities from a transponder device trigger.""" + await init_integration(hass, entry) + address = (0, 7, False) + device = get_device(hass, entry, address) + + capabilities = await device_trigger.async_get_trigger_capabilities( + hass, + { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_TYPE: "transponder", + CONF_DEVICE_ID: device.id, + }, + ) + assert capabilities and "extra_fields" in capabilities + + assert voluptuous_serialize.convert( + capabilities["extra_fields"], custom_serializer=cv.custom_serializer + ) == [{"name": "code", "optional": True, "type": "string", "lower": True}] + + +@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager) +async def test_get_fingerprint_trigger_capabilities(hass, entry): + """Test we get the expected capabilities from a fingerprint device trigger.""" + await init_integration(hass, entry) + address = (0, 7, False) + device = get_device(hass, entry, address) + + capabilities = await device_trigger.async_get_trigger_capabilities( + hass, + { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_TYPE: "fingerprint", + CONF_DEVICE_ID: device.id, + }, + ) + assert capabilities and "extra_fields" in capabilities + + assert voluptuous_serialize.convert( + capabilities["extra_fields"], custom_serializer=cv.custom_serializer + ) == [{"name": "code", "optional": True, "type": "string", "lower": True}] + + +@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager) +async def test_get_transmitter_trigger_capabilities(hass, entry): + """Test we get the expected capabilities from a transmitter device trigger.""" + await init_integration(hass, entry) + address = (0, 7, False) + device = get_device(hass, entry, address) + + capabilities = await device_trigger.async_get_trigger_capabilities( + hass, + { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_TYPE: "transmitter", + CONF_DEVICE_ID: device.id, + }, + ) + assert capabilities and "extra_fields" in capabilities + + assert voluptuous_serialize.convert( + capabilities["extra_fields"], custom_serializer=cv.custom_serializer + ) == [ + {"name": "code", "type": "string", "optional": True, "lower": True}, + {"name": "level", "type": "integer", "optional": True, "valueMin": 0}, + {"name": "key", "type": "integer", "optional": True, "valueMin": 0}, + { + "name": "action", + "type": "select", + "optional": True, + "options": [("hit", "hit"), ("make", "make"), ("break", "break")], + }, + ] + + +@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager) +async def test_get_send_keys_trigger_capabilities(hass, entry): + """Test we get the expected capabilities from a send_keys device trigger.""" + await init_integration(hass, entry) + address = (0, 7, False) + device = get_device(hass, entry, address) + + capabilities = await device_trigger.async_get_trigger_capabilities( + hass, + { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_TYPE: "send_keys", + CONF_DEVICE_ID: device.id, + }, + ) + assert capabilities and "extra_fields" in capabilities + + assert voluptuous_serialize.convert( + capabilities["extra_fields"], custom_serializer=cv.custom_serializer + ) == [ + { + "name": "key", + "type": "select", + "optional": True, + "options": [(send_key.lower(), send_key.lower()) for send_key in SENDKEYS], + }, + { + "name": "action", + "type": "select", + "options": [ + (key_action.lower(), key_action.lower()) for key_action in KEY_ACTIONS + ], + "optional": True, + }, + ] + + +@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager) +async def test_unknown_trigger_capabilities(hass, entry): + """Test we get empty capabilities if trigger is unknown.""" + await init_integration(hass, entry) + address = (0, 7, False) + device = get_device(hass, entry, address) + + capabilities = await device_trigger.async_get_trigger_capabilities( + hass, + { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_TYPE: "dummy", + CONF_DEVICE_ID: device.id, + }, + ) + assert capabilities == {} diff --git a/tests/components/lcn/test_events.py b/tests/components/lcn/test_events.py new file mode 100644 index 00000000000..f977d586dad --- /dev/null +++ b/tests/components/lcn/test_events.py @@ -0,0 +1,153 @@ +"""Tests for LCN events.""" +from unittest.mock import patch + +from pypck.inputs import Input, ModSendKeysHost, ModStatusAccessControl +from pypck.lcn_addr import LcnAddr +from pypck.lcn_defs import AccessControlPeriphery, KeyAction, SendKeyCommand + +from .conftest import MockPchkConnectionManager, init_integration + +from tests.common import async_capture_events + + +@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager) +async def test_fire_transponder_event(hass, entry): + """Test the transponder event is fired.""" + await init_integration(hass, entry) + + events = async_capture_events(hass, "lcn_transponder") + + inp = ModStatusAccessControl( + LcnAddr(0, 7, False), + periphery=AccessControlPeriphery.TRANSPONDER, + code="aabbcc", + ) + + lcn_connection = MockPchkConnectionManager.return_value + await lcn_connection.async_process_input(inp) + await hass.async_block_till_done() + + assert len(events) == 1 + assert events[0].event_type == "lcn_transponder" + assert events[0].data["code"] == "aabbcc" + + +@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager) +async def test_fire_fingerprint_event(hass, entry): + """Test the fingerprint event is fired.""" + await init_integration(hass, entry) + + events = async_capture_events(hass, "lcn_fingerprint") + + inp = ModStatusAccessControl( + LcnAddr(0, 7, False), + periphery=AccessControlPeriphery.FINGERPRINT, + code="aabbcc", + ) + + lcn_connection = MockPchkConnectionManager.return_value + await lcn_connection.async_process_input(inp) + await hass.async_block_till_done() + + assert len(events) == 1 + assert events[0].event_type == "lcn_fingerprint" + assert events[0].data["code"] == "aabbcc" + + +@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager) +async def test_fire_transmitter_event(hass, entry): + """Test the transmitter event is fired.""" + await init_integration(hass, entry) + + events = async_capture_events(hass, "lcn_transmitter") + + inp = ModStatusAccessControl( + LcnAddr(0, 7, False), + periphery=AccessControlPeriphery.TRANSMITTER, + code="aabbcc", + level=0, + key=0, + action=KeyAction.HIT, + ) + + lcn_connection = MockPchkConnectionManager.return_value + await lcn_connection.async_process_input(inp) + await hass.async_block_till_done() + + assert len(events) == 1 + assert events[0].event_type == "lcn_transmitter" + assert events[0].data["code"] == "aabbcc" + assert events[0].data["level"] == 0 + assert events[0].data["key"] == 0 + assert events[0].data["action"] == "hit" + + +@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager) +async def test_fire_sendkeys_event(hass, entry): + """Test the send_keys event is fired.""" + await init_integration(hass, entry) + + events = async_capture_events(hass, "lcn_send_keys") + + inp = ModSendKeysHost( + LcnAddr(0, 7, False), + actions=[SendKeyCommand.HIT, SendKeyCommand.MAKE, SendKeyCommand.DONTSEND], + keys=[True, True, False, False, False, False, False, False], + ) + + lcn_connection = MockPchkConnectionManager.return_value + await lcn_connection.async_process_input(inp) + await hass.async_block_till_done() + + assert len(events) == 4 + assert events[0].event_type == "lcn_send_keys" + assert events[0].data["key"] == "a1" + assert events[0].data["action"] == "hit" + assert events[1].event_type == "lcn_send_keys" + assert events[1].data["key"] == "a2" + assert events[1].data["action"] == "hit" + assert events[2].event_type == "lcn_send_keys" + assert events[2].data["key"] == "b1" + assert events[2].data["action"] == "make" + assert events[3].event_type == "lcn_send_keys" + assert events[3].data["key"] == "b2" + assert events[3].data["action"] == "make" + + +@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager) +async def test_dont_fire_on_non_module_input(hass, entry): + """Test for no event is fired if a non-module input is received.""" + await init_integration(hass, entry) + + inp = Input() + lcn_connection = MockPchkConnectionManager.return_value + + for event_name in ( + "lcn_transponder", + "lcn_fingerprint", + "lcn_transmitter", + "lcn_send_keys", + ): + events = async_capture_events(hass, event_name) + await lcn_connection.async_process_input(inp) + await hass.async_block_till_done() + assert len(events) == 0 + + +@patch("pypck.connection.PchkConnectionManager", MockPchkConnectionManager) +async def test_dont_fire_on_unknown_module(hass, entry): + """Test for no event is fired if an input from an unknown module is received.""" + await init_integration(hass, entry) + + inp = ModStatusAccessControl( + LcnAddr(0, 10, False), # unknown module + periphery=AccessControlPeriphery.FINGERPRINT, + code="aabbcc", + ) + + lcn_connection = MockPchkConnectionManager.return_value + + events = async_capture_events(hass, "lcn_transmitter") + await lcn_connection.async_process_input(inp) + await hass.async_block_till_done() + assert len(events) == 0 diff --git a/tests/components/lcn/test_init.py b/tests/components/lcn/test_init.py index e4fb5beef0d..40f655dd695 100644 --- a/tests/components/lcn/test_init.py +++ b/tests/components/lcn/test_init.py @@ -19,6 +19,7 @@ from .conftest import MockPchkConnectionManager, init_integration, setup_compone async def test_async_setup_entry(hass, entry): """Test a successful setup entry and unload of entry.""" await init_integration(hass, entry) + assert len(hass.config_entries.async_entries(DOMAIN)) == 1 assert entry.state == ConfigEntryState.LOADED