mirror of https://github.com/home-assistant/core
426 lines
14 KiB
Python
426 lines
14 KiB
Python
"""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,
|
|
{},
|
|
)
|