Repository: XKNX/xknx
Pull request #1221 (2023-03-09 20:26:59)

Move Telegram handling from CEMIFrame to CEMILData
<!--
  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.
-->
- Move `init_from_telegram()` from CEMIFrame to CEMILData. 
- `telegram()` is now a method of CEMILData instead of a property of CEMIFrame.

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)
- [x] Code quality improvements to existing code or addition of tests

## Checklist

- [ ] The documentation has been adjusted accordingly
- [x] 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)


Patch:
diff --git a/docs/changelog.md b/docs/changelog.md
index 3781d74b..88516e9f 100644
--- a/docs/changelog.md
+++ b/docs/changelog.md
@@ -16,6 +16,7 @@ nav_order: 2
 ### Internals
 
 - Use CEMILData instead of CEMIFrame in DataSecure.
+- Move `init_from_telegram()` from CEMIFrame to CEMILData. `telegram()` is now a method of CEMILData instead of a property of CEMIFrame.
 
 # 2.6.0 Connection information 2023-02-27
 
diff --git a/xknx/cemi/cemi_frame.py b/xknx/cemi/cemi_frame.py
index 7825d5e5..f0f89dde 100644
--- a/xknx/cemi/cemi_frame.py
+++ b/xknx/cemi/cemi_frame.py
@@ -125,6 +125,51 @@ class CEMILData(CEMIData):
             return 8
         raise TypeError("Data TPDU must have a payload; control TPDU must not.")
 
+    @staticmethod
+    def init_from_telegram(
+        telegram: Telegram,
+        src_addr: IndividualAddress | None = None,
+    ) -> CEMILData:
+        """Return CEMILData from a Telegram."""
+        flags = (
+            CEMIFlags.FRAME_TYPE_STANDARD
+            | CEMIFlags.DO_NOT_REPEAT
+            | CEMIFlags.BROADCAST
+            | CEMIFlags.NO_ACK_REQUESTED
+            | CEMIFlags.CONFIRM_NO_ERROR
+            | CEMIFlags.HOP_COUNT_1ST
+        )
+        if isinstance(telegram.destination_address, GroupAddress):
+            flags |= CEMIFlags.DESTINATION_GROUP_ADDRESS
+            if isinstance(telegram.tpci, TDataBroadcast):
+                flags |= CEMIFlags.PRIORITY_SYSTEM
+            else:
+                flags |= CEMIFlags.PRIORITY_LOW
+        elif isinstance(telegram.destination_address, IndividualAddress):
+            flags |= (
+                CEMIFlags.DESTINATION_INDIVIDUAL_ADDRESS | CEMIFlags.PRIORITY_SYSTEM
+            )
+        else:
+            raise TypeError()
+
+        return CEMILData(
+            flags=flags,
+            src_addr=src_addr or telegram.source_address,
+            dst_addr=telegram.destination_address,
+            tpci=telegram.tpci,
+            payload=telegram.payload,
+        )
+
+    def telegram(self) -> Telegram:
+        """Return Telegram from a CEMILData."""
+
+        return Telegram(
+            destination_address=self.dst_addr,
+            payload=self.payload,
+            source_address=self.src_addr,
+            tpci=self.tpci,
+        )
+
     def to_knx(self) -> bytes:
         """Serialize to KNX/IP raw data."""
         if self.tpci.control:
@@ -147,17 +192,6 @@ class CEMILData(CEMIData):
             + tpdu
         )
 
-    @property
-    def telegram(self) -> Telegram:
-        """Return telegram."""
-
-        return Telegram(
-            destination_address=self.dst_addr,
-            payload=self.payload,
-            source_address=self.src_addr,
-            tpci=self.tpci,
-        )
-
     @classmethod
     def from_knx(cls, raw: bytes) -> CEMILData:
         """Parse L_DATA_IND, CEMIMessageCode.L_DATA_REQ, CEMIMessageCode.L_DATA_CON."""
@@ -556,45 +590,6 @@ class CEMIFrame:
             CEMIMessageCode.M_RESET_IND,
         )
 
-    @staticmethod
-    def init_from_telegram(
-        telegram: Telegram,
-        code: CEMIMessageCode = CEMIMessageCode.L_DATA_IND,
-        src_addr: IndividualAddress | None = None,
-    ) -> CEMIFrame:
-        """Return CEMIFrame from a Telegram."""
-        flags = (
-            CEMIFlags.FRAME_TYPE_STANDARD
-            | CEMIFlags.DO_NOT_REPEAT
-            | CEMIFlags.BROADCAST
-            | CEMIFlags.NO_ACK_REQUESTED
-            | CEMIFlags.CONFIRM_NO_ERROR
-            | CEMIFlags.HOP_COUNT_1ST
-        )
-        if isinstance(telegram.destination_address, GroupAddress):
-            flags |= CEMIFlags.DESTINATION_GROUP_ADDRESS
-            if isinstance(telegram.tpci, TDataBroadcast):
-                flags |= CEMIFlags.PRIORITY_SYSTEM
-            else:
-                flags |= CEMIFlags.PRIORITY_LOW
-        elif isinstance(telegram.destination_address, IndividualAddress):
-            flags |= (
-                CEMIFlags.DESTINATION_INDIVIDUAL_ADDRESS | CEMIFlags.PRIORITY_SYSTEM
-            )
-        else:
-            raise TypeError()
-
-        return CEMIFrame(
-            code=code,
-            data=CEMILData(
-                flags=flags,
-                src_addr=src_addr or telegram.source_address,
-                dst_addr=telegram.destination_address,
-                tpci=telegram.tpci,
-                payload=telegram.payload,
-            ),
-        )
-
     def calculated_length(self) -> int:
         """Get length of KNX/IP body."""
         length = 1
@@ -671,11 +666,3 @@ class CEMIFrame:
     def __eq__(self, other: object) -> bool:
         """Equal operator."""
         return self.__dict__ == other.__dict__
-
-    @property
-    def telegram(self) -> Telegram:
-        """Return telegram."""
-        if not isinstance(self.data, CEMILData):
-            raise TypeError()
-
-        return self.data.telegram
diff --git a/xknx/cemi/cemi_handler.py b/xknx/cemi/cemi_handler.py
index 7ec0712f..0361bb84 100644
--- a/xknx/cemi/cemi_handler.py
+++ b/xknx/cemi/cemi_handler.py
@@ -54,20 +54,20 @@ class CEMIHandler:
 
     async def send_telegram(self, telegram: Telegram) -> None:
         """Create a CEMIFrame from a Telegram and send it to the CEMI Server."""
-        cemi = CEMIFrame.init_from_telegram(
+        cemi_data = CEMILData.init_from_telegram(
             telegram=telegram,
-            code=CEMIMessageCode.L_DATA_REQ,
             src_addr=(
                 self.xknx.current_address if telegram.source_address.raw == 0 else None
             ),
         )
-        if not isinstance(cemi.data, CEMILData):
-            logger.warning("Invalid CEMI data from Telegram: %s", cemi)
-            return
+        cemi = CEMIFrame(
+            code=CEMIMessageCode.L_DATA_REQ,
+            data=cemi_data,
+        )
 
         logger.debug("Outgoing CEMI: %s", cemi)
         if self.data_secure is not None:
-            cemi.data = self.data_secure.outgoing_cemi(cemi_data=cemi.data)
+            cemi.data = self.data_secure.outgoing_cemi(cemi_data=cemi_data)
         self._l_data_confirmation_event.clear()
         try:
             await self.xknx.knxip_interface.send_cemi(cemi)
@@ -139,8 +139,7 @@ class CEMIHandler:
                 )
                 return
         self.xknx.connection_manager.cemi_count_incoming += 1
-        # TODO: remove telegram init from CEMIFrame class and move it here?
-        telegram = cemi.telegram
+        telegram = cemi.data.telegram()
         telegram.direction = TelegramDirection.INCOMING
         self.telegram_received(telegram)
 
diff --git a/xknx/secure/data_secure.py b/xknx/secure/data_secure.py
index a6f55c51..f8a18f94 100644
--- a/xknx/secure/data_secure.py
+++ b/xknx/secure/data_secure.py
@@ -233,7 +233,7 @@ class DataSecure:
             plain_apdu_raw = cemi_data.payload.to_knx()
         else:
             # TODO: test if this is correct
-            plain_apdu_raw = b""  # used ein point-to-point eg. TConnect
+            plain_apdu_raw = b""  # used in point-to-point eg. TConnect
         secure_asdu = SecureData.init_from_plain_apdu(
             key=key,
             apdu=plain_apdu_raw,


Test patch:
diff --git a/test/cemi_tests/cemi_frame_test.py b/test/cemi_tests/cemi_frame_test.py
index 4443c770..2ba90321 100644
--- a/test/cemi_tests/cemi_frame_test.py
+++ b/test/cemi_tests/cemi_frame_test.py
@@ -186,42 +186,54 @@ def test_from_knx_individual_address():
 def test_telegram_group_address():
     """Test telegram conversion flags with a group address."""
     _telegram = Telegram(destination_address=GroupAddress(1))
-    frame = CEMIFrame.init_from_telegram(_telegram)
+    frame = CEMIFrame(
+        code=CEMIMessageCode.L_DATA_IND,
+        data=CEMILData.init_from_telegram(_telegram),
+    )
     assert isinstance(frame.data, CEMILData)
     assert frame.data.flags & 0x0080 == CEMIFlags.DESTINATION_GROUP_ADDRESS
     assert frame.data.flags & 0x0C00 == CEMIFlags.PRIORITY_LOW
     # test CEMIFrame.telegram property
-    assert frame.telegram == _telegram
+    assert frame.data.telegram() == _telegram
 
 
 def test_telegram_broadcast():
     """Test telegram conversion flags with a group address."""
     _telegram = Telegram(destination_address=GroupAddress(0))
-    frame = CEMIFrame.init_from_telegram(_telegram)
+    frame = CEMIFrame(
+        code=CEMIMessageCode.L_DATA_IND,
+        data=CEMILData.init_from_telegram(_telegram),
+    )
     assert isinstance(frame.data, CEMILData)
     assert frame.data.flags & 0x0080 == CEMIFlags.DESTINATION_GROUP_ADDRESS
     assert frame.data.flags & 0x0C00 == CEMIFlags.PRIORITY_SYSTEM
     assert frame.data.tpci == TDataBroadcast()
     # test CEMIFrame.telegram property
-    assert frame.telegram == _telegram
+    assert frame.data.telegram() == _telegram
 
 
 def test_telegram_individual_address():
     """Test telegram conversion flags with a individual address."""
     _telegram = Telegram(destination_address=IndividualAddress(0), tpci=TConnect())
-    frame = CEMIFrame.init_from_telegram(_telegram)
+    frame = CEMIFrame(
+        code=CEMIMessageCode.L_DATA_IND,
+        data=CEMILData.init_from_telegram(_telegram),
+    )
     assert isinstance(frame.data, CEMILData)
     assert frame.data.flags & 0x0080 == CEMIFlags.DESTINATION_INDIVIDUAL_ADDRESS
     assert frame.data.flags & 0x0C00 == CEMIFlags.PRIORITY_SYSTEM
     assert frame.data.flags & 0x0200 == CEMIFlags.NO_ACK_REQUESTED
     # test CEMIFrame.telegram property
-    assert frame.telegram == _telegram
+    assert frame.data.telegram() == _telegram
 
 
 def test_telegram_unsupported_address():
     """Test telegram conversion flags with an unsupported address."""
     with pytest.raises(TypeError):
-        CEMIFrame.init_from_telegram(Telegram(destination_address=object()))
+        CEMIFrame(
+            code=CEMIMessageCode.L_DATA_IND,
+            data=CEMILData.init_from_telegram(Telegram(destination_address=object())),
+        )
 
 
 def get_prop(code, obj_id, obj_inst, prop_id, num, six, payload):
@@ -259,8 +271,8 @@ def test_valid_read_req():
     assert frame.data.property_info.start_index == 1
     assert frame.calculated_length() == 7
     assert frame.to_knx() == raw
-    with pytest.raises(TypeError):
-        frame.telegram
+    with pytest.raises(AttributeError):
+        frame.data.telegram()
 
 
 def test_valid_read_con():
@@ -284,8 +296,6 @@ def test_valid_read_con():
     assert IndividualAddress.from_knx(frame.data.data) == IndividualAddress("1.2.3")
     assert frame.calculated_length() == 9
     assert frame.to_knx() == raw
-    with pytest.raises(TypeError):
-        frame.telegram
 
 
 def test_valid_error_read_con():
@@ -308,8 +318,6 @@ def test_valid_error_read_con():
     assert frame.data.error_code == CEMIErrorCode.CEMI_ERROR_VOID_DP
     assert frame.calculated_length() == 8
     assert frame.to_knx() == raw
-    with pytest.raises(TypeError):
-        frame.telegram
 
 
 def test_valid_write_req():
@@ -332,8 +340,6 @@ def test_valid_write_req():
     assert IndividualAddress.from_knx(frame.data.data) == IndividualAddress("1.2.3")
     assert frame.calculated_length() == 9
     assert frame.to_knx() == raw
-    with pytest.raises(TypeError):
-        frame.telegram
 
 
 def test_valid_empty_write_con():
@@ -356,8 +362,6 @@ def test_valid_empty_write_con():
     assert frame.data.error_code == None
     assert frame.calculated_length() == 7
     assert frame.to_knx() == raw
-    with pytest.raises(TypeError):
-        frame.telegram
 
 
 def test_valid_error_write_con():
@@ -380,8 +384,6 @@ def test_valid_error_write_con():
     assert frame.data.error_code == CEMIErrorCode.CEMI_ERROR_VOID_DP
     assert frame.calculated_length() == 8
     assert frame.to_knx() == raw
-    with pytest.raises(TypeError):
-        frame.telegram
 
 
 @pytest.mark.parametrize(
diff --git a/test/cemi_tests/cemi_handler_test.py b/test/cemi_tests/cemi_handler_test.py
index 32272523..f45ad022 100644
--- a/test/cemi_tests/cemi_handler_test.py
+++ b/test/cemi_tests/cemi_handler_test.py
@@ -5,7 +5,7 @@ from unittest.mock import AsyncMock, patch
 import pytest
 
 from xknx import XKNX
-from xknx.cemi import CEMIFrame, CEMIMessageCode
+from xknx.cemi import CEMIFrame, CEMILData, CEMIMessageCode
 from xknx.dpt import DPTArray
 from xknx.exceptions import ConfirmationError
 from xknx.telegram import GroupAddress, IndividualAddress, Telegram, apci, tpci
@@ -20,13 +20,15 @@ async def test_wait_for_l2_confirmation(time_travel):
         destination_address=GroupAddress(1),
         payload=apci.GroupValueWrite(DPTArray((1,))),
     )
-    test_cemi = CEMIFrame.init_from_telegram(
-        test_telegram,
+    test_cemi = CEMIFrame(
         code=CEMIMessageCode.L_DATA_REQ,
+        data=CEMILData.init_from_telegram(test_telegram),
     )
-    test_cemi_confirmation = CEMIFrame.init_from_telegram(
-        test_telegram,
+    test_cemi_confirmation = CEMIFrame(
         code=CEMIMessageCode.L_DATA_CON,
+        data=CEMILData.init_from_telegram(
+            test_telegram,
+        ),
     )
     task = asyncio.create_task(xknx.cemi_handler.send_telegram(test_telegram))
     await time_travel(0)
@@ -64,25 +66,25 @@ def test_incoming_cemi():
         destination_address=GroupAddress(1),
         payload=apci.GroupValueWrite(DPTArray((1,))),
     )
-    test_group_cemi = CEMIFrame.init_from_telegram(
-        test_telegram,
+    test_group_cemi = CEMIFrame(
         code=CEMIMessageCode.L_DATA_IND,
+        data=CEMILData.init_from_telegram(test_telegram),
     )
     xknx.cemi_handler.handle_cemi_frame(test_group_cemi)
     assert xknx.telegrams.qsize() == 1
 
     # L_DATA_CON and L_DATA_REQ should not be forwarded to the telegram queue or management
     with patch.object(xknx.cemi_handler, "telegram_received") as mock_telegram_received:
-        test_incoming_l_data_con = CEMIFrame.init_from_telegram(
-            test_telegram,
+        test_incoming_l_data_con = CEMIFrame(
             code=CEMIMessageCode.L_DATA_CON,
+            data=CEMILData.init_from_telegram(test_telegram),
         )
         xknx.cemi_handler.handle_cemi_frame(test_incoming_l_data_con)
         mock_telegram_received.assert_not_called()
 
-        test_incoming_l_data_req = CEMIFrame.init_from_telegram(
-            test_telegram,
+        test_incoming_l_data_req = CEMIFrame(
             code=CEMIMessageCode.L_DATA_REQ,
+            data=CEMILData.init_from_telegram(test_telegram),
         )
         xknx.cemi_handler.handle_cemi_frame(test_incoming_l_data_req)
         mock_telegram_received.assert_not_called()
@@ -112,8 +114,9 @@ def test_incoming_management_telegram(telegram):
     xknx.current_address = IndividualAddress("1.1.1")
 
     with patch.object(xknx.management, "process") as mock_management_process:
-        test_cemi = CEMIFrame.init_from_telegram(
-            telegram, code=CEMIMessageCode.L_DATA_IND
+        test_cemi = CEMIFrame(
+            code=CEMIMessageCode.L_DATA_IND,
+            data=CEMILData.init_from_telegram(telegram),
         )
         xknx.cemi_handler.handle_cemi_frame(test_cemi)
         mock_management_process.assert_called_once()
diff --git a/test/io_tests/request_response_tests/tunnelling_test.py b/test/io_tests/request_response_tests/tunnelling_test.py
index 068abfe9..25770bc0 100644
--- a/test/io_tests/request_response_tests/tunnelling_test.py
+++ b/test/io_tests/request_response_tests/tunnelling_test.py
@@ -1,7 +1,7 @@
 """Unit test for KNX/IP Tunnelling Request/Response."""
 from unittest.mock import patch
 
-from xknx.cemi import CEMIFrame
+from xknx.cemi import CEMIFrame, CEMILData, CEMIMessageCode
 from xknx.dpt import DPTArray
 from xknx.io.request_response import Tunnelling
 from xknx.io.transport import UDPTransport
@@ -24,11 +24,14 @@ 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))
-        raw_cemi = CEMIFrame.init_from_telegram(
-            Telegram(
-                destination_address=GroupAddress("1/2/3"),
-                payload=GroupValueWrite(DPTArray((0x1, 0x2, 0x3))),
-            )
+        raw_cemi = CEMIFrame(
+            code=CEMIMessageCode.L_DATA_IND,
+            data=CEMILData.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,
diff --git a/test/io_tests/routing_test.py b/test/io_tests/routing_test.py
index e81ff743..5cfa664d 100644
--- a/test/io_tests/routing_test.py
+++ b/test/io_tests/routing_test.py
@@ -3,7 +3,7 @@ import asyncio
 from unittest.mock import AsyncMock, Mock, call, patch
 
 from xknx import XKNX
-from xknx.cemi import CEMIFrame, CEMIMessageCode
+from xknx.cemi import CEMIFrame, CEMILData, CEMIMessageCode
 from xknx.io import Routing
 from xknx.io.const import DEFAULT_INDIVIDUAL_ADDRESS
 from xknx.io.routing import (
@@ -42,16 +42,19 @@ class TestRouting:
         # communication_channel_id: 0x02   sequence_counter: 0x81
         raw_ind = bytes.fromhex("0610 0530 0010 2900b06010fa10ff0080")
         _cemi = CEMIFrame.from_knx(raw_ind[6:])
-        test_telegram = _cemi.telegram
+        test_telegram = _cemi.data.telegram()
         test_telegram.direction = TelegramDirection.INCOMING
 
         response_telegram = Telegram(
             destination_address=IndividualAddress(test_telegram.source_address),
             tpci=tpci.TDisconnect(),
         )
-        response_cemi = CEMIFrame.init_from_telegram(
-            telegram=response_telegram,
-            src_addr=IndividualAddress("1.0.255"),
+        response_cemi = CEMIFrame(
+            code=CEMIMessageCode.L_DATA_IND,
+            data=CEMILData.init_from_telegram(
+                telegram=response_telegram,
+                src_addr=IndividualAddress("1.0.255"),
+            ),
         )
         response_frame = KNXIPFrame.init_from_body(
             RoutingIndication(raw_cemi=response_cemi.to_knx())
diff --git a/test/io_tests/secure_group_test.py b/test/io_tests/secure_group_test.py
index fafbeeef..f32d1479 100644
--- a/test/io_tests/secure_group_test.py
+++ b/test/io_tests/secure_group_test.py
@@ -2,7 +2,7 @@
 import asyncio
 from unittest.mock import Mock, patch
 
-from xknx.cemi import CEMIFrame
+from xknx.cemi import CEMIFrame, CEMILData, CEMIMessageCode
 from xknx.io.const import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT, XKNX_SERIAL_NUMBER
 from xknx.io.ip_secure import SecureGroup
 from xknx.knxip import HPAI, KNXIPFrame, RoutingIndication, SecureWrapper, TimerNotify
@@ -442,11 +442,14 @@ class TestSecureGroup:
         secure_timer = secure_group.secure_timer
         secure_timer.timer_authenticated = True
 
-        raw_test_cemi = CEMIFrame.init_from_telegram(
-            Telegram(
-                destination_address=GroupAddress("1/2/3"),
-                payload=apci.GroupValueRead(),
-            )
+        raw_test_cemi = CEMIFrame(
+            code=CEMIMessageCode.L_DATA_IND,
+            data=CEMILData.init_from_telegram(
+                Telegram(
+                    destination_address=GroupAddress("1/2/3"),
+                    payload=apci.GroupValueRead(),
+                )
+            ),
         ).to_knx()
 
         with patch.object(
diff --git a/test/io_tests/tunnel_test.py b/test/io_tests/tunnel_test.py
index df5da6fe..323f1503 100644
--- a/test/io_tests/tunnel_test.py
+++ b/test/io_tests/tunnel_test.py
@@ -6,7 +6,7 @@ from unittest.mock import AsyncMock, Mock, call, patch
 import pytest
 
 from xknx import XKNX
-from xknx.cemi import CEMIFrame, CEMIMessageCode
+from xknx.cemi import CEMIFrame, CEMILData, CEMIMessageCode
 from xknx.dpt import DPTArray
 from xknx.exceptions import CommunicationError
 from xknx.io import TCPTunnel, UDPTunnel
@@ -101,7 +101,7 @@ class TestUDPTunnel:
         raw_ind = bytes.fromhex("0610 0420 0014 04 02 81 00 2900b06010fa10ff0080")
 
         test_cemi = CEMIFrame.from_knx(raw_ind[10:])
-        test_telegram = test_cemi.telegram
+        test_telegram = test_cemi.data.telegram()
         test_telegram.direction = TelegramDirection.INCOMING
         self.tunnel.expected_sequence_number = 0x81
 
@@ -112,10 +112,12 @@ class TestUDPTunnel:
 
         self.tunnel.transport.data_received_callback(raw_ind, ("192.168.1.2", 3671))
         await asyncio.sleep(0)
-        response_cemi = CEMIFrame.init_from_telegram(
-            response_telegram,
+        response_cemi = CEMIFrame(
             code=CEMIMessageCode.L_DATA_REQ,
-            src_addr=IndividualAddress("1.0.255"),
+            data=CEMILData.init_from_telegram(
+                response_telegram,
+                src_addr=IndividualAddress("1.0.255"),
+            ),
         )
         assert send_cemi_mock.call_args_list == [
             call(response_cemi),
@@ -132,8 +134,9 @@ class TestUDPTunnel:
             destination_address=GroupAddress(1),
             payload=GroupValueWrite(DPTArray((1,))),
         )
-        cemi = CEMIFrame.init_from_telegram(
-            test_telegram, code=CEMIMessageCode.L_DATA_IND
+        cemi = CEMIFrame(
+            code=CEMIMessageCode.L_DATA_IND,
+            data=CEMILData.init_from_telegram(test_telegram),
         )
         test_frame = KNXIPFrame.init_from_body(
             TunnellingRequest(
@@ -181,10 +184,12 @@ class TestUDPTunnel:
             destination_address=GroupAddress(1),
             payload=GroupValueWrite(DPTArray((1,))),
         )
-        test_cemi = CEMIFrame.init_from_telegram(
-            test_telegram,
+        test_cemi = CEMIFrame(
             code=CEMIMessageCode.L_DATA_REQ,
-            src_addr=self.tunnel._src_address,
+            data=CEMILData.init_from_telegram(
+                test_telegram,
+                src_addr=self.tunnel._src_address,
+            ),
         )
         request = KNXIPFrame.init_from_body(
             TunnellingRequest(
@@ -196,8 +201,9 @@ class TestUDPTunnel:
         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_cemi = CEMIFrame(
+            code=CEMIMessageCode.L_DATA_CON,
+            data=CEMILData.init_from_telegram(test_telegram),
         )
         confirmation = KNXIPFrame.init_from_body(
             TunnellingRequest(
@@ -290,10 +296,12 @@ class TestUDPTunnel:
             destination_address=GroupAddress(1),
             payload=GroupValueWrite(DPTArray((1,))),
         )
-        test_cemi = CEMIFrame.init_from_telegram(
-            test_telegram,
+        test_cemi = CEMIFrame(
             code=CEMIMessageCode.L_DATA_REQ,
-            src_addr=IndividualAddress(7),
+            data=CEMILData.init_from_telegram(
+                test_telegram,
+                src_addr=IndividualAddress(7),
+            ),
         )
         test_telegram_frame = KNXIPFrame.init_from_body(
             TunnellingRequest(
diff --git a/test/knxip_tests/routing_indication_test.py b/test/knxip_tests/routing_indication_test.py
index 97197d17..3689e0db 100644
--- a/test/knxip_tests/routing_indication_test.py
+++ b/test/knxip_tests/routing_indication_test.py
@@ -1,11 +1,11 @@
 """Unit test for KNX/IP RountingIndication objects."""
 import time
 
-from xknx.cemi import CEMIFrame, CEMILData
-from xknx.dpt import DPTArray, DPTBinary, DPTTemperature, DPTTime
-from xknx.knxip import KNXIPFrame, KNXIPServiceType, RoutingIndication
+from xknx.cemi import CEMIFrame, CEMILData, CEMIMessageCode
+from xknx.dpt import DPTArray, DPTTime
+from xknx.knxip import KNXIPFrame, RoutingIndication
 from xknx.telegram import GroupAddress, IndividualAddress, Telegram
-from xknx.telegram.apci import GroupValueRead, GroupValueResponse, GroupValueWrite
+from xknx.telegram.apci import GroupValueWrite
 
 
 class TestKNXIPRountingIndication:
@@ -37,9 +37,12 @@ class TestKNXIPRountingIndication:
                 DPTArray(DPTTime().to_knx(time.strptime("13:23:42", "%H:%M:%S")))
             ),
         )
-        cemi = CEMIFrame.init_from_telegram(
-            telegram,
-            src_addr=IndividualAddress("1.2.2"),
+        cemi = CEMIFrame(
+            code=CEMIMessageCode.L_DATA_IND,
+            data=CEMILData.init_from_telegram(
+                telegram,
+                src_addr=IndividualAddress("1.2.2"),
+            ),
         )
         assert isinstance(cemi.data, CEMILData)
         cemi.data.hops = 5
diff --git a/test/knxip_tests/tunnelling_request_test.py b/test/knxip_tests/tunnelling_request_test.py
index 1d155585..21484e6e 100644
--- a/test/knxip_tests/tunnelling_request_test.py
+++ b/test/knxip_tests/tunnelling_request_test.py
@@ -1,7 +1,7 @@
 """Unit test for KNX/IP TunnellingRequest objects."""
 import pytest
 
-from xknx.cemi import CEMIFrame, CEMIMessageCode
+from xknx.cemi import CEMIFrame, CEMILData, CEMIMessageCode
 from xknx.dpt import DPTBinary
 from xknx.exceptions import CouldNotParseKNXIP
 from xknx.knxip import KNXIPFrame, TunnellingRequest
@@ -25,17 +25,19 @@ class TestKNXIPTunnellingRequest:
         assert isinstance(knxipframe.body.raw_cemi, bytes)
 
         incoming_cemi = CEMIFrame.from_knx(knxipframe.body.raw_cemi)
-        assert incoming_cemi.telegram == Telegram(
+        assert incoming_cemi.data.telegram() == Telegram(
             destination_address=GroupAddress("9/0/8"),
             payload=GroupValueWrite(DPTBinary(1)),
         )
 
-        outgoing_cemi = CEMIFrame.init_from_telegram(
-            Telegram(
-                destination_address=GroupAddress("9/0/8"),
-                payload=GroupValueWrite(DPTBinary(1)),
-            ),
+        outgoing_cemi = CEMIFrame(
             code=CEMIMessageCode.L_DATA_REQ,
+            data=CEMILData.init_from_telegram(
+                Telegram(
+                    destination_address=GroupAddress("9/0/8"),
+                    payload=GroupValueWrite(DPTBinary(1)),
+                ),
+            ),
         )
         tunnelling_request = TunnellingRequest(
             communication_channel_id=1,
diff --git a/test/secure_tests/data_secure_test.py b/test/secure_tests/data_secure_test.py
index 4439d746..b3314d66 100644
--- a/test/secure_tests/data_secure_test.py
+++ b/test/secure_tests/data_secure_test.py
@@ -105,7 +105,10 @@ class TestDataSecure:
             self.xknx.cemi_handler._l_data_confirmation_event.set()
             await task
 
-        test_cemi = CEMIFrame.init_from_telegram(test_telegram)
+        test_cemi = CEMIFrame(
+            code=CEMIMessageCode.L_DATA_IND,
+            data=CEMILData.init_from_telegram(test_telegram),
+        )
         with patch.object(
             self.data_secure,
             "received_cemi",
@@ -129,13 +132,15 @@ class TestDataSecure:
         """Test outgoing DataSecure group communication."""
         self.data_secure._sequence_number_sending = 160170101607
 
-        test_cemi = CEMIFrame.init_from_telegram(
-            Telegram(
-                destination_address=GroupAddress("0/4/0"),
-                payload=apci.GroupValueRead(),
-            ),
+        test_cemi = CEMIFrame(
             code=CEMIMessageCode.L_DATA_REQ,
-            src_addr=self.xknx.current_address,
+            data=CEMILData.init_from_telegram(
+                Telegram(
+                    destination_address=GroupAddress("0/4/0"),
+                    payload=apci.GroupValueRead(),
+                ),
+                src_addr=self.xknx.current_address,
+            ),
         )
         secured_frame_data = self.data_secure.outgoing_cemi(test_cemi.data)
         assert isinstance(secured_frame_data, CEMILData)
@@ -256,14 +261,16 @@ class TestDataSecure:
     def test_data_secure_group_receive_plain_frame(self):
         """Test incoming DataSecure group communication with plain frame."""
         src_addr = IndividualAddress("4.0.9")
-        test_cemi = CEMIFrame.init_from_telegram(
-            Telegram(
-                destination_address=GroupAddress("0/4/0"),
-                direction=TelegramDirection.INCOMING,
-                payload=apci.GroupValueResponse(DPTArray((116, 41, 41))),
-            ),
+        test_cemi = CEMIFrame(
             code=CEMIMessageCode.L_DATA_IND,
-            src_addr=src_addr,
+            data=CEMILData.init_from_telegram(
+                Telegram(
+                    destination_address=GroupAddress("0/4/0"),
+                    direction=TelegramDirection.INCOMING,
+                    payload=apci.GroupValueResponse(DPTArray((116, 41, 41))),
+                ),
+                src_addr=src_addr,
+            ),
         )
         assert src_addr in self.data_secure._individual_address_table
         with pytest.raises(
@@ -275,14 +282,16 @@ class TestDataSecure:
     def test_non_secure_group_receive_plain_frame(self):
         """Test incoming non-secure group communication with plain frame."""
         dst_addr = GroupAddress("1/2/3")
-        test_cemi = CEMIFrame.init_from_telegram(
-            Telegram(
-                destination_address=dst_addr,
-                direction=TelegramDirection.INCOMING,
-                payload=apci.GroupValueResponse(DPTArray((116, 41, 41))),
-            ),
+        test_cemi = CEMIFrame(
             code=CEMIMessageCode.L_DATA_IND,
-            src_addr=IndividualAddress("4.0.9"),
+            data=CEMILData.init_from_telegram(
+                Telegram(
+                    destination_address=dst_addr,
+                    direction=TelegramDirection.INCOMING,
+                    payload=apci.GroupValueResponse(DPTArray((116, 41, 41))),
+                ),
+                src_addr=IndividualAddress("4.0.9"),
+            ),
         )
         assert dst_addr not in self.data_secure._group_key_table
         assert self.data_secure.received_cemi(test_cemi.data) == test_cemi.data
@@ -290,14 +299,16 @@ class TestDataSecure:
     def test_non_secure_group_send_plain_frame(self):
         """Test outgoing non-secure group communication with plain frame."""
         dst_addr = GroupAddress("1/2/3")
-        test_cemi = CEMIFrame.init_from_telegram(
-            Telegram(
-                destination_address=dst_addr,
-                direction=TelegramDirection.OUTGOING,
-                payload=apci.GroupValueResponse(DPTArray((116, 41, 41))),
-            ),
+        test_cemi = CEMIFrame(
             code=CEMIMessageCode.L_DATA_REQ,
-            src_addr=self.xknx.current_address,
+            data=CEMILData.init_from_telegram(
+                Telegram(
+                    destination_address=dst_addr,
+                    direction=TelegramDirection.OUTGOING,
+                    payload=apci.GroupValueResponse(DPTArray((116, 41, 41))),
+                ),
+                src_addr=self.xknx.current_address,
+            ),
         )
         assert dst_addr not in self.data_secure._group_key_table
         assert self.data_secure.outgoing_cemi(test_cemi.data) == test_cemi.data
@@ -305,23 +316,25 @@ class TestDataSecure:
     def test_non_secure_individual_receive_plain_frame(self):
         """Test incoming non-secure group communication with plain frame."""
         src_addr = IndividualAddress("1.2.3")
-        test_cemi = CEMIFrame.init_from_telegram(
-            Telegram(
-                destination_address=self.xknx.current_address,
-                direction=TelegramDirection.INCOMING,
-                payload=apci.PropertyValueWrite(
-                    object_index=5,
-                    property_id=0x35,
-                    count=1,
-                    start_index=1,
-                    data=bytes.fromhex(
-                        "20 21 22 23 24 25 26 27 28 29 2A 2B 2C 2D 2E 2F"
+        test_cemi = CEMIFrame(
+            code=CEMIMessageCode.L_DATA_IND,
+            data=CEMILData.init_from_telegram(
+                Telegram(
+                    destination_address=self.xknx.current_address,
+                    direction=TelegramDirection.INCOMING,
+                    payload=apci.PropertyValueWrite(
+                        object_index=5,
+                        property_id=0x35,
+                        count=1,
+                        start_index=1,
+                        data=bytes.fromhex(
+                            "20 21 22 23 24 25 26 27 28 29 2A 2B 2C 2D 2E 2F"
+                        ),
                     ),
+                    tpci=tpci.TDataIndividual(),
                 ),
-                tpci=tpci.TDataIndividual(),
+                src_addr=IndividualAddress("4.0.9"),
             ),
-            code=CEMIMessageCode.L_DATA_IND,
-            src_addr=IndividualAddress("4.0.9"),
         )
         assert src_addr not in self.data_secure._individual_address_table
         assert self.data_secure.received_cemi(test_cemi.data) == test_cemi.data
@@ -329,23 +342,25 @@ class TestDataSecure:
     def test_non_secure_individual_send_plain_frame(self):
         """Test outgoing non-secure group communication with plain frame."""
         dst_addr = IndividualAddress("1.2.3")
-        test_cemi = CEMIFrame.init_from_telegram(
-            Telegram(
-                destination_address=dst_addr,
-                direction=TelegramDirection.INCOMING,
-                payload=apci.PropertyValueWrite(
-                    object_index=5,
-                    property_id=0x35,
-                    count=1,
-                    start_index=1,
-                    data=bytes.fromhex(
-                        "20 21 22 23 24 25 26 27 28 29 2A 2B 2C 2D 2E 2F"
+        test_cemi = CEMIFrame(
+            code=CEMIMessageCode.L_DATA_REQ,
+            data=CEMILData.init_from_telegram(
+                Telegram(
+                    destination_address=dst_addr,
+                    direction=TelegramDirection.INCOMING,
+                    payload=apci.PropertyValueWrite(
+                        object_index=5,
+                        property_id=0x35,
+                        count=1,
+                        start_index=1,
+                        data=bytes.fromhex(
+                            "20 21 22 23 24 25 26 27 28 29 2A 2B 2C 2D 2E 2F"
+                        ),
                     ),
+                    tpci=tpci.TDataIndividual(),
                 ),
-                tpci=tpci.TDataIndividual(),
+                src_addr=self.xknx.current_address,
             ),
-            code=CEMIMessageCode.L_DATA_REQ,
-            src_addr=self.xknx.current_address,
         )
         assert dst_addr not in self.data_secure._individual_address_table
         assert self.data_secure.outgoing_cemi(test_cemi.data) == test_cemi.data
@@ -360,10 +375,12 @@ class TestDataSecure:
             direction=TelegramDirection.OUTGOING,
             payload=apci.GroupValueWrite(DPTArray((1, 2))),
         )
-        test_cemi = CEMIFrame.init_from_telegram(
-            telegram=test_telegram,
-            src_addr=self.xknx.current_address,
+        test_cemi = CEMIFrame(
             code=CEMIMessageCode.L_DATA_REQ,
+            data=CEMILData.init_from_telegram(
+                telegram=test_telegram,
+                src_addr=self.xknx.current_address,
+            ),
         )
         scf = SecurityControlField(
             algorithm=SecurityAlgorithmIdentifier.CCM_AUTHENTICATION,
diff --git a/test/str_test.py b/test/str_test.py
index a1a402a7..c8cdfbd7 100644
--- a/test/str_test.py
+++ b/test/str_test.py
@@ -2,7 +2,13 @@
 from unittest.mock import patch
 
 from xknx import XKNX
-from xknx.cemi import CEMIFrame, CEMIMessageCode, CEMIMPropInfo, CEMIMPropReadResponse
+from xknx.cemi import (
+    CEMIFrame,
+    CEMILData,
+    CEMIMessageCode,
+    CEMIMPropInfo,
+    CEMIMPropReadResponse,
+)
 from xknx.devices import (
     BinarySensor,
     Climate,
@@ -645,12 +651,15 @@ class TestStringRepresentations:
 
     def test_cemi_ldata_frame(self):
         """Test string representation of KNX/IP CEMI Frame."""
-        cemi_frame = CEMIFrame.init_from_telegram(
-            Telegram(
-                destination_address=GroupAddress("1/2/5"),
-                payload=GroupValueWrite(DPTBinary(7)),
+        cemi_frame = CEMIFrame(
+            code=CEMIMessageCode.L_DATA_IND,
+            data=CEMILData.init_from_telegram(
+                telegram=Telegram(
+                    destination_address=GroupAddress("1/2/5"),
+                    payload=GroupValueWrite(DPTBinary(7)),
+                ),
+                src_addr=IndividualAddress("1.2.3"),
             ),
-            src_addr=IndividualAddress("1.2.3"),
         )
         assert (
             str(cemi_frame)