Repository: XKNX/xknx
Pull request #821 (2021-12-16 10:08:41)

Bump mypy from 0.910 to 0.920
Bumps [mypy](https://github.com/python/mypy) from 0.910 to 0.920.
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/python/mypy/commit/b944746a428a9476b7118893e8e7b94ef745bcc4"><code>b944746</code></a> Bump version</li>
<li><a href="https://github.com/python/mypy/commit/ae93837f7ea34221e0689d085bb8f16e4420dd1b"><code>ae93837</code></a> Fix crash when encountering match statement (<a href="https://github-redirect.dependabot.com/python/mypy/issues/11749">#11749</a>)</li>
<li><a href="https://github.com/python/mypy/commit/69dbb47da0af74af25e1b60a50d4ba2b0d8b2683"><code>69dbb47</code></a> Add Python 3.10 to trove classifiers (<a href="https://github-redirect.dependabot.com/python/mypy/issues/11747">#11747</a>)</li>
<li><a href="https://github.com/python/mypy/commit/842044e171b15caad33f41ffb88c5ec41df99947"><code>842044e</code></a> Don't crash on unsupported Python 3.10 match statement (<a href="https://github-redirect.dependabot.com/python/mypy/issues/11738">#11738</a>)</li>
<li><a href="https://github.com/python/mypy/commit/9d1d033ce73048677b2566d43a1370c2e45ccbfb"><code>9d1d033</code></a> Upgrade tomli (<a href="https://github-redirect.dependabot.com/python/mypy/issues/11729">#11729</a>)</li>
<li><a href="https://github.com/python/mypy/commit/fdae42fee9231f8605d1d0fa7d635f1e8748769b"><code>fdae42f</code></a> Fix crash involving explicit reexport, import cycle, wildcard (<a href="https://github-redirect.dependabot.com/python/mypy/issues/11632">#11632</a>)</li>
<li><a href="https://github.com/python/mypy/commit/dd086bc75c8b846ecd8a086dbf2067b92f5d432b"><code>dd086bc</code></a> Fix crash with overload and callable object decorators (<a href="https://github-redirect.dependabot.com/python/mypy/issues/11630">#11630</a>)</li>
<li><a href="https://github.com/python/mypy/commit/524b60e07156fbfb5ba666af47c1f22aac54c660"><code>524b60e</code></a> Disable universal newlines when reading TOML (<a href="https://github-redirect.dependabot.com/python/mypy/issues/10893">#10893</a>)</li>
<li><a href="https://github.com/python/mypy/commit/5a19bfbb2e7b34667606e4c7ba8901c4e94c591d"><code>5a19bfb</code></a> Properly check <code>*CustomType</code> and <code>**CustomType</code> arguments (<a href="https://github-redirect.dependabot.com/python/mypy/issues/11151">#11151</a>)</li>
<li><a href="https://github.com/python/mypy/commit/f08d72b7d620b2ed573fa91941fc93055e852efa"><code>f08d72b</code></a> Add getattr overload variants to help mypy type inference</li>
<li>Additional commits viewable in <a href="https://github.com/python/mypy/compare/v0.910...v0.920">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=mypy&package-manager=pip&previous-version=0.910&new-version=0.920)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>


Patch:
diff --git a/.coveragerc b/.coveragerc
index 519778a6..e75ab96e 100644
--- a/.coveragerc
+++ b/.coveragerc
@@ -14,3 +14,12 @@ exclude_lines =
 
     # Don't complain if tests don't hit defensive assertion code:
     raise NotImplementedError
+
+    # Don't complain about logs
+    logger.debug
+    logger.info
+    logger.warning
+    logger.error
+
+    # Don't complain about missing debug-only code:
+    def __repr__
diff --git a/changelog.md b/changelog.md
index 2c7802de..575b7045 100644
--- a/changelog.md
+++ b/changelog.md
@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased changes
+
+### Internals
+
+- Tunnel: Implement flow control according to KNX spec recommendations: wait for L_DATA.con frame before sending next L_DATA.req with 3 second timeout
+
 ## 0.18.13 Hold your colour 2021-11-13
 
 ### Internals
diff --git a/xknx/io/tunnel.py b/xknx/io/tunnel.py
index e2ff3107..bdc55c1a 100644
--- a/xknx/io/tunnel.py
+++ b/xknx/io/tunnel.py
@@ -34,6 +34,9 @@ TelegramCallbackType = Callable[[Telegram], None]
 
 logger = logging.getLogger("xknx.log")
 
+# See 3/6/3 EMI_IMI §4.1.5 Data Link Layer messages
+REQUEST_TO_CONFIRMATION_TIMEOUT = 3
+
 
 class Tunnel(Interface):
     """Class for handling KNX/IP tunnels."""
@@ -59,6 +62,8 @@ class Tunnel(Interface):
         self.route_back = route_back
         self.telegram_received_callback = telegram_received_callback
 
+        self._tunnelling_request_confirmation_event = asyncio.Event()
+
         self.udp_client: UDPClient
         self.init_udp_client()
 
@@ -118,10 +123,10 @@ class Tunnel(Interface):
             await self.udp_client.stop()
             raise ex
         else:
+            self._tunnel_established()
             await self.xknx.connection_manager.connection_state_changed(
                 XknxConnectionState.CONNECTED
             )
-            self._tunnel_established()
             return True
 
     def _tunnel_established(self) -> None:
@@ -273,7 +278,20 @@ class Tunnel(Interface):
             self.sequence_number,
             self.communication_channel,
         )
-        await tunnelling.start()
+        self._tunnelling_request_confirmation_event.clear()
+        send_and_wait_for_confirmation = asyncio.gather(
+            tunnelling.start(), self._tunnelling_request_confirmation_event.wait()
+        )
+        try:
+            await asyncio.wait_for(
+                send_and_wait_for_confirmation, timeout=REQUEST_TO_CONFIRMATION_TIMEOUT
+            )
+        except asyncio.TimeoutError:
+            # REQUEST_TO_CONFIRMATION_TIMEOUT is longer than tunnelling timeout of 1 second
+            # so exception should always be from self._tunnelling_request_confirmation_event
+            logger.warning(
+                "L_DATA_CON Data Link Layer confirmation timed out for %s", telegram
+            )
         return tunnelling.success
 
     def _increase_sequence_number(self) -> None:
@@ -301,20 +319,29 @@ class Tunnel(Interface):
         self, tunneling_request: TunnellingRequest
     ) -> None:
         """Handle incoming tunnel request."""
+        # we should only ACK if the request matches the expected sequence number or one less
+        # we should not ACK and discard the request if the sequence number is higher than the expected sequence number
+        #   or if the sequence number lower thatn (expected -1)
         self._send_tunnelling_ack(
             tunneling_request.communication_channel_id,
             tunneling_request.sequence_counter,
         )
-        # Don't handle invalid cemi frames (None) and only handle incoming L_DATA_IND frames.
-        # Ignore L_DATA_CON confirmation frames. L_DATA_REQ frames should only be outgoing.
-        if (
-            tunneling_request.cemi is not None
-            and tunneling_request.cemi.code is CEMIMessageCode.L_DATA_IND
-        ):
+        if tunneling_request.cemi is None:
+            # Don't handle invalid cemi frames (None)
+            return
+        if tunneling_request.cemi.code is CEMIMessageCode.L_DATA_IND:
             telegram = tunneling_request.cemi.telegram
             telegram.direction = TelegramDirection.INCOMING
             if self.telegram_received_callback is not None:
                 self.telegram_received_callback(telegram)
+        elif tunneling_request.cemi.code is CEMIMessageCode.L_DATA_CON:
+            # L_DATA_CON confirmation frame signals ready to send next telegram
+            self._tunnelling_request_confirmation_event.set()
+        elif tunneling_request.cemi.code is CEMIMessageCode.L_DATA_REQ:
+            # L_DATA_REQ frames should only be outgoing.
+            logger.warning(
+                "Tunnel received unexpected L_DATA_REQ frame: %s", tunneling_request
+            )
 
     def _send_tunnelling_ack(
         self, communication_channel_id: int, sequence_counter: int
diff --git a/xknx/io/udp_client.py b/xknx/io/udp_client.py
index 1cfecb02..b84a887e 100644
--- a/xknx/io/udp_client.py
+++ b/xknx/io/udp_client.py
@@ -145,7 +145,7 @@ class UDPClient:
     @staticmethod
     def create_multicast_sock(
         own_ip: str, remote_addr: tuple[str, int]
-    ) -> socket.SocketType:
+    ) -> socket.socket:
         """Create UDP multicast socket."""
         sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
diff --git a/xknx/remote_value/remote_value_control.py b/xknx/remote_value/remote_value_control.py
index 7a9e6896..edddedf2 100644
--- a/xknx/remote_value/remote_value_control.py
+++ b/xknx/remote_value/remote_value_control.py
@@ -71,4 +71,4 @@ class RemoteValueControl(RemoteValue[DPTBinary, Any]):
     @property
     def ha_device_class(self) -> str | None:
         """Return a string representing the home assistant device class."""
-        return getattr(self.dpt_class, "ha_device_class", None)  # type: ignore
+        return getattr(self.dpt_class, "ha_device_class", None)
diff --git a/xknx/remote_value/remote_value_sensor.py b/xknx/remote_value/remote_value_sensor.py
index ad91e1a2..b23aa298 100644
--- a/xknx/remote_value/remote_value_sensor.py
+++ b/xknx/remote_value/remote_value_sensor.py
@@ -80,7 +80,7 @@ class _RemoteValueGeneric(RemoteValue[DPTArray, ValueType]):
     @property
     def ha_device_class(self) -> str | None:
         """Return a string representing the home assistant device class."""
-        return getattr(self.dpt_class, "ha_device_class", None)  # type: ignore
+        return getattr(self.dpt_class, "ha_device_class", None)
 
 
 class RemoteValueSensor(_RemoteValueGeneric[Union[int, float, str]]):


Test patch:
diff --git a/requirements/testing.txt b/requirements/testing.txt
index b3c61cf2..d453bb5e 100644
--- a/requirements/testing.txt
+++ b/requirements/testing.txt
@@ -8,6 +8,6 @@ pylint==2.12.2
 pytest==6.2.5
 pytest-asyncio==0.16.0
 pytest-cov==3.0.0
-setuptools==59.6.0
+setuptools==60.0.0
 tox==3.24.4
-mypy==0.910
+mypy==0.920
diff --git a/test/io_tests/tunnel_test.py b/test/io_tests/tunnel_test.py
index d958169b..bce28aeb 100644
--- a/test/io_tests/tunnel_test.py
+++ b/test/io_tests/tunnel_test.py
@@ -1,14 +1,20 @@
 """Unit test for KNX/IP Tunnelling Request/Response."""
+import asyncio
 from unittest.mock import Mock, patch
 
+import pytest
 from xknx import XKNX
+from xknx.dpt import DPTArray
 from xknx.io import Tunnel
-from xknx.knxip import CEMIFrame
-from xknx.telegram import TelegramDirection
+from xknx.knxip import CEMIFrame, KNXIPFrame, TunnellingAck, TunnellingRequest
+from xknx.knxip.knxip_enum import CEMIMessageCode
+from xknx.telegram import Telegram, TelegramDirection
+from xknx.telegram.apci import GroupValueWrite
 
 
-class TestTunnelling:
-    """Test class for xknx/io/Tunnelling objects."""
+@pytest.mark.asyncio
+class TestTunnel:
+    """Test class for xknx/io/Tunnel objects."""
 
     def setup_method(self):
         """Set up test class."""
@@ -65,3 +71,35 @@ class TestTunnelling:
         self.tunnel.udp_client.data_received_callback(raw)
         self.tg_received_mock.assert_not_called()
         send_ack_mock.assert_called_once_with(0x02, 0x4F)
+
+    async def test_tunnel_wait_for_l2_confirmation(self, time_travel):
+        """Test tunnel waits for L_DATA.con before sending another L_DATA.req."""
+        self.tunnel.udp_client.send = Mock()
+        self.tunnel.communication_channel = 1
+
+        test_telegram = Telegram(payload=GroupValueWrite(DPTArray((1,))))
+        test_ack = KNXIPFrame.init_from_body(
+            TunnellingAck(self.xknx, sequence_counter=23)
+        )
+        confirmation = KNXIPFrame.init_from_body(
+            TunnellingRequest(
+                self.xknx,
+                communication_channel_id=1,
+                sequence_counter=23,
+                cemi=CEMIFrame.init_from_telegram(
+                    self.xknx, test_telegram, code=CEMIMessageCode.L_DATA_CON
+                ),
+            )
+        )
+        task = asyncio.create_task(self.tunnel.send_telegram(test_telegram))
+        await time_travel(0)
+        self.tunnel.udp_client.handle_knxipframe(test_ack)
+        await time_travel(0)
+        assert not task.done()
+        assert self.tunnel.udp_client.send.call_count == 1
+        self.tunnel.udp_client.handle_knxipframe(confirmation)
+        await time_travel(0)
+        assert task.done()
+        # one call for the outgoing request and one for the ACK for the confirmation
+        assert self.tunnel.udp_client.send.call_count == 2
+        await task