Fix hardware test serial controls and offline DM queue matching

tools/hil/companion_usb_client.py

@@ -101,7 +101,7 @@ def run(args):
             assert not initial, 'Fresh connection emitted unsolicited text or stale data'
             tty.write(b'ver\r')
             response = drain(tty, 0.4)
-            assert b'Companion v' in response, 'Default ASCII terminal did not answer'
+            assert b'Companion ' in response and b'(protocol ' in response, 'Default ASCII terminal did not answer'
             results['checks'].append('default ASCII accepts ver without START')
             tty.write(b'+++MESHCORE-TERM-STOP\r')
             assert b'OK - Binary mode' in drain(tty, 0.3), 'STOP acknowledgement missing'

tools/hil/esp32_companion_serial_stress.py

@@ -92,6 +92,7 @@ class StressConfig:
     cycle_delay: float = 0.05
     close_delay: float = 0.10
     verbose: bool = False
+    dtr: bool = False
 
 
 @dataclass
@@ -431,6 +432,8 @@ def _validate_config(config: StressConfig) -> None:
         raise ValueError("count must be at least 1")
     if config.baudrate < 1:
         raise ValueError("baudrate must be positive")
+    if not isinstance(config.dtr, bool):
+        raise ValueError("dtr must be a boolean")
     for name in config.queries:
         if name not in SAFE_QUERY_SPECS:
             raise ValueError(f"unknown or unsafe query {name!r}")
@@ -463,14 +466,15 @@ def _make_pyserial_factory() -> Callable[[StressConfig], Any]:
         ) from exc
 
     def factory(config: StressConfig) -> Any:
-        # Configure DTR/RTS while closed so opening a port never deliberately
-        # asserts a reset/boot strap control line.
+        # Native USB/JTAG devices may require DTR for the CDC session. Set both
+        # control lines while closed and keep RTS deasserted, including when
+        # opting into DTR, to avoid deliberately driving the ESP reset line.
         port = serial.Serial()
         port.port = config.port
         port.baudrate = config.baudrate
         port.timeout = min(config.read_poll_timeout, config.response_timeout)
         port.write_timeout = config.write_timeout
-        port.dtr = False
+        port.dtr = config.dtr
         port.rts = False
         port.open()
         return port
@@ -771,6 +775,10 @@ def build_argument_parser() -> argparse.ArgumentParser:
         help="cycles per selected mode (default: 100)",
     )
     parser.add_argument("--baudrate", type=_positive_int, default=115200)
+    parser.add_argument(
+        "--dtr", choices=("on", "off"), default="off",
+        help="CDC DTR state set before opening; RTS stays off (default: off)",
+    )
     parser.add_argument("--app-name", default="MeshCore-HIL-Stress")
     parser.add_argument(
         "--queries", type=_safe_query_list,
@@ -837,6 +845,7 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
         cycle_delay=args.cycle_delay,
         close_delay=args.close_delay,
         verbose=args.verbose,
+        dtr=args.dtr == "on",
     )
     summary = run_stress(config)
     print(json.dumps(summary, indent=2 if args.pretty else None, sort_keys=True))

tools/hil/one_key_dm.py

@@ -13,8 +13,6 @@ import json
 import struct
 import time
 
-import serial
-
 from esp32_companion_serial_stress import (
     DeviceFrameReader,
     FrameTimeout,
@@ -27,7 +25,16 @@ from esp32_companion_serial_stress import (
 
 class Link:
     def __init__(self, port: str) -> None:
-        self.port = serial.Serial(port, 115200, timeout=0.1, write_timeout=2)
+        import serial
+
+        # Native USB/JTAG CDC needs DTR. Configure while closed and leave RTS
+        # deasserted so the ESP reset line is not deliberately driven.
+        self.port = serial.Serial(baudrate=115200, timeout=0.1,
+                                  write_timeout=2, exclusive=True)
+        self.port.port = port
+        self.port.dtr = True
+        self.port.rts = False
+        self.port.open()
         self.reader = DeviceFrameReader(TransportCounters())
         self.pushes: list[bytes] = []
 
@@ -83,8 +90,34 @@ class Link:
         return self.request(bytes([0x42]) + command.encode(), (0x1D,))[1:].decode()
 
 
+def consume_expected_dm(recipient: Link, sender_key: bytes, timestamp: int,
+                        text: bytes, queue_capacity: int) -> dict:
+    """Find this test's DM in a bounded FIFO scan without exposing other text."""
+    if queue_capacity not in (256, 512):
+        raise ValueError("queue capacity must be 256 or 512")
+    skipped = 0
+    for _ in range(queue_capacity):
+        queued = recipient.request(bytes([10]), (7, 8, 10, 16, 17))
+        if queued[0] == 10:
+            return {"matched": False, "skipped": skipped, "empty": True}
+        if queued[0] in (7, 16):
+            start = 1 if queued[0] == 7 else 4
+            if len(queued) < start + 12:
+                raise RuntimeError("truncated queued contact message")
+            if (queued[start:start + 6] == sender_key[:6]
+                    and queued[start + 7] == 0  # plaintext DM
+                    and struct.unpack_from("<I", queued, start + 8)[0] == timestamp
+                    and queued[start + 12:] == text):
+                return {"matched": True, "skipped": skipped, "empty": False}
+        skipped += 1
+    return {"matched": False, "skipped": skipped, "empty": False}
+
+
 def run(sender_port: str, recipient_port: str, reset_contact: bool,
-        zero_hop: bool, invalid_signature_first: bool) -> dict:
+        zero_hop: bool, invalid_signature_first: bool,
+        queue_capacity: int = 256) -> dict:
+    if queue_capacity not in (256, 512):
+        raise ValueError("queue capacity must be 256 or 512")
     sender = Link(sender_port)
     recipient = Link(recipient_port)
     prior_dm_setting = None
@@ -188,8 +221,9 @@ def run(sender_port: str, recipient_port: str, reset_contact: bool,
         )
         waiting = any(frame[0] == 0x83 for frame in recipient.pushes)
         learned = recipient.contact(sender_key) is not None
-        queued = recipient.request(bytes([10]), (7, 10, 16))
-        replayed = queued[0] in (7, 16) and text in queued
+        queue_scan = consume_expected_dm(recipient, sender_key, timestamp,
+                                         text, queue_capacity)
+        replayed = queue_scan["matched"]
         result = {
             "sender_prefix": sender_key[:6].hex(),
             "recipient_prefix": recipient_key[:6].hex(),
@@ -200,6 +234,9 @@ def run(sender_port: str, recipient_port: str, reset_contact: bool,
             "synthetic_advert_offered": offered is not None,
             "signed_refusal_received": rejected,
             "held_message_replayed": replayed,
+            "offline_messages_skipped": queue_scan["skipped"],
+            "offline_queue_empty": queue_scan["empty"],
+            "offline_queue_scan_limit": queue_capacity,
             "timeout_ms": timeout_ms,
         }
         if invalid_signature_first:
@@ -227,7 +264,10 @@ if __name__ == "__main__":
                         help="set a direct zero-hop path; use only when both radios are nearby")
     parser.add_argument("--invalid-signature-first", action="store_true",
                         help="verify an encrypted introduction with a bad signature is rejected")
+    parser.add_argument("--queue-capacity", type=int, choices=(256, 512), default=256,
+                        help="recipient offline queue capacity; 512 for PSRAM builds (default: 256)")
     args = parser.parse_args()
     print(json.dumps(run(args.sender, args.recipient, args.reset_contact,
-                         args.zero_hop, args.invalid_signature_first),
+                         args.zero_hop, args.invalid_signature_first,
+                         args.queue_capacity),
                      sort_keys=True))

tools/hil/test_esp32_companion_serial_stress.py

@@ -7,6 +7,8 @@ from contextlib import redirect_stdout
 import io
 import json
 import struct
+import sys
+import types
 import unittest
 from unittest import mock
 
@@ -212,6 +214,35 @@ class PayloadValidationTest(unittest.TestCase):
 
 
 class StressModesTest(unittest.TestCase):
+    def test_selected_control_lines_are_set_while_port_is_closed(self) -> None:
+        class Serial:
+            def __init__(self):
+                self.is_open = False
+
+            def __setattr__(self, name, value):
+                if name in ("dtr", "rts"):
+                    assert not self.is_open
+                object.__setattr__(self, name, value)
+
+            def open(self):
+                assert self.rts is False
+                self.is_open = True
+
+        with mock.patch.dict(sys.modules, {"serial": types.SimpleNamespace(Serial=Serial)}):
+            factory = hil._make_pyserial_factory()
+            for dtr in (False, True):
+                with self.subTest(dtr=dtr):
+                    port = factory(fast_config(dtr=dtr))
+                    self.assertIs(port.dtr, dtr)
+                    self.assertTrue(port.is_open)
+
+    def test_dtr_cli_opt_in_preserves_default_and_rejects_non_boolean_config(self) -> None:
+        parser = hil.build_argument_parser()
+        self.assertEqual(parser.parse_args(["--port", "FAKE"]).dtr, "off")
+        self.assertEqual(parser.parse_args(["--port", "FAKE", "--dtr", "on"]).dtr, "on")
+        with self.assertRaisesRegex(ValueError, "dtr must be a boolean"):
+            hil._validate_config(fast_config(dtr="on"))
+
     def test_persistent_mode_uses_one_synchronously_closed_port(self) -> None:
         factory = FakeFactory(prefix=b"startup> ")
         config = fast_config(mode="persistent", count=3)

tools/hil/test_one_key_dm.py

@@ -0,0 +1,116 @@
+"""Offline queue and serial-control regressions for the one-key DM HIL run."""
+
+import struct
+import sys
+import types
+import unittest
+from unittest import mock
+
+import one_key_dm as hil
+
+
+SENDER = bytes(range(32))
+TIMESTAMP = 1_791_096_123
+TEXT = b"one-key DM HIL 1791096123"
+
+
+def contact_frame(code=7, text=TEXT, sender=SENDER, timestamp=TIMESTAMP):
+    header = bytes([code]) + (b"\0\0\0" if code == 16 else b"")
+    return header + sender[:6] + b"\xff\0" + struct.pack("<I", timestamp) + text
+
+
+class Queue:
+    def __init__(self, frames):
+        self.frames = iter(frames)
+        self.reads = 0
+
+    def request(self, payload, expected):
+        assert payload == bytes([10])
+        assert expected == (7, 8, 10, 16, 17)
+        self.reads += 1
+        return next(self.frames)
+
+
+class OneKeyDmTest(unittest.TestCase):
+    def scan(self, queue, capacity=256):
+        return hil.consume_expected_dm(queue, SENDER, TIMESTAMP, TEXT, capacity)
+
+    def test_channel_messages_before_dm_are_skipped_for_both_protocol_versions(self):
+        for code in (7, 16):
+            with self.subTest(code=code):
+                queue = Queue([b"\x08ambient channel", b"\x11another channel",
+                               contact_frame(code)])
+                self.assertEqual(self.scan(queue),
+                                 {"matched": True, "skipped": 2, "empty": False})
+                self.assertEqual(queue.reads, 3)
+
+    def test_earlier_unrelated_dms_cannot_match_by_substring_or_other_sender(self):
+        queue = Queue([
+            contact_frame(text=b"earlier " + TEXT),
+            contact_frame(sender=bytes(range(1, 33))),
+            contact_frame(timestamp=TIMESTAMP - 1),
+            contact_frame(16),
+        ])
+        self.assertEqual(self.scan(queue),
+                         {"matched": True, "skipped": 3, "empty": False})
+
+    def test_empty_queue_stops_without_reading_another_frame(self):
+        queue = Queue([b"\x0a", contact_frame()])
+        self.assertEqual(self.scan(queue),
+                         {"matched": False, "skipped": 0, "empty": True})
+        self.assertEqual(queue.reads, 1)
+
+    def test_nonmatching_queue_stops_at_each_supported_capacity(self):
+        for capacity in (256, 512):
+            with self.subTest(capacity=capacity):
+                queue = Queue([contact_frame(text=b"unrelated")] * capacity
+                              + [contact_frame()])
+                self.assertEqual(self.scan(queue, capacity),
+                                 {"matched": False, "skipped": capacity, "empty": False})
+                self.assertEqual(queue.reads, capacity)
+
+    def test_dm_in_last_queue_slot_is_still_found(self):
+        queue = Queue([b"\x08ambient"] * 255 + [contact_frame()])
+        self.assertTrue(self.scan(queue)["matched"])
+        self.assertEqual(queue.reads, 256)
+
+    def test_truncated_contact_fails_without_exposing_its_contents(self):
+        with self.assertRaisesRegex(RuntimeError, "^truncated queued contact message$"):
+            self.scan(Queue([b"\x07sensitive"]))
+
+    def test_invalid_capacity_does_not_consume_queue(self):
+        queue = Queue([contact_frame()])
+        with self.assertRaises(ValueError):
+            self.scan(queue, 1024)
+        self.assertEqual(queue.reads, 0)
+
+    def test_serial_control_lines_and_exclusivity_are_set_before_open(self):
+        events = []
+
+        class Serial:
+            def __init__(self, **kwargs):
+                self.kwargs = kwargs
+                self.is_open = False
+                events.append(("create", kwargs))
+
+            def __setattr__(self, name, value):
+                if name in ("dtr", "rts", "port"):
+                    assert not self.is_open
+                    events.append((name, value))
+                object.__setattr__(self, name, value)
+
+            def open(self):
+                assert self.dtr is True and self.rts is False
+                assert self.kwargs["exclusive"] is True
+                self.is_open = True
+                events.append(("open", None))
+
+        with mock.patch.dict(sys.modules, {"serial": types.SimpleNamespace(Serial=Serial)}):
+            link = hil.Link("/dev/ttyFAKE")
+        self.assertTrue(link.port.is_open)
+        self.assertLess(events.index(("dtr", True)), events.index(("open", None)))
+        self.assertLess(events.index(("rts", False)), events.index(("open", None)))
+
+
+if __name__ == "__main__":
+    unittest.main()