225 lines
7.1 KiB
Python
225 lines
7.1 KiB
Python
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
from csaxs_bec.devices.npoint import NPointAxis, NPointController
|
|
from csaxs_bec.devices.npoint.npoint import NpointError
|
|
|
|
# pylint: disable=protected-access
|
|
# pylint: disable=redefined-outer-name
|
|
|
|
|
|
@pytest.fixture
|
|
def controller():
|
|
"""
|
|
Fixture to create a NPointController object.
|
|
"""
|
|
with mock.patch("ophyd_devices.utils.socket.SocketIO") as socket_cls:
|
|
controller = NPointController(
|
|
socket_cls=socket_cls,
|
|
socket_host="localhost",
|
|
socket_port=1234,
|
|
device_manager=mock.MagicMock(),
|
|
)
|
|
controller.on()
|
|
controller.sock.reset_mock()
|
|
yield controller
|
|
controller.off()
|
|
|
|
|
|
@pytest.fixture
|
|
def npointx(dm_with_devices):
|
|
"""
|
|
Fixture to create a NPointAxis object.
|
|
"""
|
|
controller = mock.MagicMock()
|
|
npointx = NPointAxis(
|
|
axis_Id="A",
|
|
name="npointx",
|
|
host="localhost",
|
|
port=1234,
|
|
socket_cls=controller,
|
|
device_manager=dm_with_devices,
|
|
)
|
|
npointx.controller.on()
|
|
npointx.controller.sock.reset_mock()
|
|
npointx.controller.sock.receive.reset_mock()
|
|
yield npointx
|
|
npointx.controller.off()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"pos,msg",
|
|
[
|
|
(5, b"\xa2\x18\x12\x83\x11\xcd\xcc\x00\x00U"),
|
|
(0, b"\xa2\x18\x12\x83\x11\x00\x00\x00\x00U"),
|
|
(-5, b"\xa2\x18\x12\x83\x1133\xff\xffU"),
|
|
],
|
|
)
|
|
def test_axis_put(npointx, pos, msg):
|
|
"""
|
|
Test that the set target position sends the correct message to the controller.
|
|
"""
|
|
npointx.controller.set_target_pos(npointx.axis_Id_numeric, pos)
|
|
npointx.controller.sock.put.assert_called_with(msg)
|
|
|
|
|
|
def test_npoint_axis_move(npointx):
|
|
"""
|
|
Test that the move method sends the correct messages to the controller.
|
|
It should send the set target position, followed by 2 get current position messages.
|
|
"""
|
|
npointx.controller.sock.receive.side_effect = [
|
|
b"\xa0\x34\x13\x83\x11\x00\x00\x00\x00U", # pos 0
|
|
b"\xa0\x34\x13\x83\x11\xcd\xcc\x00\x00U", # pos 5
|
|
]
|
|
npointx.move(5)
|
|
assert (
|
|
mock.call(b"\xa2\x18\x12\x83\x11\xcd\xcc\x00\x00U")
|
|
in npointx.controller.sock.put.mock_calls
|
|
)
|
|
assert len(npointx.controller.sock.put.mock_calls) == 3
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"pos, msg_in, msg_out",
|
|
[
|
|
(5.0, b"\xa04\x13\x83\x11U", b"\xa0\x34\x13\x83\x11\xcd\xcc\x00\x00U"),
|
|
(0, b"\xa04\x13\x83\x11U", b"\xa0\x34\x13\x83\x11\x00\x00\x00\x00U"),
|
|
(-5, b"\xa04\x13\x83\x11U", b"\xa0\x34\x13\x83\x1133\xff\xffU"),
|
|
],
|
|
)
|
|
def test_axis_get_out(npointx, pos, msg_in, msg_out):
|
|
"""
|
|
Test that the readback value is correctly read from the controller.
|
|
"""
|
|
npointx.controller.sock.receive.side_effect = [msg_out]
|
|
assert pytest.approx(npointx.readback.get(), rel=0.01) == pos
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"axis, msg_in, msg_out",
|
|
[
|
|
(0, b"\xa04\x13\x83\x11U", b"\xa0\x34\x13\x83\x11\xcd\xcc\x00\x00U"),
|
|
(1, b"\xa04#\x83\x11U", b"\xa0\x34\x23\x83\x11\x00\x00\x00\x00U"),
|
|
(2, b"\xa043\x83\x11U", b"\xa0\x34\x33\x83\x1133\xff\xffU"),
|
|
],
|
|
)
|
|
def test_axis_get_in(npointx, axis, msg_in, msg_out):
|
|
"""
|
|
Test that the readback value is correctly read from the controller by directly calling the
|
|
controller's method.
|
|
"""
|
|
npointx.controller.sock.receive.side_effect = [msg_out]
|
|
npointx.controller.get_current_pos(axis)
|
|
npointx.controller.sock.put.assert_called_once_with(msg_in)
|
|
|
|
|
|
def test_axis_out_of_range(dm_with_devices):
|
|
"""
|
|
Test that an error is raised when trying to create an NPointAxis object with an invalid axis ID.
|
|
"""
|
|
with pytest.raises(ValueError):
|
|
npointx = NPointAxis(
|
|
axis_Id="G",
|
|
name="npointx",
|
|
host="localhost",
|
|
port=1234,
|
|
socket_cls=mock.MagicMock(),
|
|
device_manager=dm_with_devices,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("channel", [-1, 3])
|
|
def test_get_axis_out_of_range(controller, channel):
|
|
"""
|
|
Test that an error is raised when trying to get the current position of an invalid axis.
|
|
"""
|
|
with pytest.raises(ValueError):
|
|
controller.get_current_pos(channel=channel)
|
|
controller.sock.put.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("channel", [-1, 3])
|
|
def test_set_axis_out_of_range(controller, channel):
|
|
"""
|
|
Test that an error is raised when trying to set the target position of an invalid axis.
|
|
"""
|
|
with pytest.raises(ValueError):
|
|
controller.set_target_pos(channel=channel, pos=5)
|
|
controller.sock.put.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"axis, msg_in, msg_out",
|
|
[
|
|
(0, b"\xa0x\x10\x83\x11U", b"\xa0\x78\x10\x83\x11\x64\x00\x00\x00U"),
|
|
(1, b"\xa0x \x83\x11U", b"\xa0\x78\x20\x83\x11\x64\x00\x00\x00U"),
|
|
(2, b"\xa0x0\x83\x11U", b"\xa0\x78\x30\x83\x11\x64\x00\x00\x00U"),
|
|
],
|
|
)
|
|
def test_get_range(npointx, axis, msg_in, msg_out):
|
|
"""
|
|
Test that the range is correctly read from the controller by directly calling the
|
|
controller's method.
|
|
"""
|
|
npointx.controller.sock.receive.side_effect = [msg_out]
|
|
val = npointx.controller._get_range(axis)
|
|
npointx.controller.sock.put.assert_called_once_with(msg_in)
|
|
assert val == 100
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"response",
|
|
[
|
|
b"\xa2\x34\x13\x83\x11\x00\x00\x00\x00U", # Wrong opcode
|
|
b"\xa0\x34\x23\x83\x11\x00\x00\x00\x00U", # Wrong channel
|
|
b"\xa0\x34\x13\x82\x11\x00\x00\x00\x00U", # Wrong address
|
|
b"\xa0\x34\x13\x83\x11\x00\x00\x00\x00X", # Wrong end marker
|
|
],
|
|
)
|
|
def test_read_rejects_invalid_frame(controller, response):
|
|
controller.sock.receive.side_effect = [response]
|
|
with pytest.raises(NpointError, match="Invalid response"):
|
|
controller.get_current_pos(0)
|
|
|
|
|
|
def test_read_fragmented_payload_with_end_marker(controller):
|
|
controller.sock.receive.side_effect = [b"\xa0\x34\x13\x83\x11U", b"\x00\x00", b"\x00U"]
|
|
assert controller.get_current_pos(0) == pytest.approx(85 / 1048574 * 100)
|
|
assert [call.kwargs["buffer_length"] for call in controller.sock.receive.call_args_list] == [
|
|
10,
|
|
4,
|
|
2,
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"method,command,response,expected",
|
|
[
|
|
(
|
|
"get_target_pos",
|
|
b"\xa0\x18\x22\x83\x11U",
|
|
b"\xa0\x18\x22\x83\x1133\xff\xffU",
|
|
-52429 / 1048574 * 100,
|
|
),
|
|
("_get_servo", b"\xa0\x84\x20\x83\x11U", b"\xa0\x84\x20\x83\x11\x01\x00\x00\x00U", 1),
|
|
(
|
|
"_get_range",
|
|
b"\xa0\x78\x20\x83\x11U",
|
|
b"\xa0\x78\x20\x83\x11\xff\xff\xff\xffU",
|
|
4294967295,
|
|
),
|
|
],
|
|
)
|
|
def test_register_read_types(controller, method, command, response, expected):
|
|
controller.sock.receive.side_effect = [response]
|
|
assert getattr(controller, method)(channel=1) == pytest.approx(expected)
|
|
controller.sock.put.assert_called_once_with(command)
|
|
|
|
|
|
def test_target_write_channel_offset(controller):
|
|
controller.set_target_pos(channel=2, pos=-5)
|
|
controller.sock.put.assert_called_once_with(b"\xa2\x18\x32\x83\x1133\xff\xffU")
|
|
controller.sock.receive.assert_not_called()
|