"""Storage + embedded WebUI tests. Pins the agreed semantics: * NPZ set persistence round-trips traces + metadata (float32 wire precision); * vna_history export fails loud (ValueError) when no matching traces / bad stage; * the web bridge rejects unknown live-settings fields (HTTP 400), serves immutable snapshot copies, and forwards controls as Qt signals; * the web fan-out is latest-wins — a full client queue drops its oldest frame. """ from __future__ import annotations import asyncio import os import tempfile import unittest from pathlib import Path from unittest import mock import numpy as np os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") # before any Qt import from PyQt6.QtWidgets import QApplication # noqa: E402 from python_app.gui.controllers.app_window_web_mixin import ( # noqa: E402 AppWindowWebController, AppWindowWebMixin, ) from python_app.models.dataset_model import ComboKey, SweepCollection, TraceData # noqa: E402 from python_app.storage.npz.store import NpzStore # noqa: E402 from python_app.storage.npz.vna_history_json import ( # noqa: E402 _complex_to_points, _normalize_channel, build_vna_history_payload, ) from python_app.webui.streaming import RingBroadcaster # noqa: E402 def setUpModule() -> None: global _app _app = QApplication.instance() or QApplication([]) def _trace(in_pos: int, out_pos: int) -> TraceData: return TraceData( combo=ComboKey(input=in_pos, output=out_pos), frequency_hz=np.array([1.0, 2.0], dtype=np.float32), s11=np.array([1 + 1j, 2 + 2j], dtype=np.complex64), s21=np.array([3 + 3j, 4 - 4j], dtype=np.complex64), ) # --------------------------------------------------------------------------- # # NPZ set persistence # --------------------------------------------------------------------------- # class NpzStoreRoundTripTest(unittest.TestCase): def setUp(self) -> None: self._dir = tempfile.TemporaryDirectory() self.addCleanup(self._dir.cleanup) self.store = NpzStore(Path(self._dir.name)) def test_save_then_load_round_trips_traces_and_metadata(self) -> None: col = SweepCollection(collection_id=7, monotonic_ns=11, traces=[_trace(0, 0), _trace(1, 0)], capture_start_ns=100, capture_end_ns=200) self.store.save_set("calibration", "radar1", "set1", col) loaded = self.store.load_set("calibration", "radar1", "set1") self.assertEqual(loaded.collection_id, 7) self.assertEqual((loaded.capture_start_ns, loaded.capture_end_ns), (100, 200)) self.assertEqual(len(loaded.traces), 2) by_combo = {(t.combo.input, t.combo.output): t for t in loaded.traces} self.assertTrue(np.array_equal(by_combo[(0, 0)].s21, _trace(0, 0).s21)) self.assertTrue(np.array_equal(by_combo[(1, 0)].frequency_hz, _trace(1, 0).frequency_hz)) def test_set_appears_in_listing_and_leaves_no_tmp(self) -> None: self.store.save_set("calibration", "radar1", "set1", SweepCollection(collection_id=1, monotonic_ns=1, traces=[_trace(0, 0)])) self.assertIn("set1", self.store.list_sets("calibration", "radar1")) leftover = list(Path(self._dir.name).rglob("*.tmp")) self.assertEqual(leftover, []) def test_load_missing_set_raises(self) -> None: with self.assertRaises(Exception): self.store.load_set("calibration", "radar1", "absent") # --------------------------------------------------------------------------- # # vna_history export # --------------------------------------------------------------------------- # class VnaHistoryTest(unittest.TestCase): def _sweeps(self, *combos) -> list[SweepCollection]: return [SweepCollection(collection_id=1, monotonic_ns=1, traces=[_trace(i, o) for i, o in combos])] def test_builds_payload_for_matching_combo(self) -> None: payload = build_vna_history_payload([], self._sweeps((0, 0)), [], input_index=0, output_index=0) self.assertEqual(payload["input_index"], 0) self.assertEqual(payload["channel"], "s21") self.assertEqual(payload["preprocessed_record_count"], 1) self.assertTrue(payload["sweep_history"]) def test_no_matching_traces_raises(self) -> None: with self.assertRaises(ValueError): build_vna_history_payload([], self._sweeps((0, 0)), [], input_index=9, output_index=9) def test_invalid_primary_stage_raises(self) -> None: with self.assertRaisesRegex(ValueError, "primary_stage"): build_vna_history_payload([], self._sweeps((0, 0)), [], input_index=0, output_index=0, primary_stage="x") def test_normalize_channel(self) -> None: self.assertEqual(_normalize_channel("S21"), "s21") self.assertEqual(_normalize_channel(" s11 "), "s11") with self.assertRaises(ValueError): _normalize_channel("s99") def test_complex_to_points(self) -> None: self.assertEqual(_complex_to_points(np.array([1 + 2j, 3 - 4j])), [[1.0, 2.0], [3.0, -4.0]]) # --------------------------------------------------------------------------- # # Web bridge controller # --------------------------------------------------------------------------- # class WebControllerTest(unittest.TestCase): def setUp(self) -> None: self.controller = AppWindowWebController() self.addCleanup(self.controller.deleteLater) def test_rejects_unknown_field(self) -> None: with self.assertRaisesRegex(ValueError, "Unknown live-settings"): self.controller.apply_live_settings({"definitely_not_a_field": 1}) def test_known_field_emits_and_returns_snapshot(self) -> None: received: list[dict] = [] self.controller.apply_settings_requested.connect(received.append) out = self.controller.apply_live_settings({"gpr_min_visible_score": 0.5}) self.assertEqual(received, [{"gpr_min_visible_score": 0.5}]) self.assertIsInstance(out, list) def test_snapshot_is_replaced_and_returned_as_copy(self) -> None: self.controller.update_snapshot(status={"running": True}, live_settings=[{"name": "x"}], frame={"seq": 1}) status = self.controller.status() self.assertEqual(status, {"running": True}) status["running"] = False # mutating the copy must not affect the controller self.assertTrue(self.controller.status()["running"]) self.assertEqual(self.controller.peek_frame(), {"seq": 1}) def test_frame_only_updates_when_present(self) -> None: self.controller.update_snapshot(status={}, live_settings=[], frame={"seq": 1}) self.controller.update_snapshot(status={}, live_settings=[], frame=None) # no new grab self.assertEqual(self.controller.peek_frame(), {"seq": 1}) # keeps the last frame def test_controls_emit_signals(self) -> None: fired: list[str] = [] self.controller.start_requested.connect(lambda: fired.append("start")) self.controller.stop_requested.connect(lambda: fired.append("stop")) self.controller.single_capture_requested.connect(lambda: fired.append("single")) self.controller.capture_requested.connect(lambda: fired.append("capture")) self.controller.start() self.controller.stop() self.controller.single_capture() self.controller.capture_tmp_reference() self.assertEqual(fired, ["start", "stop", "single", "capture"]) class WebPortTest(unittest.TestCase): def _port_with_env(self, value: str | None) -> int: env = {} if value is None else {"RADAR_SYSTEM_WEBUI_PORT": value} with mock.patch.dict(os.environ, env, clear=False): if value is None: os.environ.pop("RADAR_SYSTEM_WEBUI_PORT", None) return AppWindowWebMixin._web_ui_port() def test_default_when_unset_or_invalid(self) -> None: self.assertEqual(self._port_with_env(None), 8080) self.assertEqual(self._port_with_env("abc"), 8080) self.assertEqual(self._port_with_env("99999"), 8080) # out of range self.assertEqual(self._port_with_env("0"), 8080) def test_valid_port(self) -> None: self.assertEqual(self._port_with_env("9000"), 9000) # --------------------------------------------------------------------------- # # Latest-wins fan-out # --------------------------------------------------------------------------- # class RingBroadcasterFanOutTest(unittest.TestCase): def test_full_client_queue_drops_oldest(self) -> None: async def scenario() -> None: broadcaster = RingBroadcaster(controller=object()) queue: asyncio.Queue = asyncio.Queue(maxsize=1) broadcaster.register(queue) broadcaster._publish({"type": "frame", "seq": 1}) broadcaster._publish({"type": "frame", "seq": 2}) # evicts seq 1 self.assertEqual(queue.qsize(), 1) self.assertEqual(queue.get_nowait(), {"type": "frame", "seq": 2}) # newest survives asyncio.run(scenario()) def test_unregister_stops_delivery(self) -> None: async def scenario() -> None: broadcaster = RingBroadcaster(controller=object()) queue: asyncio.Queue = asyncio.Queue(maxsize=4) broadcaster.register(queue) broadcaster.unregister(queue) broadcaster.unregister(queue) # idempotent broadcaster._publish({"type": "frame", "seq": 1}) self.assertEqual(queue.qsize(), 0) asyncio.run(scenario()) if __name__ == "__main__": unittest.main()