Test coverage of framer 100%.

pymodbus/framer/rtu.py

@@ -111,7 +111,8 @@ def decode(self, data: bytes) -> tuple[int, int, int, bytes]:
             if not (pdu_class := self.decoder.lookupPduClass(data[used_len:])):
                 continue
             if not (size := pdu_class.calculateRtuFrameSize(data[used_len:])):
-                size = data_len +1
+                Log.debug("Frame - rtu_byte_count_pos wrong")
+                return 0, dev_id, 0, self.EMPTY
             if data_len < used_len +size:
                 Log.debug("Frame - not ready")
                 return 0, dev_id, 0, self.EMPTY

pymodbus/pdu/__init__.py

@@ -1,12 +1,24 @@
 """Framer."""
 __all__ = [
     "DecodePDU",
+    "DiagnosticBase",
     "ExceptionResponse",
     "FileRecord",
     "ModbusPDU",
+    "ReadCoilsRequest",
+    "ReadDeviceInformationRequest",
+    "ReadExceptionStatusRequest",
+    "ReadHoldingRegistersRequest",
 ]
 
+from .bit_message import ReadCoilsRequest
 from .decoders import DecodePDU
+from .diag_message import DiagnosticBase
 from .exceptionresponse import ExceptionResponse
 from .file_message import FileRecord
+from .mei_message import ReadDeviceInformationRequest
+from .other_message import ReadExceptionStatusRequest
 from .pdu import ModbusPDU
+from .register_message import ReadHoldingRegistersRequest
+
+

test/framer/test_extras.py

@@ -1,5 +1,9 @@
 """Test transaction."""
+from unittest.mock import patch
 
+import pytest
+
+from pymodbus.exceptions import ModbusIOException
 from pymodbus.framer import (
     FramerAscii,
     FramerRTU,
@@ -12,7 +16,7 @@
 TEST_MESSAGE = b"\x7b\x01\x03\x00\x00\x00\x05\x85\xC9\x7d"
 
 
-class TestExtas:
+class TestExtras:
     """Test for the framer module."""
 
     client = None
@@ -73,6 +77,26 @@ def test_tcp_framer_transaction_short(self):
         assert used_len == len(msg1) + len(msg2)
         assert pdu.function_code.to_bytes(1,'big') + pdu.encode() == msg2[7:]
 
+    def test_tcp_framer_transaction_wrong_id(self):
+        """Test a half completed tcp frame transaction."""
+        msg = b"\x00\x01\x12\x34\x00\x06\xff\x02\x01\x02\x00\x08"
+        used_len, pdu = self._tcp.handleFrame(msg, 1, 0)
+        assert not pdu
+        assert used_len == len(msg)
+
+    def test_tcp_framer_transaction_wrong_tid(self):
+        """Test a half completed tcp frame transaction."""
+        msg = b"\x00\x01\x12\x34\x00\x06\xff\x02\x01\x02\x00\x08"
+        used_len, pdu = self._tcp.handleFrame(msg, 0, 10)
+        assert not pdu
+        assert used_len == len(msg)
+
+    def test_tcp_framer_transaction_wrong_fc(self):
+        """Test a half completed tcp frame transaction."""
+        msg = b"\x00\x01\x12\x34\x00\x06\xff\x70\x01\x02\x00\x08"
+        with pytest.raises(ModbusIOException):
+            self._tcp.handleFrame(msg, 0, 0)
+
     def test_tls_incoming_packet(self):
         """Framer tls incoming packet."""
         msg = b"\x00\x01\x12\x34\x00\x06\xff\x02\x01\x02\x00\x08"
@@ -85,6 +109,32 @@ def test_rtu_process_incoming_packets(self):
         _, pdu = self._rtu.handleFrame(msg, 0, 0)
         assert pdu
 
+    def test_rtu_short_packets(self):
+        """Test rtu process incoming packets."""
+        msg1 = b"\x00\x01"
+        msg2 = b"\x00\x00\x00\x01\xfc\x1b"
+        used_len, pdu = self._rtu.handleFrame(msg1, 0, 0)
+        assert not used_len
+        assert not pdu
+        used_len, pdu = self._rtu.handleFrame(msg1+msg2, 0, 0)
+        assert used_len == len(msg1) + len(msg2)
+        assert pdu
+
+    def test_rtu_calculate(self):
+        """Test rtu process incoming packets."""
+        msg = b"\x00\x01\x00\x00\x00\x01\xfc\x1b"
+        with patch("pymodbus.pdu.ReadCoilsRequest.calculateRtuFrameSize", return_value=0):
+            used_len, pdu = self._rtu.handleFrame(msg, 0, 0)
+            assert not used_len
+            assert not pdu
+
+    def test_rtu_wrong_fc(self):
+        """Test rtu process incoming packets."""
+        msg = b"\x00\x70\x00\x00\x00\x71\xfc\x1b"
+        used_len, pdu = self._rtu.handleFrame(msg, 0, 0)
+        assert not pdu
+        assert not used_len
+
     def test_ascii_process_incoming_packets(self):
         """Test ascii process incoming packet."""
         msg = b":F7031389000A60\r\n"

test/framer/test_framer.py

@@ -470,7 +470,3 @@ def test_framer_encode(self, test_framer, msg):
 
             actual = test_framer.buildFrame(message)
             assert msg == actual
-
-
-
-#    @pytest.mark.parametrize(("entry"), list(FramerType))