jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 03:07:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add knx.telegram integration specific trigger; update KNX Interface device trigger (#107592)

Browse Source 
* 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
This commit is contained in:
Matthias Alphart 2024-05-14 14:45:49 +02:00 committed by GitHub
parent ba48da7678
commit bca277a027
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 660 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
homeassistant/components/knx/device_trigger.py
View File

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

101
homeassistant/components/knx/trigger.py Normal file
View File

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

268
tests/components/knx/test_device_trigger.py
View File

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

290
tests/components/knx/test_trigger.py Normal file
View File

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