CI for csaxs_bec / test (push) Failing after 2m10s
device.read() defaults to cached=False, i.e. one RPC round trip to the device server per device (24 for the six slits). Use cached=True, which reads the last published readback from redis; a missing value still shows as '---'. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017szgTwuHG65YhjoBGiK2Gj
46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
from unittest import mock
|
|
|
|
from csaxs_bec.bec_ipython_client.plugins.cSAXS.slits import cSAXSSlits
|
|
|
|
|
|
def _make_slits(devices: dict):
|
|
client = mock.MagicMock()
|
|
client.device_manager.devices = devices
|
|
return cSAXSSlits(client)
|
|
|
|
|
|
def test_sl_get_position_uses_cached_read():
|
|
dev = mock.MagicMock()
|
|
dev.read.return_value = {"sl4xc": {"value": 1.25}}
|
|
slits = _make_slits({"sl4xc": dev})
|
|
|
|
assert slits._sl_get_position("sl4xc") == 1.25
|
|
dev.read.assert_called_once_with(cached=True)
|
|
|
|
|
|
def test_sl_get_position_returns_none_without_cached_value():
|
|
dev = mock.MagicMock()
|
|
dev.read.return_value = None # nothing published to redis yet
|
|
slits = _make_slits({"sl4xc": dev})
|
|
|
|
assert slits._sl_get_position("sl4xc") is None
|
|
assert slits._sl_get_position("unknown_device") is None
|
|
|
|
|
|
def test_slits_show_all_reads_every_slit_device_cached(capsys):
|
|
devices = {}
|
|
for prefix in cSAXSSlits._SLIT_LABELS:
|
|
for suffix in ("xc", "xs", "yc", "ys"):
|
|
name = f"{prefix}{suffix}"
|
|
dev = mock.MagicMock()
|
|
dev.read.return_value = {name: {"value": 0.5}}
|
|
devices[name] = dev
|
|
slits = _make_slits(devices)
|
|
|
|
slits.slits_show_all()
|
|
|
|
assert len(devices) == 24
|
|
for dev in devices.values():
|
|
dev.read.assert_called_once_with(cached=True)
|
|
assert "cSAXS Slits Overview" in capsys.readouterr().out
|