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,52 @@
+from time import sleep
+from unittest.mock import ANY
+
+import pytest
+import requests_mock
+
+from pxii_bec.devices.smargopolo_smargon import Smargon
+
+
+def mock_server(requests_mock):
+    mock_data = {"SHX": 1.0, "SHY": 1.0, "SHZ": 1.0, "PHI": 1.0, "CHI": 1.0}
+
+    def get_cb(request, context):
+        nonlocal mock_data
+        return mock_data
+
+    def put_cb(request, context):
+        nonlocal mock_data
+        update = {k.upper(): float(v[0]) for k, v in request.qs.items()}
+        mock_data.update(update)
+
+    requests_mock.get("http://test-smargopolo.psi.ch/readbackSCS", json=get_cb)
+    requests_mock.put("http://test-smargopolo.psi.ch/targetSCS", json=put_cb)
+
+
+@pytest.fixture
+def smargon():
+    s = Smargon(name="smargon", prefix="http://test-smargopolo.psi.ch")
+    s.controller._monitor = lambda *_: sleep(0.1)
+    yield s
+
+
+class TestSmargon:
+    def test_smargon_read(self, smargon: Smargon, requests_mock):
+        mock_server(requests_mock)
+        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, requests_mock):
+        mock_server(requests_mock)
+        smargon.wait_for_connection()
+        st = smargon.x.set(5.0)
+        smargon.controller.manual_update()
+        st.wait(timeout=1)
+        assert smargon.x.get() == 5.0