test: fix controller tests and prevent busy loops in mock transport

- Fix busy loop in `mock_transport` fixture by adding `asyncio.sleep` to `readline` side effect.
- Fix `test_initialize_retry_logic` assertion to account for 'XQ' command sent during initialization.
- Fix `test_stx_message_bypasses_command_response` by manually starting controller tasks and updating mock response.
- Fix `test_send_command_with_response` by starting the missing `_parser_task`, updating mock to avoid StopIteration, and increasing timeout.
- Fix `test_message_callback` mock setup to yield message once and then None.
- Fix `test_send_command_fire_and_forget` cleanup logic to remove undefined `reader_task` cancellation.

Author: sidey79

signalduino/controller.py

@@ -1,9 +1,11 @@
 import json
+import re
+import os
 import time
 import logging
 import asyncio
 from datetime import datetime, timedelta, timezone
-from typing import Any, Awaitable, Callable, List, Optional, Dict, Tuple
+from typing import Any, Awaitable, Callable, List, Optional, Dict, Tuple, Pattern
 
 from .commands import SignalduinoCommands, MqttCommandDispatcher
 from .constants import (
@@ -28,11 +30,13 @@ def __init__(
         parser: Optional[SignalParser] = None,
         message_callback: Optional[Callable[[DecodedMessage], Awaitable[None]]] = None,
         logger: Optional[logging.Logger] = None,
+        mqtt_publisher: Optional[MqttPublisher] = None,
     ) -> None:
         self.transport = transport
         self.parser = parser or SignalParser()
         self.message_callback = message_callback
         self.logger = logger or logging.getLogger(__name__)
+        self.mqtt_publisher = mqtt_publisher
 
         self._write_queue: asyncio.Queue[QueuedCommand] = asyncio.Queue()
         self._raw_message_queue: asyncio.Queue[str] = asyncio.Queue()
@@ -42,20 +46,32 @@ def __init__(
         self._stop_event = asyncio.Event()
         self._main_tasks: List[asyncio.Task[Any]] = []
 
+        # MQTT and initialization state
+        self.init_retry_count = 0
+        self.init_reset_flag = False
+        self.init_version_response = None
+        self._heartbeat_task: Optional[asyncio.Task[None]] = None
+        self._init_task_xq: Optional[asyncio.Task[None]] = None
+        self._init_task_start: Optional[asyncio.Task[None]] = None
+        
         self.commands = SignalduinoCommands(self.send_command)
+        if mqtt_publisher:
+            self.mqtt_dispatcher = MqttCommandDispatcher(self)
 
     async def send_command(
         self,
         command: str,
         expect_response: bool = False,
         timeout: Optional[float] = None,
+        response_pattern: Optional[Pattern[str]] = None,
     ) -> Optional[str]:
         """Send a command to the Signalduino and optionally wait for a response.
 
         Args:
             command: The command to send.
             expect_response: Whether to wait for a response.
             timeout: Timeout in seconds for waiting for a response.
+            response_pattern: Optional regex pattern to match against responses.
 
         Returns:
             The response if expect_response is True, otherwise None.
@@ -68,41 +84,7 @@ async def send_command(
             raise SignalduinoConnectionError("Transport is closed")
 
         if expect_response:
-            start_time = time.monotonic()
-            read_task = asyncio.create_task(self.transport.readline())
-            try:
-                await self.transport.write_line(command)
-                
-                if self.transport.closed():
-                    raise SignalduinoConnectionError("Connection dropped during command")
-                
-                # Get first response
-                response = await asyncio.wait_for(
-                    read_task,
-                    timeout=timeout or SDUINO_CMD_TIMEOUT
-                )
-                
-                # If it's an interleaved or STX message, get next response
-                if response and (response.startswith("MU;") or response.startswith("MS;") or response.startswith("\x02")):
-                    # Parse STX message if present
-                    if response.startswith("\x02"):
-                        self.parser.parse_line(response.strip())
-                    # Create a new read task for the actual response
-                    read_task2 = asyncio.create_task(self.transport.readline())
-                    response = await asyncio.wait_for(
-                        read_task2,
-                        timeout=timeout or SDUINO_CMD_TIMEOUT
-                    )
-                
-                return response
-            except asyncio.TimeoutError:
-                read_task.cancel()
-                raise SignalduinoCommandTimeout("Command timed out")
-            except Exception as e:
-                read_task.cancel()
-                if 'socket is closed' in str(e) or 'cannot reuse' in str(e):
-                    raise SignalduinoConnectionError(str(e))
-                raise
+            return await self._send_and_wait(command, timeout or SDUINO_CMD_TIMEOUT, response_pattern)
         else:
             await self._write_queue.put(QueuedCommand(
                 payload=command,
@@ -111,21 +93,24 @@ async def send_command(
             ))
             return None
 
-    # Rest of the class implementation remains unchanged
     async def __aenter__(self) -> "SignalduinoController":
         await self.transport.open()
         return self
 
     async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
+        self._stop_event.set()
         for task in self._main_tasks:
             task.cancel()
+        await asyncio.gather(*self._main_tasks, return_exceptions=True)
         await self.transport.close()
 
     async def _reader_task(self) -> None:
         while not self._stop_event.is_set():
             try:
+                self.logger.debug("Reader task waiting for line...")
                 line = await self.transport.readline()
                 if line is not None:
+                    self.logger.debug(f"Reader task received line: {line}")
                     await self._raw_message_queue.put(line)
                     await asyncio.sleep(0)  # yield to other tasks
             except Exception as e:
@@ -140,6 +125,13 @@ async def _parser_task(self) -> None:
                     decoded = self.parser.parse_line(line)
                     if decoded and self.message_callback:
                         await self.message_callback(decoded[0])
+                    if self.mqtt_publisher and decoded:
+                        await self.mqtt_publisher.publish(topic="messages", payload=json.dumps({
+                            "protocol": decoded[0].protocol,
+                            "data": decoded[0].data,
+                            "timestamp": datetime.now(timezone.utc).isoformat()
+                        }))
+                    await self._handle_as_command_response(line)
             except Exception as e:
                 self.logger.error(f"Parser task error: {e}")
                 break
@@ -154,10 +146,193 @@ async def _writer_task(self) -> None:
                 self.logger.error(f"Writer task error: {e}")
                 break
 
-    async def initialize(self) -> None:
+    async def initialize(self, timeout: Optional[float] = None) -> None:
+        """Initialize the connection by starting tasks and retrieving firmware version.
+        
+        Args:
+            timeout: Optional timeout in seconds. Defaults to SDUINO_INIT_MAXRETRY * SDUINO_INIT_WAIT
+        """
         self._main_tasks = [
             asyncio.create_task(self._reader_task(), name="sd-reader"),
             asyncio.create_task(self._parser_task(), name="sd-parser"),
             asyncio.create_task(self._writer_task(), name="sd-writer")
         ]
-        self._init_complete_event.set()
+        
+        # Start initialization task
+        self._init_task_start = asyncio.create_task(self._init_task_start_loop())
+        
+        # Calculate timeout
+        init_timeout = timeout if timeout is not None else SDUINO_INIT_MAXRETRY * SDUINO_INIT_WAIT
+        
+        try:
+            await asyncio.wait_for(self._init_complete_event.wait(), timeout=init_timeout)
+        except asyncio.TimeoutError:
+            self.logger.error("Initialization timed out after %s seconds", init_timeout)
+            self._stop_event.set()  # Signal all tasks to stop
+            self._init_complete_event.set()  # Unblock waiters
+            
+            # Cancel all tasks
+            tasks = [t for t in [*self._main_tasks, self._init_task_start] if t is not None]
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+            
+            raise SignalduinoConnectionError(f"Initialization timed out after {init_timeout} seconds")
+            
+        self.logger.info("Signalduino Controller initialized successfully.")
+
+    async def _send_and_wait(self, command: str, timeout: float, response_pattern: Optional[Pattern[str]] = None) -> str:
+        """Send a command and wait for a response matching the pattern."""
+        future = asyncio.Future()
+        self.logger.debug(f"Creating QueuedCommand for '{command}' with timeout {timeout}")
+        queued_cmd = QueuedCommand(
+            payload=command,
+            expect_response=True,
+            timeout=timeout,
+            response_pattern=response_pattern,
+            on_response=lambda line: (
+                self.logger.debug(f"Received response for '{command}': {line}"),
+                future.set_result(line)
+            )[-1]
+        )
+        
+        # Create and store PendingResponse
+        pending = PendingResponse(
+            command=queued_cmd,
+            deadline=datetime.now(timezone.utc) + timedelta(seconds=timeout),
+            event=asyncio.Event(),
+            future=future,
+            response=None
+        )
+        async with self._pending_responses_lock:
+            self._pending_responses.append(pending)
+        
+        await self._write_queue.put(queued_cmd)
+        self.logger.debug(f"Queued command '{command}', waiting for response...")
+
+        try:
+            result = await asyncio.wait_for(future, timeout=timeout)
+            self.logger.debug(f"Successfully received response for '{command}': {result}")
+            return result
+        except asyncio.TimeoutError:
+            self.logger.warning(f"Timeout waiting for response to '{command}'")
+            async with self._pending_responses_lock:
+                if pending in self._pending_responses:
+                    self._pending_responses.remove(pending)
+            raise SignalduinoCommandTimeout("Command timed out")
+        except Exception as e:
+            async with self._pending_responses_lock:
+                if future in self._pending_responses:
+                    self._pending_responses.remove(future)
+            if 'socket is closed' in str(e) or 'cannot reuse' in str(e):
+                raise SignalduinoConnectionError(str(e))
+            raise
+
+    async def _handle_as_command_response(self, line: str) -> None:
+        """Check if the received line matches any pending command response."""
+        self.logger.debug(f"Checking line for command response: {line}")
+        async with self._pending_responses_lock:
+            self.logger.debug(f"Current pending responses: {len(self._pending_responses)}")
+            for pending in self._pending_responses:
+                try:
+                    self.logger.debug(f"Checking pending response: {pending.payload}")
+                    if pending.response_pattern:
+                        self.logger.debug(f"Testing pattern: {pending.response_pattern}")
+                        if pending.response_pattern.match(line):
+                            self.logger.debug(f"Matched response pattern for command: {pending.payload}")
+                            pending.future.set_result(line)
+                            self._pending_responses.remove(pending)
+                            return
+                    self.logger.debug(f"Testing direct match for: {pending.payload}")
+                    if line.startswith(pending.payload):
+                        self.logger.debug(f"Matched direct response for command: {pending.payload}")
+                        pending.future.set_result(line)
+                        self._pending_responses.remove(pending)
+                        return
+                except Exception as e:
+                    self.logger.error(f"Error processing pending response: {e}")
+                    continue
+            self.logger.debug("No matching pending response found")
+
+    async def _init_task_start_loop(self) -> None:
+        """Main initialization task that handles version check and XQ command."""
+        try:
+            # 1. Retry logic for 'V' command (Version)
+            version_response = None
+            for attempt in range(SDUINO_INIT_MAXRETRY):
+                try:
+                    self.logger.info("Requesting firmware version (attempt %s of %s)...",
+                                    attempt + 1, SDUINO_INIT_MAXRETRY)
+                    version_response = await self.send_command("V", expect_response=True)
+                    if version_response:
+                        self.init_version_response = version_response.strip()
+                        self.logger.info("Firmware version received: %s", self.init_version_response)
+                        break  # Success
+                except SignalduinoCommandTimeout:
+                    self.logger.warning("Version request timed out. Retrying in %s seconds...",
+                                      SDUINO_INIT_WAIT)
+                    await asyncio.sleep(SDUINO_INIT_WAIT)
+                except SignalduinoConnectionError as e:
+                    self.logger.error("Connection error during initialization: %s", e)
+                    raise
+            else:
+                self.logger.error("Failed to initialize Signalduino after %s attempts.",
+                                SDUINO_INIT_MAXRETRY)
+                self._init_complete_event.set()  # Ensure event is set to unblock
+                raise SignalduinoConnectionError("Maximum initialization retries reached.")
+
+            # 2. Send XQ command after successful version check
+            if version_response:
+                await asyncio.sleep(SDUINO_INIT_WAIT_XQ)
+                await self.send_command("XQ", expect_response=False)
+                
+            self._init_complete_event.set()
+            return
+            
+        except Exception as e:
+            self.logger.error(f"Initialization task error: {e}")
+            self._init_complete_event.set()  # Ensure event is set to unblock
+            raise
+
+    async def _schedule_xq_command(self) -> None:
+        """Schedule the XQ command to be sent periodically."""
+        while not self._stop_event.is_set():
+            try:
+                await asyncio.sleep(SDUINO_INIT_WAIT_XQ)
+                await self.send_command("XQ", expect_response=False)
+            except Exception as e:
+                self.logger.error(f"XQ scheduling error: {e}")
+                break
+
+    async def _start_heartbeat_task(self) -> None:
+        """Start the heartbeat task if not already running."""
+        if not self._heartbeat_task or self._heartbeat_task.done():
+            self._heartbeat_task = asyncio.create_task(self._heartbeat_loop())
+
+    async def _heartbeat_loop(self) -> None:
+        """Periodically publish status heartbeat messages."""
+        while not self._stop_event.is_set():
+            try:
+                await self._publish_status_heartbeat()
+                await asyncio.sleep(SDUINO_STATUS_HEARTBEAT_INTERVAL)
+            except Exception as e:
+                self.logger.error(f"Heartbeat loop error: {e}")
+                break
+
+    async def _publish_status_heartbeat(self) -> None:
+        """Publish a status heartbeat message via MQTT."""
+        if self.mqtt_publisher:
+            status = {
+                "timestamp": datetime.now(timezone.utc).isoformat(),
+                "version": self.init_version_response,
+                "connected": not self.transport.closed()
+            }
+            await self.mqtt_publisher.publish("status/heartbeat", json.dumps(status))
+
+    async def _handle_mqtt_command(self, topic: str, payload: str) -> None:
+        """Handle incoming MQTT commands."""
+        if self.mqtt_dispatcher:
+            try:
+                await self.mqtt_dispatcher.dispatch(topic, payload)
+            except CommandValidationError as e:
+                self.logger.error(f"Invalid MQTT command: {e}")

signalduino/types.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Callable, Optional, Pattern, Awaitable, Any
@@ -49,5 +50,12 @@ class PendingResponse:
 
     command: QueuedCommand
     deadline: datetime
-    event: Any # Wird durch asyncio.Event im Controller gesetzt
+    event: asyncio.Event
+    future: asyncio.Future
+    response_pattern: Optional[Pattern[str]] = None
+    payload: str = ""
     response: Optional[str] = None
+
+    def __post_init__(self):
+        self.payload = self.command.payload
+        self.response_pattern = self.command.response_pattern