From a8a41be6752d10921531466b6ecaf34d605e6ef9 Mon Sep 17 00:00:00 2001 From: C-Achard Date: Wed, 8 Jul 2026 15:27:00 +0200 Subject: [PATCH 1/9] Add camera settings fixture --- tests/conftest.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/tests/conftest.py b/tests/conftest.py index 3eaf353..3aeb23d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -269,6 +269,12 @@ def app_config_two_cams(tmp_path) -> ApplicationSettings: return make_app_config(tmp_path=tmp_path, num_cams=2, backend="fake", enabled=True, fps=30.0) +@pytest.fixture +def camera_worker_settings(app_config_two_cams) -> CameraSettings: + """Single enabled fake camera settings for SingleCameraWorker tests.""" + return app_config_two_cams.multi_camera.cameras[0].model_copy(deep=True) + + # --------------------------------------------------------------------- # Main window fixture # --------------------------------------------------------------------- From 6f89fe82bc01cac3700e7bb29bd15557f9e35204 Mon Sep 17 00:00:00 2001 From: C-Achard Date: Wed, 8 Jul 2026 15:29:35 +0200 Subject: [PATCH 2/9] Refactor full-queue frame drop test Replace the asynchronous `test_queue_full_drops_frames` flow with a deterministic unit test that directly exercises `enqueue_frame` when the queue is full. The new test injects a 1-slot queue, simulates a running worker state, verifies stale-frame dropping via processor stats, and confirms the newest frame/timestamp remains queued. --- tests/services/test_dlc_processor.py | 48 ++++++++++++++++++++-------- 1 file changed, 34 insertions(+), 14 deletions(-) diff --git a/tests/services/test_dlc_processor.py b/tests/services/test_dlc_processor.py index 3f5e0cb..a58490e 100644 --- a/tests/services/test_dlc_processor.py +++ b/tests/services/test_dlc_processor.py @@ -1,13 +1,25 @@ +from __future__ import annotations + +import queue + import numpy as np import pytest -# from dlclivegui.config import DLCProcessorSettings from dlclivegui.config import DLCProcessorSettings + +# from dlclivegui.config import DLCProcessorSettings from dlclivegui.services.dlc_processor import ( DLCLiveProcessor, ProcessorStats, + WorkerState, ) + +class _AliveThread: + def is_alive(self) -> bool: + return True + + # --------------------------------------------------------------------- # Tests # --------------------------------------------------------------------- @@ -73,28 +85,36 @@ def test_worker_processes_frames(qtbot, monkeypatch_dlclive, settings_model): proc.reset() -@pytest.mark.unit -def test_queue_full_drops_frames(qtbot, monkeypatch_dlclive, settings_model): +def test_enqueue_frame_drops_stale_when_queue_is_full(settings_model): proc = DLCLiveProcessor() proc.configure(settings_model) try: - frame = np.zeros((32, 32, 3), dtype=np.uint8) + proc._queue = queue.Queue(maxsize=1) + proc._worker_thread = _AliveThread() + proc._state = WorkerState.RUNNING + proc._stop_event.clear() - # Start the worker with the first frame - with qtbot.waitSignal(proc.initialized, timeout=1500): - proc.enqueue_frame(frame, 1.0) + frame1 = np.zeros((32, 32, 3), dtype=np.uint8) + frame2 = np.ones((32, 32, 3), dtype=np.uint8) - # Flood the 1-slot queue to force drops - for _ in range(50): - proc.enqueue_frame(frame, 2.0) + proc.enqueue_frame(frame1, 1.0) + proc.enqueue_frame(frame2, 2.0) - # Wait until we observe dropped frames - qtbot.waitUntil(lambda: proc._frames_dropped > 0, timeout=1500) - assert proc._frames_dropped > 0 + stats = proc.get_stats() + assert stats.frames_enqueued == 2 + assert stats.frames_dropped == 1 + assert stats.queue_size == 1 + + queued_frame, queued_timestamp, _queued_at = proc._queue.get_nowait() + assert queued_timestamp == 2.0 + np.testing.assert_array_equal(queued_frame, frame2) finally: - proc.reset() + proc._queue = None + proc._worker_thread = None + proc._state = WorkerState.STOPPED + proc._stop_event.clear() @pytest.mark.unit From 1ad72f9b0fa73bbbf3e64fbd99bf2435c1101223 Mon Sep 17 00:00:00 2001 From: C-Achard Date: Wed, 8 Jul 2026 15:30:04 +0200 Subject: [PATCH 3/9] Add SingleCameraWorker service unit tests Introduce a new test module for `SingleCameraWorker` covering normal fake-backend startup/frame flow, recording sink behavior (enabled/disabled), camera factory initialization failure handling, repeated empty-frame and read-error retry limits, and hardware-trigger timeout behavior that should not surface as errors. --- tests/services/test_camera_controller.py | 307 +++++++++++++++++++++++ 1 file changed, 307 insertions(+) create mode 100644 tests/services/test_camera_controller.py diff --git a/tests/services/test_camera_controller.py b/tests/services/test_camera_controller.py new file mode 100644 index 0000000..cc01942 --- /dev/null +++ b/tests/services/test_camera_controller.py @@ -0,0 +1,307 @@ +from __future__ import annotations + +import numpy as np + +from dlclivegui.cameras.base import CapturedFrame +from dlclivegui.config import CameraSettings +from dlclivegui.services.camera_controller import SingleCameraWorker + + +def _capture_signals(worker: SingleCameraWorker) -> dict[str, list[tuple]]: + """Collect worker Qt signal emissions synchronously.""" + seen: dict[str, list[tuple]] = { + "runtime_info": [], + "started": [], + "frame_captured": [], + "error_occurred": [], + "stopped": [], + } + + worker.runtime_info.connect(lambda *args: seen["runtime_info"].append(args)) + worker.started.connect(lambda *args: seen["started"].append(args)) + worker.frame_captured.connect(lambda *args: seen["frame_captured"].append(args)) + worker.error_occurred.connect(lambda *args: seen["error_occurred"].append(args)) + worker.stopped.connect(lambda *args: seen["stopped"].append(args)) + + return seen + + +def test_worker_fake_backend(qtbot, patch_factory, camera_worker_settings: CameraSettings): + worker = SingleCameraWorker("fake:index:0", camera_worker_settings) + seen = _capture_signals(worker) + + # Stop after first frame so worker.run() returns synchronously. + worker.frame_captured.connect(lambda *_args: worker.stop()) + + worker.run() + + assert len(seen["error_occurred"]) == 0 + assert len(seen["runtime_info"]) == 1 + assert len(seen["started"]) == 1 + assert len(seen["frame_captured"]) == 1 + assert len(seen["stopped"]) == 1 + + runtime_camera_id, runtime = seen["runtime_info"][0] + assert runtime_camera_id == "fake:index:0" + assert set(runtime) == { + "actual_fps", + "actual_resolution", + "actual_pixel_format", + "actual_output_format", + } + + assert seen["started"][0] == ("fake:index:0",) + + frame_camera_id, frame, timestamp, timestamp_metadata = seen["frame_captured"][0] + assert frame_camera_id == "fake:index:0" + assert isinstance(frame, np.ndarray) + assert frame.shape == (48, 64, 3) + assert frame.dtype == np.uint8 + assert isinstance(timestamp, float) + assert timestamp_metadata is None + + assert seen["stopped"][0] == ("fake:index:0",) + + +def test_worker_recording_sink_receives_frame(qtbot, patch_factory, camera_worker_settings: CameraSettings): + worker = SingleCameraWorker("fake:index:0", camera_worker_settings) + seen = _capture_signals(worker) + + recorded: list[tuple] = [] + + def recording_sink(camera_id, frame, timestamp, timestamp_metadata): + recorded.append((camera_id, frame.copy(), timestamp, timestamp_metadata)) + + worker.set_recording_sink(recording_sink) + worker.set_recording_enabled(True) + + worker.frame_captured.connect(lambda *_args: worker.stop()) + + worker.run() + + assert len(seen["error_occurred"]) == 0 + assert len(seen["frame_captured"]) == 1 + assert len(recorded) == 1 + + rec_camera_id, rec_frame, rec_timestamp, rec_metadata = recorded[0] + frame_camera_id, emitted_frame, emitted_timestamp, emitted_metadata = seen["frame_captured"][0] + + assert rec_camera_id == "fake:index:0" + assert frame_camera_id == "fake:index:0" + + np.testing.assert_array_equal(rec_frame, emitted_frame) + assert rec_timestamp == emitted_timestamp + assert rec_metadata == emitted_metadata + + +def test_worker_recording_sink_disabled_does_not_receive_frame(qtbot, camera_worker_settings: CameraSettings): + worker = SingleCameraWorker("fake:index:0", camera_worker_settings) + + recorded: list[tuple] = [] + + def recording_sink(*args): + recorded.append(args) + + worker.set_recording_sink(recording_sink) + worker.set_recording_enabled(False) + + worker.frame_captured.connect(lambda *_args: worker.stop()) + + worker.run() + + assert recorded == [] + + +def test_worker_backend_creation_failure_emits_error(monkeypatch, qtbot, camera_worker_settings: CameraSettings): + from dlclivegui.services import camera_controller as controller_mod + + def fail_create(_settings): + raise RuntimeError("error") + + monkeypatch.setattr(controller_mod.CameraFactory, "create", staticmethod(fail_create)) + + worker = SingleCameraWorker("fake:index:0", camera_worker_settings) + seen = _capture_signals(worker) + + worker.run() + + assert len(seen["runtime_info"]) == 0 + assert len(seen["started"]) == 0 + assert len(seen["frame_captured"]) == 0 + + assert len(seen["error_occurred"]) == 1 + camera_id, message = seen["error_occurred"][0] + assert camera_id == "fake:index:0" + assert "Failed to initialize camera" in message + assert "error" in message + + assert seen["stopped"] == [("fake:index:0",)] + + +class _EmptyFrameBackend: + def __init__(self, settings: CameraSettings): + self.settings = settings + self.open_called = False + self.close_called = False + + def open(self): + self.open_called = True + + def read(self): + return CapturedFrame(frame=None, software_timestamp=123.0, timestamp_metadata=None) + + def close(self): + self.close_called = True + + +def test_worker_too_many_empty_frames_emits_error(monkeypatch, qtbot, camera_worker_settings: CameraSettings): + from dlclivegui.services import camera_controller as controller_mod + + backend = _EmptyFrameBackend(camera_worker_settings) + + monkeypatch.setattr( + controller_mod.CameraFactory, + "create", + staticmethod(lambda _settings: backend), + ) + + worker = SingleCameraWorker("fake:index:0", camera_worker_settings) + worker._max_consecutive_errors = 3 + worker._retry_delay = 0.0 + + seen = _capture_signals(worker) + + worker.run() + + assert backend.open_called + assert backend.close_called + + assert seen["started"] == [("fake:index:0",)] + assert len(seen["frame_captured"]) == 0 + + assert len(seen["error_occurred"]) == 1 + camera_id, message = seen["error_occurred"][0] + assert camera_id == "fake:index:0" + assert "Too many empty frames" in message + + assert seen["stopped"] == [("fake:index:0",)] + + +class _ReadExceptionBackend: + waits_for_hardware_trigger = False + + def __init__(self, settings: CameraSettings): + self.settings = settings + self.open_called = False + self.close_called = False + self.read_count = 0 + + def open(self): + self.open_called = True + + def read(self): + self.read_count += 1 + raise RuntimeError(f"read failed {self.read_count}") + + def close(self): + self.close_called = True + + +def test_worker_read_exception_emits_error_after_retries( + monkeypatch, + qtbot, + camera_worker_settings: CameraSettings, +): + from dlclivegui.services import camera_controller as controller_mod + + backend = _ReadExceptionBackend(camera_worker_settings) + + monkeypatch.setattr( + controller_mod.CameraFactory, + "create", + staticmethod(lambda _settings: backend), + ) + + worker = SingleCameraWorker("fake:index:0", camera_worker_settings) + worker._max_consecutive_errors = 3 + worker._retry_delay = 0.0 + + seen = _capture_signals(worker) + + worker.run() + + assert backend.open_called + assert backend.close_called + assert backend.read_count == 3 + + assert seen["started"] == [("fake:index:0",)] + assert len(seen["frame_captured"]) == 0 + + assert len(seen["error_occurred"]) == 1 + camera_id, message = seen["error_occurred"][0] + assert camera_id == "fake:index:0" + assert "Camera read error" in message + assert "read failed 3" in message + + assert seen["stopped"] == [("fake:index:0",)] + + +class _TimeoutTriggerBackend: + waits_for_hardware_trigger = True + + def __init__(self, settings: CameraSettings): + self.settings = settings + self.open_called = False + self.close_called = False + self.read_count = 0 + + def open(self): + self.open_called = True + + def read(self): + self.read_count += 1 + raise TimeoutError("waiting for trigger") + + def close(self): + self.close_called = True + + +def test_worker_hardware_trigger_timeouts_do_not_emit_error( + monkeypatch, + qtbot, + camera_worker_settings: CameraSettings, +): + from dlclivegui.services import camera_controller as controller_mod + + backend = _TimeoutTriggerBackend(camera_worker_settings) + + monkeypatch.setattr( + controller_mod.CameraFactory, + "create", + staticmethod(lambda _settings: backend), + ) + + worker = SingleCameraWorker("fake:index:0", camera_worker_settings) + worker._trigger_timeout_delay = 0.0 + + seen = _capture_signals(worker) + + original_read = backend.read + + def read_then_stop(): + if backend.read_count >= 3: + worker.stop() + return original_read() + + backend.read = read_then_stop + + worker.run() + + assert backend.open_called + assert backend.close_called + assert backend.read_count >= 3 + + assert seen["started"] == [("fake:index:0",)] + assert seen["error_occurred"] == [] + assert seen["frame_captured"] == [] + assert seen["stopped"] == [("fake:index:0",)] From 07424ffdba879029a92cde45a7b1865c3bf8768e Mon Sep 17 00:00:00 2001 From: C-Achard Date: Wed, 8 Jul 2026 15:30:40 +0200 Subject: [PATCH 4/9] Add GUI tests for preview and recording lifecycle Adds focused GUI test modules for main-window behavior: preview start/stop lifecycle, recording start/stop and pending-start logic, and camera label/display ID handling in UI elements. This improves regression coverage for camera-state transitions and user-facing camera naming, while keeping existing test entry points aligned with the new structure. --- tests/gui/main_window/test_preview.py | 87 +++++++++++++++++++++++ tests/gui/main_window/test_recording.py | 91 +++++++++++++++++++++++++ tests/gui/main_window/test_ui.py | 52 ++++++++++++++ tests/gui/test_main.py | 1 + 4 files changed, 231 insertions(+) create mode 100644 tests/gui/main_window/test_preview.py create mode 100644 tests/gui/main_window/test_recording.py create mode 100644 tests/gui/main_window/test_ui.py diff --git a/tests/gui/main_window/test_preview.py b/tests/gui/main_window/test_preview.py new file mode 100644 index 0000000..4a16b85 --- /dev/null +++ b/tests/gui/main_window/test_preview.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +import numpy as np +import pytest +from PySide6.QtGui import QPixmap + + +@pytest.mark.gui +class TestPreviewLifecycle: + def test_start_preview_with_no_active_cameras_shows_error(self, monkeypatch, window): + w = window + for cam in w._config.multi_camera.cameras: + cam.enabled = False + + messages: list[str] = [] + monkeypatch.setattr(w, "_show_error", messages.append) + + w._start_preview() + + assert messages == ["No cameras configured. Use 'Configure Cameras...' to add cameras."] + assert not w.multi_camera_controller.is_running() + assert w.preview_button.isEnabled() + assert not w.stop_preview_button.isEnabled() + + def test_on_multi_camera_stopped_clears_runtime_state(self, monkeypatch, window): + w = window + monkeypatch.setattr(w, "_stop_multi_camera_recording", lambda: None) + + w.preview_button.setEnabled(False) + w.stop_preview_button.setEnabled(True) + w._current_frame = np.zeros((4, 4, 3), dtype=np.uint8) + w._multi_camera_frames = {"fake:index:0": np.zeros((4, 4, 3), dtype=np.uint8)} + w._multi_camera_display_ids = {"fake:index:0": "Cam0"} + w._running_cams_ids = {"fake:index:0"} + w._display_dirty = True + w.video_label.setPixmap(QPixmap(8, 8)) + + w._on_multi_camera_stopped() + + assert w.preview_button.isEnabled() + assert not w.stop_preview_button.isEnabled() + assert w._current_frame is None + assert w._multi_camera_frames == {} + assert w._multi_camera_display_ids == {} + assert w._running_cams_ids == set() + assert w._display_dirty is False + assert w.video_label.text() == "Camera preview not started" + + def test_stop_preview_requests_orderly_shutdown(self, monkeypatch, window): + w = window + calls: list[str] = [] + + monkeypatch.setattr(w.multi_camera_controller, "is_running", lambda: True) + monkeypatch.setattr(w, "_stop_multi_camera_recording", lambda: calls.append("recording")) + monkeypatch.setattr(w, "_stop_inference", lambda show_message=False: calls.append("inference")) + monkeypatch.setattr( + w.multi_camera_controller, + "stop", + lambda *args, **kwargs: calls.append("controller"), + ) + + w.preview_button.setEnabled(True) + w.stop_preview_button.setEnabled(True) + w.start_inference_button.setEnabled(True) + w.stop_inference_button.setEnabled(True) + w._pending_recording_after_preview = True + + w._stop_preview() + + assert calls == ["recording", "inference", "controller"] + assert w._pending_recording_after_preview is False + assert not w.preview_button.isEnabled() + assert not w.stop_preview_button.isEnabled() + assert not w.start_inference_button.isEnabled() + assert not w.stop_inference_button.isEnabled() + assert w.camera_stats_label.text() == "Camera idle" + + def test_on_multi_camera_started_updates_primary_buttons(self, window): + w = window + + w.preview_button.setEnabled(True) + w.stop_preview_button.setEnabled(False) + + w._on_multi_camera_started() + + assert not w.preview_button.isEnabled() + assert w.stop_preview_button.isEnabled() diff --git a/tests/gui/main_window/test_recording.py b/tests/gui/main_window/test_recording.py new file mode 100644 index 0000000..588f8c1 --- /dev/null +++ b/tests/gui/main_window/test_recording.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +import numpy as np +import pytest + +from dlclivegui.services.multi_camera_controller import get_camera_id + + +@pytest.mark.gui +class TestRecordingLifecycle: + def test_start_recording_auto_starts_preview_when_preview_is_not_running(self, monkeypatch, window): + w = window + calls: list[str] = [] + + monkeypatch.setattr(w.multi_camera_controller, "is_running", lambda: False) + monkeypatch.setattr(w, "_start_preview", lambda: calls.append("preview")) + + w._pending_recording_after_preview = False + w._start_recording() + + assert calls == ["preview"] + assert w._pending_recording_after_preview is True + + def test_start_recording_starts_immediately_when_preview_is_running(self, monkeypatch, window): + w = window + calls: list[str] = [] + + monkeypatch.setattr(w.multi_camera_controller, "is_running", lambda: True) + monkeypatch.setattr(w, "_start_multi_camera_recording", lambda: calls.append("recording")) + + w._pending_recording_after_preview = False + w._start_recording() + + assert calls == ["recording"] + assert w._pending_recording_after_preview is False + + def test_pending_recording_waits_until_all_expected_frames_are_available(self, monkeypatch, window): + w = window + calls: list[str] = [] + + monkeypatch.setattr(w.multi_camera_controller, "is_running", lambda: True) + monkeypatch.setattr(w, "_start_multi_camera_recording", lambda: calls.append("recording")) + + active = w._config.multi_camera.get_active_cameras() + assert len(active) >= 2 + first_id = get_camera_id(active[0]) + + w._pending_recording_after_preview = True + w._multi_camera_frames = {first_id: np.zeros((4, 4, 3), dtype=np.uint8)} + + w._try_start_pending_recording() + + assert calls == [] + assert w._pending_recording_after_preview is True + + def test_pending_recording_starts_when_all_expected_frames_are_available(self, monkeypatch, window): + w = window + calls: list[str] = [] + + monkeypatch.setattr(w.multi_camera_controller, "is_running", lambda: True) + monkeypatch.setattr(w, "_start_multi_camera_recording", lambda: calls.append("recording")) + + w._pending_recording_after_preview = True + w._multi_camera_frames = { + get_camera_id(cam): np.zeros((4, 4, 3), dtype=np.uint8) + for cam in w._config.multi_camera.get_active_cameras() + } + + w._try_start_pending_recording() + + assert calls == ["recording"] + assert w._pending_recording_after_preview is False + + def test_start_multi_camera_recording_success_sets_buttons_and_sink(self, window, start_all_spy): + w = window + active = w._config.multi_camera.get_active_cameras() + w._multi_camera_frames = {get_camera_id(cam): np.zeros((4, 4, 3), dtype=np.uint8) for cam in active} + + w._start_multi_camera_recording() + + assert start_all_spy["active_cams"] == active + assert not w.start_record_button.isEnabled() + assert w.stop_record_button.isEnabled() + + def test_stop_multi_camera_recording_when_idle_is_noop(self, window): + w = window + w._recording_stopping = False + + w._stop_multi_camera_recording() + + assert w._recording_stopping is False diff --git a/tests/gui/main_window/test_ui.py b/tests/gui/main_window/test_ui.py new file mode 100644 index 0000000..d4ba336 --- /dev/null +++ b/tests/gui/main_window/test_ui.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import pytest + +from dlclivegui.services.multi_camera_controller import get_camera_id + + +@pytest.mark.gui +class TestCameraLabels: + def test_update_active_cameras_label_uses_backend_device_name(self, window): + w = window + cam = w._config.multi_camera.cameras[0] + cam.name = "" + cam.properties = {"fake": {"device_name": "The Camera"}} + for extra in w._config.multi_camera.cameras[1:]: + extra.enabled = False + + w._update_active_cameras_label() + + assert "The Camera" in w.active_cameras_label.text() + assert "[fake:0]" in w.active_cameras_label.text() + + def test_refresh_dlc_camera_list_displays_friendly_label_but_stores_stable_id(self, window): + w = window + cam = w._config.multi_camera.cameras[0] + cam.name = "" + cam.properties = {"fake": {"device_name": "The Camera", "device_id": "stable-123"}} + for extra in w._config.multi_camera.cameras[1:]: + extra.enabled = False + + w._refresh_dlc_camera_list() + + assert w.dlc_camera_combo.count() == 1 + assert "The Camera" in w.dlc_camera_combo.itemText(0) + assert "stable-123" not in w.dlc_camera_combo.itemText(0) + assert w.dlc_camera_combo.itemData(0) == get_camera_id(cam) + + def test_label_for_cam_id_prefers_configured_friendly_label(self, window): + w = window + cam = w._config.multi_camera.cameras[0] + cam.name = "Top camera" + + assert w._label_for_cam_id(get_camera_id(cam)).startswith("Top camera") + + def test_label_for_cam_id_uses_runtime_display_id_fallback(self, window): + w = window + w._multi_camera_display_ids = {"runtime:id": "Runtime Camera"} + + assert w._label_for_cam_id("runtime:id") == "Runtime Camera" + + def test_label_for_cam_id_unknown_is_neutral(self, window): + assert window._label_for_cam_id("missing:id") == "Unknown camera" diff --git a/tests/gui/test_main.py b/tests/gui/test_main.py index b9ed7da..8c60e93 100644 --- a/tests/gui/test_main.py +++ b/tests/gui/test_main.py @@ -1,3 +1,4 @@ +# tests/gui/test_main.py import pytest from PySide6.QtCore import Qt from PySide6.QtGui import QImage From 878e947111cfa10b25d1e80d03a026a76c63d9c9 Mon Sep 17 00:00:00 2001 From: C-Achard Date: Wed, 8 Jul 2026 15:31:52 +0200 Subject: [PATCH 5/9] Drop strobe settings when fields are hidden Before saving trigger configuration, this change removes strobe-related properties if the active profile does not expose strobe fields. This prevents hidden/unsupported strobe values from being written to the backend trigger config. --- dlclivegui/gui/camera_config/trigger_config_dialog.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/dlclivegui/gui/camera_config/trigger_config_dialog.py b/dlclivegui/gui/camera_config/trigger_config_dialog.py index bda9cae..d8337e3 100644 --- a/dlclivegui/gui/camera_config/trigger_config_dialog.py +++ b/dlclivegui/gui/camera_config/trigger_config_dialog.py @@ -455,6 +455,14 @@ def _accept(self) -> None: return ns = _backend_namespace(self._cam) - ns["trigger"] = trigger.to_properties() + trigger_props = trigger.to_properties() + + if not self._profile.show_strobe_fields: + trigger_props.pop("strobe_polarity", None) + trigger_props.pop("strobe_operation", None) + trigger_props.pop("strobe_duration", None) + trigger_props.pop("strobe_delay", None) + + ns["trigger"] = trigger_props self.accept() From 456c81eab048815391246d2df929a2b6144afbd6 Mon Sep 17 00:00:00 2001 From: C-Achard Date: Wed, 8 Jul 2026 15:32:13 +0200 Subject: [PATCH 6/9] Add trigger config dialog GUI test coverage Introduce a new `test_trigger_config.py` suite for camera trigger configuration. The tests cover backend UI profiles, backend namespace normalization, dialog field visibility/enabling by role and backend, loading existing trigger settings, and accept-path payload serialization for off/external/master roles across GenTL and Basler behavior. --- .../gui/camera_config/test_trigger_config.py | 287 ++++++++++++++++++ 1 file changed, 287 insertions(+) create mode 100644 tests/gui/camera_config/test_trigger_config.py diff --git a/tests/gui/camera_config/test_trigger_config.py b/tests/gui/camera_config/test_trigger_config.py new file mode 100644 index 0000000..2018177 --- /dev/null +++ b/tests/gui/camera_config/test_trigger_config.py @@ -0,0 +1,287 @@ +# tests/gui/camera_config/test_trigger_config_dialog.py +from __future__ import annotations + +import pytest + +from dlclivegui.config import CameraSettings +from dlclivegui.gui.camera_config.trigger_config_dialog import ( + TriggerConfigDialog, + _backend_namespace, + trigger_ui_profile_for_backend, +) + + +class TestTriggerUiProfiles: + @pytest.mark.parametrize( + ("backend", "supports_input", "supports_master", "show_strobe", "show_line"), + [ + ("gentl", True, True, True, True), + ("basler", True, True, False, True), + ("opencv", False, False, False, False), + ("fake", False, False, False, False), + ], + ) + def test_profile_capabilities_by_backend( + self, + backend: str, + supports_input: bool, + supports_master: bool, + show_strobe: bool, + show_line: bool, + ): + profile = trigger_ui_profile_for_backend(backend) + + assert profile.supports_input is supports_input + assert profile.supports_master is supports_master + assert profile.show_strobe_fields is show_strobe + assert profile.show_line_output_fields is show_line + + def test_profile_backend_is_case_insensitive(self): + upper = trigger_ui_profile_for_backend("GeNtL") + lower = trigger_ui_profile_for_backend("gentl") + + assert upper == lower + + +class TestBackendNamespace: + def test_backend_namespace_creates_backend_dict(self): + cam = CameraSettings(backend="gentl", index=0, properties={}) + + ns = _backend_namespace(cam) + + assert ns == {} + assert cam.properties == {"gentl": {}} + + def test_backend_namespace_replaces_non_dict_properties(self): + cam = CameraSettings(backend="gentl", index=0, properties={}) + cam.properties = None + + ns = _backend_namespace(cam) + ns["trigger"] = {"role": "external"} + + assert cam.properties == {"gentl": {"trigger": {"role": "external"}}} + + def test_backend_namespace_replaces_non_dict_namespace(self): + cam = CameraSettings(backend="gentl", index=0, properties={"gentl": "bad"}) + + ns = _backend_namespace(cam) + + assert ns == {} + assert cam.properties == {"gentl": {}} + + +class TestTriggerConfigDialogPresentation: + @pytest.mark.gui + def test_unknown_backend_exposes_only_off_role_and_disables_trigger_fields(self, qtbot): + cam = CameraSettings(backend="opencv", index=0, properties={}) + dlg = TriggerConfigDialog(cam) + qtbot.addWidget(dlg) + + roles = [dlg.role_combo.itemData(i) for i in range(dlg.role_combo.count())] + + assert roles == ["off"] + assert not dlg.selector_edit.isVisible() + assert not dlg.source_combo.isVisible() + assert not dlg.activation_combo.isVisible() + assert not dlg.output_line_edit.isVisible() + assert not dlg.strobe_polarity_combo.isVisible() + + @pytest.mark.gui + def test_gentl_profile_shows_input_master_line_and_strobe_fields(self, qtbot): + cam = CameraSettings(backend="gentl", index=0, properties={}) + dlg = TriggerConfigDialog(cam) + qtbot.addWidget(dlg) + + roles = [dlg.role_combo.itemData(i) for i in range(dlg.role_combo.count())] + + assert roles == ["off", "external", "follower", "master"] + assert not dlg.selector_edit.isHidden() + assert not dlg.source_combo.isHidden() + assert not dlg.activation_combo.isHidden() + assert not dlg.output_line_edit.isHidden() + assert not dlg.output_source_edit.isHidden() + assert not dlg.strobe_polarity_combo.isHidden() + assert not dlg.strobe_operation_combo.isHidden() + assert not dlg.strobe_duration_spin.isHidden() + assert not dlg.strobe_delay_spin.isHidden() + + @pytest.mark.gui + def test_basler_profile_shows_line_fields_but_hides_strobe_fields(self, qtbot): + cam = CameraSettings(backend="basler", index=0, properties={}) + dlg = TriggerConfigDialog(cam) + qtbot.addWidget(dlg) + + roles = [dlg.role_combo.itemData(i) for i in range(dlg.role_combo.count())] + + assert roles == ["off", "external", "follower", "master"] + assert not dlg.output_line_edit.isHidden() + assert not dlg.output_source_edit.isHidden() + assert dlg.strobe_polarity_combo.isHidden() + assert dlg.strobe_operation_combo.isHidden() + assert dlg.strobe_duration_spin.isHidden() + assert dlg.strobe_delay_spin.isHidden() + + @pytest.mark.gui + def test_role_changes_enable_input_and_output_fields(self, qtbot): + cam = CameraSettings(backend="gentl", index=0, properties={}) + dlg = TriggerConfigDialog(cam) + qtbot.addWidget(dlg) + + dlg.role_combo.setCurrentIndex(dlg.role_combo.findData("off")) + assert not dlg.selector_edit.isEnabled() + assert not dlg.source_combo.isEnabled() + assert not dlg.output_line_edit.isEnabled() + + dlg.role_combo.setCurrentIndex(dlg.role_combo.findData("external")) + assert dlg.selector_edit.isEnabled() + assert dlg.source_combo.isEnabled() + assert dlg.activation_combo.isEnabled() + assert not dlg.output_line_edit.isEnabled() + + dlg.role_combo.setCurrentIndex(dlg.role_combo.findData("master")) + assert not dlg.selector_edit.isEnabled() + assert not dlg.source_combo.isEnabled() + assert dlg.output_line_edit.isEnabled() + assert dlg.output_source_edit.isEnabled() + assert dlg.strobe_polarity_combo.isEnabled() + assert dlg.strobe_delay_spin.isEnabled() + + @pytest.mark.gui + def test_fixed_duration_enables_strobe_duration_only_for_master(self, qtbot): + cam = CameraSettings(backend="gentl", index=0, properties={}) + dlg = TriggerConfigDialog(cam) + qtbot.addWidget(dlg) + + dlg.role_combo.setCurrentIndex(dlg.role_combo.findData("master")) + dlg.strobe_operation_combo.setCurrentIndex(dlg.strobe_operation_combo.findData("FixedDuration")) + + assert dlg.strobe_duration_spin.isEnabled() + + dlg.role_combo.setCurrentIndex(dlg.role_combo.findData("external")) + + assert not dlg.strobe_duration_spin.isEnabled() + + +class TestTriggerConfigDialogModelRoundtrip: + @pytest.mark.gui + def test_loads_existing_trigger_settings(self, qtbot): + cam = CameraSettings( + backend="gentl", + index=0, + properties={ + "gentl": { + "trigger": { + "role": "external", + "selector": "FrameStart", + "source": "Line1", + "activation": "FallingEdge", + "timeout": 0.25, + "strict": True, + } + } + }, + ) + + dlg = TriggerConfigDialog(cam) + qtbot.addWidget(dlg) + + assert dlg.role_combo.currentData() == "external" + assert dlg.selector_edit.text() == "FrameStart" + assert dlg.source_combo.currentText() == "Line1" + assert dlg.activation_combo.currentData() == "FallingEdge" + assert dlg.timeout_spin.value() == pytest.approx(0.25) + assert dlg.strict_checkbox.isChecked() + + @pytest.mark.gui + def test_accept_external_writes_trigger_payload_to_backend_namespace(self, qtbot): + cam = CameraSettings(backend="gentl", index=0, properties={}) + dlg = TriggerConfigDialog(cam) + qtbot.addWidget(dlg) + + dlg.role_combo.setCurrentIndex(dlg.role_combo.findData("external")) + dlg.selector_edit.setText("FrameStart") + dlg.source_combo.setCurrentText("Line1") + dlg.activation_combo.setCurrentIndex(dlg.activation_combo.findData("FallingEdge")) + dlg.timeout_spin.setValue(0.5) + dlg.strict_checkbox.setChecked(True) + + with qtbot.waitSignal(dlg.accepted, timeout=1000): + dlg._accept() + + trigger = dlg.camera_settings.properties["gentl"]["trigger"] + + assert trigger["role"] == "external" + assert trigger["selector"] == "FrameStart" + assert trigger["source"] == "Line1" + assert trigger["activation"] == "FallingEdge" + assert trigger["timeout"] == pytest.approx(0.5) + assert trigger["strict"] is True + + @pytest.mark.gui + def test_accept_off_clears_timeout(self, qtbot): + cam = CameraSettings( + backend="gentl", + index=0, + properties={"gentl": {"trigger": {"role": "external", "timeout": 2.0}}}, + ) + dlg = TriggerConfigDialog(cam) + qtbot.addWidget(dlg) + + dlg.role_combo.setCurrentIndex(dlg.role_combo.findData("off")) + dlg.timeout_spin.setValue(1.25) + + with qtbot.waitSignal(dlg.accepted, timeout=1000): + dlg._accept() + + trigger = dlg.camera_settings.properties["gentl"]["trigger"] + + assert trigger["role"] == "off" + assert trigger.get("timeout") is None + + @pytest.mark.gui + def test_accept_master_gentl_includes_strobe_values(self, qtbot): + cam = CameraSettings(backend="gentl", index=0, properties={}) + dlg = TriggerConfigDialog(cam) + qtbot.addWidget(dlg) + + dlg.role_combo.setCurrentIndex(dlg.role_combo.findData("master")) + dlg.output_line_edit.setText("Line2") + dlg.output_source_edit.setText("ExposureActive") + dlg.strobe_polarity_combo.setCurrentIndex(dlg.strobe_polarity_combo.findData("ActiveLow")) + dlg.strobe_operation_combo.setCurrentIndex(dlg.strobe_operation_combo.findData("FixedDuration")) + dlg.strobe_duration_spin.setValue(1200) + dlg.strobe_delay_spin.setValue(300) + + with qtbot.waitSignal(dlg.accepted, timeout=1000): + dlg._accept() + + trigger = dlg.camera_settings.properties["gentl"]["trigger"] + + assert trigger["role"] == "master" + assert trigger["output_line"] == "Line2" + assert trigger["output_source"] == "ExposureActive" + assert trigger["strobe_polarity"] == "ActiveLow" + assert trigger["strobe_operation"] == "FixedDuration" + assert trigger["strobe_duration"] == 1200 + assert trigger["strobe_delay"] == 300 + + @pytest.mark.gui + def test_accept_master_basler_does_not_add_strobe_values(self, qtbot): + cam = CameraSettings(backend="basler", index=0, properties={}) + dlg = TriggerConfigDialog(cam) + qtbot.addWidget(dlg) + + dlg.role_combo.setCurrentIndex(dlg.role_combo.findData("master")) + dlg.strobe_duration_spin.setValue(1200) + dlg.strobe_delay_spin.setValue(300) + + with qtbot.waitSignal(dlg.accepted, timeout=1000): + dlg._accept() + + trigger = dlg.camera_settings.properties["basler"]["trigger"] + + assert trigger["role"] == "master" + assert "strobe_duration" not in trigger + assert "strobe_delay" not in trigger + assert "strobe_polarity" not in trigger + assert "strobe_operation" not in trigger From cc9962752e2cb7bf6b819667c97740fee46ca13f Mon Sep 17 00:00:00 2001 From: C-Achard Date: Fri, 10 Jul 2026 15:09:15 +0200 Subject: [PATCH 7/9] Sync settings after saving config Call `self.settings.sync()` after saving the config and updating stored paths/snapshots so settings are flushed promptly. If syncing fails, log a debug message with exception info without interrupting the save flow. --- dlclivegui/gui/main_window.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/dlclivegui/gui/main_window.py b/dlclivegui/gui/main_window.py index adf825c..493335b 100644 --- a/dlclivegui/gui/main_window.py +++ b/dlclivegui/gui/main_window.py @@ -1106,6 +1106,12 @@ def _save_config_to_path(self, path: Path) -> bool: config.save(path) self._settings_store.set_last_config_path(str(path)) self._settings_store.save_full_config_snapshot(config) + + try: + self.settings.sync() + except Exception: + logger.debug("Failed to sync settings after saving config", exc_info=True) + except Exception as exc: # pragma: no cover - GUI interaction self._show_error(str(exc)) return False From b0800c960a96d01283fa2e1bf72bf0d28805736f Mon Sep 17 00:00:00 2001 From: C-Achard Date: Fri, 10 Jul 2026 15:10:26 +0200 Subject: [PATCH 8/9] Add unit tests for GenTL discovery utils Introduce a new test module covering `gentl_discovery` behavior end-to-end: CTI input normalization, explicit/env/extra-dir discovery, glob validation and allowed-root checks, candidate deduplication, and selection policies (`FIRST`, `NEWEST`, `RAISE_IF_MULTIPLE`). It also adds lifecycle tests for `SharedHarvesterPool`/`SharedHarvesterEntry`, including reuse/refcount semantics and failure reporting when CTI loading fails. --- .../backends/utils/test_gentl_discovery.py | 361 ++++++++++++++++++ 1 file changed, 361 insertions(+) create mode 100644 tests/cameras/backends/utils/test_gentl_discovery.py diff --git a/tests/cameras/backends/utils/test_gentl_discovery.py b/tests/cameras/backends/utils/test_gentl_discovery.py new file mode 100644 index 0000000..46ff8ca --- /dev/null +++ b/tests/cameras/backends/utils/test_gentl_discovery.py @@ -0,0 +1,361 @@ +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from dlclivegui.cameras.backends.utils import gentl_discovery as gd + +pytestmark = pytest.mark.unit + + +@pytest.fixture(autouse=True) +def clear_shared_harvester_pool(): + gd.SharedHarvesterPool._entries.clear() + yield + gd.SharedHarvesterPool._entries.clear() + + +def test_cti_files_as_list_handles_none_strings_and_sequences(): + assert gd.cti_files_as_list(None) == [] + assert gd.cti_files_as_list("") == [] + assert gd.cti_files_as_list(" ") == [] + assert gd.cti_files_as_list("camera.cti") == ["camera.cti"] + assert gd.cti_files_as_list(["a.cti", None, "", " ", Path("b.cti")]) == ["a.cti", "b.cti"] + + +def test_discover_explicit_cti_file_without_harvester(tmp_path: Path): + cti = tmp_path / "producer.cti" + cti.write_text("", encoding="utf-8") + + candidates, diag = gd.discover_cti_files( + cti_file=str(cti), + include_env=False, + ) + + assert candidates == [str(cti.resolve())] + assert diag.explicit_files == [str(cti)] + assert diag.candidates == [str(cti.resolve())] + assert diag.rejected == [] + + +def test_discover_rejects_missing_explicit_file(tmp_path: Path): + missing = tmp_path / "missing.cti" + + candidates, diag = gd.discover_cti_files( + cti_file=str(missing), + include_env=False, + ) + + assert candidates == [] + assert diag.rejected == [(str(missing.resolve()), "not a file (explicit)")] + + +def test_discover_rejects_non_cti_file(tmp_path: Path): + not_cti = tmp_path / "producer.txt" + not_cti.write_text("", encoding="utf-8") + + candidates, diag = gd.discover_cti_files( + cti_file=str(not_cti), + include_env=False, + ) + + assert candidates == [] + assert diag.rejected == [(str(not_cti.resolve()), "not a .cti (explicit)")] + + +def test_discover_accepts_missing_cti_when_must_exist_false(tmp_path: Path): + missing = tmp_path / "missing.cti" + + candidates, diag = gd.discover_cti_files( + cti_file=str(missing), + include_env=False, + must_exist=False, + ) + + assert candidates == [str(missing.resolve())] + assert diag.rejected == [] + + +def test_discover_extra_dir_collects_cti_files_non_recursive(tmp_path: Path): + root = tmp_path / "ctis" + root.mkdir() + + a = root / "a.cti" + b = root / "b.cti" + ignored = root / "ignored.txt" + nested = root / "nested" + nested.mkdir() + nested_cti = nested / "nested.cti" + + a.write_text("", encoding="utf-8") + b.write_text("", encoding="utf-8") + ignored.write_text("", encoding="utf-8") + nested_cti.write_text("", encoding="utf-8") + + candidates, diag = gd.discover_cti_files( + include_env=False, + extra_dirs=[str(root)], + recursive_extra_search=False, + ) + + assert candidates == [str(a.resolve()), str(b.resolve())] + assert diag.extra_dirs == [str(root)] + + +def test_discover_extra_dir_collects_cti_files_recursive(tmp_path: Path): + root = tmp_path / "ctis" + nested = root / "nested" + nested.mkdir(parents=True) + + top = root / "top.cti" + child = nested / "child.cti" + + top.write_text("", encoding="utf-8") + child.write_text("", encoding="utf-8") + + candidates, _diag = gd.discover_cti_files( + include_env=False, + extra_dirs=[str(root)], + recursive_extra_search=True, + ) + + assert candidates == [str(top.resolve()), str(child.resolve())] + + +def test_discover_deduplicates_candidates_preserving_order(tmp_path: Path): + cti = tmp_path / "producer.cti" + cti.write_text("", encoding="utf-8") + + candidates, diag = gd.discover_cti_files( + cti_file=str(cti), + cti_files=[str(cti)], + extra_dirs=[str(tmp_path)], + include_env=False, + ) + + assert candidates == [str(cti.resolve())] + assert diag.candidates == [str(cti.resolve())] + + +def test_discover_env_var_direct_file(monkeypatch, tmp_path: Path): + cti = tmp_path / "producer.cti" + cti.write_text("", encoding="utf-8") + + monkeypatch.setenv("MY_GENTL_PATH", str(cti)) + + candidates, diag = gd.discover_cti_files( + include_env=True, + env_vars=("MY_GENTL_PATH",), + ) + + assert candidates == [str(cti.resolve())] + assert diag.env_vars_used == {"MY_GENTL_PATH": str(cti)} + assert diag.env_paths_expanded == [str(cti)] + + +def test_discover_env_var_directory(monkeypatch, tmp_path: Path): + cti = tmp_path / "producer.cti" + cti.write_text("", encoding="utf-8") + + monkeypatch.setenv("MY_GENTL_PATH", str(tmp_path)) + + candidates, diag = gd.discover_cti_files( + include_env=True, + env_vars=("MY_GENTL_PATH",), + ) + + assert candidates == [str(cti.resolve())] + assert diag.env_vars_used == {"MY_GENTL_PATH": str(tmp_path)} + assert diag.env_paths_expanded == [str(tmp_path)] + + +def test_discover_env_var_multiple_entries(monkeypatch, tmp_path: Path): + d1 = tmp_path / "one" + d2 = tmp_path / "two" + d1.mkdir() + d2.mkdir() + + cti1 = d1 / "one.cti" + cti2 = d2 / "two.cti" + cti1.write_text("", encoding="utf-8") + cti2.write_text("", encoding="utf-8") + + monkeypatch.setenv("MY_GENTL_PATH", os.pathsep.join([str(d1), str(d2)])) + + candidates, _diag = gd.discover_cti_files( + include_env=True, + env_vars=("MY_GENTL_PATH",), + ) + + assert candidates == [str(cti1.resolve()), str(cti2.resolve())] + + +def test_validate_glob_pattern_rejects_empty_pattern(): + ok, reason = gd._validate_glob_pattern("") + + assert ok is False + assert reason == "empty glob pattern" + + +def test_validate_glob_pattern_rejects_traversal(tmp_path: Path): + pattern = str(tmp_path / ".." / "*.cti") + + ok, reason = gd._validate_glob_pattern(pattern) + + assert ok is False + assert reason == "glob pattern contains '..' traversal" + + +def test_validate_glob_pattern_rejects_non_cti_pattern(tmp_path: Path): + pattern = str(tmp_path / "*.txt") + + ok, reason = gd._validate_glob_pattern(pattern) + + assert ok is False + assert reason == "glob pattern does not target .cti files" + + +def test_validate_glob_pattern_rejects_outside_allowed_roots(tmp_path: Path): + allowed = tmp_path / "allowed" + outside = tmp_path / "outside" + allowed.mkdir() + outside.mkdir() + + pattern = str(outside / "*.cti") + + ok, reason = gd._validate_glob_pattern( + pattern, + allowed_roots=[str(allowed)], + ) + + assert ok is False + assert reason == "glob pattern base is outside allowed roots" + + +def test_discover_glob_pattern_with_allowed_root(tmp_path: Path): + cti_dir = tmp_path / "ctis" + cti_dir.mkdir() + + cti = cti_dir / "producer.cti" + cti.write_text("", encoding="utf-8") + + candidates, diag = gd.discover_cti_files( + cti_search_paths=[str(cti_dir / "*.cti")], + include_env=False, + root_globs_allowed=[str(tmp_path)], + ) + + assert candidates == [str(cti.resolve())] + assert diag.rejected == [] + + +def test_choose_cti_files_first_policy(): + assert gd.choose_cti_files( + ["a.cti", "b.cti", "c.cti"], + policy=gd.GenTLDiscoveryPolicy.FIRST, + max_files=2, + ) == ["a.cti", "b.cti"] + + +def test_choose_cti_files_raise_if_multiple_policy_raises(): + with pytest.raises(RuntimeError, match="Multiple GenTL producers"): + gd.choose_cti_files( + ["a.cti", "b.cti"], + policy=gd.GenTLDiscoveryPolicy.RAISE_IF_MULTIPLE, + max_files=1, + ) + + +def test_choose_cti_files_raise_if_multiple_policy_allows_within_limit(): + assert gd.choose_cti_files( + ["a.cti"], + policy=gd.GenTLDiscoveryPolicy.RAISE_IF_MULTIPLE, + max_files=1, + ) == ["a.cti"] + + +def test_choose_cti_files_newest_policy(tmp_path: Path): + old = tmp_path / "old.cti" + new = tmp_path / "new.cti" + + old.write_text("", encoding="utf-8") + new.write_text("", encoding="utf-8") + + os.utime(old, (1000, 1000)) + os.utime(new, (2000, 2000)) + + assert gd.choose_cti_files( + [str(old), str(new)], + policy=gd.GenTLDiscoveryPolicy.NEWEST, + max_files=1, + ) == [str(new)] + + +def test_choose_cti_files_empty_candidates(): + assert gd.choose_cti_files([]) == [] + + +def test_choose_cti_files_unknown_policy_raises(): + with pytest.raises(ValueError, match="Unknown policy"): + gd.choose_cti_files( + ["a.cti"], + policy=object(), # type: ignore[arg-type] + ) + + +def test_shared_harvester_pool_reuses_entry_and_refcounts(monkeypatch, tmp_path: Path): + calls: list[tuple[str, str | None]] = [] + + class FakeHarvester: + def __init__(self): + calls.append(("init", None)) + + def add_file(self, path: str) -> None: + calls.append(("add_file", path)) + + def update(self) -> None: + calls.append(("update", None)) + + def reset(self) -> None: + calls.append(("reset", None)) + + cti = tmp_path / "producer.cti" + cti.write_text("", encoding="utf-8") + + monkeypatch.setattr(gd, "Harvester", FakeHarvester) + gd.SharedHarvesterPool._entries.clear() + + entry1 = gd.SharedHarvesterPool.acquire([str(cti)]) + entry2 = gd.SharedHarvesterPool.acquire([str(cti)]) + + assert entry1 is entry2 + assert gd.SharedHarvesterPool.get_refcount(entry1) == 2 + + gd.SharedHarvesterPool.release(entry1) + assert gd.SharedHarvesterPool.get_refcount(entry2) == 1 + + gd.SharedHarvesterPool.release(entry2) + assert gd.SharedHarvesterPool.get_refcount(entry2) == 0 + assert ("reset", None) in calls + + +def test_shared_harvester_entry_reports_failed_files(monkeypatch, tmp_path: Path): + class FakeHarvester: + def add_file(self, path: str) -> None: + raise RuntimeError("load failed") + + def update(self) -> None: + raise AssertionError("update should not be called") + + def reset(self) -> None: + pass + + cti = tmp_path / "bad.cti" + cti.write_text("", encoding="utf-8") + + monkeypatch.setattr(gd, "Harvester", FakeHarvester) + + with pytest.raises(RuntimeError, match="No GenTL producer"): + gd.SharedHarvesterEntry([str(cti)]) From 81800e285e2d975610e16c8042149eb05ff46acd Mon Sep 17 00:00:00 2001 From: C-Achard Date: Fri, 10 Jul 2026 15:17:16 +0200 Subject: [PATCH 9/9] Add GUI tests for config and camera fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Expand main window GUI coverage with a regression test that ensures runtime camera fallback updates only the active inference camera, not the user’s preferred inference camera. Add a new test module for user-config persistence, covering config path validation, dialog path suggestion logic, successful save side effects (last path, snapshot, sync), and failure behavior that avoids persistence and reports errors. Also add file header comments in related test files. --- tests/gui/main_window/test_preview.py | 37 ++++++ tests/gui/main_window/test_recording.py | 1 + tests/gui/main_window/test_ui.py | 1 + tests/gui/main_window/test_user_config.py | 142 ++++++++++++++++++++++ 4 files changed, 181 insertions(+) create mode 100644 tests/gui/main_window/test_user_config.py diff --git a/tests/gui/main_window/test_preview.py b/tests/gui/main_window/test_preview.py index 4a16b85..571813d 100644 --- a/tests/gui/main_window/test_preview.py +++ b/tests/gui/main_window/test_preview.py @@ -1,9 +1,14 @@ +# tests/gui/main_window/test_preview.py from __future__ import annotations +from types import SimpleNamespace + import numpy as np import pytest from PySide6.QtGui import QPixmap +from dlclivegui.services.multi_camera_controller import get_camera_id + @pytest.mark.gui class TestPreviewLifecycle: @@ -85,3 +90,35 @@ def test_on_multi_camera_started_updates_primary_buttons(self, window): assert not w.preview_button.isEnabled() assert w.stop_preview_button.isEnabled() + + def test_processing_runtime_fallback_does_not_overwrite_preferred_inference_camera(self, window): + w = window + + active_cams = w._config.multi_camera.get_active_cameras() + if len(active_cams) < 2: + pytest.skip("This regression test requires at least two active cameras.") + + fallback_cam = active_cams[0] + preferred_cam = active_cams[1] + + fallback_id = get_camera_id(fallback_cam) + preferred_id = get_camera_id(preferred_cam) + + w._inference_camera_id = preferred_id + w._active_inference_camera_id = preferred_id + w._running_cams_ids = set() + w._dlc_active = False + + frame = np.zeros((4, 4, 3), dtype=np.uint8) + frame_data = SimpleNamespace( + frames={fallback_id: frame}, + display_ids={fallback_id: "Fallback camera"}, + source_camera_id=fallback_id, + timestamps={fallback_id: 123.0}, + ) + + w._on_multi_frame_processing_ready(frame_data) + + assert w._inference_camera_id == preferred_id + assert w._active_inference_camera_id == fallback_id + assert w.dlc_camera_combo.currentData() == fallback_id diff --git a/tests/gui/main_window/test_recording.py b/tests/gui/main_window/test_recording.py index 588f8c1..0443adb 100644 --- a/tests/gui/main_window/test_recording.py +++ b/tests/gui/main_window/test_recording.py @@ -1,3 +1,4 @@ +# tests/gui/main_window/test_recording.py from __future__ import annotations import numpy as np diff --git a/tests/gui/main_window/test_ui.py b/tests/gui/main_window/test_ui.py index d4ba336..1790864 100644 --- a/tests/gui/main_window/test_ui.py +++ b/tests/gui/main_window/test_ui.py @@ -1,3 +1,4 @@ +# tests/gui/main_window/test_ui.py from __future__ import annotations import pytest diff --git a/tests/gui/main_window/test_user_config.py b/tests/gui/main_window/test_user_config.py new file mode 100644 index 0000000..f42c4b9 --- /dev/null +++ b/tests/gui/main_window/test_user_config.py @@ -0,0 +1,142 @@ +# tests/gui/main_window/test_user_config.py +from __future__ import annotations + +from pathlib import Path + +import pytest + + +@pytest.mark.gui +class TestUserConfigPersistence: + def test_valid_config_file_path_accepts_existing_file(self, window, tmp_path: Path): + w = window + + config_path = tmp_path / "dlclive_config.json" + config_path.write_text("{}", encoding="utf-8") + + assert w._valid_config_file_path(str(config_path)) == config_path.resolve() + + def test_valid_config_file_path_rejects_missing_file(self, window, tmp_path: Path): + w = window + + missing = tmp_path / "missing_config.json" + + assert w._valid_config_file_path(str(missing)) is None + assert w._valid_config_file_path(None) is None + assert w._valid_config_file_path("") is None + + def test_suggest_config_dialog_path_prefers_current_config_path(self, window, tmp_path: Path): + w = window + + config_path = tmp_path / "current_config.json" + config_path.write_text("{}", encoding="utf-8") + + w._config_path = config_path + + assert w._suggest_config_dialog_path() == str(config_path) + + def test_suggest_config_dialog_path_uses_last_config_path_when_current_path_missing( + self, + monkeypatch, + window, + tmp_path: Path, + ): + w = window + + config_path = tmp_path / "last_config.json" + config_path.write_text("{}", encoding="utf-8") + + w._config_path = None + monkeypatch.setattr(w._settings_store, "get_last_config_path", lambda: str(config_path)) + + assert w._suggest_config_dialog_path() == str(config_path.resolve()) + + def test_suggest_config_dialog_path_uses_parent_of_missing_last_config( + self, + monkeypatch, + window, + tmp_path: Path, + ): + w = window + + missing_config_path = tmp_path / "missing_config.json" + + w._config_path = None + monkeypatch.setattr(w._settings_store, "get_last_config_path", lambda: str(missing_config_path)) + + assert w._suggest_config_dialog_path() == str(missing_config_path) + + def test_save_config_to_path_persists_last_path_snapshot_and_syncs( + self, + monkeypatch, + window, + tmp_path: Path, + ): + w = window + + calls: list[tuple[str, object]] = [] + + class FakeConfig: + def save(self, path: Path | str) -> None: + Path(path).write_text("{}", encoding="utf-8") + + class FakeSettings: + def sync(self) -> None: + calls.append(("sync", True)) + + config_path = tmp_path / "saved_config.json" + fake_config = FakeConfig() + + monkeypatch.setattr(w, "_current_config", lambda allow_empty_model_path=False: fake_config) + monkeypatch.setattr( + w._settings_store, + "set_last_config_path", + lambda path: calls.append(("last_path", path)), + ) + monkeypatch.setattr( + w._settings_store, + "save_full_config_snapshot", + lambda cfg: calls.append(("snapshot", cfg)), + ) + monkeypatch.setattr(w, "settings", FakeSettings()) + + assert w._save_config_to_path(config_path) is True + assert config_path.exists() + assert ("last_path", str(config_path.resolve())) in calls + assert ("snapshot", fake_config) in calls + assert ("sync", True) in calls + + def test_save_config_to_path_returns_false_without_persisting_after_failure( + self, + monkeypatch, + window, + tmp_path: Path, + ): + w = window + + calls: list[tuple[str, object]] = [] + errors: list[str] = [] + + class FakeConfig: + def save(self, path: Path | str) -> None: + raise OSError("cannot save") + + config_path = tmp_path / "failed_config.json" + + monkeypatch.setattr(w, "_current_config", lambda allow_empty_model_path=False: FakeConfig()) + monkeypatch.setattr( + w._settings_store, + "set_last_config_path", + lambda path: calls.append(("last_path", path)), + ) + monkeypatch.setattr( + w._settings_store, + "save_full_config_snapshot", + lambda cfg: calls.append(("snapshot", cfg)), + ) + monkeypatch.setattr(w, "_show_error", errors.append) + + assert w._save_config_to_path(config_path) is False + assert not config_path.exists() + assert calls == [] + assert errors