Add system health check to all tests.

Author: janiversen

.github/workflows/ci.yml

@@ -94,7 +94,7 @@ jobs:
         run: pip install -e . -r requirements.txt
 
       - name: pytest
-        run: pytest -n0  -v --full-trace --timeout=20
+        run: pytest -n0  -v --full-trace --timeout=120
 
   analyze:
     name: Analyze Python

examples/simulator.py

@@ -64,6 +64,8 @@ async def run_simulator():
     cmdline = [
         "--modbus_device",
         "device_try",
+        "--modbus_server",
+        "server",
     ]
     cmd_args = get_simulator_commandline(cmdline=cmdline)
     task = ModbusSimulatorServer(**cmd_args)
@@ -82,6 +84,7 @@ async def run_simulator():
     await run_calls(client, 1)
 
     _logger.info("### shutdown client")
+    client.close()
 
     _logger.info("### shutdown server")
     await task.stop()

pymodbus/server/async_io.py

@@ -53,7 +53,7 @@ def __init__(self, owner):
             host=owner.comm_params.source_address[0],
             port=owner.comm_params.source_address[1],
         )
-        super().__init__(params, True)
+        super().__init__(params, False)
         self.server = owner
         self.running = False
         self.receive_queue = asyncio.Queue()
@@ -170,6 +170,7 @@ async def handle(self):
                         self.comm_params.comm_name,
                     )
                     self.transport_close()
+                    self.callback_disconnected(exc)
                 else:
                     Log.error("Unknown error occurred {}", exc)
                     reset_frame = True  # graceful recovery
@@ -252,7 +253,66 @@ def callback_data(self, data: bytes, addr: tuple = None) -> int:
 # --------------------------------------------------------------------------- #
 
 
-class ModbusTcpServer(ModbusProtocol):
+class ModbusBaseServer(ModbusProtocol):
+    """Common functionality for all server classes"""
+
+    def __init__(
+        self,
+        params: CommParams,
+        context,
+        ignore_missing_slaves,
+        broadcast_enable,
+        response_manipulator,
+        request_tracer,
+        identity,
+        framer,
+    ) -> None:
+        """Initialize base server."""
+        super().__init__(
+            params,
+            True,
+        )
+        self.loop = asyncio.get_running_loop()
+        self.decoder = ServerDecoder()
+        self.context = context or ModbusServerContext()
+        self.control = ModbusControlBlock()
+        self.ignore_missing_slaves = ignore_missing_slaves
+        self.broadcast_enable = broadcast_enable
+        self.response_manipulator = response_manipulator
+        self.request_tracer = request_tracer
+        self.handle_local_echo = False
+        if isinstance(identity, ModbusDeviceIdentification):
+            self.control.Identity.update(identity)
+        self.framer = framer
+        self.serving: asyncio.Future = asyncio.Future()
+
+    def callback_new_connection(self):
+        """Handle incoming connect."""
+        return ModbusServerRequestHandler(self)
+
+    async def shutdown(self):
+        """Shutdown server."""
+        await self.server_close()
+
+    async def server_close(self):
+        """Close server."""
+        if not self.serving.done():
+            self.serving.set_result(True)
+        self.transport_close()
+
+    async def serve_forever(self):
+        """Start endless loop."""
+        if self.transport:
+            raise RuntimeError(
+                "Can't call serve_forever on an already running server object"
+            )
+        await self.transport_listen()
+        Log.info("Server listening.")
+        await self.serving
+        Log.info("Server graceful shutdown.")
+
+
+class ModbusTcpServer(ModbusBaseServer):
     """A modbus threaded tcp socket server.
 
     We inherit and overload the socket server so that we
@@ -263,7 +323,7 @@ class ModbusTcpServer(ModbusProtocol):
     def __init__(
         self,
         context,
-        framer=None,
+        framer=ModbusSocketFramer,
         identity=None,
         address=("", 502),
         ignore_missing_slaves=False,
@@ -300,60 +360,16 @@ def __init__(
             ),
         )
         params.source_address = address
-
         super().__init__(
             params,
-            True,
+            context,
+            ignore_missing_slaves,
+            broadcast_enable,
+            response_manipulator,
+            request_tracer,
+            identity,
+            framer,
         )
-        self.decoder = ServerDecoder()
-        self.framer = framer or ModbusSocketFramer
-        self.context = context or ModbusServerContext()
-        self.control = ModbusControlBlock()
-        self.ignore_missing_slaves = ignore_missing_slaves
-        self.broadcast_enable = broadcast_enable
-        self.response_manipulator = response_manipulator
-        self.request_tracer = request_tracer
-        if isinstance(identity, ModbusDeviceIdentification):
-            self.control.Identity.update(identity)
-
-        # asyncio future that will be done once server has started
-        self.serving = asyncio.Future()
-        self.serving_done = asyncio.Future()
-        # constructors cannot be declared async, so we have to
-        # defer the initialization of the server
-        self.handle_local_echo = False
-
-    def handle_new_connection(self):
-        """Handle incoming connect."""
-        return ModbusServerRequestHandler(self)
-
-    async def serve_forever(self):
-        """Start endless loop."""
-        if self.transport:
-            raise RuntimeError(
-                "Can't call serve_forever on an already running server object"
-            )
-
-        await self.transport_listen()
-        self.serving.set_result(True)
-        Log.info("Server(TCP) listening.")
-        try:
-            await self.transport.serve_forever()
-        except asyncio.exceptions.CancelledError:
-            self.serving_done.set_result(False)
-            raise
-        except Exception as exc:  # pylint: disable=broad-except
-            Log.error("Server unexpected exception {}", exc)
-        self.serving_done.set_result(True)
-        Log.info("Server graceful shutdown.")
-
-    async def shutdown(self):
-        """Shutdown server."""
-        await self.server_close()
-
-    async def server_close(self):
-        """Close server."""
-        self.transport_close()
 
 
 class ModbusTlsServer(ModbusTcpServer):
@@ -367,7 +383,7 @@ class ModbusTlsServer(ModbusTcpServer):
     def __init__(  # pylint: disable=too-many-arguments
         self,
         context,
-        framer=None,
+        framer=ModbusTlsFramer,
         identity=None,
         address=("", 502),
         sslctx=None,
@@ -422,7 +438,7 @@ def __init__(  # pylint: disable=too-many-arguments
         )
 
 
-class ModbusUdpServer(ModbusProtocol):
+class ModbusUdpServer(ModbusBaseServer):
     """A modbus threaded udp socket server.
 
     We inherit and overload the socket server so that we
@@ -433,7 +449,7 @@ class ModbusUdpServer(ModbusProtocol):
     def __init__(
         self,
         context,
-        framer=None,
+        framer=ModbusSocketFramer,
         identity=None,
         address=("", 502),
         ignore_missing_slaves=False,
@@ -468,62 +484,17 @@ def __init__(
                 reconnect_delay_max=0.0,
                 timeout_connect=0.0,
             ),
-            True,
+            context,
+            ignore_missing_slaves,
+            broadcast_enable,
+            response_manipulator,
+            request_tracer,
+            identity,
+            framer,
         )
 
-        self.loop = asyncio.get_running_loop()
-        self.decoder = ServerDecoder()
-        self.framer = framer or ModbusSocketFramer
-        self.context = context or ModbusServerContext()
-        self.control = ModbusControlBlock()
-        self.ignore_missing_slaves = ignore_missing_slaves
-        self.broadcast_enable = broadcast_enable
-        self.response_manipulator = response_manipulator
-        self.request_tracer = request_tracer
-        if isinstance(identity, ModbusDeviceIdentification):
-            self.control.Identity.update(identity)
-        self.stop_serving = self.loop.create_future()
-        # asyncio future that will be done once server has started
-        self.serving = asyncio.Future()
-        self.serving_done = asyncio.Future()
-        self.handle_local_echo = False
-
-    def handle_new_connection(self):
-        """Handle incoming connect."""
-        return ModbusServerRequestHandler(self)
-
-    async def serve_forever(self):
-        """Start endless loop."""
-        if self.transport:
-            raise RuntimeError(
-                "Can't call serve_forever on an already running server object"
-            )
-        try:
-            await self.transport_listen()
-        except asyncio.exceptions.CancelledError:
-            self.serving_done.set_result(False)
-            raise
-        except Exception as exc:
-            Log.error("Server unexpected exception {}", exc)
-            self.serving_done.set_result(False)
-            raise RuntimeError(exc) from exc
-        Log.info("Server(UDP) listening.")
-        self.serving.set_result(True)
-        await self.stop_serving
-        self.serving_done.set_result(True)
-
-    async def shutdown(self):
-        """Shutdown server."""
-        await self.server_close()
-
-    async def server_close(self):
-        """Close server."""
-        self.transport_close()
-        if not self.stop_serving.done():
-            self.stop_serving.set_result(True)
-
 
-class ModbusSerialServer(ModbusProtocol):
+class ModbusSerialServer(ModbusBaseServer):
     """A modbus threaded serial socket server.
 
     We inherit and overload the socket server so that we
@@ -570,56 +541,15 @@ def __init__(
                 baudrate=kwargs.get("baudrate", 19200),
                 stopbits=kwargs.get("stopbits", 1),
             ),
-            True,
+            context,
+            kwargs.get("ignore_missing_slaves", False),
+            kwargs.get("broadcast_enable", False),
+            kwargs.get("request_tracer", None),
+            kwargs.get("response_manipulator", None),
+            kwargs.get("identity", None),
+            framer,
         )
-
-        self.loop = asyncio.get_running_loop()
         self.handle_local_echo = kwargs.get("handle_local_echo", False)
-        self.ignore_missing_slaves = kwargs.get("ignore_missing_slaves", False)
-        self.broadcast_enable = kwargs.get("broadcast_enable", False)
-        self.framer = framer or ModbusRtuFramer
-        self.decoder = ServerDecoder()
-        self.context = context or ModbusServerContext()
-        self.response_manipulator = kwargs.get("response_manipulator", None)
-        self.control = ModbusControlBlock()
-        if isinstance(identity, ModbusDeviceIdentification):
-            self.control.Identity.update(identity)
-        self.request_tracer = kwargs.get("request_tracer", None)
-        self.server = None
-        self.control = ModbusControlBlock()
-        identity = kwargs.get("identity")
-        if isinstance(identity, ModbusDeviceIdentification):
-            self.control.Identity.update(identity)
-
-    async def start(self):
-        """Start connecting."""
-
-    def handle_new_connection(self):
-        """Handle incoming connect."""
-        return ModbusServerRequestHandler(self)
-
-    async def shutdown(self):
-        """Terminate server."""
-        self.transport_close()
-        if self.server:
-            self.server.close()
-            await asyncio.wait_for(self.server.wait_closed(), 10)
-            self.server = None
-
-    async def serve_forever(self):
-        """Start endless loop."""
-        if self.server:
-            raise RuntimeError(
-                "Can't call serve_forever on an already running server object"
-            )
-        Log.info("Server(Serial) listening.")
-        await self.transport_listen()
-        try:
-            await self.transport.serve_forever()
-        except asyncio.exceptions.CancelledError:
-            raise
-        except Exception as exc:  # pylint: disable=broad-except
-            Log.error("Server unexpected exception {}", exc)
 
 
 # --------------------------------------------------------------------------- #
@@ -779,7 +709,6 @@ async def StartAsyncSerialServer(  # pylint: disable=invalid-name,dangerous-defa
     server = ModbusSerialServer(
         context, kwargs.pop("framer", ModbusAsciiFramer), identity=identity, **kwargs
     )
-    await server.start()
     await _serverList.run(server, custom_functions)
 
 

pymodbus/server/simulator/http_server.py

@@ -167,7 +167,7 @@ def __init__(
                 info_name=server["identity"]
             )
         self.modbus_server = comm(framer=framer, context=datastore, **server)
-
+        self.serving: asyncio.Future = asyncio.Future()
         self.log_file = log_file
         self.site = None
         self.http_host = http_host
@@ -234,6 +234,7 @@ async def start_modbus_server(self, app):
     async def stop_modbus_server(self, app):
         """Stop modbus server."""
         Log.info("Stopping modbus server")
+        await self.modbus_server.shutdown()
         app["modbus_server"].cancel()
         with contextlib.suppress(asyncio.exceptions.CancelledError):
             await app["modbus_server"]
@@ -254,13 +255,15 @@ async def run_forever(self, only_start=False):
         if only_start:
             return
         while True:
-            await asyncio.sleep(1)
+            await self.serving()
 
     async def stop(self):
         """Stop modbus and http servers."""
         await self.site.stop()
         self.site = None
-        await asyncio.sleep(1)
+        if not self.serving.done():
+            self.serving.set_result(True)
+        await asyncio.sleep(0.1)
 
     async def handle_html_static(self, request):
         """Handle static html."""