222 lines
7.2 KiB
Python
222 lines
7.2 KiB
Python
import copy
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
from bec_server.device_server.tests.utils import DMMock
|
|
from ophyd_devices.tests.utils import SocketMock
|
|
|
|
from csaxs_bec.devices.npoint.npoint import NPointAxis, NPointController
|
|
from csaxs_bec.devices.omny.galil.fgalil_ophyd import FlomniGalilController, FlomniGalilMotor
|
|
from csaxs_bec.devices.omny.galil.fupr_ophyd import FuprGalilController, FuprGalilMotor
|
|
from csaxs_bec.devices.omny.galil.lgalil_ophyd import LamniGalilController, LamniGalilMotor
|
|
from csaxs_bec.devices.omny.galil.ogalil_ophyd import OMNYGalilController, OMNYGalilMotor
|
|
from csaxs_bec.devices.omny.galil.sgalil_ophyd import GalilController, SGalilMotor
|
|
from csaxs_bec.devices.omny.rt.rt_flomni_ophyd import RtFlomniController, RtFlomniMotor
|
|
from csaxs_bec.devices.omny.rt.rt_lamni_ophyd import RtLamniController, RtLamniMotor
|
|
from csaxs_bec.devices.omny.rt.rt_omny_ophyd import RtOMNYController, RtOMNYMotor
|
|
from csaxs_bec.devices.smaract.smaract_ophyd import SmaractController, SmaractMotor
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def leyey(dm_with_devices):
|
|
LamniGalilController._reset_controller()
|
|
leyey_motor = LamniGalilMotor(
|
|
"H",
|
|
name="leyey",
|
|
host="mpc2680.psi.ch",
|
|
port=8081,
|
|
socket_cls=SocketMock,
|
|
device_manager=dm_with_devices,
|
|
)
|
|
leyey_motor.controller.on()
|
|
yield leyey_motor
|
|
leyey_motor.controller.off()
|
|
leyey_motor.controller._reset_controller()
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def leyex(dm_with_devices):
|
|
LamniGalilController._reset_controller()
|
|
leyex_motor = LamniGalilMotor(
|
|
"A",
|
|
name="leyey",
|
|
host="mpc2680.psi.ch",
|
|
port=8081,
|
|
socket_cls=SocketMock,
|
|
device_manager=dm_with_devices,
|
|
)
|
|
leyex_motor.controller.on()
|
|
yield leyex_motor
|
|
leyex_motor.controller.off()
|
|
leyex_motor.controller._reset_controller()
|
|
|
|
|
|
@pytest.mark.parametrize("pos,msg,sign", [(1, b" -12800\n\r", 1), (-1, b" -12800\n\r", -1)])
|
|
def test_axis_get(leyey, pos, msg, sign):
|
|
leyey.sign = sign
|
|
leyey.controller.sock.flush_buffer()
|
|
leyey.controller.sock.buffer_recv = msg
|
|
val = leyey.read()
|
|
assert val["leyey"]["value"] == pos
|
|
assert leyey.readback.get() == pos
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"target_pos,socket_put_messages,socket_get_messages",
|
|
[
|
|
(
|
|
0,
|
|
[
|
|
b"MG allaxref\r",
|
|
b"MG_XQ0\r",
|
|
b"naxis=7\r",
|
|
b"ntarget=0.000\r",
|
|
b"movereq=1\r",
|
|
b"XQ#NEWPAR\r",
|
|
b"MG_XQ0\r",
|
|
],
|
|
[b"1.00", b"-1", b":", b":", b":", b":", b"-1"],
|
|
)
|
|
],
|
|
)
|
|
def test_axis_put(leyey, target_pos, socket_put_messages, socket_get_messages):
|
|
leyey.controller.sock.flush_buffer()
|
|
leyey.controller.sock.buffer_recv = copy.deepcopy(socket_get_messages)
|
|
leyey.user_setpoint.put(target_pos)
|
|
assert leyey.controller.sock.buffer_put == socket_put_messages
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"axis_nr,direction,socket_put_messages,socket_get_messages",
|
|
[
|
|
(
|
|
0,
|
|
"forward",
|
|
[
|
|
b"naxis=0\r",
|
|
b"ndir=1\r",
|
|
b"XQ#NEWPAR\r",
|
|
b"XQ#FES\r",
|
|
b"MG_BGA\r",
|
|
b"MGbcklact[0]\r",
|
|
b"MG_XQ0\r",
|
|
b"MG_XQ2\r",
|
|
b"MG _LRA, _LFA\r",
|
|
],
|
|
[b":", b":", b":", b":", b"0", b"0", b"-1", b"-1", b"1.000 0.000"],
|
|
),
|
|
(
|
|
1,
|
|
"reverse",
|
|
[
|
|
b"naxis=1\r",
|
|
b"ndir=-1\r",
|
|
b"XQ#NEWPAR\r",
|
|
b"XQ#FES\r",
|
|
b"MG_BGB\r",
|
|
b"MGbcklact[1]\r",
|
|
b"MG_XQ0\r",
|
|
b"MG_XQ2\r",
|
|
b"MG _LRB, _LFB\r",
|
|
],
|
|
[b":", b":", b":", b":", b"0", b"0", b"-1", b"-1", b"0.000 1.000"],
|
|
),
|
|
],
|
|
)
|
|
def test_drive_axis_to_limit(leyex, axis_nr, direction, socket_put_messages, socket_get_messages):
|
|
leyex.controller.sock.flush_buffer()
|
|
leyex.controller.sock.buffer_recv = copy.deepcopy(socket_get_messages)
|
|
leyex.controller.drive_axis_to_limit(axis_nr, direction)
|
|
assert leyex.controller.sock.buffer_put == socket_put_messages
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"axis_nr,socket_put_messages,socket_get_messages",
|
|
[
|
|
(
|
|
0,
|
|
[
|
|
b"naxis=0\r",
|
|
b"XQ#NEWPAR\r",
|
|
b"XQ#FRM\r",
|
|
b"MG_BGA\r",
|
|
b"MGbcklact[0]\r",
|
|
b"MG_XQ0\r",
|
|
b"MG_XQ2\r",
|
|
b"MG axisref[0]\r",
|
|
],
|
|
[b":", b":", b":", b"0", b"0", b"-1", b"-1", b"1.00"],
|
|
),
|
|
(
|
|
1,
|
|
[
|
|
b"naxis=1\r",
|
|
b"XQ#NEWPAR\r",
|
|
b"XQ#FRM\r",
|
|
b"MG_BGB\r",
|
|
b"MGbcklact[1]\r",
|
|
b"MG_XQ0\r",
|
|
b"MG_XQ2\r",
|
|
b"MG axisref[1]\r",
|
|
],
|
|
[b":", b":", b":", b"0", b"0", b"-1", b"-1", b"1.00"],
|
|
),
|
|
],
|
|
)
|
|
def test_find_reference(leyex, axis_nr, socket_put_messages, socket_get_messages):
|
|
leyex.controller.sock.flush_buffer()
|
|
leyex.controller.sock.buffer_recv = copy.deepcopy(socket_get_messages)
|
|
try:
|
|
leyex.controller.find_reference(axis_nr)
|
|
except Exception as e:
|
|
print(e)
|
|
assert leyex.controller.sock.buffer_put == socket_put_messages
|
|
|
|
|
|
def test_wait_for_connection_called():
|
|
"""Test that wait_for_connection is called on all motors that have a socket controller."""
|
|
dm = DMMock()
|
|
testable_connections = [
|
|
(NPointAxis, NPointController),
|
|
(FlomniGalilMotor, FlomniGalilController),
|
|
(FuprGalilMotor, FuprGalilController),
|
|
(LamniGalilMotor, LamniGalilController),
|
|
(OMNYGalilMotor, OMNYGalilController),
|
|
(SGalilMotor, GalilController),
|
|
(RtFlomniMotor, RtFlomniController),
|
|
(RtLamniMotor, RtLamniController),
|
|
(RtOMNYMotor, RtOMNYController),
|
|
(SmaractMotor, SmaractController),
|
|
]
|
|
for motor_cls, controller_cls in testable_connections:
|
|
# Store values to restore later
|
|
ctrl_axis_backup = controller_cls._axes_per_controller
|
|
try:
|
|
controller_cls._reset_controller()
|
|
controller_cls._axes_per_controller = 3
|
|
|
|
motor = motor_cls(
|
|
"C",
|
|
name="test_motor",
|
|
host="mpc2680.psi.ch",
|
|
port=8081,
|
|
socket_cls=SocketMock,
|
|
device_manager=dm,
|
|
)
|
|
with mock.patch.object(motor.controller, "on") as mock_on:
|
|
|
|
motor.wait_for_connection(timeout=5.0)
|
|
assert mock_on.call_args_list[-1] == mock.call(timeout=5.0)
|
|
|
|
# Make sure destroy calls controller off
|
|
|
|
with mock.patch.object(motor.controller, "off") as mock_off:
|
|
motor.destroy()
|
|
assert mock_off.call_count == 1
|
|
assert mock_off.call_args_list[0] == mock.call(update_config=False)
|
|
assert motor._destroyed is True
|
|
|
|
finally:
|
|
controller_cls._reset_controller()
|
|
controller_cls._axes_per_controller = ctrl_axis_backup
|