"""Deleting an acquisition and changing startup settings respect stage ownership.""" from __future__ import annotations from collections import deque from dataclasses import fields import unittest from unittest import mock from python_app.gui.controllers.history_controller import HistoryController from python_app.gui.runtime.history import ( build_processor_run_signature, build_run_history_signature, record_result_history, ) from python_app.models.dataset_model import ResultCollection, SweepCollection from python_app.models.run_config_schema import RunConfigModel def sweep(cid: int, ns: int, window: tuple[int, int] = (0, 0)) -> SweepCollection: return SweepCollection(cid, ns, capture_start_ns=window[0], capture_end_ns=window[1]) class RemoveLastHistoryTest(unittest.TestCase): def test_newest_capture_is_removed_across_stages_despite_late_processing(self) -> None: old, fresh = sweep(1, 900, (10, 20)), sweep(1, 100, (30, 40)) pre_old, pre_fresh = sweep(1, 950, (10, 20)), sweep(1, 150, (30, 40)) late_old = ResultCollection(1, 1000, capture_start_ns=10, capture_end_ns=20) fresh_result = ResultCollection(1, 200, capture_start_ns=30, capture_end_ns=40) history = HistoryController() history.raw.extend([fresh, old]) history.preprocessed.extend([pre_old, pre_fresh]) history.results.extend([fresh_result, late_old]) history.delete(remove_last=True) self.assertEqual((list(history.raw), list(history.preprocessed), list(history.results)), ([old], [pre_old], [late_old])) self.assertEqual(history.removed, [{"collection_id": 1, "capture_start_ns": 30}]) self.assertFalse(history.record_result(fresh_result)) def test_missing_stages_do_not_delete_another_acquisition_with_reused_id(self) -> None: old, fresh = sweep(1, 100, (10, 20)), sweep(1, 200, (30, 40)) late = ResultCollection(1, 900, capture_start_ns=10, capture_end_ns=20) for raw, pre in (([old], [fresh]), ([old, fresh], [])): history = HistoryController() history.raw.extend(raw) history.preprocessed.extend(pre) history.results.append(late) history.delete(remove_last=True) self.assertEqual((list(history.raw), list(history.preprocessed), list(history.results)), ([old], [], [late])) def test_results_only_and_empty_histories(self) -> None: result = ResultCollection(1, 100, capture_start_ns=10, capture_end_ns=20) history = HistoryController() history.results.append(result) history.replay_results.append(result) history.delete(remove_last=True) history.delete(remove_last=True) self.assertFalse(history.results or history.replay_results) self.assertEqual(history.removed, [{"collection_id": 1, "capture_start_ns": 10}]) with mock.patch("python_app.gui.controllers.history_controller.time.monotonic_ns", return_value=50): history.delete(remove_last=False) self.assertEqual(history.clear_before_ns, 50) self.assertEqual(history.removed, []) self.assertFalse(history.record_result(result)) def test_results_sort_by_capture_and_reprocessed_identity_is_replaced(self) -> None: old = ResultCollection(1, 300, capture_start_ns=10, capture_end_ns=20) middle = ResultCollection(2, 500, capture_start_ns=30, capture_end_ns=40) fresh = ResultCollection(1, 100, capture_start_ns=50, capture_end_ns=60) history = deque([old, fresh], maxlen=3) self.assertTrue(record_result_history(history, middle)) self.assertEqual(list(history), [old, middle, fresh]) reprocessed = ResultCollection(1, 900, capture_start_ns=10, capture_end_ns=20) self.assertFalse(record_result_history(history, reprocessed)) self.assertIs(history[0], reprocessed) self.assertFalse(record_result_history(history, ResultCollection(3, 999, capture_start_ns=1))) self.assertFalse(record_result_history(deque(maxlen=0), fresh)) class RuntimeSignatureTest(unittest.TestCase): def test_reference_clock_changes_multi_device_history_only(self) -> None: for model in ("librevna_multi", "librevna", "rph_radar"): config = RunConfigModel() config.radar.model = model before = build_run_history_signature(config) config.radar.multi_device.force_external_reference = not config.radar.multi_device.force_external_reference with self.subTest(model=model): self.assertEqual(before != build_run_history_signature(config), model == "librevna_multi") def test_adc_key_and_history_follow_measurement_not_launch_paths(self) -> None: original = RunConfigModel() original.radar.model, original.radar.driver_mode = "rph_radar", "native" for section, field, value, changes_identity in ( ("rph_radar.phase_calibration", "phase1_rad", 301.0, True), ("rph_radar.band", "points", 1024, True), ("rph_radar.band", "start_hz", 2.2e9, True), ("rph_radar", "executable_path", "/new/collector", False), ("rph_radar", "tty_path", "/new/tty", False), ("sweep", "if_bandwidth_hz", 999.0, False), ("", "serial", "unused ADC serial", False), ): changed = original.clone() target = changed.radar for part in section.split(".") if section else (): target = getattr(target, part) setattr(target, field, value) with self.subTest(section=section, field=field): self.assertEqual(original.radar_key() != changed.radar_key(), changes_identity) self.assertEqual(build_run_history_signature(original) != build_run_history_signature(changed), changes_identity) def test_every_ring_endpoint_setting_requires_restart(self) -> None: original = RunConfigModel() signature = build_processor_run_signature(original) for ring_field in fields(original.rings): for field, value in (("name", "/other"), ("capacity", 7), ("slot_size_bytes", 8192)): with self.subTest(ring=ring_field.name, field=field): changed = original.clone() setattr(getattr(changed.rings, ring_field.name), field, value) self.assertNotEqual(build_processor_run_signature(changed), signature) def test_locator_and_live_reader_settings_require_restart(self) -> None: original = RunConfigModel() signature = build_processor_run_signature(original) for field in fields(original.runtime.locator_server): with self.subTest(field=field.name): changed = original.clone() value = getattr(changed.runtime.locator_server, field.name) setattr(changed.runtime.locator_server, field.name, value + "x" if isinstance(value, str) else value + 1) self.assertNotEqual(build_processor_run_signature(changed), signature) for field, value in (("processing_live_config_path", "other.json"), ("idle_sleep_ms", 8)): changed = original.clone() setattr(changed.runtime, field, value) self.assertNotEqual(build_processor_run_signature(changed), signature) def test_notch_and_preprocess_assets_reset_history_without_restarting_processor(self) -> None: original = RunConfigModel() for section, field, value in ( ("notch", "enabled", True), ("notch", "bands_hz", [(1e8, 2e8)]), ("notch", "taper_width_hz", 1e7), ("notch", "taper_type", "linear"), ("s21.calibration", "set_name", "other"), ("s21.reference", "bundle_path", "other.bin"), ): with self.subTest(section=section, field=field): changed = original.clone() target = changed.preprocess for part in section.split("."): target = getattr(target, part) setattr(target, field, value) self.assertEqual(build_processor_run_signature(changed), build_processor_run_signature(original)) self.assertNotEqual(build_run_history_signature(changed), build_run_history_signature(original)) if __name__ == "__main__": unittest.main()