"""Acceptance for the acquisition session's state and its UI-free mechanics. ``SessionController`` owns which worker runs, in which phase, and what the session reports (frame rate, cost, capture rate and drops, how many readouts failed). It touches no widget, so it can be driven directly — which is what these tests do, with a fake recording coordinator and fake workers. """ from __future__ import annotations import pytest from orisys_studio import session_controller from orisys_studio.session_controller import SessionController class _Recording: """The record channel as the controller uses it (no files, no Qt).""" def __init__(self) -> None: self.packets: list = [] self.calibrations: list = [] self.resets: list = [] self.invalid = 0 def on_packet(self, packet): self.packets.append(packet) return type("Report", (), {"valid": packet != "invalid"})() def note_calibration_applied(self, path, version, frame_id, mono_t, session) -> None: self.calibrations.append((path, version, frame_id, mono_t, session)) def note_reset(self, frame_id, mono_t, session) -> None: self.resets.append((frame_id, mono_t, session)) class _Worker: def __init__(self, **kwargs) -> None: self.kwargs = dict(kwargs) self.sinks: list = [] self.running = False def add_sink(self, sink) -> None: if sink not in self.sinks: self.sinks.append(sink) def remove_sink(self, sink) -> None: self.sinks.remove(sink) @pytest.fixture def workers(monkeypatch): """Replace both worker classes with recording fakes.""" import orisys_app_shared.acquisition.finger_worker as finger_module import orisys_app_shared.acquisition.sdk_worker as sdk_module created: list[_Worker] = [] def factory(**kwargs): worker = _Worker(**kwargs) created.append(worker) return worker monkeypatch.setattr(sdk_module, "SdkAcquisitionWorker", factory) monkeypatch.setattr(finger_module, "FingerAcquisitionWorker", factory) return created def _controller(recording=None, **kwargs): recording = recording or _Recording() controller = SessionController( recording=recording, packet_for_3d=lambda packet: None, **kwargs ) return controller, recording # --------------------------------------------------------------------- tests def test_a_confirmed_zeroing_reaches_the_record(qt_app, workers): """The device's own "the zeroing took effect" must not be dropped on the floor. Nothing was wired to it before: the reset request went out, the confirmation came back, and the session recorded no event for it. """ controller, recording = _controller() controller.ensure_workers() assert len(workers) == 2, "both workers are created (the sensor can change)" for worker in workers: callback = worker.kwargs.get("on_reset_applied") assert callable(callback), "the worker was not told where to report a zeroing" callback(7, 1.5, 3) assert recording.resets == [(7, 1.5, 3), (7, 1.5, 3)] def test_a_confirmed_calibration_reaches_the_record(qt_app, workers): controller, recording = _controller() controller.ensure_workers() for worker in workers: worker.kwargs["on_calibration_applied"]("/tmp/cal.json", "abc123", 4, 0.2, 1) assert recording.calibrations == [ ("/tmp/cal.json", "abc123", 4, 0.2, 1), ("/tmp/cal.json", "abc123", 4, 0.2, 1), ] def test_the_record_channel_is_a_sink_on_both_workers(qt_app, workers): controller, recording = _controller() controller.ensure_workers() for worker in workers: # Without a window-provided sink, the controller's own method is registered. assert controller.note_packet_recorded in worker.sinks assert len(worker.sinks) == 2, "the 3D pane is a second sink on the same thread" # A failed readout is counted; a good one is not. controller.note_packet_recorded("invalid") controller.note_packet_recorded("fine") assert controller.record_error_count == 1 assert recording.packets == ["invalid", "fine"] controller.detach_sinks() for worker in workers: assert worker.sinks == [] def test_a_run_connects_and_then_reports_its_own_telemetry(qt_app, workers): controller, recording = _controller() controller.ensure_workers() controller.reset_for_run(source="/dev/video0", sensor_type="box") first = controller.session_counter assert controller.phase == "idle" and not controller.is_running controller.mark_running() assert controller.is_running and controller.phase == "connecting" assert controller.run_timer.isActive(), "the run timer is the session's cadence" packet = type( "Packet", (), {"fps": 16.7, "process_ms": 58.0, "capture_fps": 30.0, "capture_dropped": 2}, )() controller.note_sample_telemetry(packet) assert (controller.latest_fps, controller.latest_process_ms) == (16.7, 58.0) assert controller.latest_capture_fps == 30.0 assert controller.latest_capture_dropped == 2 assert controller.note_first_frame() is True assert controller.phase == "running" assert controller.note_first_frame() is False, "the phase change happens once" assert controller.mark_stopped("user_stop") is False assert not controller.is_running and controller.phase == "idle" assert not controller.run_timer.isActive() # The telemetry of a session that is over must not be readable as this one's. assert controller.latest_fps is None and controller.latest_capture_fps == 0.0 assert controller.last_sample_valid is None controller.reset_for_run(source="/dev/video0", sensor_type="box") assert controller.session_counter == first + 1, "a new run is a new session id" def test_a_stop_while_connecting_says_so(qt_app, workers): """The caller uses this to tell "still releasing the device" from "already idle".""" controller, recording = _controller() controller.ensure_workers() controller.mark_running() assert controller.mark_stopped("user_stop") is True controller.mark_running() assert controller.note_first_frame() is True assert controller.mark_stopped("failed") is False assert controller.phase == "error" def test_the_slow_connect_hint_is_said_once(qt_app, monkeypatch, workers): controller, recording = _controller() controller.ensure_workers() clock = {"now": 100.0} monkeypatch.setattr(session_controller.time, "perf_counter", lambda: clock["now"]) controller.mark_running() clock["now"] = 100.0 + session_controller.CONNECT_SLOW_HINT_S - 0.1 assert controller.note_connect_slow() is False clock["now"] = 100.0 + session_controller.CONNECT_SLOW_HINT_S assert controller.note_connect_slow() is True assert controller.note_connect_slow() is False, "the hint was repeated" def test_a_dropped_batch_is_reported_once(qt_app, workers): controller, recording = _controller() assert controller.take_dropped_message(0) is None first = controller.take_dropped_message(3) assert first is not None and "3" in first assert controller.take_dropped_message(3) is None, "the same batch was reported twice" assert controller.take_dropped_message(5) is not None def test_a_capture_shortfall_is_judged_only_with_a_measured_rate( qt_app, monkeypatch, workers ): """The measured rate is the evidence; the driver's own report is an echo.""" monkeypatch.setattr( session_controller.session_metadata, "capture_metadata", lambda worker: {"requested_fps": 30.0, "negotiated_fps": 30.0}, ) controller, recording = _controller() controller.ensure_workers() assert controller.check_capture_negotiation() is None, "nothing measured yet" assert controller.capture_checked is False, "it decided without evidence" controller.note_sample_telemetry( type("Packet", (), {"capture_fps": 15.0, "capture_dropped": 0})() ) controller.check_capture_negotiation() assert controller.capture_checked is True, "the check never latched with evidence" assert controller.check_capture_negotiation() is None, "it is said once per session"