tests: add test for smargon

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:

pyproject.toml

@@ -25,6 +25,7 @@ dev = [
     "pytest-random-order",
     "ophyd_devices",
     "bec_server",
+    "requests-mock",
 ]
 
 [project.entry-points."bec"]

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