Test coverage server 100%.

Author: janiversen

pymodbus/server/__init__.py

@@ -1,6 +1,7 @@
 """**Server classes**."""
 
 __all__ = [
+    "ModbusBaseServer",
     "ModbusSerialServer",
     "ModbusSimulatorServer",
     "ModbusTcpServer",
@@ -25,6 +26,7 @@
     ModbusTlsServer,
     ModbusUdpServer,
 )
+from pymodbus.server.base import ModbusBaseServer
 from pymodbus.server.simulator.http_server import ModbusSimulatorServer
 from pymodbus.server.simulator.main import get_commandline as get_simulator_commandline
 from pymodbus.server.startstop import (

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/simulator/custom_actions.py

@@ -158,10 +158,12 @@ 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 TimeoutError:
+        return

pymodbus/server/startstop.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()

test/server/test_server_asyncio.py

@@ -0,0 +1,107 @@
+"""Test server asyncio."""
+from unittest import mock
+
+import pytest
+
+from pymodbus.server import (
+    ModbusBaseServer,
+    ServerAsyncStop,
+    ServerStop,
+    StartAsyncSerialServer,
+    StartAsyncTcpServer,
+    StartAsyncTlsServer,
+    StartAsyncUdpServer,
+    StartSerialServer,
+    StartTcpServer,
+    StartTlsServer,
+    StartUdpServer,
+)
+from pymodbus.datastore import ModbusDeviceContext, ModbusServerContext
+from pymodbus.transport import NULLMODEM_HOST
+
+SERV_ADDR = ("127.0.0.1", 0)
+
+
+class TestStartStopServer:
+    """Test for the pymodbus.server.startstop module."""
+
+    async def test_ServerAsyncStop(self):
+        """Test  ServerAsyncStop."""
+        with pytest.raises(RuntimeError):
+            await ServerAsyncStop()
+        with pytest.raises(RuntimeError):
+            ServerStop()
+        ModbusBaseServer.active_server = mock.AsyncMock()
+        await ServerAsyncStop()
+        ModbusBaseServer.active_server = mock.AsyncMock()
+        ServerStop()
+        ModbusBaseServer.active_server = None
+
+    @mock.patch('pymodbus.server.ModbusBaseServer.serve_forever')
+    async def test_StartAsyncSerialServer(self, mock_method):
+        """Test  StartAsyncSerialServer."""
+        mock_method.return_value=True
+        await StartAsyncSerialServer(
+            ModbusServerContext(devices=ModbusDeviceContext(), single=True),
+            port="/dev/tty01",
+        )
+
+    @mock.patch('pymodbus.server.ModbusBaseServer.serve_forever')
+    async def test_StartAsyncTcpServer(self, mock_method):
+        """Test  StartAsyncTcpServer."""
+        mock_method.return_value=True
+        await StartAsyncTcpServer(
+            ModbusServerContext(devices=ModbusDeviceContext(), single=True)
+        )
+
+    @mock.patch('pymodbus.server.ModbusBaseServer.serve_forever')
+    async def test_StartAsyncTlsServer(self, mock_method):
+        """Test  StartAsyncTlsServer."""
+        mock_method.return_value=True
+        await StartAsyncTlsServer(
+            ModbusServerContext(devices=ModbusDeviceContext(), single=True)
+        )
+
+    @mock.patch('pymodbus.server.ModbusBaseServer.serve_forever')
+    async def test_StartAsyncUdpServer(self, mock_method):
+        """Test  StartAsyncUdpServer."""
+        mock_method.return_value=True
+        await StartAsyncUdpServer(
+            ModbusServerContext(devices=ModbusDeviceContext(), single=True)
+        )
+
+    def test_ServerStop(self):
+        """Test  ServerStop."""
+
+    @mock.patch('pymodbus.server.ModbusBaseServer.serve_forever')
+    def test_StartSerialServer(self, mock_method):
+        """Test  StartSerialServer."""
+        mock_method.return_value=True
+        StartSerialServer(
+            ModbusServerContext(devices=ModbusDeviceContext(), single=True),
+            port="/dev/tty01",
+        )
+
+    @mock.patch('pymodbus.server.ModbusBaseServer.serve_forever')
+    def test_StartTcpServer(self, mock_method):
+        """Test  StartTcpServer."""
+        mock_method.return_value=True
+        StartTcpServer(
+            ModbusServerContext(devices=ModbusDeviceContext(), single=True)
+        )
+
+    @mock.patch('pymodbus.server.ModbusBaseServer.serve_forever')
+    def test_StartTlsServer(self, mock_method):
+        """Test  StartTlsServer."""
+        mock_method.return_value=True
+        StartTlsServer(
+            ModbusServerContext(devices=ModbusDeviceContext(), single=True)
+        )
+
+    @mock.patch('pymodbus.server.ModbusBaseServer.serve_forever')
+    def test_StartUdpServer(self, mock_method):
+        """Test  StartUdpServer."""
+        mock_method.return_value=True
+        StartUdpServer(
+            ModbusServerContext(devices=ModbusDeviceContext(), single=True)
+        )