test(rt_flomni): added tests

ophyd_devices/rt_lamni/rt_flomni_ophyd.py

@@ -1,4 +1,3 @@
-import functools
 import threading
 import time
 from typing import List
@@ -8,7 +7,7 @@ from bec_lib import MessageEndpoints, bec_logger, messages
 from ophyd import Component as Cpt
 from ophyd import Device, PositionerBase, Signal
 from ophyd.status import wait as status_wait
-from ophyd.utils import LimitError, ReadOnlyError
+from ophyd.utils import LimitError
 from prettytable import PrettyTable
 
 from ophyd_devices.rt_lamni.rt_ophyd import (
@@ -21,8 +20,8 @@ from ophyd_devices.rt_lamni.rt_ophyd import (
     RtSignalRO,
     retry_once,
 )
-from ophyd_devices.utils.controller import Controller, threadlocked
+from ophyd_devices.utils.controller import threadlocked
-from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected
+from ophyd_devices.utils.socket import SocketIO, raise_if_disconnected
 
 logger = bec_logger.logger
 

tests/test_rt_flomni.py

@@ -0,0 +1,58 @@
+from unittest import mock
+
+import pytest
+from utils import SocketMock
+
+from ophyd_devices.rt_lamni import RtFlomniController, RtFlomniMotor
+
+
+@pytest.fixture()
+def rt_flomni():
+    rt_flomni = RtFlomniController(
+        name="rt_flomni", socket_cls=SocketMock, socket_host="localhost", socket_port=8081
+    )
+    with mock.patch.object(rt_flomni, "get_device_manager"):
+        with mock.patch.object(rt_flomni, "sock"):
+            rtx = mock.MagicMock(spec=RtFlomniMotor)
+            rtx.name = "rtx"
+            rty = mock.MagicMock(spec=RtFlomniMotor)
+            rty.name = "rty"
+            rtz = mock.MagicMock(spec=RtFlomniMotor)
+            rtz.name = "rtz"
+            rt_flomni.set_axis(rtx, 0)
+            rt_flomni.set_axis(rty, 1)
+            rt_flomni.set_axis(rtz, 2)
+            yield rt_flomni
+
+
+def test_rt_flomni_move_to_zero(rt_flomni):
+    rt_flomni.move_to_zero()
+    assert rt_flomni.sock.mock_calls == [
+        mock.call.put(b"pa0,0\n"),
+        mock.call.put(b"pa1,0\n"),
+        mock.call.put(b"pa2,0\n"),
+    ]
+
+
+@pytest.mark.parametrize("return_value,is_running", [(b"1.00\n", False), (b"0.00\n", True)])
+def test_rt_flomni_feedback_is_running(rt_flomni, return_value, is_running):
+    rt_flomni.sock.receive.return_value = return_value
+    assert rt_flomni.feedback_is_running() == is_running
+    assert mock.call.put(b"l2\n") in rt_flomni.sock.mock_calls
+
+
+def test_feedback_enable_with_reset(rt_flomni):
+
+    device_manager = rt_flomni.get_device_manager()
+    device_manager.devices.fsamx.user_parameter.get.return_value = 0.05
+    device_manager.devices.fsamx.obj.readback.get.return_value = 0.05
+
+    with mock.patch.object(rt_flomni, "feedback_is_running", return_value=True):
+        with mock.patch.object(rt_flomni, "laser_tracker_on") as laser_tracker_on:
+            with mock.patch.object(rt_flomni, "pid_y", return_value=0.05):
+                with mock.patch.object(
+                    rt_flomni, "slew_rate_limiters_on_target", return_value=True
+                ) as slew_rate_limiters_on_target:
+
+                    rt_flomni.feedback_enable_with_reset()
+                    laser_tracker_on.assert_called_once()