"""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
async def test_telegram_trigger(
hass: HomeAssistant,
service_calls: list[ServiceCall],
knx: KNXTestKit,
) -> None:
"""Test 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(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
@pytest.mark.parametrize(
("payload", "type_option", "expected_value", "expected_unit"),
[
((0x4C,), {"type": "percent"}, 30, "%"),
((0x03,), {}, None, None), # "dpt" omitted defaults to None
((0x0C, 0x1A), {"type": "temperature"}, 21.00, "°C"),
],
)
async def test_telegram_trigger_dpt_option(
hass: HomeAssistant,
service_calls: list[ServiceCall],
knx: KNXTestKit,
payload: tuple[int, ...],
type_option: dict[str, bool],
expected_value: int | None,
expected_unit: str | None,
) -> None:
"""Test telegram trigger type option."""
await knx.setup_integration({})
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
# "catch_all" trigger
{
"trigger": {
"platform": "knx.telegram",
**type_option,
},
"action": {
"service": "test.automation",
"data_template": {
"catch_all": ("telegram - {{ trigger.destination }}"),
"trigger": (" {{ trigger }}"),
},
},
},
]
},
)
await knx.receive_write("0/0/1", payload)
assert len(service_calls) == 1
test_call = service_calls.pop()
assert test_call.data["catch_all"] == "telegram - 0/0/1"
assert test_call.data["trigger"]["value"] == expected_value
assert test_call.data["trigger"]["unit"] == expected_unit
await knx.receive_read("0/0/1")
assert len(service_calls) == 1
test_call = service_calls.pop()
assert test_call.data["catch_all"] == "telegram - 0/0/1"
assert test_call.data["trigger"]["value"] is None
assert test_call.data["trigger"]["unit"] is None
@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,
service_calls: list[ServiceCall],
knx: KNXTestKit,
group_value_options: dict[str, bool],
direction_options: dict[str, bool],
) -> None:
"""Test 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 }}"),
},
},
},
]
},
)
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(service_calls) == 1
assert service_calls.pop().data["catch_all"] == "telegram - 0/0/1"
else:
assert len(service_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(service_calls) == 1
assert service_calls.pop().data["catch_all"] == "telegram - 0/0/1"
else:
assert len(service_calls) == 0
await knx.receive_read("0/0/1")
if group_value_options["group_value_read"] and direction_options.get(
"incoming", True
):
assert len(service_calls) == 1
assert service_calls.pop().data["catch_all"] == "telegram - 0/0/1"
else:
assert len(service_calls) == 0
await hass.services.async_call(
"knx",
"send",
{"address": "0/0/1", "payload": True},
blocking=True,
)
assert len(service_calls) == 1
await knx.assert_write("0/0/1", True)
if (
group_value_options.get("group_value_write", True)
and direction_options["outgoing"]
):
assert len(service_calls) == 2
assert service_calls.pop().data["catch_all"] == "telegram - 0/0/1"
else:
assert len(service_calls) == 1
async def test_remove_telegram_trigger(
hass: HomeAssistant,
service_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(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_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 }}"),
},
},
},
]
},
)
assert (
"Unnamed automation failed to setup triggers and has been disabled: "
"extra keys not allowed @ data['invalid']. Got None"
in caplog.records[0].message
)