Refactor tests to use asyncio and AsyncMock for asynchronous behavior

- Updated test_mqtt_commands.py to utilize AsyncMock and asyncio for handling MQTT commands.
- Refactored test_set_commands.py to support async operations and ensure proper command handling.
- Modified test_transport_tcp.py to replace unittest with pytest and implement async testing for TCP transport.
- Enhanced test_version_command.py to use asyncio for simulating command responses and handling timeouts.
- Improved mock transport fixtures across tests to support async context management and operations.
- Updateed test_version_command
- feat: add asyncio support and refactor tests for async behavior

Author: sidey79

pyproject.toml

@@ -14,4 +14,7 @@ dependencies = [
 ]
 
 [tool.pytest.ini_options]
-testpaths = ["tests"]
+testpaths = ["tests"]
+
+[tool.pytest-asyncio]
+mode = "auto"

signalduino/controller.py

@@ -60,6 +60,7 @@ def __init__(
         self._write_queue: asyncio.Queue[QueuedCommand] = asyncio.Queue()
         self._pending_responses: List[PendingResponse] = []
         self._pending_responses_lock = asyncio.Lock()
+        self._init_complete_event = asyncio.Event() # NEU: Event für den Abschluss der Initialisierung
 
         # Timer-Handles (jetzt asyncio.Task anstelle von threading.Timer)
         self._heartbeat_task: Optional[asyncio.Task[Any]] = None
@@ -150,6 +151,8 @@ async def initialize(self) -> None:
         self.logger.info("Initializing device...")
         self.init_retry_count = 0
         self.init_reset_flag = False
+        self.init_version_response = None
+        self._init_complete_event.clear() # NEU: Event für erneute Initialisierung zurücksetzen
 
         if self._stop_event.is_set():
             self.logger.warning("initialize called but stop event is set.")
@@ -255,6 +258,9 @@ async def _check_version_resp(self, msg: Optional[str]) -> None:
 
             # NEU: Starte Heartbeat-Task
             await self._start_heartbeat_task()
+            
+            # NEU: Signalisiere den Abschluss der Initialisierung
+            self._init_complete_event.set()
 
         else:
             self.logger.warning("StartInit: No valid version response.")
@@ -273,8 +279,11 @@ async def _reset_device(self) -> None:
         await asyncio.sleep(2.0)
         # NEU: Der Controller ist neu gestartet und muss wieder in den async Kontext eintreten
         await self.__aenter__()
-
+        
         # Manuell die Initialisierung starten
+        self.init_version_response = None
+        self._init_complete_event.clear() # NEU: Event für erneute Initialisierung zurücksetzen
+        
         try:
             await self._send_xq()
             await self._start_init()
@@ -491,21 +500,25 @@ def on_response(response: str):
             # Warte auf das Future mit Timeout
             return await asyncio.wait_for(response_future, timeout=timeout)
         except asyncio.TimeoutError:
+            await asyncio.sleep(0)  # Gib dem Event-Loop eine Chance, _stop_event zu setzen.
             # Code Refactor: Timeout vs. dead connection
-            if self._stop_event.is_set():
+            self.logger.debug("Command timeout reached for %s", payload)
+            # Differentiate between connection drop and normal command timeout
+            # Check for a closed transport or a stopped controller
+            if self._stop_event.is_set() or (self.transport and self.transport.closed()):
                 self.logger.error(
-                    "Command '%s' timed out. Connection appears to be dead (controller stopping).", payload
+                    "Command '%s' timed out. Connection appears to be dead (transport closed or controller stopping).", payload
                 )
                 raise SignalduinoConnectionError(
                     f"Command '{payload}' failed: Connection dropped."
                 ) from None
-            
-            # Annahme: Transport-API wirft SignalduinoConnectionError bei Trennung.
-            # Wenn dies nicht der Fall ist, wird ein Timeout angenommen.
-            self.logger.warning(
-                "Command '%s' timed out. Treating as no response from device.", payload
-            )
-            raise SignalduinoCommandTimeout(f"Command '{payload}' timed out") from None
+            else:
+                # Annahme: Transport-API wirft SignalduinoConnectionError bei Trennung.
+                # Wenn dies nicht der Fall ist, wird ein Timeout angenommen.
+                self.logger.warning(
+                    "Command '%s' timed out. Treating as no response from device.", payload
+                )
+                raise SignalduinoCommandTimeout(f"Command '{payload}' timed out") from None
 
     async def _start_heartbeat_task(self) -> None:
         """Schedules the periodic status heartbeat task."""
@@ -670,17 +683,29 @@ async def run(self, timeout: Optional[float] = None) -> None:
         """
         self.logger.info("Starting main controller tasks...")
 
-        # 1. Initialisierung starten (führt Versionsprüfung durch und startet Heartbeat)
-        await self.initialize()
-
-        # 2. Haupt-Tasks erstellen und starten
+        # 1. Haupt-Tasks erstellen und starten (Muss VOR initialize() erfolgen, damit der Reader
+        # die Initialisierungsantwort empfangen kann)
         reader_task = asyncio.create_task(self._reader_task(), name="sd-reader")
         parser_task = asyncio.create_task(self._parser_task(), name="sd-parser")
         writer_task = asyncio.create_task(self._writer_task(), name="sd-writer")
 
         self._main_tasks = [reader_task, parser_task, writer_task]
+        
+        # 2. Initialisierung starten (führt Versionsprüfung durch und startet Heartbeat)
+        await self.initialize()
+        
+        # 3. Auf den Abschluss der Initialisierung warten (mit zusätzlichem Timeout)
+        try:
+            self.logger.info("Waiting for initialization to complete...")
+            await asyncio.wait_for(self._init_complete_event.wait(), timeout=SDUINO_CMD_TIMEOUT * 2)
+            self.logger.info("Initialization complete.")
+        except asyncio.TimeoutError:
+            self.logger.error("Initialization timed out after %s seconds.", SDUINO_CMD_TIMEOUT * 2)
+            # Wenn die Initialisierung fehlschlägt, stoppen wir den Controller (aexit)
+            self._stop_event.set()
+            # Der Timeout kann dazu führen, dass die await-Kette unterbrochen wird. Wir fahren fort.
 
-        # 3. Auf eine der Haupt-Tasks warten (Reader/Writer werden bei Verbindungsabbruch beendet)
+        # 4. Auf eine der kritischen Haupt-Tasks warten (Reader/Writer werden bei Verbindungsabbruch beendet)
         # Parser sollte weiterlaufen, bis die Queue leer ist. Reader/Writer sind die kritischen Tasks.
         critical_tasks = [reader_task, writer_task]
 

signalduino/mqtt.py

@@ -105,7 +105,8 @@ async def _command_listener(self) -> None:
                         if "commands" in parts:
                             cmd_index = parts.index("commands")
                             if len(parts) > cmd_index + 1:
-                                command_name = parts[cmd_index + 1]
+                                # Nimm den Rest des Pfades als Command-Name (für Unterbefehle wie set/XE)
+                                command_name = "/".join(parts[cmd_index + 1:])
                                 # Callback ist jetzt async
                                 await self.command_callback(command_name, payload)
                             else:

signalduino/transport.py

@@ -34,6 +34,10 @@ async def readline(self) -> Optional[str]:  # pragma: no cover - interface
         # Wir entfernen das Timeout-Argument, da wir dies mit asyncio.wait_for im Controller handhaben
         raise NotImplementedError
 
+    def closed(self) -> bool:  # pragma: no cover - interface
+        """Returns True if the transport is closed, False otherwise."""
+        raise NotImplementedError
+
     # is_open wird entfernt, da es in async-Umgebungen schwer zu implementieren ist
     # und die Transportfehler (SignalduinoConnectionError) zur Beendigung führen.
 
@@ -66,6 +70,9 @@ async def readline(self) -> Optional[str]:
         await asyncio.Future() # Hängt die Coroutine auf
         raise NotImplementedError("Asynchronous serial transport is not implemented yet.")
 
+    def closed(self) -> bool:
+        return self._serial is None
+
 
 class TCPTransport(BaseTransport):
     """Asynchronous TCP transport using asyncio streams."""
@@ -93,6 +100,9 @@ async def close(self) -> None:
             self._reader = None
             logger.info("TCPTransport closed.")
 
+    def closed(self) -> bool:
+        return self._writer is None
+
     async def write_line(self, data: str) -> None:
         if not self._writer:
             raise SignalduinoConnectionError("TCPTransport is not open")

tests/conftest.py

@@ -1,12 +1,13 @@
 import logging
-from unittest.mock import MagicMock
+import asyncio
+from unittest.mock import MagicMock, Mock, AsyncMock
 
 import pytest
+import pytest_asyncio
 
 from sd_protocols import SDProtocols
 from signalduino.types import DecodedMessage
-
-
+from signalduino.controller import SignalduinoController
 
 
 @pytest.fixture
@@ -24,4 +25,50 @@ def proto():
 def mock_protocols(mocker):
     """Fixture for a mocked SDProtocols instance."""
     mock = mocker.patch("signalduino.parser.mc.SDProtocols", autospec=True)
-    return mock.return_value
+    return mock.return_value
+
+
+@pytest.fixture
+def mock_transport():
+    """Fixture for a mocked async transport layer."""
+    transport = AsyncMock()
+    transport.is_open = True
+    transport.write_line = AsyncMock()
+    
+    async def aopen_mock():
+        transport.is_open = True
+    
+    async def aclose_mock():
+        transport.is_open = False
+
+    transport.aopen.side_effect = aopen_mock
+    transport.aclose.side_effect = aclose_mock
+    transport.__aenter__.return_value = transport
+    transport.__aexit__.return_value = None
+    transport.readline.return_value = None
+    return transport
+
+
+@pytest_asyncio.fixture
+async def controller(mock_transport):
+    """Fixture for a SignalduinoController with a mocked transport."""
+    ctrl = SignalduinoController(transport=mock_transport)
+
+    # Verwende eine interne Queue, um das Verhalten zu simulieren
+    # Da die Tests die Queue direkt mocken, lasse ich die Mock-Logik so, wie sie ist.
+    
+    async def mock_put(queued_command):
+        # Simulate an immediate async response for commands that expect one.
+        if queued_command.expect_response and queued_command.on_response:
+            # For Set-Commands, the response is often an echo of the command itself or 'OK'.
+            queued_command.on_response(queued_command.payload)
+
+    # We mock the queue to directly call the response callback (now async)
+    ctrl._write_queue = AsyncMock()
+    ctrl._write_queue.put.side_effect = mock_put
+
+    # Da der Controller ein async-Kontextmanager ist, müssen wir ihn im Test
+    # als solchen verwenden, was nicht in der Fixture selbst geschehen kann.
+    # Wir geben das Objekt zurück und erwarten, dass der Test await/async with verwendet.
+    async with ctrl:
+        yield ctrl

tests/test_connection_drop.py

@@ -1,8 +1,9 @@
-import queue
-import threading
-import time
+import asyncio
 import unittest
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, AsyncMock
+from typing import Optional
+
+import pytest
 
 from signalduino.controller import SignalduinoController
 from signalduino.exceptions import SignalduinoCommandTimeout, SignalduinoConnectionError
@@ -11,72 +12,78 @@
 class MockTransport(BaseTransport):
     def __init__(self):
         self.is_open_flag = False
-        self.output_queue = queue.Queue()
+        self.output_queue = asyncio.Queue()
 
-    def open(self):
+    async def aopen(self):
         self.is_open_flag = True
 
-    def close(self):
+    async def aclose(self):
         self.is_open_flag = False
 
+    async def __aenter__(self):
+        await self.aopen()
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        await self.aclose()
+        
     @property
-    def is_open(self):
+    def is_open(self) -> bool:
         return self.is_open_flag
+    
+    def closed(self) -> bool:
+        return not self.is_open_flag
 
-    def write_line(self, data):
+    async def write_line(self, data: str) -> None:
         if not self.is_open_flag:
             raise SignalduinoConnectionError("Closed")
 
-    def readline(self, timeout=None):
+    async def readline(self, timeout: Optional[float] = None) -> Optional[str]:
         if not self.is_open_flag:
              raise SignalduinoConnectionError("Closed")
         try:
-            return self.output_queue.get(timeout=timeout or 0.1)
-        except queue.Empty:
+            # await output_queue.get with timeout
+            line = await asyncio.wait_for(self.output_queue.get(), timeout=timeout or 0.1)
+            return line
+        except asyncio.TimeoutError:
             return None
 
-class TestConnectionDrop(unittest.TestCase):
-    def test_timeout_normally(self):
-        """Test that a simple timeout raises SignalduinoCommandTimeout."""
-        transport = MockTransport()
-        controller = SignalduinoController(transport)
-        controller.connect()
-        
-        # Expect SignalduinoCommandTimeout because transport sends nothing
-        with self.assertRaises(SignalduinoCommandTimeout):
-            controller.send_command("V", expect_response=True, timeout=0.5)
-            
-        controller.disconnect()
-
-    def test_connection_drop_during_command(self):
-        """Test that if connection dies during command wait, we get ConnectionError."""
-        transport = MockTransport()
-        controller = SignalduinoController(transport)
-        controller.connect()
-
-        # We need to simulate the reader loop crashing or transport closing
-        # signalduino controller checks transport.is_open or _stop_event
-        
-        # Hook into write_line to close transport immediately after sending
-        # simulating a crash right after send
-        original_write = transport.write_line
-        def side_effect(data):
-            original_write(data)
-            # Simulate connection loss
-            transport.close()
-            # Also set stop event as reader loop would
-            controller._stop_event.set()
-            
-        transport.write_line = side_effect
-
-        # Current behavior: Raises SignalduinoCommandTimeout because it just waits on queue
-        # Desired behavior: Raises SignalduinoConnectionError because connection is dead
-        
-        try:
+@pytest.mark.asyncio
+async def test_timeout_normally():
+    """Test that a simple timeout raises SignalduinoCommandTimeout."""
+    transport = MockTransport()
+    controller = SignalduinoController(transport)
+    
+    # Expect SignalduinoCommandTimeout because transport sends nothing
+    async with controller:
+        with pytest.raises(SignalduinoCommandTimeout):
+            await controller.send_command("V", expect_response=True, timeout=0.5)
+
+
+@pytest.mark.asyncio
+async def test_connection_drop_during_command():
+    """Test that if connection dies during command wait, we get ConnectionError."""
+    transport = MockTransport()
+    controller = SignalduinoController(transport)
+
+    # The synchronous exception handler must be replaced by try/except within an async context
+    
+    async with controller:
+        cmd_task = asyncio.create_task(
             controller.send_command("V", expect_response=True, timeout=1.0)
-        except Exception as e:
-            print(f"Caught exception: {type(e).__name__}: {e}")
-            # validating what it currently raises
-            # self.assertIsInstance(e, SignalduinoConnectionError) 
+        )
+
+        # Give the command a chance to be sent and be in a waiting state
+        await asyncio.sleep(0.001)
+
+        # Simulate connection loss and cancel main task to trigger cleanup
+        await transport.aclose()
+        # controller._main_task.cancel() # Entfernt, da es in der neuen Controller-Version nicht mehr notwendig ist und Fehler verursacht.
+        
+        # Introduce a small delay to allow the event loop to process the connection drop
+        # and set the controller's _stop_event before the command times out.
+        await asyncio.sleep(0.01)
 
-        controller.disconnect()
+        with pytest.raises((SignalduinoConnectionError, asyncio.CancelledError, asyncio.TimeoutError)):
+             # send_command should raise an exception because the connection is dead
+            await cmd_task