ykcheungab
234 lines
7.8 KiB
Python
234 lines
7.8 KiB
Python
"""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()
|