From bca277a027ad9715d21d8636a4cb1a86ca841082 Mon Sep 17 00:00:00 2001 From: Matthias Alphart Date: Tue, 14 May 2024 14:45:49 +0200 Subject: [PATCH] Add `knx.telegram` integration specific trigger; update KNX Interface device trigger (#107592) * Add `knx.telegram` integration specific trigger * Move implementation to trigger.py, use it from device_trigger * test device_trigger * test trigger.py * Add "incoming" and "outgoing" and handle legacy device triggers * work with mixed group address styles * improve coverage * Add no-op option * apply changed linting rules * Don't distinguish legacy device triggers from new ones that's now supported since frontend has fixed default values of extra_fields * review suggestion: reuse trigger schema for device trigger extra fields * cleanup for readability * Remove no-op option --- .../components/knx/device_trigger.py | 56 ++-- homeassistant/components/knx/trigger.py | 101 ++++++ tests/components/knx/test_device_trigger.py | 268 ++++++++++++++-- tests/components/knx/test_trigger.py | 290 ++++++++++++++++++ 4 files changed, 660 insertions(+), 55 deletions(-) create mode 100644 homeassistant/components/knx/trigger.py create mode 100644 tests/components/knx/test_trigger.py diff --git a/homeassistant/components/knx/device_trigger.py b/homeassistant/components/knx/device_trigger.py index 93e1623f88c..5551aa1d439 100644 --- a/homeassistant/components/knx/device_trigger.py +++ b/homeassistant/components/knx/device_trigger.py @@ -7,26 +7,32 @@ from typing import Any, Final import voluptuous as vol from homeassistant.components.device_automation import DEVICE_TRIGGER_BASE_SCHEMA +from homeassistant.components.device_automation.exceptions import ( + InvalidDeviceAutomationConfig, +) from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_PLATFORM, CONF_TYPE -from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback +from homeassistant.core import CALLBACK_TYPE, HomeAssistant from homeassistant.helpers import selector -from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo from homeassistant.helpers.typing import ConfigType -from . import KNXModule -from .const import DOMAIN, SIGNAL_KNX_TELEGRAM_DICT +from . import KNXModule, trigger +from .const import DOMAIN from .project import KNXProject -from .schema import ga_list_validator -from .telegrams import TelegramDict +from .trigger import ( + CONF_KNX_DESTINATION, + PLATFORM_TYPE_TRIGGER_TELEGRAM, + TELEGRAM_TRIGGER_OPTIONS, + TELEGRAM_TRIGGER_SCHEMA, + TRIGGER_SCHEMA as TRIGGER_TRIGGER_SCHEMA, +) TRIGGER_TELEGRAM: Final = "telegram" -EXTRA_FIELD_DESTINATION: Final = "destination" # no translation support -TRIGGER_SCHEMA = DEVICE_TRIGGER_BASE_SCHEMA.extend( +TRIGGER_SCHEMA: Final = DEVICE_TRIGGER_BASE_SCHEMA.extend( { - vol.Optional(EXTRA_FIELD_DESTINATION): ga_list_validator, vol.Required(CONF_TYPE): TRIGGER_TELEGRAM, + **TELEGRAM_TRIGGER_SCHEMA, } ) @@ -42,11 +48,10 @@ async def async_get_triggers( # Add trigger for KNX telegrams to interface device triggers.append( { - # Required fields of TRIGGER_BASE_SCHEMA + # Default fields when initializing the trigger CONF_PLATFORM: "device", CONF_DOMAIN: DOMAIN, CONF_DEVICE_ID: device_id, - # Required fields of TRIGGER_SCHEMA CONF_TYPE: TRIGGER_TELEGRAM, } ) @@ -66,7 +71,7 @@ async def async_get_trigger_capabilities( return { "extra_fields": vol.Schema( { - vol.Optional(EXTRA_FIELD_DESTINATION): selector.SelectSelector( + vol.Optional(CONF_KNX_DESTINATION): selector.SelectSelector( selector.SelectSelectorConfig( mode=selector.SelectSelectorMode.DROPDOWN, multiple=True, @@ -74,6 +79,7 @@ async def async_get_trigger_capabilities( options=options, ), ), + **TELEGRAM_TRIGGER_OPTIONS, } ) } @@ -86,22 +92,16 @@ async def async_attach_trigger( trigger_info: TriggerInfo, ) -> CALLBACK_TYPE: """Attach a trigger.""" - trigger_data = trigger_info["trigger_data"] - dst_addresses: list[str] = config.get(EXTRA_FIELD_DESTINATION, []) - job = HassJob(action, f"KNX device trigger {trigger_info}") + # Remove device trigger specific fields and add trigger platform identifier + trigger_config = { + key: config[key] for key in (config.keys() & TELEGRAM_TRIGGER_SCHEMA.keys()) + } | {CONF_PLATFORM: PLATFORM_TYPE_TRIGGER_TELEGRAM} - @callback - def async_call_trigger_action(telegram: TelegramDict) -> None: - """Filter Telegram and call trigger action.""" - if dst_addresses and telegram["destination"] not in dst_addresses: - return - hass.async_run_hass_job( - job, - {"trigger": {**trigger_data, **telegram}}, - ) + try: + TRIGGER_TRIGGER_SCHEMA(trigger_config) + except vol.Invalid as err: + raise InvalidDeviceAutomationConfig(f"{err}") from err - return async_dispatcher_connect( - hass, - signal=SIGNAL_KNX_TELEGRAM_DICT, - target=async_call_trigger_action, + return await trigger.async_attach_trigger( + hass, config=trigger_config, action=action, trigger_info=trigger_info ) diff --git a/homeassistant/components/knx/trigger.py b/homeassistant/components/knx/trigger.py new file mode 100644 index 00000000000..16907fa9748 --- /dev/null +++ b/homeassistant/components/knx/trigger.py @@ -0,0 +1,101 @@ +"""Offer knx telegram automation triggers.""" + +from typing import Final + +import voluptuous as vol +from xknx.telegram.address import DeviceGroupAddress, parse_device_group_address + +from homeassistant.const import CONF_PLATFORM +from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback +from homeassistant.helpers import config_validation as cv +from homeassistant.helpers.dispatcher import async_dispatcher_connect +from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo +from homeassistant.helpers.typing import ConfigType + +from .const import DOMAIN, SIGNAL_KNX_TELEGRAM_DICT +from .schema import ga_validator +from .telegrams import TelegramDict + +TRIGGER_TELEGRAM: Final = "telegram" + +PLATFORM_TYPE_TRIGGER_TELEGRAM: Final = f"{DOMAIN}.{TRIGGER_TELEGRAM}" + +CONF_KNX_DESTINATION: Final = "destination" +CONF_KNX_GROUP_VALUE_WRITE: Final = "group_value_write" +CONF_KNX_GROUP_VALUE_READ: Final = "group_value_read" +CONF_KNX_GROUP_VALUE_RESPONSE: Final = "group_value_response" +CONF_KNX_INCOMING: Final = "incoming" +CONF_KNX_OUTGOING: Final = "outgoing" + +TELEGRAM_TRIGGER_OPTIONS: Final = { + vol.Optional(CONF_KNX_GROUP_VALUE_WRITE, default=True): cv.boolean, + vol.Optional(CONF_KNX_GROUP_VALUE_RESPONSE, default=True): cv.boolean, + vol.Optional(CONF_KNX_GROUP_VALUE_READ, default=True): cv.boolean, + vol.Optional(CONF_KNX_INCOMING, default=True): cv.boolean, + vol.Optional(CONF_KNX_OUTGOING, default=True): cv.boolean, +} +TELEGRAM_TRIGGER_SCHEMA: Final = { + vol.Optional(CONF_KNX_DESTINATION): vol.All( + cv.ensure_list, + [ga_validator], + ), + **TELEGRAM_TRIGGER_OPTIONS, +} + +TRIGGER_SCHEMA = cv.TRIGGER_BASE_SCHEMA.extend( + { + vol.Required(CONF_PLATFORM): PLATFORM_TYPE_TRIGGER_TELEGRAM, + **TELEGRAM_TRIGGER_SCHEMA, + } +) + + +async def async_attach_trigger( + hass: HomeAssistant, + config: ConfigType, + action: TriggerActionType, + trigger_info: TriggerInfo, +) -> CALLBACK_TYPE: + """Listen for telegrams based on configuration.""" + _addresses: list[str] = config.get(CONF_KNX_DESTINATION, []) + dst_addresses: list[DeviceGroupAddress] = [ + parse_device_group_address(address) for address in _addresses + ] + job = HassJob(action, f"KNX trigger {trigger_info}") + trigger_data = trigger_info["trigger_data"] + + @callback + def async_call_trigger_action(telegram: TelegramDict) -> None: + """Filter Telegram and call trigger action.""" + if telegram["telegramtype"] == "GroupValueWrite": + if config[CONF_KNX_GROUP_VALUE_WRITE] is False: + return + elif telegram["telegramtype"] == "GroupValueResponse": + if config[CONF_KNX_GROUP_VALUE_RESPONSE] is False: + return + elif telegram["telegramtype"] == "GroupValueRead": + if config[CONF_KNX_GROUP_VALUE_READ] is False: + return + + if telegram["direction"] == "Incoming": + if config[CONF_KNX_INCOMING] is False: + return + elif config[CONF_KNX_OUTGOING] is False: + return + + if ( + dst_addresses + and parse_device_group_address(telegram["destination"]) not in dst_addresses + ): + return + + hass.async_run_hass_job( + job, + {"trigger": {**trigger_data, **telegram}}, + ) + + return async_dispatcher_connect( + hass, + signal=SIGNAL_KNX_TELEGRAM_DICT, + target=async_call_trigger_action, + ) diff --git a/tests/components/knx/test_device_trigger.py b/tests/components/knx/test_device_trigger.py index 3c8bf58169b..278267c4f8a 100644 --- a/tests/components/knx/test_device_trigger.py +++ b/tests/components/knx/test_device_trigger.py @@ -1,10 +1,15 @@ """Tests for KNX device triggers.""" +import logging + import pytest import voluptuous_serialize from homeassistant.components import automation from homeassistant.components.device_automation import DeviceAutomationType +from homeassistant.components.device_automation.exceptions import ( + InvalidDeviceAutomationConfig, +) from homeassistant.components.knx import DOMAIN, device_trigger from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_OFF from homeassistant.core import HomeAssistant, ServiceCall @@ -22,36 +27,13 @@ def calls(hass: HomeAssistant) -> list[ServiceCall]: return async_mock_service(hass, "test", "automation") -async def test_get_triggers( - hass: HomeAssistant, - device_registry: dr.DeviceRegistry, - knx: KNXTestKit, -) -> None: - """Test we get the expected triggers from knx.""" - await knx.setup_integration({}) - device_entry = device_registry.async_get_device( - identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")} - ) - expected_trigger = { - "platform": "device", - "domain": DOMAIN, - "device_id": device_entry.id, - "type": "telegram", - "metadata": {}, - } - triggers = await async_get_device_automations( - hass, DeviceAutomationType.TRIGGER, device_entry.id - ) - assert expected_trigger in triggers - - async def test_if_fires_on_telegram( hass: HomeAssistant, calls: list[ServiceCall], device_registry: dr.DeviceRegistry, knx: KNXTestKit, ) -> None: - """Test for telegram triggers firing.""" + """Test telegram device triggers firing.""" await knx.setup_integration({}) device_entry = device_registry.async_get_device( identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")} @@ -63,6 +45,102 @@ async def test_if_fires_on_telegram( automation.DOMAIN, { automation.DOMAIN: [ + # "catch_all" trigger + { + "trigger": { + "platform": "device", + "domain": DOMAIN, + "device_id": device_entry.id, + "type": "telegram", + "group_value_write": True, + "group_value_response": True, + "group_value_read": True, + "incoming": True, + "outgoing": True, + }, + "action": { + "service": "test.automation", + "data_template": { + "catch_all": ("telegram - {{ trigger.destination }}"), + "id": (" {{ trigger.id }}"), + }, + }, + }, + # "specific" trigger + { + "trigger": { + "platform": "device", + "domain": DOMAIN, + "device_id": device_entry.id, + "id": "test-id", + "type": "telegram", + "destination": [ + "1/2/3", + "1/516", # "1/516" -> "1/2/4" in 2level format + ], + "group_value_write": True, + "group_value_response": False, + "group_value_read": False, + "incoming": True, + "outgoing": False, + }, + "action": { + "service": "test.automation", + "data_template": { + "specific": ("telegram - {{ trigger.destination }}"), + "id": (" {{ trigger.id }}"), + }, + }, + }, + ] + }, + ) + + # "specific" shall ignore destination address + await knx.receive_write("0/0/1", (0x03, 0x2F)) + assert len(calls) == 1 + test_call = calls.pop() + assert test_call.data["catch_all"] == "telegram - 0/0/1" + assert test_call.data["id"] == 0 + + await knx.receive_write("1/2/4", (0x03, 0x2F)) + assert len(calls) == 2 + test_call = calls.pop() + assert test_call.data["specific"] == "telegram - 1/2/4" + assert test_call.data["id"] == "test-id" + test_call = calls.pop() + assert test_call.data["catch_all"] == "telegram - 1/2/4" + assert test_call.data["id"] == 0 + + # "specific" shall ignore GroupValueRead + await knx.receive_read("1/2/4") + assert len(calls) == 1 + test_call = calls.pop() + assert test_call.data["catch_all"] == "telegram - 1/2/4" + assert test_call.data["id"] == 0 + + +async def test_default_if_fires_on_telegram( + hass: HomeAssistant, + calls: list[ServiceCall], + device_registry: dr.DeviceRegistry, + knx: KNXTestKit, +) -> None: + """Test default telegram device triggers firing.""" + # by default (without a user changing any) extra_fields are not added to the trigger and + # pre 2024.2 device triggers did only support "destination" field so they didn't have + # "group_value_write", "group_value_response", "group_value_read", "incoming", "outgoing" + await knx.setup_integration({}) + device_entry = device_registry.async_get_device( + identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")} + ) + + assert await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + # "catch_all" trigger { "trigger": { "platform": "device", @@ -78,6 +156,7 @@ async def test_if_fires_on_telegram( }, }, }, + # "specific" trigger { "trigger": { "platform": "device", @@ -114,6 +193,16 @@ async def test_if_fires_on_telegram( assert test_call.data["catch_all"] == "telegram - 1/2/4" assert test_call.data["id"] == 0 + # "specific" shall catch GroupValueRead as it is not set explicitly + await knx.receive_read("1/2/4") + assert len(calls) == 2 + test_call = calls.pop() + assert test_call.data["specific"] == "telegram - 1/2/4" + assert test_call.data["id"] == "test-id" + test_call = calls.pop() + assert test_call.data["catch_all"] == "telegram - 1/2/4" + assert test_call.data["id"] == 0 + async def test_remove_device_trigger( hass: HomeAssistant, @@ -165,12 +254,35 @@ async def test_remove_device_trigger( assert len(calls) == 0 -async def test_get_trigger_capabilities_node_status( +async def test_get_triggers( hass: HomeAssistant, device_registry: dr.DeviceRegistry, knx: KNXTestKit, ) -> None: - """Test we get the expected capabilities from a node_status trigger.""" + """Test we get the expected device triggers from knx.""" + await knx.setup_integration({}) + device_entry = device_registry.async_get_device( + identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")} + ) + expected_trigger = { + "platform": "device", + "domain": DOMAIN, + "device_id": device_entry.id, + "type": "telegram", + "metadata": {}, + } + triggers = await async_get_device_automations( + hass, DeviceAutomationType.TRIGGER, device_entry.id + ) + assert expected_trigger in triggers + + +async def test_get_trigger_capabilities( + hass: HomeAssistant, + device_registry: dr.DeviceRegistry, + knx: KNXTestKit, +) -> None: + """Test we get the expected capabilities telegram device trigger.""" await knx.setup_integration({}) device_entry = device_registry.async_get_device( identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")} @@ -202,5 +314,107 @@ async def test_get_trigger_capabilities_node_status( "sort": False, }, }, - } + }, + { + "name": "group_value_write", + "optional": True, + "default": True, + "type": "boolean", + }, + { + "name": "group_value_response", + "optional": True, + "default": True, + "type": "boolean", + }, + { + "name": "group_value_read", + "optional": True, + "default": True, + "type": "boolean", + }, + { + "name": "incoming", + "optional": True, + "default": True, + "type": "boolean", + }, + { + "name": "outgoing", + "optional": True, + "default": True, + "type": "boolean", + }, ] + + +async def test_invalid_device_trigger( + hass: HomeAssistant, + device_registry: dr.DeviceRegistry, + knx: KNXTestKit, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test invalid telegram device trigger configuration.""" + await knx.setup_integration({}) + device_entry = device_registry.async_get_device( + identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")} + ) + caplog.clear() + with caplog.at_level(logging.ERROR): + assert await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + { + "trigger": { + "platform": "device", + "domain": DOMAIN, + "device_id": device_entry.id, + "type": "telegram", + "invalid": True, + }, + "action": { + "service": "test.automation", + "data_template": { + "catch_all": ("telegram - {{ trigger.destination }}"), + "id": (" {{ trigger.id }}"), + }, + }, + }, + ] + }, + ) + await hass.async_block_till_done() + assert ( + "Unnamed automation failed to setup triggers and has been disabled: " + "extra keys not allowed @ data['invalid']. Got None" + in caplog.records[0].message + ) + + +async def test_invalid_trigger_configuration( + hass: HomeAssistant, + device_registry: dr.DeviceRegistry, + knx: KNXTestKit, +): + """Test invalid telegram device trigger configuration at attach_trigger.""" + await knx.setup_integration({}) + device_entry = device_registry.async_get_device( + identifiers={(DOMAIN, f"_{knx.mock_config_entry.entry_id}_interface")} + ) + # After changing the config in async_attach_trigger, the config is validated again + # against the integration trigger. This test checks if this validation works. + with pytest.raises(InvalidDeviceAutomationConfig): + await device_trigger.async_attach_trigger( + hass, + { + "platform": "device", + "domain": DOMAIN, + "device_id": device_entry.id, + "type": "telegram", + "group_value_write": "invalid", + }, + None, + {}, + ) diff --git a/tests/components/knx/test_trigger.py b/tests/components/knx/test_trigger.py new file mode 100644 index 00000000000..3eab7d58a00 --- /dev/null +++ b/tests/components/knx/test_trigger.py @@ -0,0 +1,290 @@ +"""Tests for KNX integration specific triggers.""" + +import logging + +import pytest + +from homeassistant.components import automation +from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_OFF +from homeassistant.core import HomeAssistant, ServiceCall +from homeassistant.setup import async_setup_component + +from .conftest import KNXTestKit + +from tests.common import async_mock_service + + +@pytest.fixture +def calls(hass: HomeAssistant) -> list[ServiceCall]: + """Track calls to a mock service.""" + return async_mock_service(hass, "test", "automation") + + +async def test_telegram_trigger( + hass: HomeAssistant, + calls: list[ServiceCall], + knx: KNXTestKit, +) -> None: + """Test telegram telegram triggers firing.""" + await knx.setup_integration({}) + + # "id" field added to action to test if `trigger_data` passed correctly in `async_attach_trigger` + assert await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + # "catch_all" trigger + { + "trigger": { + "platform": "knx.telegram", + }, + "action": { + "service": "test.automation", + "data_template": { + "catch_all": ("telegram - {{ trigger.destination }}"), + "id": (" {{ trigger.id }}"), + }, + }, + }, + # "specific" trigger + { + "trigger": { + "platform": "knx.telegram", + "id": "test-id", + "destination": ["1/2/3", 2564], # 2564 -> "1/2/4" in raw format + "group_value_write": True, + "group_value_response": False, + "group_value_read": False, + "incoming": True, + "outgoing": True, + }, + "action": { + "service": "test.automation", + "data_template": { + "specific": ("telegram - {{ trigger.destination }}"), + "id": (" {{ trigger.id }}"), + }, + }, + }, + ] + }, + ) + + # "specific" shall ignore destination address + await knx.receive_write("0/0/1", (0x03, 0x2F)) + assert len(calls) == 1 + test_call = calls.pop() + assert test_call.data["catch_all"] == "telegram - 0/0/1" + assert test_call.data["id"] == 0 + + await knx.receive_write("1/2/4", (0x03, 0x2F)) + assert len(calls) == 2 + test_call = calls.pop() + assert test_call.data["specific"] == "telegram - 1/2/4" + assert test_call.data["id"] == "test-id" + test_call = calls.pop() + assert test_call.data["catch_all"] == "telegram - 1/2/4" + assert test_call.data["id"] == 0 + + # "specific" shall ignore GroupValueRead + await knx.receive_read("1/2/4") + assert len(calls) == 1 + test_call = calls.pop() + assert test_call.data["catch_all"] == "telegram - 1/2/4" + assert test_call.data["id"] == 0 + + +@pytest.mark.parametrize( + "group_value_options", + [ + { + "group_value_write": True, + "group_value_response": True, + "group_value_read": False, + }, + { + "group_value_write": False, + "group_value_response": False, + "group_value_read": True, + }, + { + # "group_value_write": True, # omitted defaults to True + "group_value_response": False, + "group_value_read": False, + }, + ], +) +@pytest.mark.parametrize( + "direction_options", + [ + { + "incoming": True, + "outgoing": True, + }, + { + # "incoming": True, # omitted defaults to True + "outgoing": False, + }, + { + "incoming": False, + "outgoing": True, + }, + ], +) +async def test_telegram_trigger_options( + hass: HomeAssistant, + calls: list[ServiceCall], + knx: KNXTestKit, + group_value_options: dict[str, bool], + direction_options: dict[str, bool], +) -> None: + """Test telegram telegram trigger options.""" + await knx.setup_integration({}) + assert await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + # "catch_all" trigger + { + "trigger": { + "platform": "knx.telegram", + **group_value_options, + **direction_options, + }, + "action": { + "service": "test.automation", + "data_template": { + "catch_all": ("telegram - {{ trigger.destination }}"), + "id": (" {{ trigger.id }}"), + }, + }, + }, + ] + }, + ) + await knx.receive_write("0/0/1", 1) + if group_value_options.get("group_value_write", True) and direction_options.get( + "incoming", True + ): + assert len(calls) == 1 + assert calls.pop().data["catch_all"] == "telegram - 0/0/1" + else: + assert len(calls) == 0 + + await knx.receive_response("0/0/1", 1) + if group_value_options["group_value_response"] and direction_options.get( + "incoming", True + ): + assert len(calls) == 1 + assert calls.pop().data["catch_all"] == "telegram - 0/0/1" + else: + assert len(calls) == 0 + + await knx.receive_read("0/0/1") + if group_value_options["group_value_read"] and direction_options.get( + "incoming", True + ): + assert len(calls) == 1 + assert calls.pop().data["catch_all"] == "telegram - 0/0/1" + else: + assert len(calls) == 0 + + await hass.services.async_call( + "knx", + "send", + {"address": "0/0/1", "payload": True}, + blocking=True, + ) + await knx.assert_write("0/0/1", True) + if ( + group_value_options.get("group_value_write", True) + and direction_options["outgoing"] + ): + assert len(calls) == 1 + assert calls.pop().data["catch_all"] == "telegram - 0/0/1" + else: + assert len(calls) == 0 + + +async def test_remove_telegram_trigger( + hass: HomeAssistant, + calls: list[ServiceCall], + knx: KNXTestKit, +) -> None: + """Test for removed callback when telegram trigger not used.""" + automation_name = "telegram_trigger_automation" + await knx.setup_integration({}) + + assert await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + { + "alias": automation_name, + "trigger": { + "platform": "knx.telegram", + }, + "action": { + "service": "test.automation", + "data_template": { + "catch_all": ("telegram - {{ trigger.destination }}") + }, + }, + } + ] + }, + ) + + await knx.receive_write("0/0/1", (0x03, 0x2F)) + assert len(calls) == 1 + assert calls.pop().data["catch_all"] == "telegram - 0/0/1" + + await hass.services.async_call( + automation.DOMAIN, + SERVICE_TURN_OFF, + {ATTR_ENTITY_ID: f"automation.{automation_name}"}, + blocking=True, + ) + await knx.receive_write("0/0/1", (0x03, 0x2F)) + assert len(calls) == 0 + + +async def test_invalid_trigger( + hass: HomeAssistant, + knx: KNXTestKit, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test invalid telegram trigger configuration.""" + await knx.setup_integration({}) + caplog.clear() + with caplog.at_level(logging.ERROR): + assert await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + { + "trigger": { + "platform": "knx.telegram", + "invalid": True, + }, + "action": { + "service": "test.automation", + "data_template": { + "catch_all": ("telegram - {{ trigger.destination }}"), + "id": (" {{ trigger.id }}"), + }, + }, + }, + ] + }, + ) + await hass.async_block_till_done() + assert ( + "Unnamed automation failed to setup triggers and has been disabled: " + "extra keys not allowed @ data['invalid']. Got None" + in caplog.records[0].message + )