Repository: XKNX/xknx
Pull request #1096 (2022-12-06 12:54:59)

Decouple CEMI parsing from KNXIPFrame parsing
<!--
  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.
-->
- Decouple KNXIPFrame parsing from CEMIFrame parsing. TunnellingRequest and RoutingIndication now carry the raw cemi frame payload as bytes. This allows decoupled CEMIFrame parsing at a later time (in Interface class rather than in KNXIPTransport class).
- Remove some tests for RoutingIndication that actually tested CEMIFrame / Telegram functions.

We are now able to inject data into the cemi parser more easily without having to pass an xknx instance throughout the IPFrame parsers. This paves the way to pass DataSecure keys.
It would also be an option to add a separate CEMI-Handler in front of `Tunnel / Routing` classes - eg. for USB support (@kistlin may be interested).

A noticeable change of this is that the `xknx.knx` logger now doesn't show the CEMI contents anymore, but the hex representation of the raw bytes 😬
```log
<KNXIPFrame <KNXIPHeader HeaderLength="6" ProtocolVersion="16" KNXIPServiceType="TUNNELLING_REQUEST" Reserve="0" TotalLength="23" />
 body="<TunnellingRequest communication_channel_id="2" sequence_counter="34" cemi="2900bcd0111529150300800c5b" />" />
```

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.
-->

- [ ] 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
- [ ] The changes are documented in the changelog (docs/changelog.md)

codecov[bot]: # [Codecov](https://codecov.io/gh/XKNX/xknx/pull/1096?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
> Merging [#1096](https://codecov.io/gh/XKNX/xknx/pull/1096?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (65f7416) into [main](https://codecov.io/gh/XKNX/xknx/commit/1474ac6f66be56d8e799077f35dd13b02bf44295?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (1474ac6) will **decrease** coverage by `0.01%`.
> The diff coverage is `88.09%`.

[![Impacted file tree graph](https://codecov.io/gh/XKNX/xknx/pull/1096/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/1096?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)

```diff
@@            Coverage Diff             @@
##             main    #1096      +/-   ##
==========================================
- Coverage   96.29%   96.27%   -0.02%     
==========================================
  Files         138      138              
  Lines        8641     8626      -15     
==========================================
- Hits         8321     8305      -16     
- Misses        320      321       +1     
```


| [Impacted Files](https://codecov.io/gh/XKNX/xknx/pull/1096?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/knxip/tunnelling\_request.py](https://codecov.io/gh/XKNX/xknx/pull/1096/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9rbnhpcC90dW5uZWxsaW5nX3JlcXVlc3QucHk=) | `91.66% <81.81%> (-0.44%)` | :arrow_down: |
| [xknx/io/tunnel.py](https://codecov.io/gh/XKNX/xknx/pull/1096/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby90dW5uZWwucHk=) | `88.29% <85.71%> (-0.43%)` | :arrow_down: |
| [xknx/io/routing.py](https://codecov.io/gh/XKNX/xknx/pull/1096/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9yb3V0aW5nLnB5) | `90.37% <87.50%> (-0.54%)` | :arrow_down: |
| [xknx/knxip/cemi\_frame.py](https://codecov.io/gh/XKNX/xknx/pull/1096/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9rbnhpcC9jZW1pX2ZyYW1lLnB5) | `100.00% <100.00%> (ø)` | |
| [xknx/knxip/routing\_indication.py](https://codecov.io/gh/XKNX/xknx/pull/1096/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9rbnhpcC9yb3V0aW5nX2luZGljYXRpb24ucHk=) | `100.00% <100.00%> (+4.00%)` | :arrow_up: |


Patch:
diff --git a/docs/changelog.md b/docs/changelog.md
index e48b0614..5d6d6875 100644
--- a/docs/changelog.md
+++ b/docs/changelog.md
@@ -8,6 +8,10 @@ nav_order: 2
 
 # Unreleased changes
 
+### Internal
+
+- Decouple KNXIPFrame parsing from CEMIFrame parsing. TunnellingRequest and RoutingIndication now carry the raw cemi frame payload as bytes. This allows decoupled CEMIFrame parsing at a later time (in Interface class rather than in KNXIPTransport class) for better error handling and upcoming features.
+
 # 2.1.0 Enhance notification device
 
 ### Devices
diff --git a/xknx/io/routing.py b/xknx/io/routing.py
index 4cbf6b53..aa41c43a 100644
--- a/xknx/io/routing.py
+++ b/xknx/io/routing.py
@@ -13,7 +13,7 @@ import random
 from typing import TYPE_CHECKING, Final
 
 from xknx.core import XknxConnectionState
-from xknx.exceptions import CommunicationError
+from xknx.exceptions import CommunicationError, UnsupportedCEMIMessage
 from xknx.knxip import (
     HPAI,
     CEMIFrame,
@@ -223,7 +223,7 @@ class Routing(Interface):
             code=CEMIMessageCode.L_DATA_IND,
             src_addr=self.individual_address,
         )
-        routing_indication = RoutingIndication(cemi=cemi)
+        routing_indication = RoutingIndication(raw_cemi=cemi.to_knx())
 
         async with self._flow_control.throttle():
             self._send_knxipframe(KNXIPFrame.init_from_body(routing_indication))
@@ -257,15 +257,17 @@ class Routing(Interface):
 
     def _handle_routing_indication(self, routing_indication: RoutingIndication) -> None:
         """Handle incoming RoutingIndication."""
-        if routing_indication.cemi is None:
-            # Don't handle invalid cemi frames (None)
+        cemi = CEMIFrame()
+        try:
+            cemi.from_knx(routing_indication.raw_cemi)
+        except UnsupportedCEMIMessage as unsupported_cemi_err:
+            logger.warning("CEMI not supported: %s", unsupported_cemi_err)
             return
-        if routing_indication.cemi.src_addr == self.individual_address:
+        if cemi.src_addr == self.individual_address:
             logger.debug("Ignoring own packet")
             return
-
         # TODO: is cemi message code L_DATA.req or .con valid for routing? if not maybe warn and ignore
-        asyncio.create_task(self.handle_cemi_frame(routing_indication.cemi))
+        asyncio.create_task(self.handle_cemi_frame(cemi))
 
     async def handle_cemi_frame(self, cemi: CEMIFrame) -> None:
         """Handle incoming telegram and send responses if applicable (device management)."""
diff --git a/xknx/io/tunnel.py b/xknx/io/tunnel.py
index 202bc083..95cc4cf8 100644
--- a/xknx/io/tunnel.py
+++ b/xknx/io/tunnel.py
@@ -12,7 +12,13 @@ import logging
 from typing import TYPE_CHECKING
 
 from xknx.core import XknxConnectionState
-from xknx.exceptions import CommunicationError, ConfirmationError, TunnellingAckError
+from xknx.exceptions import (
+    CommunicationError,
+    ConfirmationError,
+    ConversionError,
+    TunnellingAckError,
+    UnsupportedCEMIMessage,
+)
 from xknx.knxip import (
     HPAI,
     CEMIFrame,
@@ -268,10 +274,17 @@ class _Tunnel(Interface):
         A transport layer confirmation shall be awaited before sending the next telegram.
         """
         async with self._send_telegram_lock:
+            skip_increase_sequence_number = False
             try:
                 await self._tunnelling_request(cemi)
+            except ConversionError as ex:
+                # invalid CEMIFrame configurations preventing sending any frame
+                # shall not increase the sequence number
+                logger.warning("Could not send CEMI frame: %s", ex)
+                skip_increase_sequence_number = True
             finally:
-                self._increase_sequence_number()
+                if not skip_increase_sequence_number:
+                    self._increase_sequence_number()
 
     async def _tunnelling_request(self, cemi: CEMIFrame) -> None:
         """Send CEMI Frame to tunnelling server."""
@@ -282,7 +295,7 @@ class _Tunnel(Interface):
         tunnelling_request = TunnellingRequest(
             communication_channel_id=self.communication_channel,
             sequence_counter=self.sequence_number,
-            cemi=cemi,
+            raw_cemi=cemi.to_knx(),
         )
 
         if cemi.code is CEMIMessageCode.L_DATA_REQ:
@@ -338,20 +351,24 @@ class _Tunnel(Interface):
         self, tunneling_request: TunnellingRequest
     ) -> None:
         """Handle incoming tunnel request."""
-        if tunneling_request.cemi is None:
-            # Don't handle invalid cemi frames (None)
+        cemi = CEMIFrame()
+        try:
+            cemi.from_knx(tunneling_request.raw_cemi)
+        except UnsupportedCEMIMessage as unsupported_cemi_err:
+            logger.warning("CEMI not supported: %s", unsupported_cemi_err)
             return
-        if tunneling_request.cemi.code is CEMIMessageCode.L_DATA_CON:
+
+        if cemi.code is CEMIMessageCode.L_DATA_CON:
             # L_DATA_CON confirmation frame signals ready to send next telegram
             self._tunnelling_request_confirmation_event.set()
             return
 
-        if tunneling_request.cemi.code is CEMIMessageCode.L_DATA_REQ:
-            confirmation = copy(tunneling_request.cemi)
+        if cemi.code is CEMIMessageCode.L_DATA_REQ:
+            confirmation = copy(cemi)
             confirmation.code = CEMIMessageCode.L_DATA_CON
             asyncio.create_task(self._send_cemi(confirmation))
 
-        asyncio.create_task(self.handle_cemi_frame(tunneling_request.cemi))
+        asyncio.create_task(self.handle_cemi_frame(cemi))
 
     async def handle_cemi_frame(self, cemi: CEMIFrame) -> None:
         """Handle incoming telegram and send responses if applicable (device management)."""
diff --git a/xknx/knxip/cemi_frame.py b/xknx/knxip/cemi_frame.py
index 03432742..2af249bf 100644
--- a/xknx/knxip/cemi_frame.py
+++ b/xknx/knxip/cemi_frame.py
@@ -13,7 +13,7 @@ Documentation within:
 """
 from __future__ import annotations
 
-from xknx.exceptions import ConversionError, CouldNotParseKNXIP, UnsupportedCEMIMessage
+from xknx.exceptions import ConversionError, UnsupportedCEMIMessage
 from xknx.telegram import GroupAddress, IndividualAddress, Telegram
 from xknx.telegram.apci import APCI
 from xknx.telegram.tpci import TPCI, TDataGroup
@@ -159,7 +159,7 @@ class CEMIFrame:
         apdu = bytes([tpdu[0] & 0b11]) + tpdu[1:]  # clear TPCI bits
 
         if len(apdu) != (npdu_len + 1):  # TCPI octet not included in NPDU length
-            raise CouldNotParseKNXIP(
+            raise UnsupportedCEMIMessage(
                 f"APDU LEN should be {npdu_len} but is {len(apdu) - 1} in CEMI: {cemi.hex()}"
             )
 
diff --git a/xknx/knxip/routing_indication.py b/xknx/knxip/routing_indication.py
index fd069671..c7bf444f 100644
--- a/xknx/knxip/routing_indication.py
+++ b/xknx/knxip/routing_indication.py
@@ -5,15 +5,8 @@ Routing indications are used to transport CEMI Messages.
 """
 from __future__ import annotations
 
-import logging
-
-from xknx.exceptions import CouldNotParseKNXIP, UnsupportedCEMIMessage
-
 from .body import KNXIPBody
-from .cemi_frame import CEMIFrame
-from .knxip_enum import CEMIMessageCode, KNXIPServiceType
-
-logger = logging.getLogger("xknx.log")
+from .knxip_enum import KNXIPServiceType
 
 
 class RoutingIndication(KNXIPBody):
@@ -21,34 +14,23 @@ class RoutingIndication(KNXIPBody):
 
     SERVICE_TYPE = KNXIPServiceType.ROUTING_INDICATION
 
-    def __init__(self, cemi: CEMIFrame | None = None):
+    def __init__(self, raw_cemi: bytes = b""):
         """Initialize RoutingIndication object."""
-        self.cemi: CEMIFrame | None = (
-            cemi if cemi is not None else CEMIFrame(code=CEMIMessageCode.L_DATA_IND)
-        )
+        self.raw_cemi = raw_cemi
 
     def calculated_length(self) -> int:
         """Get length of KNX/IP body."""
-        assert self.cemi is not None
-        return self.cemi.calculated_length()
+        return len(self.raw_cemi)
 
     def from_knx(self, raw: bytes) -> int:
         """Parse/deserialize from KNX/IP raw data."""
-        assert self.cemi is not None
-
-        try:
-            return self.cemi.from_knx(raw)
-        except UnsupportedCEMIMessage as unsupported_cemi_err:
-            logger.warning("CEMI not supported: %s", unsupported_cemi_err)
-            self.cemi = None
-            return len(raw)
+        self.raw_cemi = raw
+        return len(raw)
 
     def to_knx(self) -> bytes:
         """Serialize to KNX/IP raw data."""
-        if self.cemi is None:
-            raise CouldNotParseKNXIP("No CEMIFrame defined.")
-        return self.cemi.to_knx()
+        return self.raw_cemi
 
     def __repr__(self) -> str:
         """Return object as readable string."""
-        return f'<RoutingIndication cemi="{self.cemi}" />'
+        return f'<RoutingIndication cemi="{self.raw_cemi.hex()}" />'
diff --git a/xknx/knxip/tunnelling_request.py b/xknx/knxip/tunnelling_request.py
index 8864e082..45a0b498 100644
--- a/xknx/knxip/tunnelling_request.py
+++ b/xknx/knxip/tunnelling_request.py
@@ -5,16 +5,11 @@ Tunnelling requests are used to transmit a KNX telegram within an existing KNX t
 """
 from __future__ import annotations
 
-import logging
-
-from xknx.exceptions import CouldNotParseKNXIP, UnsupportedCEMIMessage
+from xknx.exceptions import CouldNotParseKNXIP
 
 from .body import KNXIPBody
-from .cemi_frame import CEMIFrame, CEMIMessageCode
 from .knxip_enum import KNXIPServiceType
 
-logger = logging.getLogger("xknx.log")
-
 
 class TunnellingRequest(KNXIPBody):
     """Representation of a KNX Tunnelling Request."""
@@ -26,46 +21,31 @@ class TunnellingRequest(KNXIPBody):
         self,
         communication_channel_id: int = 1,
         sequence_counter: int = 0,
-        cemi: CEMIFrame | None = None,
+        raw_cemi: bytes = b"",
     ):
         """Initialize TunnellingRequest object."""
         self.communication_channel_id = communication_channel_id
         self.sequence_counter = sequence_counter
-        self.cemi: CEMIFrame | None = cemi or CEMIFrame(code=CEMIMessageCode.L_DATA_REQ)
+        self.raw_cemi = raw_cemi
 
     def calculated_length(self) -> int:
         """Get length of KNX/IP body."""
-        assert self.cemi is not None
-        return TunnellingRequest.HEADER_LENGTH + self.cemi.calculated_length()
+        return TunnellingRequest.HEADER_LENGTH + len(self.raw_cemi)
 
     def from_knx(self, raw: bytes) -> int:
         """Parse/deserialize from KNX/IP raw data."""
-        assert self.cemi is not None
-
-        def header_from_knx(header: bytes) -> int:
-            """Parse header bytes."""
-            if header[0] != TunnellingRequest.HEADER_LENGTH:
-                raise CouldNotParseKNXIP("connection header wrong length")
-            if len(header) < TunnellingRequest.HEADER_LENGTH:
-                raise CouldNotParseKNXIP("connection header wrong length")
-            self.communication_channel_id = header[1]
-            self.sequence_counter = header[2]
-            return TunnellingRequest.HEADER_LENGTH
-
-        pos = header_from_knx(raw)
-        try:
-            pos += self.cemi.from_knx(raw[pos:])
-        except UnsupportedCEMIMessage as unsupported_cemi_err:
-            logger.debug("CEMI not supported: %s", unsupported_cemi_err)
-            # Set cemi to None - this is checked in Tunnel() to send Ack even for unsupported CEMI messages.
-            self.cemi = None
-            return len(raw)
-        return pos
+        if raw[0] != TunnellingRequest.HEADER_LENGTH:
+            raise CouldNotParseKNXIP("connection header wrong length")
+        if len(raw) < TunnellingRequest.HEADER_LENGTH:
+            raise CouldNotParseKNXIP("connection header wrong length")
+        self.communication_channel_id = raw[1]
+        self.sequence_counter = raw[2]
+        # raw[3] is reserved
+        self.raw_cemi = raw[TunnellingRequest.HEADER_LENGTH :]
+        return len(raw)
 
     def to_knx(self) -> bytes:
         """Serialize to KNX/IP raw data."""
-        if self.cemi is None:
-            raise CouldNotParseKNXIP("No CEMIFrame defined.")
         return (
             bytes(
                 (
@@ -75,7 +55,7 @@ class TunnellingRequest(KNXIPBody):
                     0x00,  # Reserved
                 )
             )
-            + self.cemi.to_knx()
+            + self.raw_cemi
         )
 
     def __repr__(self) -> str:
@@ -84,5 +64,5 @@ class TunnellingRequest(KNXIPBody):
             "<TunnellingRequest "
             f'communication_channel_id="{self.communication_channel_id}" '
             f'sequence_counter="{self.sequence_counter}" '
-            f'cemi="{self.cemi}" />'
+            f'cemi="{self.raw_cemi.hex()}" />'
         )


Test patch:
diff --git a/requirements/testing.txt b/requirements/testing.txt
index b1082e16..3157d24c 100644
--- a/requirements/testing.txt
+++ b/requirements/testing.txt
@@ -6,8 +6,8 @@ flake8-isort==5.0.3
 pydocstyle==6.1.1
 pylint==2.15.8
 pytest==7.2.0
-pytest-asyncio==0.20.2
+pytest-asyncio==0.20.3
 pytest-cov==4.0.0
 setuptools==65.6.3
-tox==3.27.1
+tox==4.0.8
 mypy==0.991
diff --git a/test/io_tests/request_response_tests/tunnelling_test.py b/test/io_tests/request_response_tests/tunnelling_test.py
index e3331bda..ea45d048 100644
--- a/test/io_tests/request_response_tests/tunnelling_test.py
+++ b/test/io_tests/request_response_tests/tunnelling_test.py
@@ -24,16 +24,16 @@ class TestTunnelling:
         """Test tunnelling from KNX bus."""
         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(
+        raw_cemi = CEMIFrame.init_from_telegram(
             Telegram(
                 destination_address=GroupAddress("1/2/3"),
                 payload=GroupValueWrite(DPTArray((0x1, 0x2, 0x3))),
             )
-        )
+        ).to_knx()
         tunnelling_request = TunnellingRequest(
             communication_channel_id=23,
             sequence_counter=42,
-            cemi=cemi,
+            raw_cemi=raw_cemi,
         )
         tunnelling = Tunnelling(udp_transport, data_endpoint, tunnelling_request)
         tunnelling.timeout_in_seconds = 0
diff --git a/test/io_tests/routing_test.py b/test/io_tests/routing_test.py
index b98c0346..e2154816 100644
--- a/test/io_tests/routing_test.py
+++ b/test/io_tests/routing_test.py
@@ -37,20 +37,18 @@ class TestRouting:
         # L_Data.ind T_Connect from 1.0.250 to 1.0.255 (xknx tunnel endpoint)
         # communication_channel_id: 0x02   sequence_counter: 0x81
         raw_ind = bytes.fromhex("0610 0530 0010 2900b06010fa10ff0080")
-
         _cemi = CEMIFrame()
         _cemi.from_knx(raw_ind[6:])
         test_telegram = _cemi.telegram
         test_telegram.direction = TelegramDirection.INCOMING
 
         response_telegram = Telegram(tpci=tpci.TDisconnect())
+        response_cemi = CEMIFrame.init_from_telegram(
+            telegram=response_telegram,
+            src_addr=DEFAULT_INDIVIDUAL_ADDRESS,
+        )
         response_frame = KNXIPFrame.init_from_body(
-            RoutingIndication(
-                cemi=CEMIFrame.init_from_telegram(
-                    telegram=response_telegram,
-                    src_addr=DEFAULT_INDIVIDUAL_ADDRESS,
-                )
-            )
+            RoutingIndication(raw_cemi=response_cemi.to_knx())
         )
 
         async def tg_received_mock(telegram):
@@ -65,6 +63,23 @@ class TestRouting:
             call(response_frame),
         ]
 
+    @patch("logging.Logger.warning")
+    async def test_routing_indication_received_apci_unsupported(self, logging_mock):
+        """Test Tunnel sending ACK for unsupported frames."""
+        routing = Routing(
+            self.xknx,
+            individual_address=None,
+            telegram_received_callback=self.tg_received_mock,
+            local_ip="192.168.1.1",
+        )
+        # LDataInd Unsupported Extended APCI from 0.0.1 to 0/0/0 broadcast
+        # <UnsupportedCEMIMessage description="APCI not supported: 0b1111111000 in CEMI: 2900b0d0000100000103f8" />
+        raw = bytes.fromhex("0610 0530 0010 2900b0d0000100000103f8")
+
+        routing.transport.data_received_callback(raw, ("192.168.1.2", 3671))
+        self.tg_received_mock.assert_not_called()
+        logging_mock.assert_called_once()
+
     @patch("logging.Logger.warning")
     async def test_routing_lost_message(self, logging_mock):
         """Test class for received RoutingLostMessage frames."""
diff --git a/test/io_tests/secure_group_test.py b/test/io_tests/secure_group_test.py
index 45fe7012..6fdb3dd1 100644
--- a/test/io_tests/secure_group_test.py
+++ b/test/io_tests/secure_group_test.py
@@ -410,23 +410,27 @@ class TestSecureGroup:
         secure_timer = secure_group.secure_timer
         secure_timer.timer_authenticated = True
 
-        test_cemi = CEMIFrame.init_from_telegram(
+        raw_test_cemi = CEMIFrame.init_from_telegram(
             Telegram(
                 destination_address=GroupAddress("1/2/3"),
                 payload=apci.GroupValueRead(),
             )
-        )
+        ).to_knx()
 
         with patch.object(
             secure_timer, "reschedule", wraps=secure_timer.reschedule
         ) as mock_reschedule:
             assert not secure_timer.sched_update
-            secure_group.send(KNXIPFrame.init_from_body(RoutingIndication(test_cemi)))
+            secure_group.send(
+                KNXIPFrame.init_from_body(RoutingIndication(raw_cemi=raw_test_cemi))
+            )
             mock_reschedule.assert_called_once()
             mock_reschedule.reset_mock()
             # no reschedule when sched_update is true
             secure_timer.reschedule(update=(bytes(2), bytes(6)))
             assert secure_timer.sched_update
             mock_reschedule.reset_mock()
-            secure_group.send(KNXIPFrame.init_from_body(RoutingIndication(test_cemi)))
+            secure_group.send(
+                KNXIPFrame.init_from_body(RoutingIndication(raw_cemi=raw_test_cemi))
+            )
             mock_reschedule.assert_not_called()
diff --git a/test/io_tests/tunnel_test.py b/test/io_tests/tunnel_test.py
index 3ee259fb..cd177980 100644
--- a/test/io_tests/tunnel_test.py
+++ b/test/io_tests/tunnel_test.py
@@ -126,13 +126,14 @@ class TestUDPTunnel:
         self.tunnel.expected_sequence_number = 10
 
         test_telegram = Telegram(payload=GroupValueWrite(DPTArray((1,))))
+        cemi = CEMIFrame.init_from_telegram(
+            test_telegram, code=CEMIMessageCode.L_DATA_IND
+        )
         test_frame = KNXIPFrame.init_from_body(
             TunnellingRequest(
                 communication_channel_id=1,
                 sequence_counter=10,
-                cemi=CEMIFrame.init_from_telegram(
-                    test_telegram, code=CEMIMessageCode.L_DATA_IND
-                ),
+                raw_cemi=cemi.to_knx(),
             )
         )
         test_ack = KNXIPFrame.init_from_body(TunnellingAck(sequence_counter=10))
@@ -193,14 +194,15 @@ class TestUDPTunnel:
         self.tunnel.expected_sequence_number = 23
 
         test_telegram = Telegram(payload=GroupValueWrite(DPTArray((1,))))
+        cemi = CEMIFrame.init_from_telegram(
+            test_telegram, code=CEMIMessageCode.L_DATA_CON
+        )
         test_ack = KNXIPFrame.init_from_body(TunnellingAck(sequence_counter=23))
         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
-                ),
+                raw_cemi=cemi.to_knx(),
             )
         )
         task = asyncio.create_task(self.tunnel.send_telegram(test_telegram))
@@ -224,27 +226,29 @@ class TestUDPTunnel:
         self.tunnel.expected_sequence_number = 15
 
         test_telegram = Telegram(payload=GroupValueWrite(DPTArray((1,))))
+        cemi = CEMIFrame.init_from_telegram(
+            test_telegram,
+            code=CEMIMessageCode.L_DATA_REQ,
+            src_addr=self.tunnel._src_address,
+        )
         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,
-                ),
+                raw_cemi=cemi.to_knx(),
             )
         )
         test_ack = KNXIPFrame.init_from_body(
             TunnellingAck(sequence_counter=self.tunnel.sequence_number)
         )
+        confirmation_cemi = CEMIFrame.init_from_telegram(
+            test_telegram, code=CEMIMessageCode.L_DATA_CON
+        )
         confirmation = KNXIPFrame.init_from_body(
             TunnellingRequest(
                 communication_channel_id=1,
                 sequence_counter=15,
-                cemi=CEMIFrame.init_from_telegram(
-                    test_telegram, code=CEMIMessageCode.L_DATA_CON
-                ),
+                raw_cemi=confirmation_cemi.to_knx(),
             )
         )
         confirmation_ack = KNXIPFrame.init_from_body(TunnellingAck(sequence_counter=15))
@@ -328,15 +332,16 @@ class TestUDPTunnel:
         # Send - use data endpoint
         self.tunnel.transport.send.reset_mock()
         test_telegram = Telegram(payload=GroupValueWrite(DPTArray((1,))))
+        cemi = CEMIFrame.init_from_telegram(
+            test_telegram,
+            code=CEMIMessageCode.L_DATA_REQ,
+            src_addr=IndividualAddress(7),
+        )
         test_telegram_frame = KNXIPFrame.init_from_body(
             TunnellingRequest(
                 communication_channel_id=23,
                 sequence_counter=0,
-                cemi=CEMIFrame.init_from_telegram(
-                    test_telegram,
-                    code=CEMIMessageCode.L_DATA_REQ,
-                    src_addr=IndividualAddress(7),
-                ),
+                raw_cemi=cemi.to_knx(),
             )
         )
         asyncio.create_task(self.tunnel.send_telegram(test_telegram))
@@ -411,13 +416,14 @@ class TestTCPTunnel:
         self.tunnel.communication_channel = 1
 
         test_telegram = Telegram(payload=GroupValueWrite(DPTArray((1,))))
+        cemi = CEMIFrame.init_from_telegram(
+            test_telegram, code=CEMIMessageCode.L_DATA_CON
+        )
         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
-                ),
+                raw_cemi=cemi.to_knx(),
             )
         )
         task = asyncio.create_task(self.tunnel.send_telegram(test_telegram))
diff --git a/test/knxip_tests/cemi_frame_test.py b/test/knxip_tests/cemi_frame_test.py
index 9c3e202b..171dcacf 100644
--- a/test/knxip_tests/cemi_frame_test.py
+++ b/test/knxip_tests/cemi_frame_test.py
@@ -1,7 +1,7 @@
 """Tests for the CEMIFrame object."""
 import pytest
 
-from xknx.exceptions import ConversionError, CouldNotParseKNXIP, UnsupportedCEMIMessage
+from xknx.exceptions import ConversionError, UnsupportedCEMIMessage
 from xknx.knxip.cemi_frame import CEMIFrame
 from xknx.knxip.knxip_enum import CEMIFlags, CEMIMessageCode
 from xknx.telegram import GroupAddress, IndividualAddress, Telegram
@@ -88,7 +88,7 @@ def test_invalid_tpci_apci(raw, err_msg):
 def test_invalid_apdu_len():
     """Test for invalid apdu len."""
     frame = CEMIFrame()
-    with pytest.raises(CouldNotParseKNXIP, match=r".*APDU LEN should be .*"):
+    with pytest.raises(UnsupportedCEMIMessage, match=r".*APDU LEN should be .*"):
         frame.from_knx(get_data(0x29, 0, 0, 0, 0, 2, 0, []))
 
 
diff --git a/test/knxip_tests/routing_indication_test.py b/test/knxip_tests/routing_indication_test.py
index 609d880b..7b09d46e 100644
--- a/test/knxip_tests/routing_indication_test.py
+++ b/test/knxip_tests/routing_indication_test.py
@@ -17,13 +17,8 @@ class TestKNXIPRountingIndication:
         knxipframe.from_knx(raw)
 
         assert isinstance(knxipframe.body, RoutingIndication)
-        assert isinstance(knxipframe.body.cemi, CEMIFrame)
-
-        assert knxipframe.body.cemi.src_addr == IndividualAddress("1.2.2")
-        assert knxipframe.body.cemi.dst_addr == GroupAddress(337)
-
-        assert len(knxipframe.body.cemi.payload.value.value) == 1
-        assert knxipframe.body.cemi.payload.value.value[0] == 0xF0
+        assert isinstance(knxipframe.body.raw_cemi, bytes)
+        assert len(knxipframe.body.raw_cemi) == 12
 
     def test_from_knx_to_knx(self):
         """Test parsing and streaming CEMIFrame KNX/IP."""
@@ -46,7 +41,7 @@ class TestKNXIPRountingIndication:
         cemi = CEMIFrame(src_addr=IndividualAddress("1.2.2"))
         cemi.telegram = telegram
         cemi.set_hops(5)
-        routing_indication = RoutingIndication(cemi=cemi)
+        routing_indication = RoutingIndication(raw_cemi=cemi.to_knx())
         knxipframe = KNXIPFrame.init_from_body(routing_indication)
 
         raw = bytes.fromhex(
@@ -57,188 +52,21 @@ class TestKNXIPRountingIndication:
         assert knxipframe.body.to_knx() == raw[6:]
         assert knxipframe.to_knx() == raw
 
-    def test_telegram_get(self):
-        """Test parsing and streaming CEMIFrame KNX/IP packet, group read."""
-        raw = bytes.fromhex("06 10 05 30 00 12 29 00 bc d0 12 02 01 51 02 00 40 f0")
-        knxipframe = KNXIPFrame()
-        knxipframe.from_knx(raw)
-
-        telegram = knxipframe.body.cemi.telegram
-
-        assert telegram.destination_address == GroupAddress(337)
-
-        assert len(telegram.payload.value.value) == 1
-        assert telegram.payload.value.value[0] == 0xF0
-
-    #
-    # end-to-end tests:
-    #
-    #   - parsing KNX telegram and check the result
-    #   - reassembling scond KNXIPFrame
-    #   - comparing both
-    #
-
-    def test_end_to_end_group_write_binary_on(self):
-        """Test parsing and streaming CEMIFrame KNX/IP packet, switch on light in my kitchen."""
-        # Switch on Kitchen-L1
-        raw = bytes.fromhex("06 10 05 30 00 11 29 00 bc d0 ff f9 01 49 01 00 81")
-        knxipframe = KNXIPFrame()
-        knxipframe.from_knx(raw)
-        telegram = knxipframe.body.cemi.telegram
-        assert telegram == Telegram(
-            destination_address=GroupAddress("329"),
-            payload=GroupValueWrite(DPTBinary(1)),
-            source_address=IndividualAddress("15.15.249"),
-        )
-
-        cemi = CEMIFrame(src_addr=IndividualAddress("15.15.249"))
-        cemi.telegram = telegram
-        cemi.set_hops(5)
-        routing_indication = RoutingIndication(cemi=cemi)
-        knxipframe2 = KNXIPFrame.init_from_body(routing_indication)
-
-        assert knxipframe2.header.to_knx() == raw[0:6]
-        assert knxipframe2.body.to_knx() == raw[6:]
-        assert knxipframe2.to_knx() == raw
-
-    def test_end_to_end_group_write_binary_off(self):
-        """Test parsing and streaming CEMIFrame KNX/IP packet, switch off light in my kitchen."""
+    def test_end_to_end_routing_indication(self):
+        """Test parsing and streaming RoutingIndication KNX/IP packet."""
         # Switch off Kitchen-L1
         raw = bytes.fromhex("06 10 05 30 00 11 29 00 bc d0 ff f9 01 49 01 00 80")
         knxipframe = KNXIPFrame()
         knxipframe.from_knx(raw)
-        telegram = knxipframe.body.cemi.telegram
-        assert telegram == Telegram(
-            destination_address=GroupAddress("329"),
-            payload=GroupValueWrite(DPTBinary(0)),
-            source_address=IndividualAddress("15.15.249"),
-        )
+        raw_cemi = knxipframe.body.raw_cemi
 
-        cemi = CEMIFrame(src_addr=IndividualAddress("15.15.249"))
-        cemi.telegram = telegram
-        cemi.set_hops(5)
-        routing_indication = RoutingIndication(cemi=cemi)
+        routing_indication = RoutingIndication(raw_cemi=raw_cemi)
         knxipframe2 = KNXIPFrame.init_from_body(routing_indication)
 
         assert knxipframe2.header.to_knx() == raw[0:6]
         assert knxipframe2.body.to_knx() == raw[6:]
         assert knxipframe2.to_knx() == raw
 
-    def test_end_to_end_group_write_1byte(self):
-        """Test parsing and streaming CEMIFrame KNX/IP packet, dimm light in my kitchen."""
-        # Dimm Kitchen L1 to 0x65
-        raw = bytes.fromhex("06 10 05 30 00 12 29 00 bc d0 ff f9 01 4b 02 00 80 65")
-        knxipframe = KNXIPFrame()
-        knxipframe.from_knx(raw)
-        telegram = knxipframe.body.cemi.telegram
-        assert telegram == Telegram(
-            destination_address=GroupAddress("331"),
-            payload=GroupValueWrite(DPTArray(0x65)),
-            source_address=IndividualAddress("15.15.249"),
-        )
-
-        cemi = CEMIFrame(src_addr=IndividualAddress("15.15.249"))
-        cemi.telegram = telegram
-        cemi.set_hops(5)
-        routing_indication = RoutingIndication(cemi=cemi)
-        knxipframe2 = KNXIPFrame.init_from_body(routing_indication)
-
-        assert knxipframe2.header.to_knx() == raw[0:6]
-        assert knxipframe2.body.to_knx() == raw[6:]
-        assert knxipframe2.to_knx() == raw
-
-    def test_end_to_end_group_write_2bytes(self):
-        """Test parsing and streaming CEMIFrame KNX/IP packet, setting value of thermostat."""
-        # Incoming Temperature from thermostat
-        raw = bytes.fromhex("06 10 05 30 00 13 29 00 bc d0 14 02 08 01 03 00 80 07 c1")
-        knxipframe = KNXIPFrame()
-        knxipframe.from_knx(raw)
-        telegram = knxipframe.body.cemi.telegram
-        assert telegram == Telegram(
-            destination_address=GroupAddress("2049"),
-            payload=GroupValueWrite(DPTArray(DPTTemperature().to_knx(19.85))),
-            source_address=IndividualAddress("1.4.2"),
-        )
-
-        cemi = CEMIFrame(src_addr=IndividualAddress("1.4.2"))
-        cemi.telegram = telegram
-        cemi.set_hops(5)
-        routing_indication = RoutingIndication(cemi=cemi)
-        knxipframe2 = KNXIPFrame.init_from_body(routing_indication)
-
-        assert knxipframe2.header.to_knx() == raw[0:6]
-        assert knxipframe2.body.to_knx() == raw[6:]
-        assert knxipframe2.to_knx() == raw
-
-    def test_end_to_end_group_read(self):
-        """Test parsing and streaming CEMIFrame KNX/IP packet, group read."""
-        # State request
-        raw = bytes.fromhex("06 10 05 30 00 11 29 00 bc d0 ff f9 01 b8 01 00 00")
-        knxipframe = KNXIPFrame()
-        knxipframe.from_knx(raw)
-        telegram = knxipframe.body.cemi.telegram
-        assert telegram == Telegram(
-            destination_address=GroupAddress("440"),
-            payload=GroupValueRead(),
-            source_address=IndividualAddress("15.15.249"),
-        )
-
-        cemi = CEMIFrame(src_addr=IndividualAddress("15.15.249"))
-        cemi.telegram = telegram
-        cemi.set_hops(5)
-        routing_indication = RoutingIndication(cemi=cemi)
-        knxipframe2 = KNXIPFrame.init_from_body(routing_indication)
-
-        assert knxipframe2.header.to_knx() == raw[0:6]
-        assert knxipframe2.body.to_knx() == raw[6:]
-        assert knxipframe2.to_knx() == raw
-
-    def test_end_to_end_group_response(self):
-        """Test parsing and streaming CEMIFrame KNX/IP packet, group response."""
-        # Incoming state
-        raw = bytes.fromhex("06 10 05 30 00 11 29 00 bc d0 13 01 01 88 01 00 41")
-        knxipframe = KNXIPFrame()
-        knxipframe.from_knx(raw)
-        telegram = knxipframe.body.cemi.telegram
-        assert telegram == Telegram(
-            destination_address=GroupAddress("392"),
-            payload=GroupValueResponse(DPTBinary(1)),
-            source_address=IndividualAddress("1.3.1"),
-        )
-
-        cemi = CEMIFrame(src_addr=IndividualAddress("1.3.1"))
-        cemi.telegram = telegram
-        cemi.set_hops(5)
-        routing_indication = RoutingIndication(cemi=cemi)
-        knxipframe2 = KNXIPFrame.init_from_body(routing_indication)
-
-        assert knxipframe2.header.to_knx() == raw[0:6]
-        assert knxipframe2.body.to_knx() == raw[6:]
-        assert knxipframe2.to_knx() == raw
-
-    def test_maximum_apci(self):
-        """Test parsing and streaming CEMIFrame KNX/IP packet, testing maximum APCI."""
-        telegram = Telegram(
-            destination_address=GroupAddress(337),
-            payload=GroupValueWrite(DPTBinary(DPTBinary.APCI_MAX_VALUE)),
-            source_address=IndividualAddress("1.3.1"),
-        )
-
-        cemi = CEMIFrame(src_addr=IndividualAddress("1.3.1"))
-        cemi.telegram = telegram
-        cemi.set_hops(5)
-        routing_indication = RoutingIndication(cemi=cemi)
-        knxipframe = KNXIPFrame.init_from_body(routing_indication)
-
-        raw = bytes.fromhex("06 10 05 30 00 11 29 00 bc d0 13 01 01 51 01 00 bf")
-
-        assert knxipframe.to_knx() == raw
-
-        knxipframe2 = KNXIPFrame()
-        knxipframe2.init(KNXIPServiceType.ROUTING_INDICATION)
-        knxipframe2.from_knx(knxipframe.to_knx())
-        assert knxipframe2.body.cemi.telegram == telegram
-
     def test_from_knx_invalid_cemi(self):
         """Test parsing and streaming CEMIFrame KNX/IP packet with unsupported CEMICode."""
         routing_indication = RoutingIndication()
diff --git a/test/knxip_tests/tunnelling_request_test.py b/test/knxip_tests/tunnelling_request_test.py
index 597af1eb..5c359630 100644
--- a/test/knxip_tests/tunnelling_request_test.py
+++ b/test/knxip_tests/tunnelling_request_test.py
@@ -22,22 +22,24 @@ class TestKNXIPTunnellingRequest:
         assert isinstance(knxipframe.body, TunnellingRequest)
         assert knxipframe.body.communication_channel_id == 1
         assert knxipframe.body.sequence_counter == 23
-        assert isinstance(knxipframe.body.cemi, CEMIFrame)
+        assert isinstance(knxipframe.body.raw_cemi, bytes)
 
-        assert knxipframe.body.cemi.telegram == Telegram(
+        incoming_cemi = CEMIFrame()
+        incoming_cemi.from_knx(knxipframe.body.raw_cemi)
+        assert incoming_cemi.telegram == Telegram(
             destination_address=GroupAddress("9/0/8"),
             payload=GroupValueWrite(DPTBinary(1)),
         )
 
-        cemi = CEMIFrame(code=CEMIMessageCode.L_DATA_REQ)
-        cemi.telegram = Telegram(
+        outgoing_cemi = CEMIFrame(code=CEMIMessageCode.L_DATA_REQ)
+        outgoing_cemi.telegram = Telegram(
             destination_address=GroupAddress("9/0/8"),
             payload=GroupValueWrite(DPTBinary(1)),
         )
         tunnelling_request = TunnellingRequest(
             communication_channel_id=1,
             sequence_counter=23,
-            cemi=cemi,
+            raw_cemi=outgoing_cemi.to_knx(),
         )
         knxipframe2 = KNXIPFrame.init_from_body(tunnelling_request)
 
diff --git a/test/str_test.py b/test/str_test.py
index 7f63a7d7..14d3a342 100644
--- a/test/str_test.py
+++ b/test/str_test.py
@@ -626,9 +626,7 @@ class TestStringRepresentations:
         tunnelling_request.sequence_counter = 42
         assert (
             str(tunnelling_request)
-            == '<TunnellingRequest communication_channel_id="23" sequence_counter="42" '
-            'cemi="<CEMIFrame code="L_DATA_REQ" src_addr="IndividualAddress("0.0.0")" dst_addr="GroupAddress("0/0/0")" '
-            'flags="               0" tpci="TDataGroup()" payload="None" />" />'
+            == '<TunnellingRequest communication_channel_id="23" sequence_counter="42" cemi="" />'
         )
 
     def test_tunnelling_ack(self):
@@ -689,8 +687,4 @@ class TestStringRepresentations:
     def test_routing_indication_str(self):
         """Test string representation of GatewayDescriptor."""
         routing_indication = RoutingIndication()
-        assert (
-            str(routing_indication)
-            == '<RoutingIndication cemi="<CEMIFrame code="L_DATA_IND" src_addr="IndividualAddress("0.0.0")" '
-            'dst_addr="GroupAddress("0/0/0")" flags="               0" tpci="TDataGroup()" payload="None" />" />'
-        )
+        assert str(routing_indication) == '<RoutingIndication cemi="" />'