some fixes

python_app/tests/test_single_capture.py

@@ -0,0 +1,94 @@
+"""Single-capture completion logic.
+
+The single capture starts a one-shot pipeline and must finish on exactly the result
+that belongs to its sweep. Because collection ids restart per orchestrator run, a stale
+result from a previous run can carry the SAME id; the start timestamp disambiguates them.
+These tests pin that boundary (fresh vs stale same-id) and the readiness state machine —
+the path that previously hung on the simulator when start_ns was recorded too late.
+"""
+
+from __future__ import annotations
+
+import os
+import unittest
+from pathlib import Path
+
+import numpy as np
+
+os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
+from PyQt6.QtWidgets import QApplication  # noqa: E402
+
+from python_app.gui.app_window import AppWindow  # noqa: E402
+from python_app.models.dataset_model import (  # noqa: E402
+    ComboKey,
+    ResultBlock,
+    ResultCollection,
+    ResultPayload,
+)
+
+_app: QApplication | None = None
+_window: AppWindow | None = None
+
+
+def setUpModule() -> None:
+    global _app, _window
+    _app = QApplication.instance() or QApplication([])
+    _window = AppWindow(Path("."))
+
+
+def tearDownModule() -> None:
+    if _window is not None:
+        _window.close()
+
+
+def _result(collection_id: int, monotonic_ns: int) -> ResultCollection:
+    payload = ResultPayload(
+        processing_name="pass_through", kind=1,
+        frequency_hz=np.array([1.0, 2.0], dtype=np.float32),
+        trace=np.array([1 + 1j, 2 - 2j], dtype=np.complex64),
+    )
+    block = ResultBlock(combo=ComboKey(input=0, output=0), payloads=[payload])
+    return ResultCollection(collection_id=collection_id, monotonic_ns=monotonic_ns, blocks=[block])
+
+
+class SingleCaptureBoundaryTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.w = _window
+        self.w._processing_mode.setCurrentText("pass_through")
+        self.w._single_capture_active = True
+        self.w._single_capture_start_ns = 1000
+        self.w._single_capture_seen_raw = True
+        self.w._single_capture_target_collection_id = 5
+        self.w._result_history.clear()
+
+    def test_picks_fresh_result_not_stale_same_id(self) -> None:
+        self.w._result_history.append(_result(5, 500))    # stale: same id, before start_ns
+        self.w._result_history.append(_result(5, 1500))   # fresh: at/after start_ns
+        target = self.w._find_single_capture_target_result()
+        self.assertIsNotNone(target)
+        self.assertEqual(target.monotonic_ns, 1500)
+
+    def test_stale_same_id_result_is_not_matched(self) -> None:
+        self.w._result_history.append(_result(5, 500))    # only a stale same-id result
+        self.assertIsNone(self.w._find_single_capture_target_result())
+
+    def test_wrong_collection_id_is_not_matched(self) -> None:
+        self.w._result_history.append(_result(9, 1500))   # fresh but a different sweep
+        self.assertIsNone(self.w._find_single_capture_target_result())
+
+    def test_not_ready_until_a_fresh_raw_is_seen(self) -> None:
+        # The readiness guard that the timing fix makes reachable: with no raw recognized
+        # as fresh yet, the capture must not complete (it must keep waiting, not hang).
+        self.w._single_capture_seen_raw = False
+        self.w._result_history.append(_result(5, 1500))
+        self.assertFalse(self.w._finish_single_capture_if_ready())
+        self.assertTrue(self.w._single_capture_active)
+
+    def test_completes_and_stops_on_the_fresh_result(self) -> None:
+        self.w._result_history.append(_result(5, 1500))
+        self.assertTrue(self.w._finish_single_capture_if_ready())
+        self.assertFalse(self.w._single_capture_active)  # _stop_run cleared the flag
+
+
+if __name__ == "__main__":
+    unittest.main()