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