Repository: XKNX/xknx
Pull request #1018 (2022-08-22 22:13:51)

Correctly retry sending an unACKed TunnellingRequest for UDP connections
<!--
  You are awesome! Thanks for contributing to our project!
  Please, DO NOT DELETE ANY TEXT from this template!.
-->
## Description
<!--
Please include a summary of the change and which issue is fixed.
Please also include relevant motivation and context.
List any dependencies that are required for this change.
-->


Fixes # (issue)

## Type of change
<!--
Please tick the applicable options.
NOTE: Ticking multiple options most likely indicates
that your change is to big and it is suggested to split it into several smaller PRs.
-->

- [x] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] This change requires a documentation update

## Checklist:

- [ ] The documentation has been adjusted accordingly
- [ ] Tests have been added that prove the fix is effective or that the feature works
- [x] The changes are documented in the changelog (docs/changelog.md)

codecov[bot]: # [Codecov](https://codecov.io/gh/XKNX/xknx/pull/1018?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
> Merging [#1018](https://codecov.io/gh/XKNX/xknx/pull/1018?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (2f4c84b) into [main](https://codecov.io/gh/XKNX/xknx/commit/49e5602778ca46199a77e93205d477ec880aafdf?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (49e5602) will **decrease** coverage by `0.05%`.
> The diff coverage is `53.48%`.

[![Impacted file tree graph](https://codecov.io/gh/XKNX/xknx/pull/1018/graphs/tree.svg?width=650&height=150&src=pr&token=irWbIygS84&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)](https://codecov.io/gh/XKNX/xknx/pull/1018?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)

```diff
@@            Coverage Diff             @@
##             main    #1018      +/-   ##
==========================================
- Coverage   96.09%   96.03%   -0.06%     
==========================================
  Files         133      133              
  Lines        8140     8147       +7     
==========================================
+ Hits         7822     7824       +2     
- Misses        318      323       +5     
```


| [Impacted Files](https://codecov.io/gh/XKNX/xknx/pull/1018?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/exceptions/\_\_init\_\_.py](https://codecov.io/gh/XKNX/xknx/pull/1018/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9leGNlcHRpb25zL19faW5pdF9fLnB5) | `100.00% <ø> (ø)` | |
| [xknx/io/tunnel.py](https://codecov.io/gh/XKNX/xknx/pull/1018/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby90dW5uZWwucHk=) | `76.15% <48.71%> (-1.14%)` | :arrow_down: |
| [xknx/exceptions/exception.py](https://codecov.io/gh/XKNX/xknx/pull/1018/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9leGNlcHRpb25zL2V4Y2VwdGlvbi5weQ==) | `100.00% <100.00%> (ø)` | |
| [xknx/io/request\_response/tunnelling.py](https://codecov.io/gh/XKNX/xknx/pull/1018/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9yZXF1ZXN0X3Jlc3BvbnNlL3R1bm5lbGxpbmcucHk=) | `100.00% <100.00%> (ø)` | |

farmio: It will increase total. It's just the patch that has less than the whole project. This module always had poor coverage - mostly the reconnect stuff. 

Patch:
diff --git a/docs/changelog.md b/docs/changelog.md
index f41e3068..f7efb5f5 100644
--- a/docs/changelog.md
+++ b/docs/changelog.md
@@ -6,6 +6,12 @@ nav_order: 2
 
 # Changelog
 
+## Unreleased changes
+
+### Bugfixes
+
+- Correctly retry sending a TunnellingRequest if no TunnellingAck was received for the first time for UDP tunnelling connections.
+
 ## 1.0.0 Support for lukewarm temperatures 2022-08-13
 
 ### Internal
diff --git a/xknx/exceptions/__init__.py b/xknx/exceptions/__init__.py
index 661ed90a..94f1b1d1 100644
--- a/xknx/exceptions/__init__.py
+++ b/xknx/exceptions/__init__.py
@@ -17,6 +17,7 @@ from .exception import (
     ManagementConnectionRefused,
     ManagementConnectionTimeout,
     SecureException,
+    TunnellingAckError,
     UnsupportedCEMIMessage,
     XKNXException,
 )
@@ -38,6 +39,7 @@ __all__ = [
     "InvalidSignature",
     "KNXSecureValidationError",
     "SecureException",
+    "TunnellingAckError",
     "UnsupportedCEMIMessage",
     "XKNXException",
 ]
diff --git a/xknx/exceptions/exception.py b/xknx/exceptions/exception.py
index e47a4429..9057d2c4 100644
--- a/xknx/exceptions/exception.py
+++ b/xknx/exceptions/exception.py
@@ -34,6 +34,10 @@ class ConfirmationError(CommunicationError):
     """No confirmation received from KNX server for sent Telegram."""
 
 
+class TunnellingAckError(CommunicationError):
+    """No ACK or error status received from UDP KNX server for sent Telegram."""
+
+
 class CouldNotParseTelegram(XKNXException):
     """Could not parse telegram error."""
 
diff --git a/xknx/io/request_response/tunnelling.py b/xknx/io/request_response/tunnelling.py
index e2c14928..68ad7c7f 100644
--- a/xknx/io/request_response/tunnelling.py
+++ b/xknx/io/request_response/tunnelling.py
@@ -3,7 +3,7 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
-from xknx.knxip import CEMIFrame, KNXIPFrame, TunnellingAck, TunnellingRequest
+from xknx.knxip import KNXIPFrame, TunnellingAck, TunnellingRequest
 
 from .request_response import RequestResponse
 
@@ -12,7 +12,7 @@ if TYPE_CHECKING:
 
 
 class Tunnelling(RequestResponse):
-    """Class to TunnelingRequest and wait for TunnelingACK (UDP only)."""
+    """Class to send TunnelingRequest and wait for TunnelingACK (UDP only)."""
 
     transport: UDPTransport
 
@@ -20,28 +20,17 @@ class Tunnelling(RequestResponse):
         self,
         transport: UDPTransport,
         data_endpoint: tuple[str, int] | None,
-        cemi: CEMIFrame,
-        sequence_counter: int,
-        communication_channel_id: int,
+        tunnelling_request: TunnellingRequest,
     ):
         """Initialize Tunnelling class."""
         self.data_endpoint_addr = data_endpoint
-
+        self.tunnelling_request = tunnelling_request
         super().__init__(transport, TunnellingAck)
 
-        self.cemi_frame = cemi
-        self.sequence_counter = sequence_counter
-        self.communication_channel_id = communication_channel_id
-
     async def send_request(self) -> None:
         """Build knxipframe (within derived class) and send via UDP."""
         self.transport.send(self.create_knxipframe(), addr=self.data_endpoint_addr)
 
     def create_knxipframe(self) -> KNXIPFrame:
         """Create KNX/IP Frame object to be sent to device."""
-        tunnelling_request = TunnellingRequest(
-            communication_channel_id=self.communication_channel_id,
-            sequence_counter=self.sequence_counter,
-            cemi=self.cemi_frame,
-        )
-        return KNXIPFrame.init_from_body(tunnelling_request)
+        return KNXIPFrame.init_from_body(self.tunnelling_request)
diff --git a/xknx/io/tunnel.py b/xknx/io/tunnel.py
index 82e06e76..7f32183b 100644
--- a/xknx/io/tunnel.py
+++ b/xknx/io/tunnel.py
@@ -7,13 +7,12 @@ from __future__ import annotations
 
 from abc import abstractmethod
 import asyncio
-from collections.abc import Awaitable
 from copy import copy
 import logging
 from typing import TYPE_CHECKING
 
 from xknx.core import XknxConnectionState
-from xknx.exceptions import CommunicationError, ConfirmationError
+from xknx.exceptions import CommunicationError, ConfirmationError, TunnellingAckError
 from xknx.knxip import (
     HPAI,
     CEMIFrame,
@@ -257,52 +256,40 @@ class _Tunnel(Interface):
 
     async def _send_cemi(self, cemi: CEMIFrame) -> None:
         """
-        Send CEMI Frame to tunnelling server - retry mechanism.
+        Send CEMI Frame to tunnelling server.
 
         A transport layer confirmation shall be awaited before sending the next telegram.
-
-        If a TUNNELLING_REQUEST frame is not confirmed within the TUNNELLING_REQUEST_TIMEOUT
-        time of one (1) second then the frame shall be repeated once with the same sequence counter
-        value by the sending KNXnet/IP device.
-
-        If the KNXnet/IP device does not receive a TUNNELLING_ACK frame within the
-        TUNNELLING_REQUEST_TIMEOUT (= 1 second) or the status of a received
-        TUNNELLING_ACK frame signals any kind of error condition, the sending device
-        shall repeat the TUNNELLING_REQUEST frame once and then terminate the
-        connection by sending a DISCONNECT_REQUEST frame to the other devices
-        control endpoint.
         """
         async with self._send_telegram_lock:
             try:
-                if await self._tunnelling_request(cemi):
-                    return
-
-                logger.debug("Sending of telegram failed. Retrying a second time.")
-                if await self._tunnelling_request(cemi):
-                    return
-
-                logger.debug("Resending telegram failed. Reconnecting to tunnel.")
-                if self._reconnect_task is None or self._reconnect_task.done():
-                    self._tunnel_lost()
-                await self.xknx.connection_manager.connected.wait()
-                if not await self._tunnelling_request(cemi):
-                    raise CommunicationError(
-                        "Resending the telegram repeatedly failed.", True
-                    )
+                await self._tunnelling_request(cemi)
             finally:
                 self._increase_sequence_number()
 
-    @abstractmethod
-    async def _tunnelling_request(self, cemi: CEMIFrame) -> bool:
-        """Send Telegram to tunnelling device."""
+    async def _tunnelling_request(self, cemi: CEMIFrame) -> None:
+        """Send CEMI Frame to tunnelling server."""
+        if self.communication_channel is None:
+            raise CommunicationError(
+                "Sending telegram failed. No active communication channel."
+            )
+        tunnelling_request = TunnellingRequest(
+            communication_channel_id=self.communication_channel,
+            sequence_counter=self.sequence_number,
+            cemi=cemi,
+        )
+
+        if cemi.code is CEMIMessageCode.L_DATA_REQ:
+            await self._wait_for_tunnelling_request_confirmation(tunnelling_request)
+        else:
+            await self._send_tunnelling_request(tunnelling_request)
 
     async def _wait_for_tunnelling_request_confirmation(
-        self, send_tunneling_request_aw: Awaitable[None], cemi: CEMIFrame
+        self, frame: TunnellingRequest
     ) -> None:
         """Wait for confirmation of tunnelling request."""
         self._tunnelling_request_confirmation_event.clear()
         send_and_wait_for_confirmation = asyncio.gather(
-            send_tunneling_request_aw,
+            self._send_tunnelling_request(frame),
             self._tunnelling_request_confirmation_event.wait(),
         )
         try:
@@ -311,12 +298,14 @@ class _Tunnel(Interface):
                 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
             raise ConfirmationError(
-                f"L_DATA_CON Data Link Layer confirmation timed out for {cemi.telegram}"
+                f"L_DATA_CON Data Link Layer confirmation timed out for {frame}"
             )
 
+    @abstractmethod
+    async def _send_tunnelling_request(self, frame: TunnellingRequest) -> None:
+        """Send TunnellingRequest frame to tunnelling device."""
+
     def _increase_sequence_number(self) -> None:
         """Increase sequence number."""
         self.sequence_number = self.sequence_number + 1 & 0xFF
@@ -477,26 +466,73 @@ class UDPTunnel(_Tunnel):
 
     # OUTGOING REQUESTS
 
-    async def _tunnelling_request(self, cemi: CEMIFrame) -> bool:
+    async def _send_cemi(self, cemi: CEMIFrame) -> None:
+        """
+        Send CEMI Frame to tunnelling server - with retry mechanism for UDP connection.
+
+        A transport layer confirmation shall be awaited before sending the next telegram.
+
+        If a TUNNELLING_REQUEST frame is not confirmed within the TUNNELLING_REQUEST_TIMEOUT
+        time of one (1) second then the frame shall be repeated once with the same sequence counter
+        value by the sending KNXnet/IP device.
+
+        If the KNXnet/IP device does not receive a TUNNELLING_ACK frame within the
+        TUNNELLING_REQUEST_TIMEOUT (= 1 second) or the status of a received
+        TUNNELLING_ACK frame signals any kind of error condition, the sending device
+        shall repeat the TUNNELLING_REQUEST frame once and then terminate the
+        connection by sending a DISCONNECT_REQUEST frame to the other devices
+        control endpoint.
+        """
+        async with self._send_telegram_lock:
+            try:
+                try:
+                    await self._tunnelling_request(cemi)
+                except TunnellingAckError as err:
+                    logger.debug("%s. Retrying a second time.", err)
+                else:
+                    return
+
+                try:
+                    await self._tunnelling_request(cemi)
+                except TunnellingAckError as err:
+                    logger.debug("%s. Reconnecting tunnel.", err)
+                else:
+                    return
+
+                if self._reconnect_task is None or self._reconnect_task.done():
+                    self._tunnel_lost()
+                await self.xknx.connection_manager.connected.wait()
+                try:
+                    await self._tunnelling_request(cemi)
+                except TunnellingAckError as err:
+                    raise CommunicationError(
+                        f"Resending the telegram repeatedly failed. {err}", True
+                    )
+
+            finally:
+                self._increase_sequence_number()
+
+    async def _send_tunnelling_request(self, frame: TunnellingRequest) -> None:
         """Send Telegram to tunnelling device."""
-        if self.communication_channel is None:
-            raise CommunicationError(
-                "Sending telegram failed. No active communication channel."
-            )
         tunnelling = Tunnelling(
             transport=self.transport,
             data_endpoint=self._data_endpoint_addr,
-            cemi=cemi,
-            sequence_counter=self.sequence_number,
-            communication_channel_id=self.communication_channel,
+            tunnelling_request=frame,
         )
-        if cemi.code is CEMIMessageCode.L_DATA_REQ:
-            await self._wait_for_tunnelling_request_confirmation(
-                send_tunneling_request_aw=tunnelling.start(), cemi=cemi
-            )
-        else:
-            await tunnelling.start()
-        return tunnelling.success
+        await tunnelling.start()
+
+        if not tunnelling.success:
+            if error_code := tunnelling.response_status_code:
+                reason = (
+                    f"Received TUNNELLING_ACK with error code: {error_code.name} "
+                    f"for frame {frame}"
+                )
+            else:
+                reason = (
+                    "Did not receive a TUNNELLING_ACK within 1 second for "
+                    f"frame with sequence_counter={frame.sequence_counter}"
+                )
+            raise TunnellingAckError(reason)
 
     # INCOMING REQUESTS
 
@@ -562,31 +598,9 @@ class TCPTunnel(_Tunnel):
     async def setup_tunnel(self) -> None:
         """Set up tunnel before sending a ConnectionRequest."""
 
-    async def _tunnelling_request(self, cemi: CEMIFrame) -> bool:
+    async def _send_tunnelling_request(self, frame: TunnellingRequest) -> None:
         """Send Telegram to tunnelling device."""
-        if self.communication_channel is None:
-            raise CommunicationError(
-                "Sending telegram failed. No active communication channel."
-            )
-        tunnelling_request = TunnellingRequest(
-            communication_channel_id=self.communication_channel,
-            sequence_counter=self.sequence_number,
-            cemi=cemi,
-        )
-
-        if cemi.code is CEMIMessageCode.L_DATA_REQ:
-
-            async def _async_wrapper() -> None:
-                self.transport.send(KNXIPFrame.init_from_body(tunnelling_request))
-
-            await self._wait_for_tunnelling_request_confirmation(
-                send_tunneling_request_aw=_async_wrapper(),
-                cemi=cemi,
-            )
-        else:
-            self.transport.send(KNXIPFrame.init_from_body(tunnelling_request))
-
-        return True
+        self.transport.send(KNXIPFrame.init_from_body(frame))
 
 
 class SecureTunnel(TCPTunnel):


Test patch:
diff --git a/test/io_tests/request_response_tests/tunnelling_test.py b/test/io_tests/request_response_tests/tunnelling_test.py
index 42b0072c..e3331bda 100644
--- a/test/io_tests/request_response_tests/tunnelling_test.py
+++ b/test/io_tests/request_response_tests/tunnelling_test.py
@@ -22,7 +22,6 @@ class TestTunnelling:
 
     async def test_tunnelling(self):
         """Test tunnelling from KNX bus."""
-        communication_channel_id = 23
         data_endpoint = ("192.168.1.2", 4567)
         udp_transport = UDPTransport(("192.168.1.1", 0), ("192.168.1.2", 1234))
         cemi = CEMIFrame.init_from_telegram(
@@ -31,25 +30,16 @@ class TestTunnelling:
                 payload=GroupValueWrite(DPTArray((0x1, 0x2, 0x3))),
             )
         )
-        sequence_counter = 42
-        tunnelling = Tunnelling(
-            udp_transport,
-            data_endpoint,
-            cemi,
-            sequence_counter,
-            communication_channel_id,
+        tunnelling_request = TunnellingRequest(
+            communication_channel_id=23,
+            sequence_counter=42,
+            cemi=cemi,
         )
+        tunnelling = Tunnelling(udp_transport, data_endpoint, tunnelling_request)
         tunnelling.timeout_in_seconds = 0
 
         assert tunnelling.awaited_response_class == TunnellingAck
-        assert tunnelling.communication_channel_id == communication_channel_id
 
-        # Expected KNX/IP-Frame:
-        tunnelling_request = TunnellingRequest(
-            communication_channel_id=communication_channel_id,
-            sequence_counter=sequence_counter,
-        )
-        tunnelling_request.cemi = cemi
         exp_knxipframe = KNXIPFrame.init_from_body(tunnelling_request)
         with patch("xknx.io.transport.UDPTransport.send") as mock_udp_send, patch(
             "xknx.io.transport.UDPTransport.getsockname"
diff --git a/test/io_tests/tunnel_test.py b/test/io_tests/tunnel_test.py
index 8d4b42c8..4e7fb2b4 100644
--- a/test/io_tests/tunnel_test.py
+++ b/test/io_tests/tunnel_test.py
@@ -6,7 +6,8 @@ import pytest
 
 from xknx import XKNX
 from xknx.dpt import DPTArray
-from xknx.io import UDPTunnel
+from xknx.exceptions import CommunicationError
+from xknx.io import TCPTunnel, UDPTunnel
 from xknx.io.gateway_scanner import GatewayDescriptor
 from xknx.knxip import (
     HPAI,
@@ -134,7 +135,7 @@ class TestUDPTunnel:
         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):
+    async def test_tunnel_send_and_wait_for_l2_confirmation(self, time_travel):
         """Test tunnel waits for L_DATA.con before sending another L_DATA.req."""
         self.tunnel.transport.send = Mock()
         self.tunnel.communication_channel = 1
@@ -163,6 +164,70 @@ class TestUDPTunnel:
         assert self.tunnel.transport.send.call_count == 2
         await task
 
+    async def test_tunnel_send_retry(self, time_travel):
+        """Test tunnel resends the telegram when no ACK was received."""
+        self.tunnel.transport.send = Mock()
+        self.tunnel.communication_channel = 1
+        self.tunnel.sequence_number = 23
+
+        test_telegram = Telegram(payload=GroupValueWrite(DPTArray((1,))))
+        request = KNXIPFrame.init_from_body(
+            TunnellingRequest(
+                communication_channel_id=self.tunnel.communication_channel,
+                sequence_counter=self.tunnel.sequence_number,
+                cemi=CEMIFrame.init_from_telegram(
+                    test_telegram,
+                    code=CEMIMessageCode.L_DATA_REQ,
+                    src_addr=self.tunnel._src_address,
+                ),
+            )
+        )
+        test_ack = KNXIPFrame.init_from_body(
+            TunnellingAck(sequence_counter=self.tunnel.sequence_number)
+        )
+        confirmation = KNXIPFrame.init_from_body(
+            TunnellingRequest(
+                communication_channel_id=1,
+                sequence_counter=self.tunnel.sequence_number,
+                cemi=CEMIFrame.init_from_telegram(
+                    test_telegram, code=CEMIMessageCode.L_DATA_CON
+                ),
+            )
+        )
+        confirmation_ack = KNXIPFrame.init_from_body(
+            TunnellingAck(sequence_counter=self.tunnel.sequence_number)
+        )
+
+        task = asyncio.create_task(self.tunnel.send_telegram(test_telegram))
+        await time_travel(0)
+        assert self.tunnel.transport.send.call_args_list == [call(request, addr=None)]
+        self.tunnel.transport.send.reset_mock()
+        # no ACK received, resend same telegram
+        await time_travel(1)
+        assert self.tunnel.transport.send.call_args_list == [call(request, addr=None)]
+        self.tunnel.transport.send.reset_mock()
+        self.tunnel.transport.handle_knxipframe(test_ack, HPAI())
+        await time_travel(0)
+        assert not task.done()
+        self.tunnel.transport.handle_knxipframe(confirmation, HPAI())
+        await time_travel(0)
+        assert task.done()
+        assert self.tunnel.transport.send.call_args_list == [
+            call(confirmation_ack, addr=None)
+        ]
+        self.tunnel.transport.send.reset_mock()
+        await task
+
+        # Test raise after 2 missed ACKs (reconnect if `auto_reconnect` was True)
+        with pytest.raises(CommunicationError):
+            task = asyncio.create_task(self.tunnel.send_telegram(test_telegram))
+            # no ACKs received, for 2x wait time (with advancing the loop in between)
+            await time_travel(1)
+            await time_travel(1)
+            assert self.tunnel.transport.send.call_count == 2
+            self.tunnel.transport.send.reset_mock()
+            await task
+
     @pytest.mark.parametrize(
         "route_back,data_endpoint_addr,local_endpoint",
         [
@@ -270,3 +335,47 @@ class TestUDPTunnel:
         assert isinstance(task.result(), GatewayDescriptor)
         assert self.tunnel.transport.send.call_count == 1
         await task
+
+
+class TestTCPTunnel:
+    """Test class for xknx/io/TCPTunnel objects."""
+
+    def setup_method(self):
+        """Set up test class."""
+        # pylint: disable=attribute-defined-outside-init
+        self.xknx = XKNX()
+        self.tg_received_mock = AsyncMock()
+        self.tunnel = TCPTunnel(
+            self.xknx,
+            gateway_ip="192.168.1.2",
+            gateway_port=3671,
+            telegram_received_callback=self.tg_received_mock,
+            auto_reconnect=False,
+            auto_reconnect_wait=3,
+        )
+
+    async def test_tunnel_send_and_wait_for_l2_confirmation(self, time_travel):
+        """Test tunnel waits for L_DATA.con before sending another L_DATA.req."""
+        self.tunnel.transport.send = Mock()
+        self.tunnel.communication_channel = 1
+
+        test_telegram = Telegram(payload=GroupValueWrite(DPTArray((1,))))
+        confirmation = KNXIPFrame.init_from_body(
+            TunnellingRequest(
+                communication_channel_id=1,
+                sequence_counter=23,
+                cemi=CEMIFrame.init_from_telegram(
+                    test_telegram, code=CEMIMessageCode.L_DATA_CON
+                ),
+            )
+        )
+        task = asyncio.create_task(self.tunnel.send_telegram(test_telegram))
+        await time_travel(0)
+        assert not task.done()
+        assert self.tunnel.transport.send.call_count == 1
+        self.tunnel.transport.handle_knxipframe(confirmation, HPAI())
+        await time_travel(0)
+        assert task.done()
+        # one call for the outgoing request. No Acks for TCP.
+        assert self.tunnel.transport.send.call_count == 1
+        await task