DummyProtocol, real usage.

Author: janiversen

test/transport/conftest.py

@@ -1,11 +1,13 @@
 """Configure pytest."""
-import asyncio
+from __future__ import annotations
+
 import os
 import sys
 from unittest import mock
 
 import pytest
 
+from pymodbus.logging import Log
 from pymodbus.transport import CommParams, CommType, ModbusProtocol
 
 
@@ -17,26 +19,18 @@ class DummyProtocol(ModbusProtocol):
 
     def __init__(self, is_server=False):
         """Initialize."""
-        self.transport = None
-        self.is_server = is_server
-        self.is_closing = False
-        self.data = b""
-        self.connection_made = mock.Mock()
-        self.connection_lost = mock.Mock()
-        self.reconnect_task: asyncio.Task = None
+        #  self.connection_made = mock.Mock()
+        #  self.connection_lost = mock.Mock()
         super().__init__(CommParams(), is_server)
 
     def callback_new_connection(self) -> ModbusProtocol:
         """Call when listener receive new connection request."""
         return DummyProtocol()
 
-    def close(self):  # pylint: disable=arguments-differ
-        """Simulate close."""
-        self.is_closing = True
-
-    def data_received(self, data):
-        """Call when some data is received."""
-        self.data += data
+    def callback_data(self, data: bytes, addr: tuple | None = None) -> int:
+        """Handle received data."""
+        Log.debug("callback_data called: {} addr={}", data, ":hex", addr)
+        return 0
 
 
 @pytest.fixture(name="dummy_protocol")

test/transport/test_nullmodem.py

@@ -1,5 +1,6 @@
 """Test transport."""
 import asyncio
+from unittest import mock
 
 import pytest
 
@@ -19,13 +20,17 @@ def get_port_in_class(base_ports):
     async def test_init(self, dummy_protocol):
         """Test initialize."""
         prot = dummy_protocol()
+        prot.connection_made = mock.Mock()
+        prot.connection_lost = mock.Mock()
         NullModem(prot)
         prot.connection_made.assert_not_called()
         prot.connection_lost.assert_not_called()
 
     async def test_close(self, dummy_protocol):
         """Test close."""
         prot = dummy_protocol()
+        prot.connection_made = mock.Mock()
+        prot.connection_lost = mock.Mock()
         modem = NullModem(prot)
         modem.close()
         prot.connection_made.assert_not_called()
@@ -45,6 +50,8 @@ async def test_close_no_protocol(self):
     async def test_close_twice(self, dummy_protocol):
         """Test close twice."""
         prot = dummy_protocol()
+        prot.connection_made = mock.Mock()
+        prot.connection_lost = mock.Mock()
         modem = NullModem(prot)
         modem.close()
         prot.connection_made.assert_not_called()
@@ -53,15 +60,17 @@ async def test_close_twice(self, dummy_protocol):
 
     async def test_listen(self, dummy_protocol, use_port):
         """Test listener (shared list)."""
-        protocol = dummy_protocol(is_server=True)
-        listen = NullModem.set_listener(use_port, protocol)
-        assert NullModem.listeners[use_port] == protocol
+        prot = dummy_protocol(is_server=True)
+        prot.connection_made = mock.Mock()
+        prot.connection_lost = mock.Mock()
+        listen = NullModem.set_listener(use_port, prot)
+        assert NullModem.listeners[use_port] == prot
         assert len(NullModem.listeners) == 1
         assert not NullModem.connections
         listen.close()
         assert not NullModem.listeners
-        protocol.connection_made.assert_not_called()
-        protocol.connection_lost.assert_not_called()
+        prot.connection_made.assert_not_called()
+        prot.connection_lost.assert_not_called()
 
     async def test_listen_twice(self, dummy_protocol, use_port):
         """Test exception when listening twice."""
@@ -101,8 +110,12 @@ async def test_is_dirty(self, dummy_protocol, use_port):
     async def test_connect(self, dummy_protocol, use_port):
         """Test connect."""
         prot_listen = dummy_protocol(is_server=True)
+        prot_listen.connection_made = mock.Mock()
+        prot_listen.connection_lost = mock.Mock()
         listen = NullModem.set_listener(use_port, prot_listen)
         prot1 = dummy_protocol()
+        prot1.connection_made = mock.Mock()
+        prot1.connection_lost = mock.Mock()
         modem, _ = NullModem.set_connection(use_port, prot1)
         modem_b = modem.other_modem
         assert modem.protocol != listen.protocol
@@ -178,8 +191,8 @@ async def test_flow_write(self, dummy_protocol, use_port):
         test_data2 = b"efgh"
         modem.write(test_data1)
         modem_b.write(test_data2)
-        assert modem_b.protocol.data == test_data1
-        assert modem.protocol.data == test_data2
+        assert modem_b.protocol.recv_buffer == test_data1
+        assert modem.protocol.recv_buffer == test_data2
         modem.close()
         listen.close()
 
@@ -193,16 +206,16 @@ async def test_flow_sendto(self, dummy_protocol, use_port):
         test_data2 = b"efgh"
         modem.sendto(test_data1)
         modem_b.sendto(test_data2)
-        assert modem_b.protocol.data == test_data1
-        assert modem.protocol.data == test_data2
+        assert modem_b.protocol.recv_buffer == test_data1
+        assert modem.protocol.recv_buffer == test_data2
         modem.close()
         listen.close()
 
     async def test_flow_not_connected(self, dummy_protocol):
         """Test flow not connected."""
         modem = NullModem(dummy_protocol())
         modem.sendto("no connection")
-        assert modem.protocol.data == b''
+        assert modem.protocol.recv_buffer == b''
         modem.close()
 
 
@@ -219,16 +232,16 @@ async def test_triangle_flow(self, dummy_protocol, use_port):
         test_data4 = b"mnop"
         modem1.write(test_data1)
         modem1b.write(test_data2)
-        assert modem1b.protocol.data == test_data1
-        assert modem1.protocol.data == test_data2
-        assert modem2b.protocol.data == b""
-        assert modem2.protocol.data == b""
+        assert modem1b.protocol.recv_buffer == test_data1
+        assert modem1.protocol.recv_buffer == test_data2
+        assert modem2b.protocol.recv_buffer == b""
+        assert modem2.protocol.recv_buffer == b""
         modem2.write(test_data3)
         modem2b.write(test_data4)
-        assert modem1b.protocol.data == test_data1
-        assert modem1.protocol.data == test_data2
-        assert modem2b.protocol.data == test_data3
-        assert modem2.protocol.data == test_data4
+        assert modem1b.protocol.recv_buffer == test_data1
+        assert modem1.protocol.recv_buffer == test_data2
+        assert modem2b.protocol.recv_buffer == test_data3
+        assert modem2.protocol.recv_buffer == test_data4
         modem1.close()
         modem2.close()
         listen.close()
@@ -262,8 +275,8 @@ def manipulator(data):
         modem_b.write(data1)
         modem_b.write(data2)
         x = b"".join(part for part in add_text)
-        assert modem.protocol.data == data1 + data2
-        assert modem_b.protocol.data == data1 + x + data2 + x
+        assert modem.protocol.recv_buffer == data1 + data2
+        assert modem_b.protocol.recv_buffer == data1 + x + data2 + x
         modem.close()
         listen.close()