Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add events and device triggers to LCN (#58745)

pull/61465/head
Andre Lengwenus 2021-12-11 06:02:17 +01:00 committed by GitHub
parent 5e750a0625
commit 74d5cbd3a9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 831 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

86
homeassistant/components/lcn/__init__.py
View File

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

29
homeassistant/components/lcn/const.py
View File

@ -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",

105
homeassistant/components/lcn/device_trigger.py Normal file
View File

@ -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], {})

2
homeassistant/components/lcn/helpers.py
View File

@ -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]})",
)

2
homeassistant/components/lcn/manifest.json
View File

@ -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"
}

10
homeassistant/components/lcn/strings.json Normal file
View File

@ -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"
}
}
}

10
homeassistant/components/lcn/translations/en.json Normal file
View File

@ -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"
}
}
}

2
requirements_all.txt
View File

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

2
requirements_test_all.txt
View File

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

26
tests/components/lcn/conftest.py
View File

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

409
tests/components/lcn/test_device_trigger.py Normal file
View File

@ -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 == {}

153
tests/components/lcn/test_events.py Normal file
View File

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

1
tests/components/lcn/test_init.py
View File

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