From 4e4bb2e5a89431a67285b98b99076e07a5836578 Mon Sep 17 00:00:00 2001 From: Matthias Alphart Date: Wed, 28 Jul 2021 19:04:11 +0200 Subject: [PATCH] Test KNX events (#53433) * test knx_event * use async_capture_events --- tests/components/knx/conftest.py | 4 +- tests/components/knx/test_events.py | 83 +++++++++++++++++++++++++++ tests/components/knx/test_services.py | 16 +++--- 3 files changed, 94 insertions(+), 9 deletions(-) create mode 100644 tests/components/knx/test_events.py diff --git a/tests/components/knx/conftest.py b/tests/components/knx/conftest.py index 26f8f1eabac..0dc8749830e 100644 --- a/tests/components/knx/conftest.py +++ b/tests/components/knx/conftest.py @@ -19,6 +19,8 @@ from homeassistant.setup import async_setup_component class KNXTestKit: """Test helper for the KNX integration.""" + INDIVIDUAL_ADDRESS = "1.2.3" + def __init__(self, hass: HomeAssistant): """Init KNX test helper class.""" self.hass: HomeAssistant = hass @@ -148,7 +150,7 @@ class KNXTestKit: destination_address=GroupAddress(group_address), direction=TelegramDirection.INCOMING, payload=payload, - source_address=IndividualAddress("1.2.3"), + source_address=IndividualAddress(self.INDIVIDUAL_ADDRESS), ) ) await self.hass.async_block_till_done() diff --git a/tests/components/knx/test_events.py b/tests/components/knx/test_events.py new file mode 100644 index 00000000000..6a9e021ff53 --- /dev/null +++ b/tests/components/knx/test_events.py @@ -0,0 +1,83 @@ +"""Test KNX events.""" + +from homeassistant.components.knx import CONF_KNX_EVENT_FILTER +from homeassistant.core import HomeAssistant + +from .conftest import KNXTestKit + +from tests.common import async_capture_events + + +async def test_knx_event(hass: HomeAssistant, knx: KNXTestKit): + """Test `knx_event` event.""" + test_group_a = "0/4/*" + test_address_a_1 = "0/4/0" + test_address_a_2 = "0/4/100" + test_group_b = "1/3-6/*" + test_address_b_1 = "1/3/0" + test_address_b_2 = "1/6/200" + test_group_c = "2/6/4,5" + test_address_c_1 = "2/6/4" + test_address_c_2 = "2/6/5" + test_address_d = "5/4/3" + events = async_capture_events(hass, "knx_event") + + async def test_event_data(address, payload): + await hass.async_block_till_done() + assert len(events) == 1 + event = events.pop() + assert event.data["data"] == payload + assert event.data["direction"] == "Incoming" + assert event.data["destination"] == address + if payload is None: + assert event.data["telegramtype"] == "GroupValueRead" + else: + assert event.data["telegramtype"] in ( + "GroupValueWrite", + "GroupValueResponse", + ) + assert event.data["source"] == KNXTestKit.INDIVIDUAL_ADDRESS + + await knx.setup_integration( + { + CONF_KNX_EVENT_FILTER: [ + test_group_a, + test_group_b, + test_group_c, + test_address_d, + ] + } + ) + + # no event received + await hass.async_block_till_done() + assert len(events) == 0 + + # receive telegrams for group addresses matching the filter + await knx.receive_write(test_address_a_1, True) + await test_event_data(test_address_a_1, True) + + await knx.receive_response(test_address_a_2, False) + await test_event_data(test_address_a_2, False) + + await knx.receive_write(test_address_b_1, (1,)) + await test_event_data(test_address_b_1, (1,)) + + await knx.receive_response(test_address_b_2, (255,)) + await test_event_data(test_address_b_2, (255,)) + + await knx.receive_write(test_address_c_1, (89, 43, 34, 11)) + await test_event_data(test_address_c_1, (89, 43, 34, 11)) + + await knx.receive_response(test_address_c_2, (255, 255, 255, 255)) + await test_event_data(test_address_c_2, (255, 255, 255, 255)) + + await knx.receive_read(test_address_d) + await test_event_data(test_address_d, None) + + # receive telegrams for group addresses not matching the filter + await knx.receive_write("0/5/0", True) + await knx.receive_write("1/7/0", True) + await knx.receive_write("2/6/6", True) + await hass.async_block_till_done() + assert len(events) == 0 diff --git a/tests/components/knx/test_services.py b/tests/components/knx/test_services.py index fe13a289a78..80ed51e6aec 100644 --- a/tests/components/knx/test_services.py +++ b/tests/components/knx/test_services.py @@ -4,6 +4,8 @@ from homeassistant.core import HomeAssistant from .conftest import KNXTestKit +from tests.common import async_capture_events + async def test_send(hass: HomeAssistant, knx: KNXTestKit): """Test `knx.send` service.""" @@ -74,17 +76,14 @@ async def test_read(hass: HomeAssistant, knx: KNXTestKit): async def test_event_register(hass: HomeAssistant, knx: KNXTestKit): """Test `knx.event_register` service.""" - events = [] + events = async_capture_events(hass, "knx_event") test_address = "1/2/3" - def listener(event): - events.append(event) - await knx.setup_integration({}) - hass.bus.async_listen("knx_event", listener) # no event registered await knx.receive_write(test_address, True) + await hass.async_block_till_done() assert len(events) == 0 # register event @@ -93,10 +92,10 @@ async def test_event_register(hass: HomeAssistant, knx: KNXTestKit): ) await knx.receive_write(test_address, True) await knx.receive_write(test_address, False) + await hass.async_block_till_done() assert len(events) == 2 - # remove event registration - events = [] + # remove event registration - no event added await hass.services.async_call( "knx", "event_register", @@ -104,7 +103,8 @@ async def test_event_register(hass: HomeAssistant, knx: KNXTestKit): blocking=True, ) await knx.receive_write(test_address, True) - assert len(events) == 0 + await hass.async_block_till_done() + assert len(events) == 2 async def test_exposure_register(hass: HomeAssistant, knx: KNXTestKit):