From 745314b11596322b2ff1dd639852e6e75918e9ed Mon Sep 17 00:00:00 2001 From: Matthias Alphart Date: Sat, 24 Jul 2021 14:03:04 +0200 Subject: [PATCH] Test KNX services (#53367) --- tests/components/knx/test_expose.py | 9 +- tests/components/knx/test_services.py | 166 ++++++++++++++++++++++++++ tests/components/knx/test_switch.py | 8 +- 3 files changed, 177 insertions(+), 6 deletions(-) create mode 100644 tests/components/knx/test_services.py diff --git a/tests/components/knx/test_expose.py b/tests/components/knx/test_expose.py index 6922b6ed926..25ec0f92604 100644 --- a/tests/components/knx/test_expose.py +++ b/tests/components/knx/test_expose.py @@ -2,9 +2,12 @@ from homeassistant.components.knx import CONF_KNX_EXPOSE, KNX_ADDRESS from homeassistant.components.knx.schema import ExposeSchema from homeassistant.const import CONF_ATTRIBUTE, CONF_ENTITY_ID, CONF_TYPE +from homeassistant.core import HomeAssistant + +from .conftest import KNXTestKit -async def test_binary_expose(hass, knx): +async def test_binary_expose(hass: HomeAssistant, knx: KNXTestKit): """Test a binary expose to only send telegrams on state change.""" entity_id = "fake.entity" await knx.setup_integration( @@ -31,7 +34,7 @@ async def test_binary_expose(hass, knx): await knx.assert_write("1/1/8", False) -async def test_expose_attribute(hass, knx): +async def test_expose_attribute(hass: HomeAssistant, knx: KNXTestKit): """Test an expose to only send telegrams on attribute change.""" entity_id = "fake.entity" attribute = "fake_attribute" @@ -76,7 +79,7 @@ async def test_expose_attribute(hass, knx): await knx.assert_telegram_count(0) -async def test_expose_attribute_with_default(hass, knx): +async def test_expose_attribute_with_default(hass: HomeAssistant, knx: KNXTestKit): """Test an expose to only send telegrams on attribute change.""" entity_id = "fake.entity" attribute = "fake_attribute" diff --git a/tests/components/knx/test_services.py b/tests/components/knx/test_services.py new file mode 100644 index 00000000000..fe13a289a78 --- /dev/null +++ b/tests/components/knx/test_services.py @@ -0,0 +1,166 @@ +"""Test KNX services.""" +from homeassistant.const import STATE_OFF, STATE_ON +from homeassistant.core import HomeAssistant + +from .conftest import KNXTestKit + + +async def test_send(hass: HomeAssistant, knx: KNXTestKit): + """Test `knx.send` service.""" + test_address = "1/2/3" + await knx.setup_integration({}) + + # send DPT 1 telegram + await hass.services.async_call( + "knx", "send", {"address": test_address, "payload": True}, blocking=True + ) + await knx.assert_write(test_address, True) + + # send raw DPT 5 telegram + await hass.services.async_call( + "knx", "send", {"address": test_address, "payload": [99]}, blocking=True + ) + await knx.assert_write(test_address, (99,)) + + # send "percent" DPT 5 telegram + await hass.services.async_call( + "knx", + "send", + {"address": test_address, "payload": 99, "type": "percent"}, + blocking=True, + ) + await knx.assert_write(test_address, (0xFC,)) + + # send "temperature" DPT 9 telegram + await hass.services.async_call( + "knx", + "send", + {"address": test_address, "payload": 21.0, "type": "temperature"}, + blocking=True, + ) + await knx.assert_write(test_address, (0x0C, 0x1A)) + + # send multiple telegrams + await hass.services.async_call( + "knx", + "send", + {"address": [test_address, "2/2/2", "3/3/3"], "payload": 99, "type": "percent"}, + blocking=True, + ) + await knx.assert_write(test_address, (0xFC,)) + await knx.assert_write("2/2/2", (0xFC,)) + await knx.assert_write("3/3/3", (0xFC,)) + + +async def test_read(hass: HomeAssistant, knx: KNXTestKit): + """Test `knx.read` service.""" + await knx.setup_integration({}) + + # send read telegram + await hass.services.async_call("knx", "read", {"address": "1/1/1"}, blocking=True) + await knx.assert_read("1/1/1") + + # send multiple read telegrams + await hass.services.async_call( + "knx", + "read", + {"address": ["1/1/1", "2/2/2", "3/3/3"]}, + blocking=True, + ) + await knx.assert_read("1/1/1") + await knx.assert_read("2/2/2") + await knx.assert_read("3/3/3") + + +async def test_event_register(hass: HomeAssistant, knx: KNXTestKit): + """Test `knx.event_register` service.""" + events = [] + 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) + assert len(events) == 0 + + # register event + await hass.services.async_call( + "knx", "event_register", {"address": test_address}, blocking=True + ) + await knx.receive_write(test_address, True) + await knx.receive_write(test_address, False) + assert len(events) == 2 + + # remove event registration + events = [] + await hass.services.async_call( + "knx", + "event_register", + {"address": test_address, "remove": True}, + blocking=True, + ) + await knx.receive_write(test_address, True) + assert len(events) == 0 + + +async def test_exposure_register(hass: HomeAssistant, knx: KNXTestKit): + """Test `knx.exposure_register` service.""" + test_address = "1/2/3" + test_entity = "fake.entity" + test_attribute = "fake_attribute" + + await knx.setup_integration({}) + + # no exposure registered + hass.states.async_set(test_entity, STATE_ON, {}) + await knx.assert_no_telegram() + + # register exposure + await hass.services.async_call( + "knx", + "exposure_register", + {"address": test_address, "entity_id": test_entity, "type": "binary"}, + blocking=True, + ) + hass.states.async_set(test_entity, STATE_OFF, {}) + await knx.assert_write(test_address, False) + + # register exposure + await hass.services.async_call( + "knx", + "exposure_register", + {"address": test_address, "remove": True}, + blocking=True, + ) + hass.states.async_set(test_entity, STATE_ON, {}) + await knx.assert_no_telegram() + + # register exposure for attribute with default + await hass.services.async_call( + "knx", + "exposure_register", + { + "address": test_address, + "entity_id": test_entity, + "attribute": test_attribute, + "type": "percentU8", + "default": 0, + }, + blocking=True, + ) + # no attribute on first change wouldn't work because no attribute change since last test + hass.states.async_set(test_entity, STATE_ON, {test_attribute: 30}) + await knx.assert_write(test_address, (30,)) + hass.states.async_set(test_entity, STATE_OFF, {}) + await knx.assert_write(test_address, (0,)) + # don't send same value sequentially + hass.states.async_set(test_entity, STATE_ON, {test_attribute: 25}) + hass.states.async_set(test_entity, STATE_ON, {test_attribute: 25}) + hass.states.async_set(test_entity, STATE_ON, {test_attribute: 25, "unrelated": 2}) + hass.states.async_set(test_entity, STATE_OFF, {test_attribute: 25}) + await knx.assert_telegram_count(1) + await knx.assert_write(test_address, (25,)) diff --git a/tests/components/knx/test_switch.py b/tests/components/knx/test_switch.py index 407d6d83267..eff34243ca8 100644 --- a/tests/components/knx/test_switch.py +++ b/tests/components/knx/test_switch.py @@ -8,10 +8,12 @@ from homeassistant.components.knx.const import ( ) from homeassistant.components.knx.schema import SwitchSchema from homeassistant.const import CONF_NAME, STATE_OFF, STATE_ON -from homeassistant.core import State +from homeassistant.core import HomeAssistant, State + +from .conftest import KNXTestKit -async def test_switch_simple(hass, knx): +async def test_switch_simple(hass: HomeAssistant, knx: KNXTestKit): """Test simple KNX switch.""" await knx.setup_integration( { @@ -50,7 +52,7 @@ async def test_switch_simple(hass, knx): await knx.assert_telegram_count(0) -async def test_switch_state(hass, knx): +async def test_switch_state(hass: HomeAssistant, knx: KNXTestKit): """Test KNX switch with state_address.""" _ADDRESS = "1/1/1" _STATE_ADDRESS = "2/2/2"