191 lines
6.4 KiB
Python
191 lines
6.4 KiB
Python
"""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
|