52 lines
1.5 KiB
Python
52 lines
1.5 KiB
Python
from copy import copy
|
|
from threading import RLock
|
|
from unittest.mock import ANY
|
|
|
|
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, params: dict | None = None, body: dict | None = None):
|
|
with self.lock:
|
|
assert params is not None
|
|
self.mock_data.update(params)
|
|
|
|
|
|
@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
|
|
s.controller._stop_monitor_readback_event.set()
|
|
|
|
|
|
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
|