test: add tests for panda, fix tests for fermat scan

tests/tests_devices/test_panda.py

@@ -0,0 +1,190 @@
+"""Module for testing the PandaBoxCSAXS and PandaBoxOMNY devices."""
+
+# pylint: skip-file
+from __future__ import annotations
+
+from unittest import mock
+
+import pytest
+from ophyd import Staged
+
+from csaxs_bec.devices.panda_box.panda_box import PandaBoxCSAXS
+from csaxs_bec.devices.panda_box.panda_box_omny import PandaBoxOMNY
+
+
+@pytest.fixture
+def panda_omny():
+    dev_name = "panda_omny"
+    dev = PandaBoxOMNY(
+        name=dev_name,
+        host="omny-panda-box.psi.ch",
+        signal_alias={
+            "FMC_IN.VAL1.Min": "cap_voltage_fzp_y_min",
+            "FMC_IN.VAL1.Max": "cap_voltage_fzp_y_max",
+            "FMC_IN.VAL1.Mean": "cap_voltage_fzp_y_mean",
+            "FMC_IN.VAL2.Min": "cap_voltage_fzp_x_min",
+            "FMC_IN.VAL2.Max": "cap_voltage_fzp_x_max",
+            "FMC_IN.VAL2.Mean": "cap_voltage_fzp_x_mean",
+        },
+    )
+    yield dev
+
+
+@pytest.fixture
+def panda_csaxs():
+    dev_name = "panda_csaxs"
+    dev = PandaBoxCSAXS(name=dev_name, host="csaxs-panda-box.psi.ch")
+    yield dev
+
+
+def test_panda_omny(panda_omny):
+    assert panda_omny.name == "panda_omny"
+    assert panda_omny.host == "omny-panda-box.psi.ch"
+
+    all_signal_names = [name for name, _ in panda_omny.data.signals]
+    # Check that the signal aliases are correctly set up
+    assert "cap_voltage_fzp_y_min" in all_signal_names
+    assert "cap_voltage_fzp_y_max" in all_signal_names
+    assert "cap_voltage_fzp_y_mean" in all_signal_names
+    assert "cap_voltage_fzp_x_min" in all_signal_names
+    assert "cap_voltage_fzp_x_max" in all_signal_names
+    assert "cap_voltage_fzp_x_mean" in all_signal_names
+
+    # Check that the original signal names are not present
+    assert "FMC_IN.VAL1.Min" not in all_signal_names
+    assert "FMC_IN.VAL1.Max" not in all_signal_names
+    assert "FMC_IN.VAL1.Mean" not in all_signal_names
+    assert "FMC_IN.VAL2.Min" not in all_signal_names
+    assert "FMC_IN.VAL2.Max" not in all_signal_names
+    assert "FMC_IN.VAL2.Mean" not in all_signal_names
+
+    assert panda_omny._acquisition_group == "burst"
+    assert panda_omny._timeout_on_completed == 10
+
+
+@pytest.mark.parametrize(
+    "scan_type, frames_per_trigger, expected_acquisition_group",
+    [
+        ("fly", 1, "fly"),
+        ("fly", 5, "fly"),
+        ("step", 10, "burst"),
+        ("step", 1, "monitored"),  # Default case
+    ],
+)
+def test_panda_omny_stage(panda_omny, scan_type, frames_per_trigger, expected_acquisition_group):
+    # Check that the stage signal is present and has the correct PV
+    assert len(panda_omny._status_callbacks) == 0
+
+    panda_omny.scan_info.msg.scan_type = scan_type
+    panda_omny.scan_info.msg.scan_parameters["frames_per_trigger"] = frames_per_trigger
+    panda_omny.stage()
+
+    assert panda_omny._acquisition_group == expected_acquisition_group
+    assert panda_omny.staged == Staged.yes
+
+
+def test_panda_omny_complete(panda_omny):
+    """Test the on_complete method of the PandaBoxCSAXS device."""
+    panda_omny.scan_info.msg.num_points = 1
+    panda_omny.scan_info.msg.scan_parameters["frames_per_trigger"] = 1
+
+    panda_omny._timeout_on_completed = 0.5  # Set a short timeout for testing
+
+    def _mock_return_captured(*args, **kwargs):
+        return ["=0"]
+
+    # Timeout Error on complete
+    with (
+        mock.patch.object(panda_omny, "send_raw", side_effect=_mock_return_captured),
+        mock.patch.object(panda_omny, "_disarm", return_value=None) as mock_disarm,
+    ):
+        status = panda_omny.on_complete()
+        assert status.done is False
+        assert status.success is False
+
+        with pytest.raises(TimeoutError):
+            status.wait(timeout=4)
+        mock_disarm.assert_called_once()
+
+    # Successful complete
+    panda_omny._timeout_on_completed = 5
+    with (
+        mock.patch.object(panda_omny, "send_raw", side_effect=[["=0"], ["=0"], ["=1"]]),
+        mock.patch.object(panda_omny, "_disarm", return_value=None) as mock_disarm,
+    ):
+        status = panda_omny.on_complete()
+        assert status.done is False
+        assert status.success is False
+
+        status.wait(timeout=4)
+        mock_disarm.assert_called_once()
+        assert status.done is True
+        assert status.success is True
+
+
+def test_panda_csaxs(panda_csaxs):
+    assert panda_csaxs.name == "panda_csaxs"
+    assert panda_csaxs.host == "csaxs-panda-box.psi.ch"
+
+    assert panda_csaxs._acquisition_group == "burst"
+    assert panda_csaxs._timeout_on_completed == 10
+
+
+@pytest.mark.parametrize(
+    "scan_type, frames_per_trigger, expected_acquisition_group",
+    [
+        ("fly", 1, "fly"),
+        ("fly", 5, "fly"),
+        ("step", 10, "burst"),
+        ("step", 1, "monitored"),  # Default case
+    ],
+)
+def test_panda_csaxs_stage(panda_csaxs, scan_type, frames_per_trigger, expected_acquisition_group):
+    """Test the on_stage method of the PandaBoxCSAXS device for different scan types and frames per trigger."""
+    assert len(panda_csaxs._status_callbacks) == 0
+
+    panda_csaxs.scan_info.msg.scan_type = scan_type
+    panda_csaxs.scan_info.msg.scan_parameters["frames_per_trigger"] = frames_per_trigger
+    panda_csaxs.stage()
+
+    assert panda_csaxs._acquisition_group == expected_acquisition_group
+    assert panda_csaxs.staged == Staged.yes
+
+
+def test_panda_csaxs_complete(panda_csaxs):
+    """Test the on_complete method of the PandaBoxCSAXS device."""
+    panda_csaxs.scan_info.msg.num_points = 1
+    panda_csaxs.scan_info.msg.scan_parameters["frames_per_trigger"] = 1
+
+    panda_csaxs._timeout_on_completed = 0.5  # Set a short timeout for testing
+
+    def _mock_return_captured(*args, **kwargs):
+        return ["=0"]
+
+    # Timeout Error on complete
+    with (
+        mock.patch.object(panda_csaxs, "send_raw", side_effect=_mock_return_captured),
+        mock.patch.object(panda_csaxs, "_disarm", return_value=None) as mock_disarm,
+    ):
+        status = panda_csaxs.on_complete()
+        assert status.done is False
+        assert status.success is False
+
+        with pytest.raises(TimeoutError):
+            status.wait(timeout=4)
+        mock_disarm.assert_called_once()
+
+    # Successful complete
+    panda_csaxs._timeout_on_completed = 5
+    with (
+        mock.patch.object(panda_csaxs, "send_raw", side_effect=[["=0"], ["=0"], ["=1"]]),
+        mock.patch.object(panda_csaxs, "_disarm", return_value=None) as mock_disarm,
+    ):
+        status = panda_csaxs.on_complete()
+        assert status.done is False
+        assert status.success is False
+
+        status.wait(timeout=4)
+        mock_disarm.assert_called_once()
+        assert status.done is True
+        assert status.success is True

tests/tests_scans/test_lamni_fermat_scan.py

@@ -302,6 +302,36 @@ def device_manager_mock():
                     action="set",
                     parameter={"value": 2.1508313829565293},
                 ),
+                messages.DeviceInstructionMessage(
+                    metadata={
+                        "readout_priority": "monitored",
+                        "RID": "1234",
+                        "device_instr_id": "diid",
+                    },
+                    device=["bpm4i", "lsamx", "lsamy", "samx", "samy"],
+                    action="pre_scan",
+                    parameter={},
+                ),
+                messages.DeviceInstructionMessage(
+                    metadata={
+                        "readout_priority": "monitored",
+                        "RID": "1234",
+                        "device_instr_id": "diid",
+                    },
+                    device="rtx",
+                    action="set",
+                    parameter={"value": 1.3681828686580249},
+                ),
+                messages.DeviceInstructionMessage(
+                    metadata={
+                        "readout_priority": "monitored",
+                        "RID": "1234",
+                        "device_instr_id": "diid",
+                    },
+                    device="rty",
+                    action="set",
+                    parameter={"value": 2.1508313829565293},
+                ),
                 None,
                 messages.DeviceInstructionMessage(
                     metadata={