Repository: XKNX/xknx
Pull request #1139 (2023-01-08 22:15:06)

CEMIFrame refactor
<!--
  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.
-->
- Remove `CEMIFrame.telegram` setter in favour of `init_from_telegram()` staticmethod; convert `from_knx()` and `from_knx_data_link_layer()` to staticmethods returning a CEMIFrame.
- Remove default values for `CEMIFrame` constructor, require keyword arguments.

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
- [x] The changes are documented in the changelog (docs/changelog.md)

farmio: ```py
    def to_knx(self) -> bytes:
        ...
        if not isinstance(self.src_addr, IndividualAddress):
            raise ConversionError("src_addr invalid")
        if not isinstance(self.dst_addr, (GroupAddress, IndividualAddress)):
            raise ConversionError("dst_addr invalid")
```
This is not really needed anymore since it should be checked by mypy.
@marvin-w shall we remove it, or keep it?
codecov[bot]: # [Codecov](https://codecov.io/gh/XKNX/xknx/pull/1139?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
> Merging [#1139](https://codecov.io/gh/XKNX/xknx/pull/1139?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (7c134c7) into [main](https://codecov.io/gh/XKNX/xknx/commit/ec053b9bed1da9008c4cfc12c7966dcab1d0c82c?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (ec053b9) will **decrease** coverage by `0.00%`.
> The diff coverage is `100.00%`.

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

```diff
@@            Coverage Diff             @@
##             main    #1139      +/-   ##
==========================================
- Coverage   96.35%   96.34%   -0.01%     
==========================================
  Files         141      141              
  Lines        8685     8678       -7     
==========================================
- Hits         8368     8361       -7     
  Misses        317      317              
```


| [Impacted Files](https://codecov.io/gh/XKNX/xknx/pull/1139?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/cemi/cemi\_handler.py](https://codecov.io/gh/XKNX/xknx/pull/1139?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9jZW1pL2NlbWlfaGFuZGxlci5weQ==) | `100.00% <ø> (ø)` | |
| [xknx/cemi/cemi\_frame.py](https://codecov.io/gh/XKNX/xknx/pull/1139?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9jZW1pL2NlbWlfZnJhbWUucHk=) | `100.00% <100.00%> (ø)` | |
| [xknx/io/routing.py](https://codecov.io/gh/XKNX/xknx/pull/1139?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby9yb3V0aW5nLnB5) | `91.66% <100.00%> (-0.07%)` | :arrow_down: |
| [xknx/io/tunnel.py](https://codecov.io/gh/XKNX/xknx/pull/1139?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9pby90dW5uZWwucHk=) | `85.77% <100.00%> (-0.06%)` | :arrow_down: |


Patch:
diff --git a/docs/changelog.md b/docs/changelog.md
index bce3274a..44bad940 100644
--- a/docs/changelog.md
+++ b/docs/changelog.md
@@ -17,6 +17,8 @@ nav_order: 2
 ### Internals
 
 - Move `CEMIFrame`, `CEMIFlags` and `CEMIMessageCode` to xknx.cemi package.
+- Remove `CEMIFrame.telegram` setter in favour of `init_from_telegram()` staticmethod; convert `from_knx()` and `from_knx_data_link_layer()` to staticmethods returning a CEMIFrame.
+- Remove default values for `CEMIFrame` constructor.
 - Parse T_Data_Broadcast TPCI. Forward these telegrams to the Management class.
 - KNXIPHeader total_length is 2 bytes long. There are no reserved bytes.
 - Revert handling L_Data.req frames for incoming device management requests.
diff --git a/xknx/cemi/cemi_frame.py b/xknx/cemi/cemi_frame.py
index cebd9256..03ea95e0 100644
--- a/xknx/cemi/cemi_frame.py
+++ b/xknx/cemi/cemi_frame.py
@@ -16,7 +16,7 @@ from __future__ import annotations
 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, TDataBroadcast, TDataGroup
+from xknx.telegram.tpci import TPCI, TDataBroadcast
 
 from .const import CEMIFlags, CEMIMessageCode
 
@@ -26,19 +26,20 @@ class CEMIFrame:
 
     def __init__(
         self,
-        code: CEMIMessageCode = CEMIMessageCode.L_DATA_IND,
-        flags: int = 0,
-        src_addr: IndividualAddress | None = None,
-        dst_addr: GroupAddress | IndividualAddress | None = None,
-        tpci: TPCI | None = None,
-        payload: APCI | None = None,
+        *,
+        code: CEMIMessageCode,
+        flags: int,
+        src_addr: IndividualAddress,
+        dst_addr: GroupAddress | IndividualAddress,
+        tpci: TPCI,
+        payload: APCI | None,
     ):
         """Initialize CEMIFrame object."""
         self.code = code
         self.flags = flags
-        self.src_addr = src_addr or IndividualAddress(None)
-        self.dst_addr = dst_addr or GroupAddress(None)
-        self.tpci = tpci or TDataGroup()
+        self.src_addr = src_addr
+        self.dst_addr = dst_addr
+        self.tpci = tpci
         self.payload = payload
 
     @staticmethod
@@ -48,31 +49,7 @@ class CEMIFrame:
         src_addr: IndividualAddress | None = None,
     ) -> CEMIFrame:
         """Return CEMIFrame from a Telegram."""
-        cemi = CEMIFrame(
-            code=code,
-            src_addr=src_addr or IndividualAddress(None),
-        )
-        # dst_addr, payload and cmd are set by telegram.setter
-        cemi.telegram = telegram
-        return cemi
-
-    @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,
-        )
-
-    @telegram.setter
-    def telegram(self, telegram: Telegram) -> None:
-        """Set telegram."""
-        # TODO: Move to separate function, together with setting of
-        # CEMIMessageCode
-        self.flags = (
+        flags = (
             CEMIFlags.FRAME_TYPE_STANDARD
             | CEMIFlags.DO_NOT_REPEAT
             | CEMIFlags.BROADCAST
@@ -80,23 +57,38 @@ class CEMIFrame:
             | CEMIFlags.CONFIRM_NO_ERROR
             | CEMIFlags.HOP_COUNT_1ST
         )
-
         if isinstance(telegram.destination_address, GroupAddress):
-            self.flags |= CEMIFlags.DESTINATION_GROUP_ADDRESS
+            flags |= CEMIFlags.DESTINATION_GROUP_ADDRESS
             if isinstance(telegram.tpci, TDataBroadcast):
-                self.flags |= CEMIFlags.PRIORITY_SYSTEM
+                flags |= CEMIFlags.PRIORITY_SYSTEM
             else:
-                self.flags |= CEMIFlags.PRIORITY_LOW
+                flags |= CEMIFlags.PRIORITY_LOW
         elif isinstance(telegram.destination_address, IndividualAddress):
-            self.flags |= (
+            flags |= (
                 CEMIFlags.DESTINATION_INDIVIDUAL_ADDRESS | CEMIFlags.PRIORITY_SYSTEM
             )
         else:
             raise TypeError()
 
-        self.dst_addr = telegram.destination_address
-        self.tpci = telegram.tpci
-        self.payload = telegram.payload
+        return CEMIFrame(
+            code=code,
+            flags=flags,
+            src_addr=src_addr or telegram.source_address,
+            dst_addr=telegram.destination_address,
+            tpci=telegram.tpci,
+            payload=telegram.payload,
+        )
+
+    @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,
+        )
 
     def set_hops(self, hops: int) -> None:
         """Set hops."""
@@ -113,27 +105,29 @@ class CEMIFrame:
             return 10
         raise TypeError("Data TPDU must have a payload; control TPDU must not.")
 
-    def from_knx(self, raw: bytes) -> int:
+    @staticmethod
+    def from_knx(raw: bytes) -> CEMIFrame:
         """Parse/deserialize from KNX/IP raw data."""
         try:
-            self.code = CEMIMessageCode(raw[0])
+            code = CEMIMessageCode(raw[0])
         except ValueError:
             raise UnsupportedCEMIMessage(
                 f"CEMIMessageCode not implemented: {raw[0]} in CEMI: {raw.hex()}"
             )
 
-        if self.code not in (
+        if code in (
             CEMIMessageCode.L_DATA_IND,
             CEMIMessageCode.L_DATA_REQ,
             CEMIMessageCode.L_DATA_CON,
         ):
-            raise UnsupportedCEMIMessage(
-                f"Could not handle CEMIMessageCode: {self.code} / {raw[0]} in CEMI: {raw.hex()}"
-            )
+            return CEMIFrame.from_knx_data_link_layer(raw, code)
 
-        return self.from_knx_data_link_layer(raw)
+        raise UnsupportedCEMIMessage(
+            f"Could not handle CEMIMessageCode: {code} / {raw[0]} in CEMI: {raw.hex()}"
+        )
 
-    def from_knx_data_link_layer(self, cemi: bytes) -> int:
+    @staticmethod
+    def from_knx_data_link_layer(cemi: bytes, code: CEMIMessageCode) -> CEMIFrame:
         """Parse L_DATA_IND, CEMIMessageCode.L_DATA_REQ, CEMIMessageCode.L_DATA_CON."""
         if len(cemi) < 10:
             raise UnsupportedCEMIMessage(
@@ -145,26 +139,24 @@ class CEMIFrame:
         # Additional information is not yet parsed.
         addil = cemi[1]
         # Control field 1 and Control field 2 - first 2 octets after Additional information
-        self.flags = cemi[2 + addil] * 256 + cemi[3 + addil]
+        flags = cemi[2 + addil] * 256 + cemi[3 + addil]
 
-        self.src_addr = IndividualAddress(cemi[4 + addil] * 256 + cemi[5 + addil])
+        src_addr = IndividualAddress(cemi[4 + addil] * 256 + cemi[5 + addil])
 
-        dst_is_group_address = bool(self.flags & CEMIFlags.DESTINATION_GROUP_ADDRESS)
-        dst_raw_address = cemi[6 + addil] * 256 + cemi[7 + addil]
-        self.dst_addr = (
-            GroupAddress(dst_raw_address)
-            if dst_is_group_address
-            else IndividualAddress(dst_raw_address)
+        _dst_is_group_address = bool(flags & CEMIFlags.DESTINATION_GROUP_ADDRESS)
+        _dst_raw_address = cemi[6 + addil] * 256 + cemi[7 + addil]
+        dst_addr: GroupAddress | IndividualAddress = (
+            GroupAddress(_dst_raw_address)
+            if _dst_is_group_address
+            else IndividualAddress(_dst_raw_address)
         )
 
-        npdu_len = cemi[8 + addil]
-
-        tpdu = cemi[9 + addil :]
-        apdu = bytes([tpdu[0] & 0b11]) + tpdu[1:]  # clear TPCI bits
-
-        if len(apdu) != (npdu_len + 1):  # TCPI octet not included in NPDU length
+        _npdu_len = cemi[8 + addil]
+        _tpdu = cemi[9 + addil :]
+        _apdu = bytes([_tpdu[0] & 0b11]) + _tpdu[1:]  # clear TPCI bits
+        if len(_apdu) != (_npdu_len + 1):  # TCPI octet not included in NPDU length
             raise UnsupportedCEMIMessage(
-                f"APDU LEN should be {npdu_len} but is {len(apdu) - 1} in CEMI: {cemi.hex()}"
+                f"APDU LEN should be {_npdu_len} but is {len(_apdu) - 1} in CEMI: {cemi.hex()}"
             )
 
         # TPCI (transport layer control information)
@@ -172,30 +164,45 @@ class CEMIFrame:
         # - no control bit set (data) -> First 6 bit
         # APCI (application layer control information) -> Last  10 bit of TPCI/APCI
         try:
-            self.tpci = TPCI.resolve(
-                raw_tpci=tpdu[0],
-                dst_is_group_address=dst_is_group_address,
-                dst_is_zero=not dst_raw_address,
+            tpci = TPCI.resolve(
+                raw_tpci=_tpdu[0],
+                dst_is_group_address=_dst_is_group_address,
+                dst_is_zero=not _dst_raw_address,
             )
         except ConversionError as err:
-            raise UnsupportedCEMIMessage(f"TPCI not supported: {tpdu[0]:#10b}") from err
+            raise UnsupportedCEMIMessage(
+                f"TPCI not supported: {_tpdu[0]:#10b}"
+            ) from err
 
-        if self.tpci.control:
-            if npdu_len:
+        if tpci.control:
+            if _npdu_len:
                 raise UnsupportedCEMIMessage(
-                    f"Invalid length for control TPDU {self.tpci}: {npdu_len}"
+                    f"Invalid length for control TPDU {tpci}: {_npdu_len}"
                 )
-            return 10 + addil
+            return CEMIFrame(
+                code=code,
+                flags=flags,
+                src_addr=src_addr,
+                dst_addr=dst_addr,
+                tpci=tpci,
+                payload=None,
+            )
 
-        _apci = apdu[0] * 256 + apdu[1]
+        _apci = _apdu[0] * 256 + _apdu[1]
         try:
-            self.payload = APCI.resolve_apci(_apci)
+            payload = APCI.resolve_apci(_apci)
         except ConversionError as err:
             raise UnsupportedCEMIMessage(f"APCI not supported: {_apci:#012b}") from err
+        payload.from_knx(_apdu)
 
-        self.payload.from_knx(apdu)
-
-        return 10 + addil + npdu_len
+        return CEMIFrame(
+            code=code,
+            flags=flags,
+            src_addr=src_addr,
+            dst_addr=dst_addr,
+            tpci=tpci,
+            payload=payload,
+        )
 
     def to_knx(self) -> bytes:
         """Serialize to KNX/IP raw data."""
@@ -205,17 +212,12 @@ class CEMIFrame:
         else:
             if not isinstance(self.payload, APCI):
                 raise ConversionError(
-                    f"Invalid payload set for data TPDU: {self.payload.__class__}"
+                    f"Invalid payload set for data TPDU: {type(self.payload)}"
                 )
             tpdu = self.payload.to_knx()
             tpdu[0] |= self.tpci.to_knx()
             npdu_len = self.payload.calculated_length()
 
-        if not isinstance(self.src_addr, IndividualAddress):
-            raise ConversionError("src_addr invalid")
-        if not isinstance(self.dst_addr, (GroupAddress, IndividualAddress)):
-            raise ConversionError("dst_addr invalid")
-
         return (
             bytes(
                 (
diff --git a/xknx/cemi/cemi_handler.py b/xknx/cemi/cemi_handler.py
index 97205c12..6188eb74 100644
--- a/xknx/cemi/cemi_handler.py
+++ b/xknx/cemi/cemi_handler.py
@@ -39,7 +39,9 @@ class CEMIHandler:
         cemi = CEMIFrame.init_from_telegram(
             telegram=telegram,
             code=CEMIMessageCode.L_DATA_REQ,
-            src_addr=self.xknx.current_address,  # TODO: if Telegram address is 0 ?
+            src_addr=(
+                self.xknx.current_address if telegram.source_address.raw == 0 else None
+            ),
         )
         self._l_data_confirmation_event.clear()
         logger.debug("Outgoing CEMI: %s", cemi)
diff --git a/xknx/io/routing.py b/xknx/io/routing.py
index 603f56ea..bf844f2a 100644
--- a/xknx/io/routing.py
+++ b/xknx/io/routing.py
@@ -257,9 +257,8 @@ class Routing(Interface):
 
     def _handle_routing_indication(self, routing_indication: RoutingIndication) -> None:
         """Handle incoming RoutingIndication."""
-        cemi = CEMIFrame()
         try:
-            cemi.from_knx(routing_indication.raw_cemi)
+            cemi = CEMIFrame.from_knx(routing_indication.raw_cemi)
         except UnsupportedCEMIMessage as unsupported_cemi_err:
             logger.warning("CEMI not supported: %s", unsupported_cemi_err)
             return
diff --git a/xknx/io/tunnel.py b/xknx/io/tunnel.py
index 4283f83f..7ca3d2da 100644
--- a/xknx/io/tunnel.py
+++ b/xknx/io/tunnel.py
@@ -307,9 +307,8 @@ class _Tunnel(Interface):
         self, tunneling_request: TunnellingRequest
     ) -> None:
         """Handle incoming tunnel request."""
-        cemi = CEMIFrame()
         try:
-            cemi.from_knx(tunneling_request.raw_cemi)
+            cemi = CEMIFrame.from_knx(tunneling_request.raw_cemi)
         except UnsupportedCEMIMessage as unsupported_cemi_err:
             logger.warning("CEMI not supported: %s", unsupported_cemi_err)
             return


Test patch:
diff --git a/requirements/testing.txt b/requirements/testing.txt
index 470d71c6..ed75c22f 100644
--- a/requirements/testing.txt
+++ b/requirements/testing.txt
@@ -3,11 +3,11 @@ pre-commit==2.21.0
 isort==5.11.4
 flake8==6.0.0
 flake8-isort==6.0.0
-pydocstyle==6.2.2
+pydocstyle==6.2.3
 pylint==2.15.9
 pytest==7.2.0
 pytest-asyncio==0.20.3
 pytest-cov==4.0.0
 setuptools==65.6.3
-tox==4.2.4
+tox==4.2.6
 mypy==0.991
diff --git a/test/cemi_tests/cemi_frame_test.py b/test/cemi_tests/cemi_frame_test.py
index 33f9a502..3abf04a3 100644
--- a/test/cemi_tests/cemi_frame_test.py
+++ b/test/cemi_tests/cemi_frame_test.py
@@ -1,8 +1,7 @@
 """Tests for the CEMIFrame object."""
 import pytest
 
-from xknx.cemi import CEMIFlags, CEMIMessageCode
-from xknx.cemi.cemi_frame import CEMIFrame
+from xknx.cemi import CEMIFlags, CEMIFrame, CEMIMessageCode
 from xknx.exceptions import ConversionError, UnsupportedCEMIMessage
 from xknx.telegram import GroupAddress, IndividualAddress, Telegram
 from xknx.telegram.apci import GroupValueRead
@@ -32,15 +31,13 @@ def get_data(code, adil, flags, src, dst, npdu_len, tpci_apci, payload):
 def test_valid_command():
     """Test for valid frame parsing."""
     raw = get_data(0x29, 0, 0x0080, 1, 1, 1, 0, [])
-    frame = CEMIFrame()
-    packet_len = frame.from_knx(raw)
+    frame = CEMIFrame.from_knx(raw)
     assert frame.code == CEMIMessageCode.L_DATA_IND
     assert frame.flags == 0x0080
     assert frame.src_addr == IndividualAddress(1)
     assert frame.dst_addr == GroupAddress(1)
     assert frame.payload == GroupValueRead()
     assert frame.tpci == TDataGroup()
-    assert packet_len == 11
     assert frame.calculated_length() == 11
     assert frame.to_knx() == raw
 
@@ -48,15 +45,13 @@ def test_valid_command():
 def test_valid_tpci_control():
     """Test for valid tpci control."""
     raw = bytes((0x29, 0, 0, 0, 0, 0, 0, 0, 0, 0x80))
-    frame = CEMIFrame()
-    packet_len = frame.from_knx(raw)
+    frame = CEMIFrame.from_knx(raw)
     assert frame.code == CEMIMessageCode.L_DATA_IND
     assert frame.flags == 0
     assert frame.payload is None
     assert frame.src_addr == IndividualAddress(0)
     assert frame.dst_addr == IndividualAddress(0)
     assert frame.tpci == TConnect()
-    assert packet_len == 10
     assert frame.calculated_length() == 10
     assert frame.to_knx() == raw
 
@@ -80,52 +75,26 @@ def test_valid_tpci_control():
 )
 def test_invalid_tpci_apci(raw, err_msg):
     """Test for invalid APCIService."""
-    frame = CEMIFrame()
     with pytest.raises(UnsupportedCEMIMessage, match=err_msg):
-        frame.from_knx_data_link_layer(raw)
+        CEMIFrame.from_knx_data_link_layer(raw, code=CEMIMessageCode.L_DATA_IND)
 
 
 def test_invalid_apdu_len():
     """Test for invalid apdu len."""
-    frame = CEMIFrame()
     with pytest.raises(UnsupportedCEMIMessage, match=r".*APDU LEN should be .*"):
-        frame.from_knx(get_data(0x29, 0, 0, 0, 0, 2, 0, []))
-
-
-def test_invalid_src_addr():
-    """Test for invalid src addr."""
-    frame = CEMIFrame()
-    frame.code = CEMIMessageCode.L_DATA_IND
-    frame.flags = 0
-    frame.payload = GroupValueRead()
-    frame.src_addr = GroupAddress(0)
-    frame.dst_addr = IndividualAddress(0)
-
-    with pytest.raises(ConversionError, match=r"src_addr invalid"):
-        frame.to_knx()
-
-
-def test_invalid_dst_addr():
-    """Test for invalid dst addr."""
-    frame = CEMIFrame()
-    frame.code = CEMIMessageCode.L_DATA_IND
-    frame.flags = 0
-    frame.payload = GroupValueRead()
-    frame.src_addr = IndividualAddress(0)
-    frame.dst_addr = None
-
-    with pytest.raises(ConversionError, match=r"dst_addr invalid"):
-        frame.to_knx()
+        CEMIFrame.from_knx(get_data(0x29, 0, 0, 0, 0, 2, 0, []))
 
 
 def test_invalid_payload():
     """Test for having wrong payload set."""
-    frame = CEMIFrame()
-    frame.code = CEMIMessageCode.L_DATA_IND
-    frame.flags = 0
-    frame.payload = None
-    frame.src_addr = IndividualAddress(0)
-    frame.dst_addr = IndividualAddress(0)
+    frame = CEMIFrame(
+        code=CEMIMessageCode.L_DATA_IND,
+        flags=0,
+        src_addr=IndividualAddress(0),
+        dst_addr=IndividualAddress(0),
+        tpci=TDataGroup(),
+        payload=None,
+    )
 
     with pytest.raises(TypeError):
         frame.calculated_length()
@@ -136,53 +105,47 @@ def test_invalid_payload():
 
 def test_from_knx_with_not_handleable_cemi():
     """Test for having unhandlebale cemi set."""
-    frame = CEMIFrame()
     with pytest.raises(
         UnsupportedCEMIMessage, match=r".*CEMIMessageCode not implemented:.*"
     ):
-        frame.from_knx(get_data(0x30, 0, 0, 0, 0, 2, 0, []))
+        CEMIFrame.from_knx(get_data(0x30, 0, 0, 0, 0, 2, 0, []))
 
 
 def test_from_knx_with_not_implemented_cemi():
     """Test for having not implemented CEMI set."""
-    frame = CEMIFrame()
     with pytest.raises(
         UnsupportedCEMIMessage, match=r".*Could not handle CEMIMessageCode:.*"
     ):
-        frame.from_knx(
+        CEMIFrame.from_knx(
             get_data(CEMIMessageCode.L_BUSMON_IND.value, 0, 0, 0, 0, 2, 0, [])
         )
 
 
 def test_invalid_invalid_len():
     """Test for invalid cemi len."""
-    frame = CEMIFrame()
     with pytest.raises(UnsupportedCEMIMessage, match=r".*CEMI too small.*"):
-        frame.from_knx_data_link_layer(get_data(0x29, 0, 0, 0, 0, 2, 0, [])[:5])
+        CEMIFrame.from_knx_data_link_layer(
+            get_data(0x29, 0, 0, 0, 0, 2, 0, [])[:5],
+            code=CEMIMessageCode.L_DATA_IND,
+        )
 
 
 def test_from_knx_group_address():
     """Test conversion for a cemi with a group address as destination."""
-    frame = CEMIFrame()
-    frame.from_knx(get_data(0x29, 0, 0x80, 0, 0, 1, 0, []))
-
+    frame = CEMIFrame.from_knx(get_data(0x29, 0, 0x80, 0, 0, 1, 0, []))
     assert frame.dst_addr == GroupAddress(0)
 
 
 def test_from_knx_individual_address():
     """Test conversion for a cemi with a individual address as destination."""
-    frame = CEMIFrame()
-    frame.from_knx(get_data(0x29, 0, 0x00, 0, 0, 1, 0, []))
-
+    frame = CEMIFrame.from_knx(get_data(0x29, 0, 0x00, 0, 0, 1, 0, []))
     assert frame.dst_addr == IndividualAddress(0)
 
 
 def test_telegram_group_address():
     """Test telegram conversion flags with a group address."""
-    frame = CEMIFrame()
     _telegram = Telegram(destination_address=GroupAddress(1))
-    # test CEMIFrame.telegram setter
-    frame.telegram = _telegram
+    frame = CEMIFrame.init_from_telegram(_telegram)
     assert frame.flags & 0x0080 == CEMIFlags.DESTINATION_GROUP_ADDRESS
     assert frame.flags & 0x0C00 == CEMIFlags.PRIORITY_LOW
     # test CEMIFrame.telegram property
@@ -191,10 +154,8 @@ def test_telegram_group_address():
 
 def test_telegram_broadcast():
     """Test telegram conversion flags with a group address."""
-    frame = CEMIFrame()
     _telegram = Telegram(destination_address=GroupAddress(0))
-    # test CEMIFrame.telegram setter
-    frame.telegram = _telegram
+    frame = CEMIFrame.init_from_telegram(_telegram)
     assert frame.flags & 0x0080 == CEMIFlags.DESTINATION_GROUP_ADDRESS
     assert frame.flags & 0x0C00 == CEMIFlags.PRIORITY_SYSTEM
     assert frame.tpci == TDataBroadcast()
@@ -204,10 +165,8 @@ def test_telegram_broadcast():
 
 def test_telegram_individual_address():
     """Test telegram conversion flags with a individual address."""
-    frame = CEMIFrame()
     _telegram = Telegram(destination_address=IndividualAddress(0), tpci=TConnect())
-    # test CEMIFrame.telegram setter
-    frame.telegram = _telegram
+    frame = CEMIFrame.init_from_telegram(_telegram)
     assert frame.flags & 0x0080 == CEMIFlags.DESTINATION_INDIVIDUAL_ADDRESS
     assert frame.flags & 0x0C00 == CEMIFlags.PRIORITY_SYSTEM
     assert frame.flags & 0x0200 == CEMIFlags.NO_ACK_REQUESTED
@@ -217,6 +176,5 @@ def test_telegram_individual_address():
 
 def test_telegram_unsupported_address():
     """Test telegram conversion flags with an unsupported address."""
-    frame = CEMIFrame()
     with pytest.raises(TypeError):
-        frame.telegram = Telegram(destination_address=object())
+        CEMIFrame.init_from_telegram(Telegram(destination_address=object()))
diff --git a/test/io_tests/routing_test.py b/test/io_tests/routing_test.py
index 571cde41..9f404ca4 100644
--- a/test/io_tests/routing_test.py
+++ b/test/io_tests/routing_test.py
@@ -41,8 +41,7 @@ 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:])
+        _cemi = CEMIFrame.from_knx(raw_ind[6:])
         test_telegram = _cemi.telegram
         test_telegram.direction = TelegramDirection.INCOMING
 
diff --git a/test/io_tests/tunnel_test.py b/test/io_tests/tunnel_test.py
index 2819c310..819e4ba4 100644
--- a/test/io_tests/tunnel_test.py
+++ b/test/io_tests/tunnel_test.py
@@ -72,8 +72,7 @@ class TestUDPTunnel:
     @patch("xknx.io.UDPTunnel._send_tunnelling_ack")
     async def test_tunnel_request_received(self, send_ack_mock, raw):
         """Test Tunnel for calling send_ack on normal frames."""
-        test_cemi = CEMIFrame()
-        test_cemi.from_knx(raw[10:])
+        test_cemi = CEMIFrame.from_knx(raw[10:])
         self.tunnel.expected_sequence_number = 0x21
 
         self.tunnel.transport.data_received_callback(raw, ("192.168.1.2", 3671))
@@ -97,8 +96,7 @@ class TestUDPTunnel:
         # communication_channel_id: 0x02   sequence_counter: 0x81
         raw_ind = bytes.fromhex("0610 0420 0014 04 02 81 00 2900b06010fa10ff0080")
 
-        test_cemi = CEMIFrame()
-        test_cemi.from_knx(raw_ind[10:])
+        test_cemi = CEMIFrame.from_knx(raw_ind[10:])
         test_telegram = test_cemi.telegram
         test_telegram.direction = TelegramDirection.INCOMING
         self.tunnel.expected_sequence_number = 0x81
diff --git a/test/knxip_tests/routing_indication_test.py b/test/knxip_tests/routing_indication_test.py
index 35b5f5a6..04126f2f 100644
--- a/test/knxip_tests/routing_indication_test.py
+++ b/test/knxip_tests/routing_indication_test.py
@@ -37,8 +37,10 @@ class TestKNXIPRountingIndication:
                 DPTArray(DPTTime().to_knx(time.strptime("13:23:42", "%H:%M:%S")))
             ),
         )
-        cemi = CEMIFrame(src_addr=IndividualAddress("1.2.2"))
-        cemi.telegram = telegram
+        cemi = CEMIFrame.init_from_telegram(
+            telegram,
+            src_addr=IndividualAddress("1.2.2"),
+        )
         cemi.set_hops(5)
         routing_indication = RoutingIndication(raw_cemi=cemi.to_knx())
         knxipframe = KNXIPFrame.init_from_body(routing_indication)
diff --git a/test/knxip_tests/tunnelling_request_test.py b/test/knxip_tests/tunnelling_request_test.py
index 220a529a..1d155585 100644
--- a/test/knxip_tests/tunnelling_request_test.py
+++ b/test/knxip_tests/tunnelling_request_test.py
@@ -24,17 +24,18 @@ class TestKNXIPTunnellingRequest:
         assert knxipframe.body.sequence_counter == 23
         assert isinstance(knxipframe.body.raw_cemi, bytes)
 
-        incoming_cemi = CEMIFrame()
-        incoming_cemi.from_knx(knxipframe.body.raw_cemi)
+        incoming_cemi = CEMIFrame.from_knx(knxipframe.body.raw_cemi)
         assert incoming_cemi.telegram == Telegram(
             destination_address=GroupAddress("9/0/8"),
             payload=GroupValueWrite(DPTBinary(1)),
         )
 
-        outgoing_cemi = CEMIFrame(code=CEMIMessageCode.L_DATA_REQ)
-        outgoing_cemi.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)),
+            ),
+            code=CEMIMessageCode.L_DATA_REQ,
         )
         tunnelling_request = TunnellingRequest(
             communication_channel_id=1,
diff --git a/test/str_test.py b/test/str_test.py
index 8214fdf2..3e9af0e0 100644
--- a/test/str_test.py
+++ b/test/str_test.py
@@ -641,11 +641,12 @@ class TestStringRepresentations:
 
     def test_cemi_frame(self):
         """Test string representation of KNX/IP CEMI Frame."""
-        cemi_frame = CEMIFrame()
-        cemi_frame.src_addr = IndividualAddress("1.2.3")
-        cemi_frame.telegram = Telegram(
-            destination_address=GroupAddress("1/2/5"),
-            payload=GroupValueWrite(DPTBinary(7)),
+        cemi_frame = CEMIFrame.init_from_telegram(
+            Telegram(
+                destination_address=GroupAddress("1/2/5"),
+                payload=GroupValueWrite(DPTBinary(7)),
+            ),
+            src_addr=IndividualAddress("1.2.3"),
         )
         assert (
             str(cemi_frame)