225 lines
7.9 KiB
Python
225 lines
7.9 KiB
Python
"""Unit tests for the Timepix Fly backend."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
from ophyd import StatusBase
|
|
|
|
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_backend import TimepixFlyBackend
|
|
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_client import (
|
|
TimePixFlyStatus,
|
|
TimePixStatusError,
|
|
)
|
|
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_interface import (
|
|
NetAddresses,
|
|
OtherConfigModel,
|
|
PixelMap,
|
|
)
|
|
|
|
|
|
class FakeTimepixFlyClient:
|
|
"""Minimal client double that can drive backend status callbacks."""
|
|
|
|
def __init__(self, rest_url: str, ws_url: str):
|
|
self.rest_url = rest_url
|
|
self.ws_url = ws_url
|
|
self.status = TimePixFlyStatus.CONFIG
|
|
self._status_callbacks = {}
|
|
self.error_message = "boom"
|
|
self.on_connected = mock.Mock()
|
|
self.shutdown = mock.Mock()
|
|
self.start = mock.Mock()
|
|
self.stop_running_collection = mock.Mock()
|
|
self.set_other_config = mock.Mock()
|
|
self.set_pixel_map = mock.Mock()
|
|
self.get_net_addresses = mock.Mock(
|
|
return_value=NetAddresses(
|
|
control="127.0.0.1:8452", address="127.0.0.1:8451", server="127.0.0.1:8080"
|
|
)
|
|
)
|
|
|
|
def add_status_callback(self, status, success, error, run=True):
|
|
"""Store callbacks and optionally resolve them immediately."""
|
|
if run:
|
|
if self.status in success:
|
|
status.set_finished()
|
|
return
|
|
if self.status in error:
|
|
status.set_exception(
|
|
TimePixStatusError(
|
|
f"TimePixFly state '{self.status.value}': {self.error_message}"
|
|
)
|
|
)
|
|
return
|
|
self._status_callbacks[id(status)] = (status, success, error)
|
|
|
|
def last_error(self):
|
|
"""Return a lightweight error object."""
|
|
return SimpleNamespace(message=self.error_message)
|
|
|
|
def emit_status(self, status_value: TimePixFlyStatus):
|
|
"""Resolve stored status callbacks as if a websocket status update arrived."""
|
|
self.status = status_value
|
|
for cb_id, (status, success, error) in list(self._status_callbacks.items()):
|
|
with status._lock:
|
|
if status.done:
|
|
self._status_callbacks.pop(cb_id, None)
|
|
continue
|
|
if status_value in success:
|
|
status.set_finished()
|
|
self._status_callbacks.pop(cb_id, None)
|
|
elif status_value in error:
|
|
status.set_exception(
|
|
TimePixStatusError(
|
|
f"TimePixFly state '{status_value.value}': {self.error_message}"
|
|
)
|
|
)
|
|
self._status_callbacks.pop(cb_id, None)
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def backend_with_states():
|
|
"""Return a backend together with a helper that emits backend states."""
|
|
with mock.patch(
|
|
"superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_backend.TimepixFlyClient",
|
|
FakeTimepixFlyClient,
|
|
):
|
|
backend = TimepixFlyBackend(backend_rest_url="localhost:8452", hostname="localhost")
|
|
yield backend, backend.timepix_fly_client
|
|
backend.on_destroy()
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def pixel_map():
|
|
"""Small valid pixel map for backend unit tests."""
|
|
return PixelMap(chips=[[{"i": 0, "p": 0, "f": [1.0]}]])
|
|
|
|
|
|
def test_timepix_fly_backend_stage_pushes_configuration(backend_with_states, pixel_map):
|
|
"""Stage should push both config objects to the client."""
|
|
backend, client = backend_with_states
|
|
other_config = OtherConfigModel(output_uri="tcp:localhost:9000", TRoiStep=2, TRoiN=16)
|
|
|
|
backend.on_stage(other_config=other_config, pixel_map=pixel_map)
|
|
|
|
client.set_other_config.assert_called_once_with(other_config)
|
|
client.set_pixel_map.assert_called_once_with(pixel_map)
|
|
|
|
|
|
def test_timepix_fly_backend_trigger_callback_success(backend_with_states):
|
|
"""Trigger status should resolve once the backend reports await_connection."""
|
|
backend, client = backend_with_states
|
|
|
|
status = backend.on_trigger()
|
|
assert status.done is False
|
|
|
|
client.emit_status(TimePixFlyStatus.AWAIT_CONNECTION)
|
|
|
|
status.wait(timeout=1)
|
|
assert status.done is True
|
|
assert status.success is True
|
|
|
|
|
|
def test_timepix_fly_backend_trigger_callback_error(backend_with_states):
|
|
"""Trigger status should fail when the backend reports an exception state."""
|
|
backend, client = backend_with_states
|
|
|
|
status = backend.on_trigger()
|
|
client.error_message = "failed to configure"
|
|
client.emit_status(TimePixFlyStatus.EXCEPT)
|
|
|
|
with pytest.raises(TimePixStatusError, match="failed to configure"):
|
|
status.wait(timeout=1)
|
|
|
|
|
|
def test_timepix_fly_backend_complete_callback_success(backend_with_states):
|
|
"""Complete status should resolve when the backend goes back to config."""
|
|
backend, client = backend_with_states
|
|
|
|
client.emit_status(TimePixFlyStatus.COLLECT)
|
|
status = backend.on_complete()
|
|
assert status.done is False
|
|
assert status.success is False
|
|
client.emit_status(TimePixFlyStatus.CONFIG)
|
|
status.wait(timeout=1)
|
|
assert status.done is True
|
|
assert status.success is True
|
|
|
|
client.emit_status(TimePixFlyStatus.COLLECT)
|
|
status = backend.on_complete()
|
|
client.error_message = "unexpected error during collection"
|
|
client.emit_status(TimePixFlyStatus.EXCEPT)
|
|
with pytest.raises(
|
|
TimePixStatusError, match="TimePixFly state 'except': unexpected error during collection"
|
|
):
|
|
status.wait(timeout=1)
|
|
|
|
|
|
def test_timepix_fly_backend_stop_cancels_tracked_statuses(backend_with_states):
|
|
"""Stopping the backend should fail all tracked statuses and stop collection."""
|
|
backend, client = backend_with_states
|
|
status = StatusBase()
|
|
backend.cancel_on_stop(status)
|
|
|
|
backend.on_stop()
|
|
|
|
client.stop_running_collection.assert_called_once()
|
|
with pytest.raises(RuntimeError, match="Stop called on device"):
|
|
status.wait(timeout=1)
|
|
|
|
|
|
def test_timepix_fly_backend_add_and_remove_callback(backend_with_states):
|
|
"""Callbacks can be registered and removed by id."""
|
|
backend, _ = backend_with_states
|
|
|
|
cb_id = backend.add_callback(lambda *_args, **_kwargs: None, kwd={"scan_id": 5})
|
|
|
|
stored_cb_id = next(iter(backend.callbacks))
|
|
assert str(stored_cb_id) == cb_id
|
|
backend.remove_callback(stored_cb_id)
|
|
assert stored_cb_id not in backend.callbacks
|
|
|
|
|
|
def test_timepix_fly_backend_decode_end_frame_runs_callbacks(backend_with_states):
|
|
"""The buffered frame callback should be invoked only once EndFrame arrives."""
|
|
backend, _ = backend_with_states
|
|
received = {}
|
|
|
|
def callback(start_frame, data_frames, end_frame, scan_id):
|
|
received["start_frame"] = start_frame
|
|
received["data_frames"] = data_frames
|
|
received["end_frame"] = end_frame
|
|
received["scan_id"] = scan_id
|
|
|
|
backend.add_callback(callback, kwd={"scan_id": 7})
|
|
backend._process_timepix_fly_msg(
|
|
{
|
|
"type": "StartFrame",
|
|
"Mode": "TOA",
|
|
"TRoiStart": 0,
|
|
"TRoiStep": 1,
|
|
"TRoiN": 2,
|
|
"NumEnergyPoints": 2,
|
|
"save_interval": 10,
|
|
}
|
|
)
|
|
backend._process_timepix_fly_msg(
|
|
{
|
|
"type": "XesData",
|
|
"period": 1,
|
|
"TDSpectra": [1, 2, 3, 4],
|
|
"totalEvents": 4,
|
|
"beforeROI": 0,
|
|
"afterROI": 0,
|
|
}
|
|
)
|
|
backend._process_timepix_fly_msg({"type": "EndFrame", "error": "", "periods": 5})
|
|
assert received["start_frame"]["type"] == "StartFrame"
|
|
assert received["data_frames"][0]["type"] == "XesData"
|
|
assert received["end_frame"]["type"] == "EndFrame"
|
|
assert received["scan_id"] == 7
|
|
assert backend._TimepixFlyBackend__msg_buffer == []
|