"""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 == []