diff --git a/docs/TestSuite.md b/docs/TestSuite.md index 9cc8d535..f185c233 100644 --- a/docs/TestSuite.md +++ b/docs/TestSuite.md @@ -33,7 +33,7 @@ known holes), **Thin** (minimal), **Gap** (little/none). |---|---|---:|---|---|---| | **1. Facade** | `MotionInterface` | 6 | sw | **Thin** | Only construction + handle wiring (`test_motion_interface.py`). No facade-level scan/connect integration test off-hardware. | | **2. Device wrappers** | `MotionConsole`, `MotionSensor` | 93 | hw (mostly) | **Strong** | ~1:1 with the console/sensor command surface (`test_console.py` 43, `test_sensor.py` 43 ×2 sides). Software units: `test_pedestal_height.py`, `test_get_pdc_buffer.py`. Holes: I2C write→read roundtrip; IMU streaming. | -| **3. Transport & protocol** | `MotionUart`, `CommInterface`, `StreamInterface`, `UartPacket`, `CommandError` | 18 | sw + hw | Adequate | `test_errors.py` (CRC, CommandError), `test_comm_paths.py` (path isolation, hw), `test_comm_transport_down.py` (cancellation, sw). Hole: UART CRC-corruption + timeout HIL. | +| **3. Transport & protocol** | `MotionUart`, `CommInterface`, `StreamInterface`, `UartPacket`, `CommandError` | 18 | sw + hw | Adequate | `test_errors.py` (CRC, CommandError), `test_comm_paths.py` (path isolation, hw), `test_comm_transport_down.py` (cancellation, sw). Back-pressure: `test_usb_drain_stall.py` (HIL sweep — how long a host stall the firmware tolerates, issue #116) + `test_usb_drain_stall_unit.py` (its classifier, sw). Hole: UART CRC-corruption + timeout HIL. | | **4. Workflows / orchestration** | `ScanWorkflow`, `CalibrationWorkflow`, `ContactQualityWorkflow` | 89 | sw + hw | **Strong** (compute/lifecycle) | Scan (`test_scan_workflow.py`, `test_run_collection_scan.py`, `test_reduced_mode.py`, `test_sequences.py` hw), contact-quality (14), calibration workflow/compute/console/procedure (45). Holes: streaming + dual-sensor **acquisition** HIL (backlog HIGH). | | **5. Science pipeline** | `omotion/pipeline/` (+ `MotionProcessing` parse shim) | 170 | sw | **Strong** | Every stage, the dark model (estimators/history), and infra (sources, sinks, runner, tee, factory) have tests; golden-replay + determinism guard the math. Largest area. No major holes. | | **6. Storage & playback** | `ScanDatabase`, `ScanDBSink`, `SessionPlayback` | 5 (+10 sink) | sw | Adequate | `test_scan_database.py` (DB roundtrips) + `test_pipeline/test_scan_db_sink.py` (sink, counted under pipeline). Hole: `SessionPlayback` has no dedicated test. | diff --git a/tests/test_usb_drain_stall.py b/tests/test_usb_drain_stall.py new file mode 100644 index 00000000..af90a12e --- /dev/null +++ b/tests/test_usb_drain_stall.py @@ -0,0 +1,314 @@ +"""HIL characterization: how long can the host stall the USB drain before +the sensor firmware gives up? + +Why this exists +--------------- +`PipelineRunner` consumes sinks inline on the same thread that drains +`LiveUsbSource._batch_queue`, so any slow sink stops `dev.read()` from being +called. With ~0.2 s of packet queue plus a *shared* 4-slot batch queue, the +host absorbs only a few hundred ms before back-pressure reaches the sensor's +deliberately shallow 4-deep (~100 ms) histogram queue. See openmotion-sdk#116. + +In the field (bloodflow-app#300) a sub-second stall on a 20-minute scan +escalated all the way to the RIGHT module dropping off the USB bus, and no +host-side warning was emitted at any point — `StreamInterface` only warns +after a *full second* of blocked `put`, so everything below that is invisible. + +This test injects a controlled stall and records which firmware failure +signatures appear at each stall length, turning "sub-second stalls are bad" +into numbers: + + a. HISTO_QUEUE_FULL host absorption exhausted + b. USB_TX_BLOCKED firmware's histogram TX blocks + c. CAMERA_OVERRUN camera receive channels starve + d. FALSE_SCAN_END firmware abandons the scan <- sensor-fw#101 + e. ENDPOINT_LOST host sees EPIPE / device lost + +(d) is the number sensor-fw#101 needs to be specified against. (a) is the +acceptance metric for #116: with the fix in PR #113 the first stall length +that produces a queue-full should move out by roughly the ratio of the new +batch-queue budget to the current one. + +Run it +------ + pytest tests/test_usb_drain_stall.py -s -m "sensor and console" + +Needs a console and both sensors. Nothing is asserted about (b)-(e): they +are recorded and printed. The only hard assertion is the #116 budget — +see STALL_BUDGET_S. +""" + +import logging +import re +import time +from dataclasses import dataclass, field + +import pytest + +from omotion.ScanWorkflow import ScanRequest +from omotion.config import DEBUG_FLAG_USB_PRINTF + +pytestmark = [pytest.mark.sensor, pytest.mark.console, pytest.mark.slow] + + +# Stall lengths to sweep, seconds. Ordered small -> large so the run +# degrades gradually; a stall that kills the endpoint ends that scan but +# not the sweep. +STALL_SWEEP_S = (0.1, 0.2, 0.3, 0.5, 1.0, 2.0) + +# Seconds of scan per stall value. Long enough for several stalls to land +# plus one dark interval close (600 frames @ 40 Hz = 15 s). +SCAN_SEC = 40 + +# One stall every Nth "live" batch. Batches are batch_size_frames=10 at +# 40 Hz = 0.25 s, so 20 -> a stall roughly every 5 s. +STALL_EVERY_N_BATCHES = 20 + +# The host stall the system must tolerate without back-pressuring the +# firmware at all. Today's shipping code is expected to FAIL this — that +# failure is the point of the test, and the number is what PR #113 has to +# move. Tighten/relax deliberately, with a bench run to back it up. +STALL_BUDGET_S = 0.3 + + +# -------------------------------------------------------------------------- +# Firmware signatures, in escalation order +# -------------------------------------------------------------------------- + +SIGNATURES = ( + ("HISTO_QUEUE_FULL", re.compile(r"HISTO enqueue fail: queue full")), + ("USB_TX_BLOCKED", re.compile(r"USBD_HISTO_SendData failed|COMM USB TX Timeout")), + ("CAMERA_OVERRUN", re.compile(r"Error in (?:USART|SPI)\d+: Overrun error" + r"|failed to setup receive for Camera")), + ("FALSE_SCAN_END", re.compile(r"Scan finished \((\d+) ms")), + ("ENDPOINT_LOST", re.compile(r"device lost|Pipe error|errno=32")), + # Host-side: the >1 s put warning. Present == the stall was so long even + # the current coarse instrumentation caught it. + ("HOST_PUT_WARNED", re.compile(r"data_queue full for >1s")), +) + + +class _SignatureCollector(logging.Handler): + """Scan SDK log records for the firmware/host failure signatures. + + Firmware printf arrives as WARNING records on openmotion.sdk.CommInterface, + often several newline-separated lines per record, so match against the + whole formatted message rather than line by line. + """ + + def __init__(self): + super().__init__(level=logging.DEBUG) + self.hits: dict[str, int] = {} + self.first_seen: dict[str, float] = {} + self.scan_finished_ms: list[int] = [] + self._t0 = time.monotonic() + + def reset(self): + self.hits.clear() + self.first_seen.clear() + self.scan_finished_ms.clear() + self._t0 = time.monotonic() + + def emit(self, record): + try: + msg = record.getMessage() + except Exception: + return + for name, pattern in SIGNATURES: + m = pattern.search(msg) + if not m: + continue + self.hits[name] = self.hits.get(name, 0) + 1 + self.first_seen.setdefault(name, time.monotonic() - self._t0) + if name == "FALSE_SCAN_END": + self.scan_finished_ms.append(int(m.group(1))) + + +class StallSink: + """Sink that blocks the drain thread for `stall_s` every Nth batch. + + Stands in for a slow storage sink (SQLite commit, CSV flush) or any + heavy work the runner does inline. Sleeping here is exactly equivalent: + PipelineRunner._dispatch calls consume() on the thread that would + otherwise be pulling from _batch_queue -> parser -> dev.read(). + """ + + channels = frozenset({"live"}) + + def __init__(self, stall_s: float, every_n: int = STALL_EVERY_N_BATCHES): + self.stall_s = float(stall_s) + self.every_n = int(every_n) + self.batches = 0 + self.stalls = 0 + + def on_scan_start(self, meta) -> None: + self.batches = 0 + self.stalls = 0 + + def consume(self, channel: str, payload) -> None: + self.batches += 1 + if self.every_n > 0 and self.batches % self.every_n == 0: + self.stalls += 1 + time.sleep(self.stall_s) + + def on_complete(self) -> None: + pass + + +@dataclass +class StallResult: + stall_s: float + stalls_injected: int + hits: dict = field(default_factory=dict) + first_seen: dict = field(default_factory=dict) + scan_finished_ms: list = field(default_factory=list) + commanded_sec: int = SCAN_SEC + + @property + def false_scan_end(self) -> bool: + """A 'Scan finished' well short of the commanded duration.""" + return any(ms < (self.commanded_sec - 5) * 1000 + for ms in self.scan_finished_ms) + + +# -------------------------------------------------------------------------- +# Fixtures +# -------------------------------------------------------------------------- + +@pytest.fixture(scope="module") +def collector(): + """Attach the signature collector to the SDK's logger tree.""" + handler = _SignatureCollector() + log = logging.getLogger("openmotion.sdk") + prev_level = log.level + log.addHandler(handler) + log.setLevel(logging.DEBUG) + try: + yield handler + finally: + log.removeHandler(handler) + log.setLevel(prev_level) + + +@pytest.fixture(scope="module") +def printf_on(sensor_left, sensor_right): + """Firmware printf must be ON — it is the only channel these signatures + travel over. Restores the previous flags afterwards.""" + prev = {} + for name, sensor in (("left", sensor_left), ("right", sensor_right)): + try: + flags = sensor.get_debug_flags() + except Exception: + pytest.skip(f"{name} sensor did not answer get_debug_flags") + prev[name] = flags + sensor.set_debug_flags(flags | DEBUG_FLAG_USB_PRINTF) + try: + yield + finally: + for name, sensor in (("left", sensor_left), ("right", sensor_right)): + try: + sensor.set_debug_flags(prev[name]) + except Exception: + logging.getLogger(__name__).warning( + "could not restore %s debug flags to 0x%02X", + name, prev[name]) + + +@pytest.fixture(scope="module") +def results() -> list: + """Collected across the sweep; printed by the summary test.""" + return [] + + +# -------------------------------------------------------------------------- +# The sweep +# -------------------------------------------------------------------------- + +def _run_stalled_scan(motion, collector, stall_s: float) -> StallResult: + sink = StallSink(stall_s) + collector.reset() + + request = ScanRequest( + subject_id=f"stall{stall_s:g}", + duration_sec=SCAN_SEC, + left_camera_mask=0xFF, + right_camera_mask=0xFF, + disable_laser=False, + sinks=[sink], + # Production-like: the default DB/CSV sinks stay in the chain, so the + # measured tolerance is the real system's, not a stripped-down one. + skip_default_storage=False, + write_corrected_csv=False, + write_telemetry_csv=False, + ) + + assert motion.start_scan(request), f"start_scan refused at stall={stall_s}s" + motion.scan_workflow.await_complete(timeout_sec=SCAN_SEC + 30) + + return StallResult( + stall_s=stall_s, + stalls_injected=sink.stalls, + hits=dict(collector.hits), + first_seen=dict(collector.first_seen), + scan_finished_ms=list(collector.scan_finished_ms), + ) + + +@pytest.mark.parametrize("stall_s", STALL_SWEEP_S) +def test_drain_stall_sweep(motion, console, printf_on, collector, results, stall_s): + """Record which firmware failure signatures a host stall of `stall_s` + produces. Characterization — only the #116 budget is asserted.""" + result = _run_stalled_scan(motion, collector, stall_s) + results.append(result) + + print(f"\n stall={stall_s:>4}s injected={result.stalls_injected:>3} " + + " ".join(f"{name}={result.hits.get(name, 0)}" + for name, _ in SIGNATURES)) + if result.scan_finished_ms: + print(f" firmware 'Scan finished' at: " + f"{[ms / 1000 for ms in result.scan_finished_ms]} s " + f"(commanded {SCAN_SEC} s)") + + if stall_s <= STALL_BUDGET_S: + assert "HISTO_QUEUE_FULL" not in result.hits, ( + f"a {stall_s}s host stall — within the {STALL_BUDGET_S}s budget — " + f"already back-pressures the sensor into queue-full. This is " + f"openmotion-sdk#116; expected to fail until PR #113 lands." + ) + + # Recovery gate: whatever happened, the next scan must still be able to + # start. If the endpoint was left halted, this is where the sweep stops + # being meaningful and the operator needs to know immediately. + if "ENDPOINT_LOST" in result.hits: + pytest.fail( + f"stall={stall_s}s took a module off the USB bus " + f"(ENDPOINT_LOST x{result.hits['ENDPOINT_LOST']}). Remaining " + f"sweep points need a device power-cycle to be trustworthy." + ) + + +def test_zz_summary(results): + """Print the escalation table. Named zz_ so it sorts after the sweep.""" + if not results: + pytest.skip("sweep did not run") + + def first(pred): + hits = [r.stall_s for r in sorted(results, key=lambda r: r.stall_s) if pred(r)] + return f"{hits[0]:g}s" if hits else "not reached" + + print("\n\n USB drain stall tolerance (openmotion-sdk#116)") + print(" " + "-" * 62) + print(f" {'stall':>7} | " + " | ".join(f"{n[:9]:>9}" for n, _ in SIGNATURES)) + print(" " + "-" * 62) + for r in sorted(results, key=lambda r: r.stall_s): + cells = " | ".join( + f"{('x' + str(r.hits[n])) if n in r.hits else '-':>9}" + for n, _ in SIGNATURES + ) + print(f" {r.stall_s:>6g}s | {cells}") + print(" " + "-" * 62) + print(f" first queue-full : {first(lambda r: 'HISTO_QUEUE_FULL' in r.hits)}") + print(f" first camera overrun : {first(lambda r: 'CAMERA_OVERRUN' in r.hits)}") + print(f" first false scan end : {first(lambda r: r.false_scan_end)} <- sensor-fw#101") + print(f" first endpoint loss : {first(lambda r: 'ENDPOINT_LOST' in r.hits)}") + print(f" host warned (>1s put) : {first(lambda r: 'HOST_PUT_WARNED' in r.hits)}") diff --git a/tests/test_usb_drain_stall_unit.py b/tests/test_usb_drain_stall_unit.py new file mode 100644 index 00000000..570d2106 --- /dev/null +++ b/tests/test_usb_drain_stall_unit.py @@ -0,0 +1,194 @@ +"""Unit coverage for the USB-drain-stall HIL harness (openmotion-sdk#116). + +The harness in test_usb_drain_stall.py only earns bench time if its +classifier actually recognises the firmware printf it will be reading. +Every line quoted below is verbatim from the bloodflow-app#300 field log +(sensor FW 1.8.1-rc.2), including the multi-line records — CommInterface +emits several newline-separated firmware lines per log record, which is +exactly where a line-oriented matcher would silently miss the escalation. +""" + +import logging +import time + +import pytest + +from test_usb_drain_stall import ( + SIGNATURES, + StallResult, + StallSink, + _SignatureCollector, +) + +pytestmark = pytest.mark.unit + + +# Verbatim from open-motion-20260702_104931.log +FIELD_QUEUE_FULL = ( + "[LEFT-COMM PRINTF] HISTO enqueue fail: queue full " + "(count=4 size=4 enq=393 deq=389 datain=10842631 txfail=0)" +) +FIELD_ESCALATION = ( + "[LEFT-COMM PRINTF] USBD_HISTO_SendData failed: 3\n" + "COMM USB TX Timeout\n" + "COMM USB TX Timeout\n" + "Error in USART2: Overrun error \n" + "failed to setup receive for Camera 1 channel\n" + "Error in USART3: Overrun error \n" + "HISTO enqueue fail: queue full (count=4 size=4 enq=393 de" +) +FIELD_SCAN_FINISHED = "[RIGHT-COMM PRINTF] Scan finished (741352 ms, 29600 frames)" +FIELD_SPURIOUS_CYCLE = "[LEFT-COMM PRINTF] Scan finished (194 ms, 2 frames)" +FIELD_SPI_OVERRUN = "[LEFT-COMM PRINTF] Error in SPI2: Overrun error" +FIELD_ENDPOINT_LOST = "RIGHT-HISTO stream error (device lost): [Errno 32] Pipe error" +FIELD_HOST_PUT_WARN = ( + "LEFT-HISTO: data_queue full for >1s during streaming " + "(parser falling behind?); dropping 32837-byte chunk" +) + +# Benign traffic that must NOT be classified as failure +FIELD_BENIGN = ( + "[LEFT-COMM PRINTF] Scan started", + "[LEFT-COMM PRINTF] HISTO flush(start): q=0 ep=0 e-d=0", + "[RIGHT-COMM PRINTF] Enabled Power for Camera 1", + "[LEFT-COMM PRINTF] [DIAG] overruns c1-c8: 0 0 0 0 0 0 0 0", + "LEFT-HISTO: Streaming started", +) + + +def _feed(collector, *messages): + log = logging.getLogger("openmotion.sdk.test_harness") + log.addHandler(collector) + log.setLevel(logging.DEBUG) + log.propagate = False + try: + for msg in messages: + log.warning("%s", msg) + finally: + log.removeHandler(collector) + + +@pytest.fixture +def collector(): + return _SignatureCollector() + + +def test_classifies_queue_full(collector): + _feed(collector, FIELD_QUEUE_FULL) + assert collector.hits == {"HISTO_QUEUE_FULL": 1} + + +def test_multiline_record_hits_every_stage(collector): + """One CommInterface record carrying several firmware lines must be + credited to each stage it contains — the field log's records look + exactly like this, and a line-oriented matcher would drop most of them.""" + _feed(collector, FIELD_ESCALATION) + assert collector.hits["HISTO_QUEUE_FULL"] == 1 + assert collector.hits["USB_TX_BLOCKED"] == 1 + assert collector.hits["CAMERA_OVERRUN"] == 1 + assert "FALSE_SCAN_END" not in collector.hits + + +def test_spi_overrun_counts_as_camera_overrun(collector): + """L7's dropout came through SPI2, not a USART — both are the same stage.""" + _feed(collector, FIELD_SPI_OVERRUN) + assert collector.hits == {"CAMERA_OVERRUN": 1} + + +def test_scan_finished_captures_duration(collector): + _feed(collector, FIELD_SCAN_FINISHED, FIELD_SPURIOUS_CYCLE) + assert collector.hits["FALSE_SCAN_END"] == 2 + assert collector.scan_finished_ms == [741352, 194] + + +def test_endpoint_lost_and_host_warning(collector): + _feed(collector, FIELD_ENDPOINT_LOST, FIELD_HOST_PUT_WARN) + assert collector.hits["ENDPOINT_LOST"] == 1 + assert collector.hits["HOST_PUT_WARNED"] == 1 + + +def test_benign_traffic_is_not_classified(collector): + _feed(collector, *FIELD_BENIGN) + assert collector.hits == {} + + +def test_reset_clears_state(collector): + _feed(collector, FIELD_QUEUE_FULL, FIELD_SCAN_FINISHED) + collector.reset() + assert collector.hits == {} and collector.scan_finished_ms == [] + + +def test_first_seen_is_recorded_once(collector): + _feed(collector, FIELD_QUEUE_FULL) + first = collector.first_seen["HISTO_QUEUE_FULL"] + time.sleep(0.01) + _feed(collector, FIELD_QUEUE_FULL) + assert collector.hits["HISTO_QUEUE_FULL"] == 2 + assert collector.first_seen["HISTO_QUEUE_FULL"] == first + + +def test_every_signature_has_a_field_example(): + """Guard against adding a signature with no known-good sample.""" + covered = _SignatureCollector() + _feed(covered, FIELD_ESCALATION, FIELD_SCAN_FINISHED, + FIELD_ENDPOINT_LOST, FIELD_HOST_PUT_WARN) + assert set(covered.hits) == {name for name, _ in SIGNATURES} + + +# -------------------------------------------------------------------------- +# StallSink +# -------------------------------------------------------------------------- + +def test_stall_sink_only_stalls_on_every_nth_batch(): + sink = StallSink(stall_s=0.05, every_n=4) + sink.on_scan_start(None) + + t0 = time.perf_counter() + for _ in range(3): + sink.consume("live", object()) + assert sink.stalls == 0 + assert time.perf_counter() - t0 < 0.04, "stalled before the Nth batch" + + sink.consume("live", object()) + assert sink.stalls == 1 + assert time.perf_counter() - t0 >= 0.05 + + +def test_stall_sink_counters_reset_between_scans(): + sink = StallSink(stall_s=0.0, every_n=2) + sink.on_scan_start(None) + for _ in range(4): + sink.consume("live", object()) + assert (sink.batches, sink.stalls) == (4, 2) + sink.on_scan_start(None) + assert (sink.batches, sink.stalls) == (0, 0) + + +def test_stall_sink_declares_only_the_live_channel(): + """'live' fires once per batch (~0.25 s); 'final' only on interval close + (~15 s), which is too coarse to sweep a 40 s scan with.""" + assert StallSink(0.0).channels == frozenset({"live"}) + + +# -------------------------------------------------------------------------- +# StallResult +# -------------------------------------------------------------------------- + +def test_false_scan_end_ignores_a_normal_completion(): + r = StallResult(stall_s=0.1, stalls_injected=8, + scan_finished_ms=[40_120], commanded_sec=40) + assert r.false_scan_end is False + + +def test_false_scan_end_flags_an_early_finish(): + r = StallResult(stall_s=2.0, stalls_injected=8, + scan_finished_ms=[21_500], commanded_sec=40) + assert r.false_scan_end is True + + +def test_false_scan_end_flags_the_spurious_restart_cycles(): + """The field failure's signature: a plausible finish followed by dozens + of millisecond-long ones.""" + r = StallResult(stall_s=2.0, stalls_injected=8, + scan_finished_ms=[39_900, 194, 430], commanded_sec=40) + assert r.false_scan_end is True