From 0acd12607a6aba673b650a33db6a712cac94dc4e Mon Sep 17 00:00:00 2001 From: appel_c Date: Mon, 2 Mar 2026 09:20:52 +0100 Subject: [PATCH] test: add tests for panda, fix tests for fermat scan --- tests/tests_devices/test_panda.py | 190 ++++++++++++++++++++ tests/tests_scans/test_lamni_fermat_scan.py | 30 ++++ 2 files changed, 220 insertions(+) create mode 100644 tests/tests_devices/test_panda.py diff --git a/tests/tests_devices/test_panda.py b/tests/tests_devices/test_panda.py new file mode 100644 index 0000000..02bca15 --- /dev/null +++ b/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 diff --git a/tests/tests_scans/test_lamni_fermat_scan.py b/tests/tests_scans/test_lamni_fermat_scan.py index 2f94ede..564233e 100644 --- a/tests/tests_scans/test_lamni_fermat_scan.py +++ b/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={