Test coverage pdu 100%. (#2755)

Author: janiversen

pymodbus/pdu/decoders.py

@@ -6,7 +6,8 @@
 from pymodbus.exceptions import MessageRegisterException, ModbusException
 from pymodbus.logging import Log
 
-from .pdu import ExceptionResponse, ModbusPDU
+from .exceptionresponse import ExceptionResponse
+from .pdu import ModbusPDU
 
 
 class DecodePDU:

pymodbus/pdu/pdu.py

@@ -111,51 +111,6 @@ def calculateRtuFrameSize(cls, data: bytes) -> int:
         )
 
 
-class ExceptionResponse(ModbusPDU):
-    """Base class for a modbus exception PDU."""
-
-    rtu_frame_size = 5
-
-    ILLEGAL_FUNCTION = 0x01
-    ILLEGAL_ADDRESS = 0x02
-    ILLEGAL_VALUE = 0x03
-    DEVICE_FAILURE = 0x04
-    ACKNOWLEDGE = 0x05
-    DEVICE_BUSY = 0x06
-    NEGATIVE_ACKNOWLEDGE = 0x07
-    MEMORY_PARITY_ERROR = 0x08
-    GATEWAY_PATH_UNAVIABLE = 0x0A
-    GATEWAY_NO_RESPONSE = 0x0B
-
-    def __init__(
-            self,
-            function_code: int,
-            exception_code: int = 0,
-            device_id: int = 1,
-            transaction: int = 0) -> None:
-        """Initialize the modbus exception response."""
-        super().__init__(transaction_id=transaction, dev_id=device_id)
-        self.function_code = function_code | 0x80
-        self.exception_code = exception_code
-
-    def __str__(self) -> str:
-        """Build a representation of an exception response."""
-        return (
-            f"{self.__class__.__name__}("
-            f"dev_id={self.dev_id}, "
-            f"function_code={self.function_code}, "
-            f"exception_code={self.exception_code})"
-        )
-
-    def encode(self) -> bytes:
-        """Encode a modbus exception response."""
-        return struct.pack(">B", self.exception_code)
-
-    def decode(self, data: bytes) -> None:
-        """Decode a modbus exception response."""
-        self.exception_code = int(data[0])
-
-
 def pack_bitstring(bits: list[bool], align_byte=True) -> bytes:
     """Create a bytestring out of a list of bits.
 

pymodbus/server/requesthandler.py

@@ -7,7 +7,7 @@
 from pymodbus.constants import ExcCodes
 from pymodbus.exceptions import ModbusIOException, NoSuchIdException
 from pymodbus.logging import Log
-from pymodbus.pdu.pdu import ExceptionResponse
+from pymodbus.pdu import ExceptionResponse
 from pymodbus.transaction import TransactionManager
 from pymodbus.transport import CommParams
 

test/conftest.py

@@ -8,6 +8,7 @@
 import pytest
 import pytest_asyncio
 
+from pymodbus.constants import ExcCodes
 from pymodbus.datastore import ModbusBaseDeviceContext
 from pymodbus.server import ServerAsyncStop
 from pymodbus.transport import NULLMODEM_HOST, CommParams, CommType
@@ -218,6 +219,8 @@ def __init__(self, valid=False, default=True):
 
         def getValues(self, _fc, _address, count=0):
             """Get values."""
+            if count > 0x100:
+                return ExcCodes.ILLEGAL_VALUE
             return [self.default] * count
 
         def setValues(self, _fc, _address, _values):

test/pdu/test_bit.py

@@ -4,6 +4,7 @@
 import pytest
 
 import pymodbus.pdu.bit_message as bit_msg
+from pymodbus.constants import ExcCodes
 
 
 class TestModbusBitMessage:
@@ -34,6 +35,18 @@ async def test_bit_read_update_datastore_value_errors(self, mock_context):
         ):
             await pdu.update_datastore(context)
 
+    async def test_bit_datastore_exceptions(self, mock_context):
+        """Test bit exception response from datastore."""
+        context = mock_context()
+        context.async_getValues = mock.AsyncMock(return_value=ExcCodes.ILLEGAL_VALUE)
+        for pdu in (
+            (bit_msg.ReadCoilsRequest(address=1, count=0x800)),
+            (bit_msg.ReadDiscreteInputsRequest(address=1, count=0x800)),
+            (bit_msg.WriteSingleCoilRequest(address=1, bits=[True])),
+            (bit_msg.WriteMultipleCoilsRequest(address=1, bits=[True] * 5)),
+        ):
+            await pdu.update_datastore(context)
+
     async def test_bit_read_update_datastore_address_errors(self, mock_context):
         """Test bit read request encoding."""
         context = mock_context()

test/pdu/test_decoders.py

@@ -100,6 +100,17 @@ def test_client_lookup(self, code, frame):
         if not code & 0x80:
             assert pdu.function_code == code
 
+    def test_client_lookup_no_fc(self):
+        """Test lookup for responses."""
+        data = b'\x01\x70'
+        pdu = self.client.lookupPduClass(data)
+        assert not pdu
+
+    def test_list_function_codes(self):
+        """Test lookup for responses."""
+        fc_list = self.client.list_function_codes()
+        assert fc_list
+
     @pytest.mark.parametrize(("code", "frame"), list(requests))
     def test_server_lookup(self, code, frame):
         """Test lookup for requests."""

test/not_updated/test_device.py test/pdu/test_device.pytest/not_updated/test_device.py renamed to test/pdu/test_device.py

@@ -362,3 +362,10 @@ def test_modbus_plus_statistics_helpers(self):
         stats_summary = list(statistics.summary())
         assert sorted(summary) == sorted(stats_summary)
         assert not sum(sum(value[1]) for value in statistics)
+
+
+    def test_device_info_name(self):
+        """Test setting of info_name."""
+        name = "Maintainer"
+        info = ModbusDeviceIdentification(info_name={"VendorName": name})
+        assert name == info.VendorName

test/not_updated/test_events.py test/pdu/test_events.pytest/not_updated/test_events.py renamed to test/pdu/test_events.py

@@ -81,6 +81,14 @@ def test_read_device_information_calc1(self):
         assert handle.calculateRtuFrameSize(b"\x0e\x01\x83") == 999
         assert handle.calculateRtuFrameSize(b"\x0e\x01\x83\x00\x00\x03\x01\x03") == 998
 
+
+    def test_read_device_information_sub_fc(self):
+        """Test calculateRtuFrameSize, short buffer."""
+        handle = ReadDeviceInformationResponse()
+        assert handle.decode_sub_function_code(b"\x0e\x01\x83") == 0x83
+        handle = ReadDeviceInformationRequest()
+        assert handle.decode_sub_function_code(b"\x0e\x01\x83") == 0x83
+
     def test_read_device_information_encode(self):
         """Test that the read fifo queue response can encode."""
         message = b"\x0e\x01\x83\x00\x00\x03"

test/pdu/test_mei_messages.py

@@ -17,11 +17,6 @@
 
 TEST_MESSAGE = b"\x06\x00\x0a\x00\x0b\x00\x0c"
 
-# ---------------------------------------------------------------------------#
-#  Fixture
-# ---------------------------------------------------------------------------#
-
-
 class TestReadRegisterMessages:
     """Register Message Test Fixture.
 
@@ -86,14 +81,15 @@ def test_register_read_response_decode_error(self):
         with pytest.raises(ModbusIOException):
             reg.decode(b'\x14\x00\x03\x00\x11')
 
-    async def test_register_read_requests_count_errors(self):
+    async def test_register_read_requests_count_errors(self, mock_context):
         """This tests that the register request messages.
 
         will break on counts that are out of range
         """
+        context = mock_context()
         requests = [
-            #ReadHoldingRegistersRequest(address=1, count=0x800),
-            #ReadInputRegistersRequest(address=1, count=0x800),
+            ReadHoldingRegistersRequest(address=1, count=0x800),
+            ReadInputRegistersRequest(address=1, count=0x800),
             ReadWriteMultipleRegistersRequest(
                 read_address=1, read_count=0x800, write_address=1, write_registers=[5]
             ),
@@ -102,7 +98,7 @@ async def test_register_read_requests_count_errors(self):
             ),
         ]
         for request in requests:
-            result = await request.update_datastore(None)
+            result = await request.update_datastore(context)
             assert result.exception_code == ExcCodes.ILLEGAL_VALUE
 
     async def test_register_read_requests_verify_errors(self, mock_context):
@@ -114,8 +110,8 @@ async def test_register_read_requests_verify_errors(self, mock_context):
         requests = [
             ReadHoldingRegistersRequest(address=-1, count=5),
             ReadInputRegistersRequest(address=-1, count=5),
-            # ReadWriteMultipleRegistersRequest(-1,5,1,5),
-            # ReadWriteMultipleRegistersRequest(1,5,-1,5),
+            ReadWriteMultipleRegistersRequest(-1,5,1,[5]),
+            ReadWriteMultipleRegistersRequest(1,5,-1,[5]),
         ]
         for request in requests:
             await request.update_datastore(context)
@@ -147,17 +143,28 @@ async def test_read_write_multiple_registers_verify(self, mock_context):
         """Test read/write multiple registers."""
         context = mock_context()
         request = ReadWriteMultipleRegistersRequest(
-            read_address=1, read_count=10, write_address=2, write_registers=[0x00]
+            read_address=1, read_count=0x200, write_address=2, write_registers=[0x00]
         )
-        await request.update_datastore(context)
-        #assert response.exception_code == ExcCodes.ILLEGAL_ADDRESS
+        response = await request.update_datastore(context)
+        assert response.exception_code == ExcCodes.ILLEGAL_VALUE
 
         await request.update_datastore(context)
-        #assert response.exception_code == ExcCodes.ILLEGAL_ADDRESS
+        assert response.exception_code == ExcCodes.ILLEGAL_VALUE
 
         request.write_byte_count = 0x100
         await request.update_datastore(context)
-        #assert response.exception_code == ExcCodes.ILLEGAL_VALUE
+        assert response.exception_code == ExcCodes.ILLEGAL_VALUE
+
+    async def test_register_datastore_exceptions(self, mock_context):
+        """Test exception response from datastore."""
+        context = mock_context()
+        context.async_getValues = mock.AsyncMock(return_value=ExcCodes.ILLEGAL_VALUE)
+        for pdu in (
+            ReadHoldingRegistersRequest(address=-1, count=5),
+            ReadInputRegistersRequest(address=-1, count=5),
+            ReadWriteMultipleRegistersRequest(-1, 5, 1, [5], 1),
+        ):
+            await pdu.update_datastore(context)
 
     def test_serializing_to_string(self):
         """Test serializing to string."""