Files
csaxs_bec/tests/tests_devices/test_fupr_ophyd.py
T
appel_c 93384b87e0
CI for csaxs_bec / test (push) Failing after 1m29s
CI for csaxs_bec / test (pull_request) Failing after 1m29s
fix(readback-timeout): fix monkeypatch for readback timeout; closes #140
2026-02-16 17:38:10 +01:00

72 lines
2.3 KiB
Python

from unittest import mock
import pytest
from ophyd_devices.tests.utils import SocketMock
from ophyd_devices.utils.socket import SocketSignal
from csaxs_bec.devices.omny.galil.fupr_ophyd import FuprGalilController, FuprGalilMotor
@pytest.fixture
def fsamroy(dm_with_devices):
FuprGalilController._reset_controller()
fsamroy_motor = FuprGalilMotor(
"A",
name="fsamroy",
host="mpc2680.psi.ch",
port=8081,
socket_cls=SocketMock,
device_manager=dm_with_devices,
)
for walk in fsamroy_motor.walk_signals():
if isinstance(walk.item, SocketSignal):
walk.item._readback_timeout = (
0.0 # Set the readback timeout to 0 to avoid waiting during tests
)
fsamroy_motor.controller.on()
assert isinstance(fsamroy_motor.controller, FuprGalilController)
yield fsamroy_motor
fsamroy_motor.controller.off()
fsamroy_motor.controller._reset_controller()
@pytest.mark.parametrize(
"pos,msg_received,msg_put,sign",
[
(-0.5, b" -12800\n\r", [b"TPA\r", b"MG_BGA\r", b"TPA\r"], 1),
(-0.5, b" 12800\n\r", [b"TPA\r", b"MG_BGA\r", b"TPA\r"], -1),
],
)
def test_axis_get(fsamroy, pos, msg_received, msg_put, sign):
fsamroy.sign = sign
fsamroy.device_manager = mock.MagicMock()
fsamroy.controller.sock.flush_buffer()
fsamroy.controller.sock.buffer_recv = msg_received
val = fsamroy.read()
assert val["fsamroy"]["value"] == pos
assert fsamroy.readback.get() == pos
assert fsamroy.controller.sock.buffer_put == msg_put
@pytest.mark.parametrize(
"target_pos,socket_put_messages,socket_get_messages",
[
(
0,
[b"MG axisref\r", b"PAA=0\r", b"PAA=0\r", b"BGA\r"],
[b"1.00", b"-1", b":", b":", b":", b":", b"-1"],
)
],
)
def test_axis_put(fsamroy, target_pos, socket_put_messages, socket_get_messages):
fsamroy.controller.sock.flush_buffer()
fsamroy.controller.sock.buffer_recv = socket_get_messages
fsamroy.user_setpoint.put(target_pos)
assert fsamroy.controller.sock.buffer_put == socket_put_messages
def test_drive_axis_to_limit(fsamroy):
fsamroy.controller.sock.flush_buffer()
with pytest.raises(NotImplementedError):
fsamroy.controller.drive_axis_to_limit(0, "forward")