orisys-app/apps/autocal/tests/test_poller.py

"""The poller thread: serialisation, failure isolation, and clean teardown.

The two things that must never happen are a controller call on the GUI thread
and two controller calls at once. Both are asserted here against a transport
that records how it was used.
"""

from __future__ import annotations

import threading
import time

import pytest
from orisys_autocal.motion import (
    ConnectionSpec,
    DemoTransport,
    MotionPoller,
    MotionService,
)


class RecordingTransport(DemoTransport):
    """A *connected* demo platform that records thread and overlap of every call.

    Connected on purpose: the service polls only ``BUS_TYPE`` and ``Bus_InitStatus``
    without a BAS program, so an unconnected transport would keep every read
    short and nothing could be observed. A ``_concurrent`` guard asserts no two
    calls ever overlap rather than merely asserting the maximum sum of started
    calls, which would be trivially satisfied.
    """

    def __init__(
        self,
        *,
        delay: float = 0.0,
        unexpected_on: set[str] | None = None,
    ) -> None:
        super().__init__()
        self.delay = delay
        self.unexpected_on = unexpected_on or set()
        self.thread_ids: set[int] = set()
        self.max_concurrent = 0
        self.overlaps: list[str] = []
        self.calls: list[str] = []
        self._lock = threading.Lock()
        self._concurrent = 0
        # A fake "BAS program has run": the bus reports EtherCAT and ready.
        self._init_started = None
        self._preloaded = True

    def open(self, spec) -> None:
        super().open(ConnectionSpec(mode="demo"))
        self._bas_loaded = True

    def user_var(self, name: str) -> float:
        if name == "BUS_TYPE":
            return 0.0
        if name == "Bus_InitStatus":
            return 1.0
        if name == "Bus_TotalAxisnum":
            return 5.0
        return super().user_var(name)

    def _enter(self, name: str) -> None:
        with self._lock:
            self._concurrent += 1
            if self._concurrent > 1:
                self.overlaps.append(name)
            self.max_concurrent = max(self.max_concurrent, self._concurrent)
            self.calls.append(name)
            self.thread_ids.add(threading.get_ident())
        if self.delay:
            time.sleep(self.delay)
        if name in self.unexpected_on:
            raise RuntimeError(f"simulated unexpected fault in {name}")

    def _exit(self) -> None:
        with self._lock:
            self._concurrent -= 1

    def get_dpos(self, axis: int) -> float:
        self._enter("get_dpos")
        try:
            return super().get_dpos(axis)
        finally:
            self._exit()

    def get_mpos(self, axis: int) -> float:
        self._enter("get_mpos")
        try:
            return super().get_mpos(axis)
        finally:
            self._exit()

    def get_axis_status(self, axis: int) -> int:
        self._enter("get_axis_status")
        try:
            return super().get_axis_status(axis)
        finally:
            self._exit()

    def get_if_idle(self, axis: int) -> int:
        self._enter("get_if_idle")
        try:
            return super().get_if_idle(axis)
        finally:
            self._exit()


def _connected_poller(transport: RecordingTransport, *, axes=(0,)) -> MotionPoller:
    service = MotionService(transport)
    assert service.connect(ConnectionSpec(mode="demo")).ok
    return MotionPoller(service, axes=list(axes))


@pytest.fixture
def qt_app_session(qt_app):
    return qt_app

def _drain(app, seconds: float = 1.0) -> None:
    deadline = time.monotonic() + seconds
    while time.monotonic() < deadline:
        app.processEvents()
        time.sleep(0.005)


def _wait_for(predicate, app, timeout: float = 5.0) -> bool:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        app.processEvents()
        if predicate():
            return True
        time.sleep(0.01)
    return False


def test_completed_commands_are_reported_and_none_are_lost(qt_app):
    transport = RecordingTransport()
    poller = _connected_poller(transport)
    try:
        repo = {"finished": [], "failed": []}
        poller.commandFinished.connect(lambda token, result: repo["finished"].append(token))
        poller.commandFailed.connect(lambda token, message: repo["failed"].append(token))

        tokens = [poller.submit(lambda service: None) for _ in range(5)]
        assert _wait_for(lambda: len(repo["finished"]) == 5, qt_app)
        assert sorted(repo["finished"]) == sorted(tokens)
        assert repo["failed"] == []
    finally:
        poller.shutdown()


def test_every_controller_call_happens_on_one_worker_thread(qt_app):
    transport = RecordingTransport()
    poller = _connected_poller(transport, axes=(0, 1))
    try:
        assert _wait_for(lambda: "get_dpos" in transport.calls, qt_app)
        poller.submit(lambda service: None)
        _drain(qt_app, 0.3)
        assert len(transport.thread_ids) == 1, transport.thread_ids
        assert threading.get_ident() not in transport.thread_ids, (
            "a controller call ran on the test's (GUI) thread"
        )
    finally:
        poller.shutdown()


def test_two_controller_calls_never_overlap(qt_app):
    """A slow poll plus queued commands must still serialise."""
    transport = RecordingTransport(delay=0.02)
    poller = _connected_poller(transport)
    try:
        for _ in range(4):
            poller.submit(lambda service: None)
        _drain(qt_app, 0.8)
        assert transport.max_concurrent == 1, (
            f"{transport.max_concurrent} controller calls overlapped at once"
        )
        assert not transport.overlaps, transport.overlaps
    finally:
        poller.shutdown()


def test_an_unexpected_fault_is_reported_and_does_not_kill_the_poller(qt_app):
    """A non-``ZAuxError`` from inside a command must not stop the timer."""
    transport = RecordingTransport(unexpected_on={"get_dpos"})
    poller = _connected_poller(transport)
    try:
        failures = []
        poller.commandFailed.connect(lambda token, message: failures.append((token, message)))

        # The submitted action asks for the position directly, so the fault
        # surfaces from the command rather than being absorbed by a poll.
        poller.submit(lambda service: service.transport.get_dpos(0))
        assert _wait_for(lambda: bool(failures), qt_app)
        assert failures[0][1]

        # The poller is still alive: later snapshots keep arriving.
        before = len(transport.calls)
        _drain(qt_app, 0.4)
        assert len(transport.calls) > before
    finally:
        poller.shutdown()


def test_shutdown_stops_the_thread_and_is_idempotent(qt_app):
    poller = _connected_poller(RecordingTransport())
    _drain(qt_app, 0.2)
    poller.shutdown(timeout_ms=3000)
    poller.shutdown(timeout_ms=3000)  # must not raise or hang


def test_a_refused_command_still_arrives_with_its_reason(qt_app):
    """An unconnected service refuses in the guard, without touching the DLL."""
    poller = MotionPoller(MotionService(RecordingTransport()), axes=[0])
    try:
        results = []
        poller.commandFinished.connect(lambda token, result: results.append(result))
        poller.submit(lambda service: service.jog(0, True))
        assert _wait_for(lambda: bool(results), qt_app)
        assert results[0].ok is False
        assert results[0].message
    finally:
        poller.shutdown()


def test_connect_and_disconnect_round_trip_through_the_poller(qt_app):
    poller = MotionPoller(MotionService(RecordingTransport()), axes=[0])
    try:
        states = []
        poller.connectionChanged.connect(states.append)
        poller.connect_controller(ConnectionSpec(mode="demo"))
        assert _wait_for(lambda: poller.service.connected, qt_app)
        poller.disconnect_controller()
        assert _wait_for(lambda: not poller.service.connected, qt_app)
    finally:
        poller.shutdown()