Test coverage server 100%.

Author: janiversen

pymodbus/server/__init__.py

@@ -1,6 +1,7 @@
 """**Server classes**."""
 
 __all__ = [
+    "ModbusBaseServer",
     "ModbusSerialServer",
     "ModbusSimulatorServer",
     "ModbusTcpServer",
@@ -19,6 +20,7 @@
     "get_simulator_commandline",
 ]
 
+from pymodbus.server.base import ModbusBaseServer
 from pymodbus.server.server import (
     ModbusSerialServer,
     ModbusTcpServer,

pymodbus/server/base.py

@@ -18,7 +18,7 @@
 class ModbusBaseServer(ModbusProtocol):
     """Common functionality for all server classes."""
 
-    active_server: ModbusBaseServer | None
+    active_server: ModbusBaseServer | None = None
 
     def __init__(  # pylint: disable=too-many-arguments
         self,

pymodbus/server/requesthandler.py

@@ -43,24 +43,17 @@ def __init__(self, owner, trace_packet, trace_pdu, trace_connect):
     def callback_disconnected(self, exc: Exception | None) -> None:
         """Call when connection is lost."""
         super().callback_disconnected(exc)
-        try:
-            if exc is None:
-                Log.debug(
-                    "Handler for stream [{}] has been canceled", self.comm_params.comm_name
-                )
-            else:
-                Log.debug(
-                    "Client Disconnection {} due to {}",
-                    self.comm_params.comm_name,
-                    exc,
-                )
-            self.running = False
-        except Exception as exc:  # pylint: disable=broad-except
-            Log.error(
-                "Datastore unable to fulfill request: {}; {}",
+        if exc is None:
+            Log.debug(
+                "Handler for stream [{}] has been canceled", self.comm_params.comm_name
+            )
+        else:
+            Log.debug(
+                "Client Disconnection {} due to {}",
+                self.comm_params.comm_name,
                 exc,
-                traceback.format_exc(),
             )
+        self.running = False
 
     def callback_data(self, data: bytes, addr: tuple | None = None) -> int:
         """Handle received data."""
@@ -89,7 +82,7 @@ async def handle_request(self):
             if self.server.broadcast_enable and not self.last_pdu.dev_id:
                 # if broadcasting then execute on all device contexts,
                 # note response will be ignored
-                for dev_id in self.server.context.device_id():
+                for dev_id in self.server.context.device_ids():
                     await self.last_pdu.update_datastore(self.server.context[dev_id])
                 return
 

pymodbus/server/simulator/custom_actions.py

@@ -48,7 +48,7 @@
 from pymodbus.server.simulator.http_server import ModbusSimulatorServer
 
 
-def get_commandline(extras=None, cmdline=None):
+def get_commandline(cmdline=None):
     """Get command line arguments."""
     parser = argparse.ArgumentParser(
         description="Modbus server with REST-API and web server"
@@ -97,9 +97,6 @@ def get_commandline(extras=None, cmdline=None):
         help="python file with custom actions, default is none",
         type=str,
     )
-    if extras:
-        for extra in extras:
-            parser.add_argument(extra[0], **extra[1])
     args = parser.parse_args(cmdline)
     pymodbus_apply_logging_config(args.log.upper())
     Log.info("Start simulator")
@@ -112,17 +109,11 @@ def get_commandline(extras=None, cmdline=None):
     return cmd_args
 
 
-async def run_main():
+async def run_main(cmdline=None):
     """Run server async."""
-    cmd_args = get_commandline()
+    cmd_args = get_commandline(cmdline=cmdline)
     task = ModbusSimulatorServer(**cmd_args)
     await task.run_forever()
 
-
-def main():
-    """Run server."""
-    asyncio.run(run_main(), debug=True)
-
-
 if __name__ == "__main__":
-    main()
+    asyncio.run(run_main(), debug=True)

pymodbus/server/simulator/main.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import asyncio
+import concurrent
 import os
 
 from pymodbus.datastore import ModbusServerContext
@@ -158,10 +159,13 @@ async def ServerAsyncStop() -> None:
     await ModbusBaseServer.active_server.shutdown()
     ModbusBaseServer.active_server = None
 
-
 def ServerStop() -> None:
     """Terminate server."""
     if not ModbusBaseServer.active_server:
         raise RuntimeError("Modbus server not running.")
-    future = asyncio.run_coroutine_threadsafe(ServerAsyncStop(), ModbusBaseServer.active_server.loop)
-    future.result(timeout=10 if os.name == 'nt' else 0.1)
+    try:
+        future = asyncio.run_coroutine_threadsafe(ServerAsyncStop(), ModbusBaseServer.active_server.loop)
+        future.result(timeout=10 if os.name == 'nt' else 0.1)
+    except concurrent.futures.TimeoutError:
+        pass
+    ModbusBaseServer.active_server = None

pymodbus/server/startstop.py

@@ -62,8 +62,8 @@ def __init__(
             self._lock = asyncio.Lock()
             self.low_level_send = self.send
             self.response_future: asyncio.Future = asyncio.Future()
-            self.last_pdu: ModbusPDU | None
-            self.last_addr: tuple | None
+            self.last_pdu: ModbusPDU | None = None
+            self.last_addr: tuple | None = None
 
     def dummy_trace_packet(self, sending: bool, data: bytes) -> bytes:
         """Do dummy trace."""

pymodbus/transaction/transaction.py

@@ -0,0 +1,67 @@
+"""Test server asyncio."""
+from unittest import mock
+
+import pytest
+
+from pymodbus.pdu import ReadHoldingRegistersRequest
+from pymodbus.server import ModbusBaseServer
+from pymodbus.transport import CommParams, CommType
+
+
+class TestBaseServer:
+    """Test for the pymodbus.server.startstop module."""
+
+    @pytest.fixture
+    async def baseserver(self):
+        """Fixture to provide base_server."""
+        server = ModbusBaseServer(
+            CommParams(
+                comm_type=CommType.TCP,
+                comm_name="server_listener",
+                reconnect_delay=0.0,
+                reconnect_delay_max=0.0,
+                timeout_connect=0.0,
+            ),
+            None,
+            False,
+            False,
+            None,
+            "socket",
+            None,
+            None,
+            None,
+            [ReadHoldingRegistersRequest],
+        )
+        return server
+
+    async def test_base(self, baseserver):
+        """Test __init__."""
+
+    async def test_base_serve_forever1(self, baseserver):
+        """Test serve_forever."""
+        baseserver.listen = mock.AsyncMock(return_value=None)
+        with pytest.raises(RuntimeError):
+            await baseserver.serve_forever()
+
+    async def test_base_serve_forever2(self, baseserver):
+        """Test serve_forever."""
+        baseserver.listen = mock.AsyncMock(return_value=True)
+        await baseserver.serve_forever(background=True)
+        baseserver.serving.set_result(True)
+        await baseserver.serve_forever()
+
+
+    async def test_base_connected(self, baseserver):
+        """Test serve_forever."""
+        with pytest.raises(RuntimeError):
+            baseserver.callback_connected()
+
+    async def test_base_disconnected(self, baseserver):
+        """Test serve_forever."""
+        with pytest.raises(RuntimeError):
+            baseserver.callback_disconnected(None)
+
+    async def test_base_data(self, baseserver):
+        """Test serve_forever."""
+        with pytest.raises(RuntimeError):
+            baseserver.callback_data(None)

test/server/test_base.py

@@ -0,0 +1,80 @@
+"""Test server asyncio."""
+from unittest import mock
+
+import pytest
+
+from pymodbus.server import ModbusBaseServer
+from pymodbus.transport import CommParams, CommType
+from pymodbus.exceptions import ModbusIOException, NoSuchIdException
+from pymodbus.pdu import ExceptionResponse
+from pymodbus.datastore import (
+    ModbusDeviceContext,
+    ModbusSequentialDataBlock,
+    ModbusServerContext,
+)
+
+
+class TestRequesthandler:
+    """Test for the pymodbus.server.startstop module."""
+
+    @pytest.fixture
+    async def requesthandler(self):
+        """Fixture to provide base_server."""
+        store = ModbusDeviceContext(
+            di=ModbusSequentialDataBlock(0, [17] * 100),
+            co=ModbusSequentialDataBlock(0, [17] * 100),
+            hr=ModbusSequentialDataBlock(0, [17] * 100),
+            ir=ModbusSequentialDataBlock(0, [17] * 100),
+        )
+        server = ModbusBaseServer(
+            CommParams(
+                comm_type=CommType.TCP,
+                comm_name="server_listener",
+                reconnect_delay=0.0,
+                reconnect_delay_max=0.0,
+                timeout_connect=0.0,
+                source_address=(0, 0),
+            ),
+            ModbusServerContext(devices=store, single=True),
+            False,
+            False,
+            None,
+            "socket",
+            None,
+            None,
+            None,
+            [],
+        )
+        conn = server.callback_new_connection()
+        conn.pdu_send = mock.Mock(return_value=True)
+        return conn
+
+    async def test_requesthandler(self, requesthandler):
+        """Test __init__."""
+
+    async def test_rh_callback_data(self, requesthandler):
+        """Test __init__."""
+        with mock.patch("pymodbus.transaction.TransactionManager.callback_data") as cb_data:
+            cb_data.side_effect=ModbusIOException
+            data = b"012"
+            assert len(data) == requesthandler.callback_data(data, None)
+
+    async def test_rh_handle_request(self, requesthandler):
+        """Test __init__."""
+        requesthandler.last_pdu = None
+        await requesthandler.handle_request()
+        requesthandler.last_pdu = ExceptionResponse(17)
+        await requesthandler.handle_request()
+        requesthandler.last_pdu.update_datastore = mock.AsyncMock()
+        requesthandler.server.broadcast_enable = True
+        requesthandler.last_pdu.dev_id = 0
+        await requesthandler.handle_request()
+        requesthandler.last_pdu.update_datastore.side_effect = NoSuchIdException
+        await requesthandler.handle_request()
+        requesthandler.server.ignore_missing_devices = True
+        await requesthandler.handle_request()
+
+    async def test_rh_server_send(self, requesthandler):
+        """Test __init__."""
+        requesthandler.server_send(None, None)
+        requesthandler.server_send(ExceptionResponse(17), None)

test/server/test_requesthandler.py

@@ -15,7 +15,11 @@
     ModbusServerContext,
 )
 from pymodbus.exceptions import NoSuchIdException
-from pymodbus.server import ModbusTcpServer, ModbusTlsServer, ModbusUdpServer
+from pymodbus.server import (
+    ModbusTcpServer,
+    ModbusTlsServer,
+    ModbusUdpServer,
+)
 
 
 _logger = logging.getLogger()