fix: some command responses with tests

Author: sidey79

signalduino/commands.py

@@ -35,11 +35,15 @@ def get_help(self) -> str:
 
     def get_free_ram(self) -> str:
         """Query free RAM (R)."""
-        return self._send("R", expect_response=True, timeout=2.0, response_pattern=None)
+        # Response is typically a number (bytes)
+        pattern = re.compile(r"^\d+$")
+        return self._send("R", expect_response=True, timeout=2.0, response_pattern=pattern)
 
     def get_uptime(self) -> str:
         """Query uptime in seconds (t)."""
-        return self._send("t", expect_response=True, timeout=2.0, response_pattern=None)
+        # Response is a number (seconds)
+        pattern = re.compile(r"^\d+$")
+        return self._send("t", expect_response=True, timeout=2.0, response_pattern=pattern)
 
     def ping(self) -> str:
         """Ping device (P)."""
@@ -65,7 +69,9 @@ def factory_reset(self) -> str:
 
     def get_config(self) -> str:
         """Read configuration (CG)."""
-        return self._send("CG", expect_response=True, timeout=2.0, response_pattern=None)
+        # Response format: MS=1;MU=1;...
+        pattern = re.compile(r"^MS=.*")
+        return self._send("CG", expect_response=True, timeout=2.0, response_pattern=pattern)
 
     def set_decoder_state(self, decoder: str, enabled: bool) -> None:
         """
@@ -133,12 +139,16 @@ def init_wmbus(self) -> str:
     def read_eeprom(self, address: int) -> str:
         """Read EEPROM byte (r<addr>)."""
         addr_hex = f"{address:02X}"
-        return self._send(f"r{addr_hex}", expect_response=True, timeout=2.0, response_pattern=None)
+        # Response format: EEPROM <addr> = <val>
+        pattern = re.compile(r"EEPROM.*", re.IGNORECASE)
+        return self._send(f"r{addr_hex}", expect_response=True, timeout=2.0, response_pattern=pattern)
 
     def read_eeprom_block(self, address: int) -> str:
         """Read EEPROM block (r<addr>n)."""
         addr_hex = f"{address:02X}"
-        return self._send(f"r{addr_hex}n", expect_response=True, timeout=2.0, response_pattern=None)
+        # Response format: EEPROM <addr> : <val> ...
+        pattern = re.compile(r"EEPROM.*", re.IGNORECASE)
+        return self._send(f"r{addr_hex}n", expect_response=True, timeout=2.0, response_pattern=pattern)
 
     def set_patable(self, value: str | int) -> str:
         """Write PA Table (x<val>)."""

tests/test_controller.py

@@ -95,22 +95,81 @@ def readline_side_effect():
     controller = SignalduinoController(transport=mock_transport, parser=mock_parser)
     controller.connect()
     try:
-        response = controller.send_command("V", expect_response=True, timeout=1)
+        response = controller.commands.get_version(timeout=1)
         mock_transport.write_line.assert_called_with("V")
         assert response is not None
         assert "SIGNALduino" in response
     finally:
         controller.disconnect()
 
 
+def test_send_command_with_interleaved_message(mock_transport, mock_parser):
+    """
+    Test sending a command and receiving an irrelevant message before the
+    expected command response. The irrelevant message must not be consumed
+    as the response, and the correct response must still be received.
+    """
+    # Queue for all messages from the device
+    response_q = queue.Queue()
+
+    # The irrelevant message (e.g., an asynchronous received signal)
+    interleaved_message = "MU;P0=353;P1=-184;D=0123456789;CP=1;SP=0;R=248;\n"
+    # The expected command response
+    command_response = "V 3.5.0-dev SIGNALduino\n"
+
+    def write_line_side_effect(payload):
+        # When the controller writes "V", simulate the device responding with
+        # an interleaved message *then* the command response.
+        if payload == "V":
+            # 1. Interleaved message
+            response_q.put(interleaved_message)
+            # 2. Command response
+            response_q.put(command_response)
+
+    def readline_side_effect():
+        # Simulate blocking read that gets a value from the queue.
+        try:
+            return response_q.get(timeout=0.5)
+        except queue.Empty:
+            return None
+
+    mock_transport.write_line.side_effect = write_line_side_effect
+    mock_transport.readline.side_effect = readline_side_effect
+
+    # Mock the parser to track if the interleaved message is passed to it
+    mock_parser.parse_line = Mock(wraps=mock_parser.parse_line)
+
+    controller = SignalduinoController(transport=mock_transport, parser=mock_parser)
+    controller.connect()
+    try:
+        response = controller.commands.get_version(timeout=1)
+        mock_transport.write_line.assert_called_with("V")
+        
+        # 1. Verify that the correct command response was received by send_command
+        assert response is not None
+        assert "SIGNALduino" in response
+        assert response.strip() == command_response.strip()
+
+        # 2. Verify that the interleaved message was passed to the parser
+        # The parser loop (_parser_loop) should attempt to parse the interleaved_message
+        # because _handle_as_command_response should return False for it.
+        mock_parser.parse_line.assert_called_with(interleaved_message.strip())
+
+        # Give the parser thread a moment to process the message
+        time.sleep(0.1)
+        
+    finally:
+        controller.disconnect()
+
+
 def test_send_command_timeout(mock_transport, mock_parser):
     """Test that a command times out if no response is received."""
     mock_transport.readline.return_value = None
     controller = SignalduinoController(transport=mock_transport, parser=mock_parser)
     controller.connect()
     try:
         with pytest.raises(SignalduinoCommandTimeout):
-            controller.send_command("V", expect_response=True, timeout=0.2)
+            controller.commands.get_version(timeout=0.2)
     finally:
         controller.disconnect()