Repository: XKNX/xknx

Issue:
ExposeSensor loopback
Outgoing Telegrams are currently only processed in devices when they are of type GroupValueWrite.
https://github.com/XKNX/xknx/blob/aa093838dea3db033a29137b5be59131d11eb3bb/xknx/core/telegram_queue.py#L166

This prevents xknx devices to read each other over the KNX bus. 
Eg. a temperature is sent to the bus via ExposeSensor from HA - this GA can not be read from a Climate or Weather device running on the same xknx instance. So the StateUpdater would fail to read even though it was sent to the bus properly. Neither would the GroupValueRead reach the ExposeSensor, nor would the GroupValueResponse reach the Climate/Weather device.

We could process every Telegram from outgoing. Is there any chance this introduces infinite loops? 🤔
There is no mechanism currently to see if the telegram comes from the receiving xknx device itself, so a device answering to a GroupValueRead would process its own GroupValueResponse...

Discussion:
coveralls: 
[![Coverage Status](https://coveralls.io/builds/38576156/badge)](https://coveralls.io/builds/38576156)

Coverage decreased (-0.002%) to 94.404% when pulling **875f011d8645ac48262611adf986746c5674a7f7 on farmio:expose-loopback** into **8784f51dba1a1fa9d1eacac9a5a678d9d067a9f6 on XKNX:main**.


Patch:
diff --git a/changelog.md b/changelog.md
index c6b058f7..d8ed8803 100644
--- a/changelog.md
+++ b/changelog.md
@@ -2,20 +2,24 @@
 
 ## Unreleased changes
 
+## Devices
+
 - Add support for cover lock
+- ExposeSensor values can now be read from other xknx devices that share a group address
 - Add more information to sensors and binary sensors in the HA integration
-- Store last Telegram and decoded value in RemoteValue
+
 
 ### Breaking Changes
 
 - Remove configuration handling from core library (use https://xknx.io/config-converter)
-- Drop support for python 3.7
 
 ### Internals
 
+- Drop support for python 3.7
 - use pytest tests instead of unittest TestCase
 - Move RequestResponse and subclasses to xknx.io.request_response.*
 - Move ConnectionConfig to xknx.io.connection
+- Store last Telegram and decoded value in RemoteValue
 
 ## 0.17.5 Add support for unique ids 2021-03-30
 
diff --git a/xknx/core/telegram_queue.py b/xknx/core/telegram_queue.py
index eba105ff..6021a016 100644
--- a/xknx/core/telegram_queue.py
+++ b/xknx/core/telegram_queue.py
@@ -15,7 +15,6 @@ from typing import TYPE_CHECKING, Awaitable, Callable
 
 from xknx.exceptions import CommunicationError, XKNXException
 from xknx.telegram import AddressFilter, GroupAddress, Telegram, TelegramDirection
-from xknx.telegram.apci import GroupValueWrite
 
 if TYPE_CHECKING:
     from xknx.xknx import XKNX
@@ -175,8 +174,7 @@ class TelegramQueue:
         telegram_logger.debug(telegram)
         if self.xknx.knxip_interface is not None:
             await self.xknx.knxip_interface.send_telegram(telegram)
-            if isinstance(telegram.payload, GroupValueWrite):
-                await self.xknx.devices.process(telegram)
+            await self.xknx.devices.process(telegram)
 
             for telegram_received_cb in self.telegram_received_cbs:
                 if telegram_received_cb.is_within_filter(telegram):
diff --git a/xknx/core/value_reader.py b/xknx/core/value_reader.py
index 2a48352e..2369f3d8 100644
--- a/xknx/core/value_reader.py
+++ b/xknx/core/value_reader.py
@@ -41,6 +41,7 @@ class ValueReader:
         cb_obj = self.xknx.telegram_queue.register_telegram_received_cb(
             self.telegram_received,
             group_addresses=[self.group_address],
+            match_for_outgoing=True,
         )
         await self.send_group_read()
 


Test patch:
diff --git a/test/devices_tests/sensor_expose_loop_test.py b/test/devices_tests/sensor_expose_loop_test.py
index c94ea7d9..a30076d5 100644
--- a/test/devices_tests/sensor_expose_loop_test.py
+++ b/test/devices_tests/sensor_expose_loop_test.py
@@ -1,10 +1,12 @@
 """Unit test for Sensor and ExposeSensor objects."""
+from unittest.mock import AsyncMock, call
+
 import pytest
 from xknx import XKNX
 from xknx.devices import BinarySensor, ExposeSensor, Sensor
 from xknx.dpt import DPTArray, DPTBinary
 from xknx.telegram import GroupAddress, Telegram, TelegramDirection
-from xknx.telegram.apci import GroupValueWrite
+from xknx.telegram.apci import GroupValueRead, GroupValueResponse, GroupValueWrite
 
 
 @pytest.mark.asyncio
@@ -178,45 +180,69 @@ class TestSensorExposeLoop:
     @pytest.mark.parametrize("value_type,test_payload,test_value", test_cases)
     async def test_array_sensor_loop(self, value_type, test_payload, test_value):
         """Test sensor and expose_sensor with different values."""
-
         xknx = XKNX()
-        sensor = Sensor(
-            xknx,
-            "TestSensor",
-            group_address_state="1/1/1",
-            value_type=value_type,
-        )
+        xknx.knxip_interface = AsyncMock()
+        xknx.rate_limit = False
+        await xknx.telegram_queue.start()
+
         expose = ExposeSensor(
             xknx,
             "TestExpose",
-            group_address="2/2/2",
+            group_address="1/1/1",
             value_type=value_type,
         )
+        assert expose.resolve_state() is None
+        # set a value from expose - HA sends strings for new values
+        stringified_value = str(test_value)
+        await expose.set(stringified_value)
 
-        incoming_telegram = Telegram(
+        outgoing_telegram = Telegram(
             destination_address=GroupAddress("1/1/1"),
-            direction=TelegramDirection.INCOMING,
+            direction=TelegramDirection.OUTGOING,
             payload=GroupValueWrite(test_payload),
         )
-        await sensor.process(incoming_telegram)
-
-        incoming_value = sensor.resolve_state()
+        await xknx.telegrams.join()
+        xknx.knxip_interface.send_telegram.assert_called_with(outgoing_telegram)
         if isinstance(test_value, float):
-            assert round(incoming_value, 4) == test_value
+            assert round(expose.resolve_state(), 4) == test_value
         else:
-            assert incoming_value == test_value
+            assert expose.resolve_state() == test_value
 
-        # HA sends strings for new values
-        stringified_value = str(test_value)
-        await expose.set(stringified_value)
+        # init sensor after expose is set - with same group address
+        sensor = Sensor(
+            xknx,
+            "TestSensor",
+            group_address_state="1/1/1",
+            value_type=value_type,
+        )
+        assert sensor.resolve_state() is None
 
-        assert xknx.telegrams.qsize() == 1
-        outgoing_telegram = xknx.telegrams.get_nowait()
-        assert outgoing_telegram == Telegram(
-            destination_address=GroupAddress("2/2/2"),
+        # read sensor state (from expose as it has the same GA)
+        # wait_for_result so we don't have to await self.xknx.telegrams.join()
+        await sensor.sync(wait_for_result=True)
+        read_telegram = Telegram(
+            destination_address=GroupAddress("1/1/1"),
             direction=TelegramDirection.OUTGOING,
-            payload=GroupValueWrite(test_payload),
+            payload=GroupValueRead(),
+        )
+        response_telegram = Telegram(
+            destination_address=GroupAddress("1/1/1"),
+            direction=TelegramDirection.OUTGOING,
+            payload=GroupValueResponse(test_payload),
+        )
+        xknx.knxip_interface.send_telegram.assert_has_calls(
+            [
+                call(read_telegram),
+                call(response_telegram),
+            ]
         )
+        # test if Sensor has successfully read from ExposeSensor
+        if isinstance(test_value, float):
+            assert round(sensor.resolve_state(), 4) == test_value
+        else:
+            assert sensor.resolve_state() == test_value
+        assert expose.resolve_state() == sensor.resolve_state()
+        await xknx.telegram_queue.stop()
 
 
 @pytest.mark.asyncio
@@ -232,31 +258,52 @@ class TestBinarySensorExposeLoop:
     )
     async def test_binary_sensor_loop(self, value_type, test_payload, test_value):
         """Test binary_sensor and expose_sensor with binary values."""
-
         xknx = XKNX()
-        sensor = BinarySensor(xknx, "TestSensor", group_address_state="1/1/1")
+        xknx.knxip_interface = AsyncMock()
+        xknx.rate_limit = False
+        await xknx.telegram_queue.start()
+
         expose = ExposeSensor(
             xknx,
             "TestExpose",
-            group_address="2/2/2",
+            group_address="1/1/1",
             value_type=value_type,
         )
+        assert expose.resolve_state() is None
 
-        incoming_telegram = Telegram(
+        await expose.set(test_value)
+        await xknx.telegrams.join()
+        outgoing_telegram = Telegram(
             destination_address=GroupAddress("1/1/1"),
-            direction=TelegramDirection.INCOMING,
+            direction=TelegramDirection.OUTGOING,
             payload=GroupValueWrite(test_payload),
         )
-        await sensor.process(incoming_telegram)
+        xknx.knxip_interface.send_telegram.assert_called_with(outgoing_telegram)
+        assert expose.resolve_state() == test_value
 
-        incoming_value = sensor.is_on()
-        assert incoming_value == test_value
+        bin_sensor = BinarySensor(xknx, "TestSensor", group_address_state="1/1/1")
+        assert bin_sensor.state is None
 
-        await expose.set(test_value)
-        assert xknx.telegrams.qsize() == 1
-        outgoing_telegram = xknx.telegrams.get_nowait()
-        assert outgoing_telegram == Telegram(
-            destination_address=GroupAddress("2/2/2"),
+        # read sensor state (from expose as it has the same GA)
+        # wait_for_result so we don't have to await self.xknx.telegrams.join()
+        await bin_sensor.sync(wait_for_result=True)
+        read_telegram = Telegram(
+            destination_address=GroupAddress("1/1/1"),
             direction=TelegramDirection.OUTGOING,
-            payload=GroupValueWrite(test_payload),
+            payload=GroupValueRead(),
+        )
+        response_telegram = Telegram(
+            destination_address=GroupAddress("1/1/1"),
+            direction=TelegramDirection.OUTGOING,
+            payload=GroupValueResponse(test_payload),
+        )
+        xknx.knxip_interface.send_telegram.assert_has_calls(
+            [
+                call(read_telegram),
+                call(response_telegram),
+            ]
         )
+        # test if Sensor has successfully read from ExposeSensor
+        assert bin_sensor.state == test_value
+        assert expose.resolve_state() == bin_sensor.state
+        await xknx.telegram_queue.stop()