"""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
from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.setup import async_setup_component
from .conftest import KNXTestKit
from tests.common import async_get_device_automations
async def test_if_fires_on_telegram(
hass: HomeAssistant,
service_calls: list[ServiceCall],
device_registry: dr.DeviceRegistry,
knx: KNXTestKit,
) -> None:
"""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")}
)
# "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": "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(service_calls) == 1
test_call = service_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(service_calls) == 2
test_call = service_calls.pop()
assert test_call.data["specific"] == "telegram - 1/2/4"
assert test_call.data["id"] == "test-id"
test_call = service_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(service_calls) == 1
test_call = service_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,
service_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",
"domain": DOMAIN,
"device_id": device_entry.id,
"type": "telegram",
},
"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,
"type": "telegram",
"destination": ["1/2/3", "1/2/4"],
"id": "test-id",
},
"action": {
"service": "test.automation",
"data_template": {
"specific": ("telegram - {{ trigger.destination }}"),
"id": (" {{ trigger.id }}"),
},
},
},
]
},
)
await knx.receive_write("0/0/1", (0x03, 0x2F))
assert len(service_calls) == 1
test_call = service_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(service_calls) == 2
test_call = service_calls.pop()
assert test_call.data["specific"] == "telegram - 1/2/4"
assert test_call.data["id"] == "test-id"
test_call = service_calls.pop()
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(service_calls) == 2
test_call = service_calls.pop()
assert test_call.data["specific"] == "telegram - 1/2/4"
assert test_call.data["id"] == "test-id"
test_call = service_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,
service_calls: list[ServiceCall],
device_registry: dr.DeviceRegistry,
knx: KNXTestKit,
) -> None:
"""Test for removed callback when device trigger not used."""
automation_name = "telegram_trigger_automation"
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: [
{
"alias": automation_name,
"trigger": {
"platform": "device",
"domain": DOMAIN,
"device_id": device_entry.id,
"type": "telegram",
},
"action": {
"service": "test.automation",
"data_template": {
"catch_all": ("telegram - {{ trigger.destination }}")
},
},
}
]
},
)
await knx.receive_write("0/0/1", (0x03, 0x2F))
assert len(service_calls) == 1
assert service_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,
)
assert len(service_calls) == 1
await knx.receive_write("0/0/1", (0x03, 0x2F))
assert len(service_calls) == 1
async def test_get_triggers(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
knx: KNXTestKit,
) -> None:
"""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")}
)
capabilities = await device_trigger.async_get_trigger_capabilities(
hass,
{
"platform": "device",
"domain": DOMAIN,
"device_id": device_entry.id,
"type": "telegram",
},
)
assert capabilities and "extra_fields" in capabilities
assert voluptuous_serialize.convert(
capabilities["extra_fields"], custom_serializer=cv.custom_serializer
) == [
{
"name": "destination",
"optional": True,
"selector": {
"select": {
"custom_value": True,
"mode": "dropdown",
"multiple": True,
"options": [],
"sort": False,
},
},
},
{
"name": "group_value_write",
"optional": True,
"default": True,
"selector": {
"boolean": {},
},
},
{
"name": "group_value_response",
"optional": True,
"default": True,
"selector": {
"boolean": {},
},
},
{
"name": "group_value_read",
"optional": True,
"default": True,
"selector": {
"boolean": {},
},
},
{
"name": "incoming",
"optional": True,
"default": True,
"selector": {
"boolean": {},
},
},
{
"name": "outgoing",
"optional": True,
"default": True,
"selector": {
"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 }}"),
},
},
},
]
},
)
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,
) -> None:
"""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,
{},
)