orisys-app/apps/studio/tests/test_session_controller.py

"""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"