Xknx unneeded expose (#48311)

homeassistant/components/knx/expose.py

@@ -114,12 +114,15 @@ class KNXExposeSensor:
         if new_state.state in (STATE_UNKNOWN, STATE_UNAVAILABLE):
             return
 
+        old_state = event.data.get("old_state")
+
         if self.expose_attribute is None:
-            await self._async_set_knx_value(new_state.state)
+            if old_state is None or old_state.state != new_state.state:
+                # don't send same value sequentially
+                await self._async_set_knx_value(new_state.state)
             return
 
         new_attribute = new_state.attributes.get(self.expose_attribute)
-        old_state = event.data.get("old_state")
 
         if old_state is not None:
             old_attribute = old_state.attributes.get(self.expose_attribute)

requirements_test_all.txt

@@ -1201,6 +1201,9 @@ wolf_smartset==0.1.8
 # homeassistant.components.xbox
 xbox-webapi==2.0.8
 
+# homeassistant.components.knx
+xknx==0.17.4
+
 # homeassistant.components.bluesound
 # homeassistant.components.rest
 # homeassistant.components.startca

tests/components/knx/__init__.py

@@ -0,0 +1 @@
+"""The tests for KNX integration."""

tests/components/knx/test_expose.py

@@ -0,0 +1,67 @@
+"""Test knx expose."""
+from unittest.mock import AsyncMock, Mock
+
+from homeassistant.components.knx.expose import KNXExposeSensor
+
+
+async def test_binary_expose(hass):
+    """Test that a binary expose sends only telegrams on state change."""
+    e_id = "fake.entity"
+    xknxMock = Mock()
+    xknxMock.telegrams = AsyncMock()
+    KNXExposeSensor(hass, xknxMock, "binary", e_id, None, "0", "1/1/8")
+    assert xknxMock.devices.add.call_count == 1, "Expected one device add"
+
+    # Change state to on
+    xknxMock.reset_mock()
+    hass.states.async_set(e_id, "on", {})
+    await hass.async_block_till_done()
+    assert xknxMock.telegrams.put.call_count == 1, "Expected telegram for state change"
+
+    # Change attribute; keep state
+    xknxMock.reset_mock()
+    hass.states.async_set(e_id, "on", {"brightness": 180})
+    await hass.async_block_till_done()
+    assert (
+        xknxMock.telegrams.put.call_count == 0
+    ), "Expected no telegram; state not changed"
+
+    # Change attribute and state
+    xknxMock.reset_mock()
+    hass.states.async_set(e_id, "off", {"brightness": 0})
+    await hass.async_block_till_done()
+    assert xknxMock.telegrams.put.call_count == 1, "Expected telegram for state change"
+
+
+async def test_expose_attribute(hass):
+    """Test that an expose sends only telegrams on attribute change."""
+    e_id = "fake.entity"
+    a_id = "fakeAttribute"
+    xknxMock = Mock()
+    xknxMock.telegrams = AsyncMock()
+    KNXExposeSensor(hass, xknxMock, "percentU8", e_id, a_id, None, "1/1/8")
+    assert xknxMock.devices.add.call_count == 1, "Expected one device add"
+
+    # Change state to on; no attribute
+    xknxMock.reset_mock()
+    hass.states.async_set(e_id, "on", {})
+    await hass.async_block_till_done()
+    assert xknxMock.telegrams.put.call_count == 0
+
+    # Change attribute; keep state
+    xknxMock.reset_mock()
+    hass.states.async_set(e_id, "on", {a_id: 1})
+    await hass.async_block_till_done()
+    assert xknxMock.telegrams.put.call_count == 1
+
+    # Change state keep attribute
+    xknxMock.reset_mock()
+    hass.states.async_set(e_id, "off", {a_id: 1})
+    await hass.async_block_till_done()
+    assert xknxMock.telegrams.put.call_count == 0
+
+    # Change state and attribute
+    xknxMock.reset_mock()
+    hass.states.async_set(e_id, "on", {a_id: 0})
+    await hass.async_block_till_done()
+    assert xknxMock.telegrams.put.call_count == 1