orisys-app/shared/tests/test_worker_cancellation.py

"""Cancellation semantics for the acquisition workers.

The GUI must never block on a device open that hangs: ``stop()`` returns
immediately, the worker thread owns its own teardown, and a new session waits
for the previous one *in its own thread* instead of opening the device twice.

These tests drive the real worker loop with a fake ``orisys`` module, so they
need no camera and no sensor bundle beyond a ``sensor.json``.
"""

from __future__ import annotations

import sys
import threading
import time
import types
from pathlib import Path

APP_ROOT = Path(__file__).resolve().parents[2]
BOX_CONFIG = APP_ROOT / "apps" / "studio" / "config" / "sensors" / "box" / "sensor.json"

STOP_BUDGET_S = 0.5


class FakeSensor:
    """Blocks in the constructor until ``release`` is set."""

    def __init__(self, release: threading.Event, tracker: "OpenTracker", **kwargs):
        tracker.opened(self)
        if not release.wait(timeout=10.0):
            raise TimeoutError("fake sensor was never released")
        self.configured: dict = {}
        self.connected = True

    def configure(self, **kwargs):
        self.configured = kwargs

    def get_img(self):
        return None

    def disconnect(self):
        self.connected = False


class OpenTracker:
    """Records how many devices are open at the same time."""

    def __init__(self) -> None:
        self.lock = threading.Lock()
        self.open_count = 0
        self.peak_open = 0
        self.instances: list[FakeSensor] = []

    def opened(self, sensor: FakeSensor) -> None:
        with self.lock:
            self.instances.append(sensor)
            self.open_count += 1
            self.peak_open = max(self.peak_open, self.open_count)

    def closed(self, sensor: FakeSensor) -> None:
        with self.lock:
            if sensor.connected:
                sensor.connected = False
                self.open_count -= 1


def _install_fake_orisys(tracker: OpenTracker, release: threading.Event) -> object:
    """Replace ``orisys`` (and ``orisys.finger``) with blocking fakes."""
    saved = {name: sys.modules.get(name) for name in ("orisys", "orisys.finger")}

    class _Sensor(FakeSensor):
        def __init__(self, *args, **kwargs):
            super().__init__(release, tracker, **kwargs)

        def disconnect(self):
            tracker.closed(self)
            super().disconnect()

    module = types.ModuleType("orisys")
    module.__version__ = "fake"
    module.Sensor = _Sensor  # type: ignore[attr-defined]

    class _Runtime:
        def __init__(self, *args, **kwargs):
            self._door = _Sensor()
            self.tactile = types.SimpleNamespace(raw_sensor=None)

        @classmethod
        def open(cls, **kwargs):
            return cls(**kwargs)

        def warmup(self, **_kwargs):
            return None

        def close(self):
            self._door.disconnect()

    finger_module = types.ModuleType("orisys.finger")
    finger_module.FingerRuntime = _Runtime  # type: ignore[attr-defined]
    module.finger = finger_module  # type: ignore[attr-defined]

    sys.modules["orisys"] = module
    sys.modules["orisys.finger"] = finger_module
    return saved


def _restore(saved) -> None:
    for name, original in saved.items():
        if original is None:
            sys.modules.pop(name, None)
        else:
            sys.modules[name] = original


def _new_worker(kind: str):
    from orisys_app_shared.acquisition.finger_worker import FingerAcquisitionWorker
    from orisys_app_shared.acquisition.sdk_worker import SdkAcquisitionWorker

    if kind == "finger":
        return FingerAcquisitionWorker()
    return SdkAcquisitionWorker()


def _wait_until(predicate, timeout=5.0, interval=0.01) -> bool:
    deadline = time.perf_counter() + timeout
    while time.perf_counter() < deadline:
        if predicate():
            return True
        time.sleep(interval)
    return predicate()


def _stop_is_immediate_and_keeps_ownership(kind: str) -> None:
    tracker = OpenTracker()
    release = threading.Event()
    saved = _install_fake_orisys(tracker, release)
    try:
        worker = _new_worker(kind)
        worker.start("0", config_ref=BOX_CONFIG)
        assert _wait_until(lambda: tracker.peak_open >= 1), "session never opened the device"

        started = time.perf_counter()
        worker.stop()
        elapsed = time.perf_counter() - started
        assert elapsed < STOP_BUDGET_S, f"stop() blocked the caller for {elapsed:.2f}s"
        assert worker._thread is not None, "stop() abandoned a still-running thread"
        assert worker.consume_end_reason() == "stopped"

        release.set()
        assert _wait_until(lambda: worker._thread is None), "thread never released ownership"
        assert _wait_until(lambda: tracker.open_count == 0), "device was not closed"
    finally:
        release.set()
        _restore(saved)


def test_sdk_stop_is_immediate_and_keeps_ownership():
    _stop_is_immediate_and_keeps_ownership("sdk")


def test_finger_stop_is_immediate_and_keeps_ownership():
    _stop_is_immediate_and_keeps_ownership("finger")


def test_second_session_waits_instead_of_double_opening():
    tracker = OpenTracker()
    release = threading.Event()
    saved = _install_fake_orisys(tracker, release)
    try:
        worker = _new_worker("sdk")
        worker.start("0", config_ref=BOX_CONFIG)
        assert _wait_until(lambda: tracker.peak_open == 1)

        worker.stop()
        worker.start("0", config_ref=BOX_CONFIG)  # user hits Run again while stuck
        time.sleep(0.3)
        assert tracker.peak_open == 1, "two sessions opened the same device at once"

        release.set()
        assert _wait_until(lambda: len(tracker.instances) == 2, timeout=5.0), (
            "the second session never opened after the stale one released the device"
        )
        assert tracker.peak_open == 1, "the device was open twice concurrently"
        assert tracker.open_count <= 1, "stale session kept its device open"
        worker.stop(wait=True)
        assert _wait_until(lambda: tracker.open_count == 0), "device leaked"
    finally:
        release.set()
        _restore(saved)


def test_stop_wait_is_bounded_and_closes_the_device():
    tracker = OpenTracker()
    release = threading.Event()
    saved = _install_fake_orisys(tracker, release)
    try:
        worker = _new_worker("sdk")
        worker.start("0", config_ref=BOX_CONFIG)
        assert _wait_until(lambda: tracker.peak_open == 1)
        release.set()
        started = time.perf_counter()
        worker.stop(wait=True)
        elapsed = time.perf_counter() - started
        assert elapsed < 5.0, f"stop(wait=True) took {elapsed:.2f}s"
        assert _wait_until(lambda: tracker.open_count == 0), "device was not closed"
        assert not worker.is_running
    finally:
        release.set()
        _restore(saved)


if __name__ == "__main__":  # pragma: no cover - manual runner
    failures = 0
    for name, function in sorted(globals().items()):
        if name.startswith("test_") and callable(function):
            try:
                function()
                print(f"  ok   {name}")
            except AssertionError as exc:
                failures += 1
                print(f"  FAIL {name}: {exc}")
            except Exception as exc:  # noqa: BLE001
                failures += 1
                print(f"  ERR  {name}: {type(exc).__name__}: {exc}")
    print(f"failures: {failures}")
    raise SystemExit(1 if failures else 0)