diff --git a/pxii_bec/devices/smargopolo_smargon.py b/pxii_bec/devices/smargopolo_smargon.py index c0eda96..9af7bdf 100644 --- a/pxii_bec/devices/smargopolo_smargon.py +++ b/pxii_bec/devices/smargopolo_smargon.py @@ -79,13 +79,19 @@ class SmargonController(OphydObject): ], ) + def manual_update(self): + self._update_reading(self._readback_endpoint, self._readback_lock, self._readbacks) + + def _update_reading(self, endpoint: str, lock: RLock, buffer: dict): + data = self._rest_get(endpoint) + timestamp = time.monotonic() + with lock: + buffer.update(data) + buffer["__timestamp"] = timestamp + def _monitor(self, endpoint: str, event: Event, lock: RLock, buffer: dict): while not event.is_set(): - data = self._rest_get(endpoint) - timestamp = time.monotonic() - with lock: - buffer.update(data) - buffer["__timestamp"] = timestamp + self._update_reading(endpoint, lock, buffer) time.sleep(self._readback_poll_interval) def _clean_monitor(self): @@ -151,6 +157,7 @@ class Smargon(PSIDeviceBase): def wait_for_connection(self, **kwargs): # type: ignore self.controller.start_monitor() + self.controller.manual_update() return super().wait_for_connection(**kwargs) def stop(self, *, success: bool = False) -> None: diff --git a/pyproject.toml b/pyproject.toml index d259c6e..133b014 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,7 @@ dev = [ "pytest-random-order", "ophyd_devices", "bec_server", + "requests-mock", ] [project.entry-points."bec"] diff --git a/tests/tests_devices/test_smargon.py b/tests/tests_devices/test_smargon.py new file mode 100644 index 0000000..ba20c0b --- /dev/null +++ b/tests/tests_devices/test_smargon.py @@ -0,0 +1,49 @@ +from copy import copy +from threading import RLock +from unittest.mock import ANY, MagicMock, patch + +import pytest + + +class MockServer: + def __init__(self) -> None: + self.lock = RLock() + self.mock_data = {"SHX": 1.0, "SHY": 1.0, "SHZ": 1.0, "PHI": 1.0, "CHI": 1.0} + + def get(self, endpoint): + with self.lock: + return copy(self.mock_data) + + def put(self, val: dict[str, float]): + with self.lock: + self.mock_data.update(val) + + +@pytest.fixture +def smargon(): + mock_server = MockServer() + from pxii_bec.devices.smargopolo_smargon import Smargon + + s = Smargon(name="smargon", prefix="http://test-smargopolo.psi.ch") + s.controller._rest_get = mock_server.get + s.controller._rest_put = mock_server.put + yield s + + +class TestSmargon: + def test_smargon_read(self, smargon): + smargon.wait_for_connection() + reading = smargon.read() + assert dict(reading) == { + "smargon_x": {"value": 1.0, "timestamp": ANY}, + "smargon_y": {"value": 1.0, "timestamp": ANY}, + "smargon_z": {"value": 1.0, "timestamp": ANY}, + "smargon_phi": {"value": 1.0, "timestamp": ANY}, + "smargon_chi": {"value": 1.0, "timestamp": ANY}, + } + + def test_smargon_set_with_status(self, smargon): + smargon.wait_for_connection() + st = smargon.x.set(5.0) + st.wait(timeout=1) + assert smargon.x.get() == 5.0