Test KNX services (#53367)

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"

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

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"