Add server/client network stub, to allow test of network packets.

pymodbus/transport/__init__.py

@@ -3,9 +3,11 @@
     "CommParams",
     "CommType",
     "ModbusProtocol",
+    "ModbusProtocolStub",
     "NULLMODEM_HOST",
 ]
 
+from pymodbus.transport.stub import ModbusProtocolStub
 from pymodbus.transport.transport import (
     NULLMODEM_HOST,
     CommParams,

pymodbus/transport/stub.py

@@ -0,0 +1,29 @@
+"""ModbusProtocol network stub."""
+from __future__ import annotations
+
+from pymodbus.transport.transport import ModbusProtocol
+
+
+class ModbusProtocolStub(ModbusProtocol):
+    """Protocol layer including transport."""
+
+    async def start_run(self):
+        """Call need functions to start server/client."""
+        if  self.is_server:
+            return await self.transport_listen()
+        return await self.transport_connect()
+
+    def callback_data(self, data: bytes, addr: tuple | None = None) -> int:
+        """Handle received data."""
+        if (response := self.stub_handle_data(data)):
+            self.transport_send(response)
+        return len(data)
+
+    # ---------------- #
+    # external methods #
+    # ---------------- #
+    def stub_handle_data(self, data: bytes) -> bytes | None:
+        """Handle received data."""
+        if len(data) > 5:
+            return data
+        return None

test/conftest.py

@@ -46,6 +46,7 @@ def pytest_configure():
     "TestReconnectModbusProtocol": 8000,
     "TestClientServerSyncExamples": 8100,
     "TestClientServerAsyncExamples": 8200,
+    "TestNetwork": 8300,
 }
 
 

test/sub_transport/test_reconnect.py

@@ -28,6 +28,7 @@ async def test_no_reconnect_call(self, client):
 
     async def test_reconnect_call(self, client):
         """Test connection_lost()."""
+        client.comm_params.on_reconnect_callback = mock.MagicMock()
         client.loop = asyncio.get_running_loop()
         client.call_create = mock.AsyncMock(return_value=(None, None))
         await client.transport_connect()
@@ -38,6 +39,7 @@ async def test_reconnect_call(self, client):
         assert client.reconnect_task
         assert client.call_create.call_count == 2
         assert client.reconnect_delay_current == client.comm_params.reconnect_delay * 2
+        assert client.comm_params.on_reconnect_callback.called
         client.transport_close()
 
     async def test_multi_reconnect_call(self, client):

test/test_network.py

@@ -0,0 +1,28 @@
+"""Test transport."""
+
+import pytest
+
+from pymodbus.client import AsyncModbusTcpClient
+from pymodbus.logging import Log
+from pymodbus.transport import NULLMODEM_HOST, ModbusProtocolStub
+
+
+class TestNetwork:
+    """Test network problems."""
+
+    @staticmethod
+    @pytest.fixture(name="use_port")
+    def get_port_in_class(base_ports):
+        """Return next port."""
+        base_ports[__class__.__name__] += 1
+        return base_ports[__class__.__name__]
+
+    async def test_stub(self, use_port, use_cls):
+        """Test double packet on network."""
+        Log.debug("test_double_packet {}", use_port)
+        client = AsyncModbusTcpClient(NULLMODEM_HOST, port=use_port)
+        stub = ModbusProtocolStub(use_cls, True)
+        assert await stub.start_run()
+        assert await client.connect()
+        client.transport_close()
+        stub.transport_close()