bec/ophyd_devices
0
0
Fork 0
mirror of https://github.com/bec-project/ophyd_devices.git synced 2025-06-23 11:27:57 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat: move csaxs devices to plugin structure, fix imports and tests

Browse Source
This commit is contained in:
appel_c
2024-04-19 12:03:30 +02:00
parent f5366e453f
commit 74f6fa7ffd
54 changed files with 19 additions and 13441 deletions
Download Patch File Download Diff File Expand all files Collapse all files

298
tests/test_delay_generator_csaxs.py
View File

@ -1,298 +0,0 @@
# pylint: skip-file
from unittest import mock
import pytest
from ophyd_devices.epics.devices.delay_generator_csaxs import DDGSetup
from ophyd_devices.epics.devices.psi_delay_generator_base import TriggerSource
def patch_dual_pvs(device):
for walk in device.walk_signals():
if not hasattr(walk.item, "_read_pv"):
continue
if not hasattr(walk.item, "_write_pv"):
continue
if walk.item._read_pv.pvname.endswith("_RBV"):
walk.item._read_pv = walk.item._write_pv
@pytest.fixture(scope="function")
def mock_DDGSetup():
mock_ddg = mock.MagicMock()
yield DDGSetup(parent=mock_ddg)
# Fixture for scaninfo
@pytest.fixture(
params=[
{
"scan_id": "1234",
"scan_type": "step",
"num_points": 500,
"frames_per_trigger": 1,
"exp_time": 0.1,
"readout_time": 0.1,
},
{
"scan_id": "1234",
"scan_type": "step",
"num_points": 500,
"frames_per_trigger": 5,
"exp_time": 0.01,
"readout_time": 0,
},
{
"scan_id": "1234",
"scan_type": "fly",
"num_points": 500,
"frames_per_trigger": 1,
"exp_time": 1,
"readout_time": 0.2,
},
{
"scan_id": "1234",
"scan_type": "fly",
"num_points": 500,
"frames_per_trigger": 5,
"exp_time": 0.1,
"readout_time": 0.4,
},
]
)
def scaninfo(request):
return request.param
# Fixture for DDG config default values
@pytest.fixture(
params=[
{
"delay_burst": 0.0,
"delta_width": 0.0,
"additional_triggers": 0,
"polarity": [0, 0, 0, 0, 0],
"amplitude": 0.0,
"offset": 0.0,
"thres_trig_level": 0.0,
},
{
"delay_burst": 0.1,
"delta_width": 0.1,
"additional_triggers": 1,
"polarity": [0, 0, 1, 0, 0],
"amplitude": 5,
"offset": 0.0,
"thres_trig_level": 2.5,
},
]
)
def ddg_config_defaults(request):
return request.param
# Fixture for DDG config scan values
@pytest.fixture(
params=[
{
"fixed_ttl_width": [0, 0, 0, 0, 0],
"trigger_width": None,
"set_high_on_exposure": False,
"set_high_on_stage": False,
"set_trigger_source": "SINGLE_SHOT",
"premove_trigger": False,
},
{
"fixed_ttl_width": [0, 0, 0, 0, 0],
"trigger_width": 0.1,
"set_high_on_exposure": True,
"set_high_on_stage": False,
"set_trigger_source": "SINGLE_SHOT",
"premove_trigger": True,
},
{
"fixed_ttl_width": [0, 0, 0, 0, 0],
"trigger_width": 0.1,
"set_high_on_exposure": False,
"set_high_on_stage": False,
"set_trigger_source": "EXT_RISING_EDGE",
"premove_trigger": False,
},
]
)
def ddg_config_scan(request):
return request.param
# Fixture for delay pairs
@pytest.fixture(
params=[
{"all_channels": ["channelAB", "channelCD"], "all_delay_pairs": ["AB", "CD"]},
{"all_channels": [], "all_delay_pairs": []},
{"all_channels": ["channelT0", "channelAB", "channelCD"], "all_delay_pairs": ["AB", "CD"]},
]
)
def channel_pairs(request):
return request.param
def test_check_scan_id(mock_DDGSetup, scaninfo, ddg_config_defaults, ddg_config_scan):
"""Test the check_scan_id method."""
# Set first attributes of parent class
for k, v in scaninfo.items():
setattr(mock_DDGSetup.parent.scaninfo, k, v)
for k, v in ddg_config_defaults.items():
getattr(mock_DDGSetup.parent, k).get.return_value = v
for k, v in ddg_config_scan.items():
getattr(mock_DDGSetup.parent, k).get.return_value = v
# Call the function you want to test
mock_DDGSetup.check_scan_id()
mock_DDGSetup.parent.scaninfo.load_scan_metadata.assert_called_once()
def test_on_pre_scan(mock_DDGSetup, scaninfo, ddg_config_defaults, ddg_config_scan):
"""Test the check_scan_id method."""
# Set first attributes of parent class
for k, v in scaninfo.items():
setattr(mock_DDGSetup.parent.scaninfo, k, v)
for k, v in ddg_config_defaults.items():
getattr(mock_DDGSetup.parent, k).get.return_value = v
for k, v in ddg_config_scan.items():
getattr(mock_DDGSetup.parent, k).get.return_value = v
# Call the function you want to test
mock_DDGSetup.on_pre_scan()
if ddg_config_scan["premove_trigger"]:
mock_DDGSetup.parent.trigger_shot.put.assert_called_once_with(1)
@pytest.mark.parametrize("source", ["SINGLE_SHOT", "EXT_RISING_EDGE"])
def test_on_trigger(mock_DDGSetup, scaninfo, ddg_config_defaults, ddg_config_scan, source):
"""Test the on_trigger method."""
# Set first attributes of parent class
for k, v in scaninfo.items():
setattr(mock_DDGSetup.parent.scaninfo, k, v)
for k, v in ddg_config_defaults.items():
getattr(mock_DDGSetup.parent, k).get.return_value = v
for k, v in ddg_config_scan.items():
getattr(mock_DDGSetup.parent, k).get.return_value = v
# Call the function you want to test
mock_DDGSetup.parent.source.name = "source"
mock_DDGSetup.parent.source.read.return_value = {
mock_DDGSetup.parent.source.name: {"value": getattr(TriggerSource, source)}
}
mock_DDGSetup.on_trigger()
if source == "SINGLE_SHOT":
mock_DDGSetup.parent.trigger_shot.put.assert_called_once_with(1)
def test_initialize_default_parameter(
mock_DDGSetup, scaninfo, ddg_config_defaults, ddg_config_scan, channel_pairs
):
"""Test the initialize_default_parameter method."""
# Set first attributes of parent class
for k, v in scaninfo.items():
setattr(mock_DDGSetup.parent.scaninfo, k, v)
for k, v in ddg_config_defaults.items():
getattr(mock_DDGSetup.parent, k).get.return_value = v
for k, v in ddg_config_scan.items():
getattr(mock_DDGSetup.parent, k).get.return_value = v
# Call the function you want to test
mock_DDGSetup.parent.all_channels = channel_pairs["all_channels"]
mock_DDGSetup.parent.all_delay_pairs = channel_pairs["all_delay_pairs"]
calls = []
calls.extend(
[
mock.call("polarity", ddg_config_defaults["polarity"][ii], [channel])
for ii, channel in enumerate(channel_pairs["all_channels"])
]
)
calls.extend([mock.call("amplitude", ddg_config_defaults["amplitude"])])
calls.extend([mock.call("offset", ddg_config_defaults["offset"])])
calls.extend(
[
mock.call(
"reference", 0, [f"channel{pair}.ch1" for pair in channel_pairs["all_delay_pairs"]]
)
]
)
calls.extend(
[
mock.call(
"reference", 0, [f"channel{pair}.ch2" for pair in channel_pairs["all_delay_pairs"]]
)
]
)
mock_DDGSetup.initialize_default_parameter()
mock_DDGSetup.parent.set_channels.assert_has_calls(calls)
def test_prepare_ddg(mock_DDGSetup, scaninfo, ddg_config_defaults, ddg_config_scan, channel_pairs):
"""Test the prepare_ddg method."""
# Set first attributes of parent class
for k, v in scaninfo.items():
setattr(mock_DDGSetup.parent.scaninfo, k, v)
for k, v in ddg_config_defaults.items():
getattr(mock_DDGSetup.parent, k).get.return_value = v
for k, v in ddg_config_scan.items():
getattr(mock_DDGSetup.parent, k).get.return_value = v
# Call the function you want to test
mock_DDGSetup.parent.all_channels = channel_pairs["all_channels"]
mock_DDGSetup.parent.all_delay_pairs = channel_pairs["all_delay_pairs"]
mock_DDGSetup.prepare_ddg()
mock_DDGSetup.parent.set_trigger.assert_called_once_with(
getattr(TriggerSource, ddg_config_scan["set_trigger_source"])
)
if scaninfo["scan_type"] == "step":
if ddg_config_scan["set_high_on_exposure"]:
num_burst_cycle = 1 + ddg_config_defaults["additional_triggers"]
exp_time = ddg_config_defaults["delta_width"] + scaninfo["frames_per_trigger"] * (
scaninfo["exp_time"] + scaninfo["readout_time"]
)
total_exposure = exp_time
delay_burst = ddg_config_defaults["delay_burst"]
else:
exp_time = ddg_config_defaults["delta_width"] + scaninfo["exp_time"]
total_exposure = exp_time + scaninfo["readout_time"]
delay_burst = ddg_config_defaults["delay_burst"]
num_burst_cycle = (
scaninfo["frames_per_trigger"] + ddg_config_defaults["additional_triggers"]
)
elif scaninfo["scan_type"] == "fly":
if ddg_config_scan["set_high_on_exposure"]:
num_burst_cycle = 1 + ddg_config_defaults["additional_triggers"]
exp_time = (
ddg_config_defaults["delta_width"]
+ scaninfo["num_points"] * scaninfo["exp_time"]
+ (scaninfo["num_points"] - 1) * scaninfo["readout_time"]
)
total_exposure = exp_time
delay_burst = ddg_config_defaults["delay_burst"]
else:
exp_time = ddg_config_defaults["delta_width"] + scaninfo["exp_time"]
total_exposure = exp_time + scaninfo["readout_time"]
delay_burst = ddg_config_defaults["delay_burst"]
num_burst_cycle = scaninfo["num_points"] + ddg_config_defaults["additional_triggers"]
# mock_DDGSetup.parent.burst_enable.assert_called_once_with(
# mock.call(num_burst_cycle, delay_burst, total_exposure, config="first")
# )
mock_DDGSetup.parent.burst_enable.assert_called_once_with(
num_burst_cycle, delay_burst, total_exposure, config="first"
)
if not ddg_config_scan["trigger_width"]:
call = mock.call("width", exp_time)
assert call in mock_DDGSetup.parent.set_channels.mock_calls
else:
call = mock.call("width", ddg_config_scan["trigger_width"])
assert call in mock_DDGSetup.parent.set_channels.mock_calls
if ddg_config_scan["set_high_on_exposure"]:
calls = [
mock.call("width", value, channels=[channel])
for value, channel in zip(
ddg_config_scan["fixed_ttl_width"], channel_pairs["all_channels"]
)
if value != 0
]
if calls:
assert all(calls in mock_DDGSetup.parent.set_channels.mock_calls)

2
tests/test_dynamic_pseudo.py
View File

@ -1,9 +1,9 @@
from unittest import mock
import pytest
from bec_server.device_server.tests.utils import DMMock
from ophyd_devices.utils.dynamic_pseudo import ComputedSignal
from tests.utils import DMMock
@pytest.fixture

455
tests/test_eiger9m_csaxs.py
View File

@ -1,455 +0,0 @@
# pylint: skip-file
import threading
from unittest import mock
import ophyd
import pytest
from bec_lib import MessageEndpoints, messages
from ophyd_devices.epics.devices.eiger9m_csaxs import Eiger9McSAXS
from tests.utils import DMMock, MockPV
def patch_dual_pvs(device):
for walk in device.walk_signals():
if not hasattr(walk.item, "_read_pv"):
continue
if not hasattr(walk.item, "_write_pv"):
continue
if walk.item._read_pv.pvname.endswith("_RBV"):
walk.item._read_pv = walk.item._write_pv
@pytest.fixture(scope="function")
def mock_det():
name = "eiger"
prefix = "X12SA-ES-EIGER9M:"
sim_mode = False
dm = DMMock()
with mock.patch.object(dm, "connector"):
with (
mock.patch("ophyd_devices.epics.devices.psi_detector_base.FileWriter"),
mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.PSIDetectorBase._update_service_config"
),
):
with mock.patch.object(ophyd, "cl") as mock_cl:
mock_cl.get_pv = MockPV
mock_cl.thread_class = threading.Thread
with mock.patch.object(Eiger9McSAXS, "_init"):
det = Eiger9McSAXS(
name=name, prefix=prefix, device_manager=dm, sim_mode=sim_mode
)
patch_dual_pvs(det)
yield det
def test_init():
"""Test the _init function:"""
name = "eiger"
prefix = "X12SA-ES-EIGER9M:"
sim_mode = False
dm = DMMock()
with mock.patch.object(dm, "connector"):
with (
mock.patch("ophyd_devices.epics.devices.psi_detector_base.FileWriter"),
mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.PSIDetectorBase._update_service_config"
),
):
with mock.patch.object(ophyd, "cl") as mock_cl:
mock_cl.get_pv = MockPV
with (
mock.patch(
"ophyd_devices.epics.devices.eiger9m_csaxs.Eiger9MSetup.initialize_default_parameter"
) as mock_default,
mock.patch(
"ophyd_devices.epics.devices.eiger9m_csaxs.Eiger9MSetup.initialize_detector"
) as mock_init_det,
mock.patch(
"ophyd_devices.epics.devices.eiger9m_csaxs.Eiger9MSetup.initialize_detector_backend"
) as mock_init_backend,
):
Eiger9McSAXS(name=name, prefix=prefix, device_manager=dm, sim_mode=sim_mode)
mock_default.assert_called_once()
mock_init_det.assert_called_once()
mock_init_backend.assert_called_once()
@pytest.mark.parametrize(
"trigger_source, detector_state, expected_exception", [(2, 1, True), (2, 0, False)]
)
def test_initialize_detector(mock_det, trigger_source, detector_state, expected_exception):
"""Test the _init function:
This includes testing the functions:
- _init_detector
- _stop_det
- _set_trigger
--> Testing the filewriter is done in test_init_filewriter
Validation upon setting the correct PVs
"""
mock_det.cam.detector_state._read_pv.mock_data = detector_state
if expected_exception:
with pytest.raises(Exception):
mock_det.timeout = 0.1
mock_det.custom_prepare.initialize_detector()
else:
mock_det.custom_prepare.initialize_detector() # call the method you want to test
assert mock_det.cam.acquire.get() == 0
assert mock_det.cam.detector_state.get() == detector_state
assert mock_det.cam.trigger_mode.get() == trigger_source
def test_trigger(mock_det):
"""Test the trigger function:
Validate that trigger calls the custom_prepare.on_trigger() function
"""
with mock.patch.object(mock_det.custom_prepare, "on_trigger") as mock_on_trigger:
mock_det.trigger()
mock_on_trigger.assert_called_once()
@pytest.mark.parametrize(
"readout_time, expected_value", [(1e-3, 3e-3), (3e-3, 3e-3), (5e-3, 5e-3), (None, 3e-3)]
)
def test_update_readout_time(mock_det, readout_time, expected_value):
if readout_time is None:
mock_det.custom_prepare.update_readout_time()
assert mock_det.readout_time == expected_value
else:
mock_det.scaninfo.readout_time = readout_time
mock_det.custom_prepare.update_readout_time()
assert mock_det.readout_time == expected_value
@pytest.mark.parametrize(
"eacc, exp_url, daq_status, daq_cfg, expected_exception",
[
("e12345", "http://xbl-daq-29:5000", {"state": "READY"}, {"writer_user_id": 12543}, False),
("e12345", "http://xbl-daq-29:5000", {"state": "READY"}, {"writer_user_id": 15421}, False),
("e12345", "http://xbl-daq-29:5000", {"state": "BUSY"}, {"writer_user_id": 15421}, True),
("e12345", "http://xbl-daq-29:5000", {"state": "READY"}, {"writer_ud": 12345}, True),
],
)
def test_initialize_detector_backend(
mock_det, eacc, exp_url, daq_status, daq_cfg, expected_exception
):
"""Test self.custom_prepare.initialize_detector_backend (std daq in this case)
This includes testing the functions:
- _update_service_config
Validation upon checking set values in mocked std_daq instance
"""
with mock.patch("ophyd_devices.epics.devices.eiger9m_csaxs.StdDaqClient") as mock_std_daq:
instance = mock_std_daq.return_value
instance.stop_writer.return_value = None
instance.get_status.return_value = daq_status
instance.get_config.return_value = daq_cfg
mock_det.scaninfo.username = eacc
# scaninfo.username.return_value = eacc
if expected_exception:
with pytest.raises(Exception):
mock_det.timeout = 0.1
mock_det.custom_prepare.initialize_detector_backend()
else:
mock_det.custom_prepare.initialize_detector_backend()
instance.stop_writer.assert_called_once()
instance.get_status.assert_called()
instance.set_config.assert_called_once_with(daq_cfg)
@pytest.mark.parametrize(
"scaninfo, daq_status, daq_cfg, detector_state, stopped, expected_exception",
[
(
{
"eacc": "e12345",
"num_points": 500,
"frames_per_trigger": 1,
"filepath": "test.h5",
"scan_id": "123",
"mokev": 12.4,
},
{"state": "READY"},
{"writer_user_id": 12543},
5,
False,
False,
),
(
{
"eacc": "e12345",
"num_points": 500,
"frames_per_trigger": 1,
"filepath": "test.h5",
"scan_id": "123",
"mokev": 12.4,
},
{"state": "BUSY"},
{"writer_user_id": 15421},
5,
False,
False,
),
(
{
"eacc": "e12345",
"num_points": 500,
"frames_per_trigger": 1,
"filepath": "test.h5",
"scan_id": "123",
"mokev": 18.4,
},
{"state": "READY"},
{"writer_user_id": 12345},
4,
False,
True,
),
],
)
def test_stage(
mock_det, scaninfo, daq_status, daq_cfg, detector_state, stopped, expected_exception
):
with (
mock.patch.object(mock_det.custom_prepare, "std_client") as mock_std_daq,
mock.patch.object(
mock_det.custom_prepare, "publish_file_location"
) as mock_publish_file_location,
):
mock_std_daq.stop_writer.return_value = None
mock_std_daq.get_status.return_value = daq_status
mock_std_daq.get_config.return_value = daq_cfg
mock_det.scaninfo.num_points = scaninfo["num_points"]
mock_det.scaninfo.frames_per_trigger = scaninfo["frames_per_trigger"]
mock_det.filewriter.compile_full_filename.return_value = scaninfo["filepath"]
# TODO consider putting energy as variable in scaninfo
mock_det.device_manager.add_device("mokev", value=12.4)
mock_det.cam.beam_energy.put(scaninfo["mokev"])
mock_det.stopped = stopped
mock_det.cam.detector_state._read_pv.mock_data = detector_state
with mock.patch.object(mock_det.custom_prepare, "prepare_detector_backend") as mock_prep_fw:
mock_det.filepath = scaninfo["filepath"]
if expected_exception:
with pytest.raises(Exception):
mock_det.timeout = 0.1
mock_det.stage()
else:
mock_det.stage()
mock_prep_fw.assert_called_once()
# Check _prep_det
assert mock_det.cam.num_images.get() == int(
scaninfo["num_points"] * scaninfo["frames_per_trigger"]
)
assert mock_det.cam.num_frames.get() == 1
mock_publish_file_location.assert_called_with(done=False)
assert mock_det.cam.acquire.get() == 1
@pytest.mark.parametrize(
"scaninfo, daq_status, expected_exception",
[
(
{
"eacc": "e12345",
"num_points": 500,
"frames_per_trigger": 1,
"filepath": "test.h5",
"scan_id": "123",
},
{"state": "BUSY", "acquisition": {"state": "WAITING_IMAGES"}},
False,
),
(
{
"eacc": "e12345",
"num_points": 500,
"frames_per_trigger": 1,
"filepath": "test.h5",
"scan_id": "123",
},
{"state": "BUSY", "acquisition": {"state": "WAITING_IMAGES"}},
False,
),
(
{
"eacc": "e12345",
"num_points": 500,
"frames_per_trigger": 1,
"filepath": "test.h5",
"scan_id": "123",
},
{"state": "BUSY", "acquisition": {"state": "ERROR"}},
True,
),
],
)
def test_prepare_detector_backend(mock_det, scaninfo, daq_status, expected_exception):
with (
mock.patch.object(mock_det.custom_prepare, "std_client") as mock_std_daq,
mock.patch.object(mock_det.custom_prepare, "filepath_exists") as mock_file_path_exists,
mock.patch.object(mock_det.custom_prepare, "stop_detector_backend") as mock_stop_backend,
mock.patch.object(mock_det, "scaninfo"),
):
mock_std_daq.start_writer_async.return_value = None
mock_std_daq.get_status.return_value = daq_status
mock_det.filewriter.compile_full_filename.return_value = scaninfo["filepath"]
mock_det.scaninfo.num_points = scaninfo["num_points"]
mock_det.scaninfo.frames_per_trigger = scaninfo["frames_per_trigger"]
if expected_exception:
with pytest.raises(Exception):
mock_det.timeout = 0.1
mock_det.custom_prepare.prepare_data_backend()
mock_file_path_exists.assert_called_once()
assert mock_stop_backend.call_count == 2
else:
mock_det.custom_prepare.prepare_data_backend()
mock_file_path_exists.assert_called_once()
mock_stop_backend.assert_called_once()
daq_writer_call = {
"output_file": scaninfo["filepath"],
"n_images": int(scaninfo["num_points"] * scaninfo["frames_per_trigger"]),
}
mock_std_daq.start_writer_async.assert_called_with(daq_writer_call)
@pytest.mark.parametrize("stopped, expected_exception", [(False, False), (True, True)])
def test_unstage(mock_det, stopped, expected_exception):
with (
mock.patch.object(mock_det.custom_prepare, "finished") as mock_finished,
mock.patch.object(
mock_det.custom_prepare, "publish_file_location"
) as mock_publish_file_location,
):
mock_det.stopped = stopped
if expected_exception:
mock_det.unstage()
assert mock_det.stopped is True
else:
mock_det.unstage()
mock_finished.assert_called_once()
mock_publish_file_location.assert_called_with(done=True, successful=True)
assert mock_det.stopped is False
def test_stop_detector_backend(mock_det):
with mock.patch.object(mock_det.custom_prepare, "std_client") as mock_std_daq:
mock_std_daq.stop_writer.return_value = None
mock_det.std_client = mock_std_daq
mock_det.custom_prepare.stop_detector_backend()
mock_std_daq.stop_writer.assert_called_once()
@pytest.mark.parametrize(
"scaninfo",
[
({"filepath": "test.h5", "successful": True, "done": False, "scan_id": "123"}),
({"filepath": "test.h5", "successful": False, "done": True, "scan_id": "123"}),
({"filepath": "test.h5", "successful": None, "done": True, "scan_id": "123"}),
],
)
def test_publish_file_location(mock_det, scaninfo):
mock_det.scaninfo.scan_id = scaninfo["scan_id"]
mock_det.filepath = scaninfo["filepath"]
mock_det.custom_prepare.publish_file_location(
done=scaninfo["done"], successful=scaninfo["successful"]
)
if scaninfo["successful"] is None:
msg = messages.FileMessage(file_path=scaninfo["filepath"], done=scaninfo["done"])
else:
msg = messages.FileMessage(
file_path=scaninfo["filepath"], done=scaninfo["done"], successful=scaninfo["successful"]
)
expected_calls = [
mock.call(
MessageEndpoints.public_file(scaninfo["scan_id"], mock_det.name),
msg,
pipe=mock_det.connector.pipeline.return_value,
),
mock.call(
MessageEndpoints.file_event(mock_det.name),
msg,
pipe=mock_det.connector.pipeline.return_value,
),
]
assert mock_det.connector.set_and_publish.call_args_list == expected_calls
def test_stop(mock_det):
with (
mock.patch.object(mock_det.custom_prepare, "stop_detector") as mock_stop_det,
mock.patch.object(
mock_det.custom_prepare, "stop_detector_backend"
) as mock_stop_detector_backend,
):
mock_det.stop()
mock_stop_det.assert_called_once()
mock_stop_detector_backend.assert_called_once()
assert mock_det.stopped is True
@pytest.mark.parametrize(
"stopped, scaninfo, cam_state, daq_status, expected_exception",
[
(
False,
{"num_points": 500, "frames_per_trigger": 4},
0,
{"acquisition": {"state": "FINISHED", "stats": {"n_write_completed": 2000}}},
False,
),
(
False,
{"num_points": 500, "frames_per_trigger": 4},
0,
{"acquisition": {"state": "FINISHED", "stats": {"n_write_completed": 1999}}},
True,
),
(
False,
{"num_points": 500, "frames_per_trigger": 1},
1,
{"acquisition": {"state": "READY", "stats": {"n_write_completed": 500}}},
True,
),
(
False,
{"num_points": 500, "frames_per_trigger": 1},
0,
{"acquisition": {"state": "FINISHED", "stats": {"n_write_completed": 500}}},
False,
),
],
)
def test_finished(mock_det, stopped, cam_state, daq_status, scaninfo, expected_exception):
with (
mock.patch.object(mock_det.custom_prepare, "std_client") as mock_std_daq,
mock.patch.object(mock_det.custom_prepare, "stop_detector_backend") as mock_stop_backend,
mock.patch.object(mock_det.custom_prepare, "stop_detector") as mock_stop_det,
):
mock_std_daq.get_status.return_value = daq_status
mock_det.cam.acquire._read_pv.mock_state = cam_state
mock_det.scaninfo.num_points = scaninfo["num_points"]
mock_det.scaninfo.frames_per_trigger = scaninfo["frames_per_trigger"]
if expected_exception:
with pytest.raises(Exception):
mock_det.timeout = 0.1
mock_det.custom_prepare.finished()
assert mock_det.stopped is stopped
else:
mock_det.custom_prepare.finished()
if stopped:
assert mock_det.stopped is stopped
mock_stop_backend.assert_called()
mock_stop_det.assert_called_once()

312
tests/test_falcon_csaxs.py
View File

@ -1,312 +0,0 @@
# pylint: skip-file
import os
import threading
from unittest import mock
import ophyd
import pytest
from bec_lib import MessageEndpoints, messages
from ophyd_devices.epics.devices.falcon_csaxs import FalconcSAXS, FalconTimeoutError
from tests.utils import DMMock, MockPV
def patch_dual_pvs(device):
for walk in device.walk_signals():
if not hasattr(walk.item, "_read_pv"):
continue
if not hasattr(walk.item, "_write_pv"):
continue
if walk.item._read_pv.pvname.endswith("_RBV"):
walk.item._read_pv = walk.item._write_pv
@pytest.fixture(scope="function")
def mock_det():
name = "falcon"
prefix = "X12SA-SITORO:"
sim_mode = False
dm = DMMock()
with mock.patch.object(dm, "connector"):
with (
mock.patch("ophyd_devices.epics.devices.psi_detector_base.FileWriter") as filemixin,
mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.PSIDetectorBase._update_service_config"
) as mock_service_config,
):
with mock.patch.object(ophyd, "cl") as mock_cl:
mock_cl.get_pv = MockPV
mock_cl.thread_class = threading.Thread
with mock.patch.object(FalconcSAXS, "_init"):
det = FalconcSAXS(
name=name, prefix=prefix, device_manager=dm, sim_mode=sim_mode
)
patch_dual_pvs(det)
yield det
@pytest.mark.parametrize(
"trigger_source, mapping_source, ignore_gate, pixels_per_buffer, detector_state,"
" expected_exception",
[(1, 1, 0, 20, 0, False), (1, 1, 0, 20, 1, True)],
)
# TODO rewrite this one, write test for init_detector, init_filewriter is tested
def test_init_detector(
mock_det,
trigger_source,
mapping_source,
ignore_gate,
pixels_per_buffer,
detector_state,
expected_exception,
):
"""Test the _init function:
This includes testing the functions:
- _init_detector
- _stop_det
- _set_trigger
--> Testing the filewriter is done in test_init_filewriter
Validation upon setting the correct PVs
"""
mock_det.value_pixel_per_buffer = pixels_per_buffer
mock_det.state._read_pv.mock_data = detector_state
if expected_exception:
with pytest.raises(FalconTimeoutError):
mock_det.timeout = 0.1
mock_det.custom_prepare.initialize_detector()
else:
mock_det.custom_prepare.initialize_detector()
assert mock_det.state.get() == detector_state
assert mock_det.collect_mode.get() == mapping_source
assert mock_det.pixel_advance_mode.get() == trigger_source
assert mock_det.ignore_gate.get() == ignore_gate
assert mock_det.preset_mode.get() == 1
assert mock_det.erase_all.get() == 1
assert mock_det.input_logic_polarity.get() == 0
assert mock_det.auto_pixels_per_buffer.get() == 0
assert mock_det.pixels_per_buffer.get() == pixels_per_buffer
@pytest.mark.parametrize(
"readout_time, expected_value", [(1e-3, 3e-3), (3e-3, 3e-3), (5e-3, 5e-3), (None, 3e-3)]
)
def test_update_readout_time(mock_det, readout_time, expected_value):
if readout_time is None:
mock_det.custom_prepare.update_readout_time()
assert mock_det.readout_time == expected_value
else:
mock_det.scaninfo.readout_time = readout_time
mock_det.custom_prepare.update_readout_time()
assert mock_det.readout_time == expected_value
def test_initialize_default_parameter(mock_det):
with mock.patch.object(
mock_det.custom_prepare, "update_readout_time"
) as mock_update_readout_time:
mock_det.custom_prepare.initialize_default_parameter()
assert mock_det.value_pixel_per_buffer == 20
mock_update_readout_time.assert_called_once()
@pytest.mark.parametrize(
"scaninfo",
[
(
{
"eacc": "e12345",
"num_points": 500,
"frames_per_trigger": 1,
"exp_time": 0.1,
"filepath": "test.h5",
"scan_id": "123",
"mokev": 12.4,
}
)
],
)
def test_stage(mock_det, scaninfo):
"""Test the stage function:
This includes testing _prep_det
"""
with (
mock.patch.object(mock_det, "set_trigger") as mock_set_trigger,
mock.patch.object(
mock_det.custom_prepare, "prepare_detector_backend"
) as mock_prep_data_backend,
mock.patch.object(
mock_det.custom_prepare, "publish_file_location"
) as mock_publish_file_location,
mock.patch.object(mock_det.custom_prepare, "arm_acquisition") as mock_arm_acquisition,
):
mock_det.scaninfo.exp_time = scaninfo["exp_time"]
mock_det.scaninfo.num_points = scaninfo["num_points"]
mock_det.scaninfo.frames_per_trigger = scaninfo["frames_per_trigger"]
mock_det.stage()
mock_set_trigger.assert_called_once()
assert mock_det.preset_real.get() == scaninfo["exp_time"]
assert mock_det.pixels_per_run.get() == int(
scaninfo["num_points"] * scaninfo["frames_per_trigger"]
)
mock_prep_data_backend.assert_called_once()
mock_publish_file_location.assert_called_once_with(done=False)
mock_arm_acquisition.assert_called_once()
@pytest.mark.parametrize(
"scaninfo",
[
(
{
"filepath": "/das/work/p18/p18533/data/S00000-S00999/S00001/data.h5",
"num_points": 500,
"frames_per_trigger": 1,
}
),
(
{
"filepath": "/das/work/p18/p18533/data/S00000-S00999/S00001/data1234.h5",
"num_points": 500,
"frames_per_trigger": 1,
}
),
],
)
def test_prepare_data_backend(mock_det, scaninfo):
mock_det.filewriter.compile_full_filename.return_value = scaninfo["filepath"]
mock_det.scaninfo.num_points = scaninfo["num_points"]
mock_det.scaninfo.frames_per_trigger = scaninfo["frames_per_trigger"]
mock_det.scaninfo.scan_number = 1
mock_det.custom_prepare.prepare_data_backend()
file_path, file_name = os.path.split(scaninfo["filepath"])
assert mock_det.hdf5.file_path.get() == file_path
assert mock_det.hdf5.file_name.get() == file_name
assert mock_det.hdf5.file_template.get() == "%s%s"
assert mock_det.hdf5.num_capture.get() == int(
scaninfo["num_points"] * scaninfo["frames_per_trigger"]
)
assert mock_det.hdf5.file_write_mode.get() == 2
assert mock_det.hdf5.array_counter.get() == 0
assert mock_det.hdf5.capture.get() == 1
@pytest.mark.parametrize(
"scaninfo",
[
({"filepath": "test.h5", "successful": True, "done": False, "scan_id": "123"}),
({"filepath": "test.h5", "successful": False, "done": True, "scan_id": "123"}),
({"filepath": "test.h5", "successful": None, "done": True, "scan_id": "123"}),
],
)
def test_publish_file_location(mock_det, scaninfo):
mock_det.scaninfo.scan_id = scaninfo["scan_id"]
mock_det.filepath = scaninfo["filepath"]
mock_det.custom_prepare.publish_file_location(
done=scaninfo["done"], successful=scaninfo["successful"]
)
if scaninfo["successful"] is None:
msg = messages.FileMessage(file_path=scaninfo["filepath"], done=scaninfo["done"])
else:
msg = messages.FileMessage(
file_path=scaninfo["filepath"], done=scaninfo["done"], successful=scaninfo["successful"]
)
expected_calls = [
mock.call(
MessageEndpoints.public_file(scaninfo["scan_id"], mock_det.name),
msg,
pipe=mock_det.connector.pipeline.return_value,
),
mock.call(
MessageEndpoints.file_event(mock_det.name),
msg,
pipe=mock_det.connector.pipeline.return_value,
),
]
assert mock_det.connector.set_and_publish.call_args_list == expected_calls
@pytest.mark.parametrize("detector_state, expected_exception", [(1, False), (0, True)])
def test_arm_acquisition(mock_det, detector_state, expected_exception):
with mock.patch.object(mock_det, "stop") as mock_stop:
mock_det.state._read_pv.mock_data = detector_state
if expected_exception:
with pytest.raises(FalconTimeoutError):
mock_det.timeout = 0.1
mock_det.custom_prepare.arm_acquisition()
mock_stop.assert_called_once()
else:
mock_det.custom_prepare.arm_acquisition()
assert mock_det.start_all.get() == 1
def test_trigger(mock_det):
with mock.patch.object(mock_det.custom_prepare, "on_trigger") as mock_on_trigger:
mock_det.trigger()
mock_on_trigger.assert_called_once()
@pytest.mark.parametrize("stopped, expected_abort", [(False, False), (True, True)])
def test_unstage(mock_det, stopped, expected_abort):
with (
mock.patch.object(mock_det.custom_prepare, "finished") as mock_finished,
mock.patch.object(
mock_det.custom_prepare, "publish_file_location"
) as mock_publish_file_location,
):
mock_det.stopped = stopped
if expected_abort:
mock_det.unstage()
assert mock_det.stopped is stopped
assert mock_publish_file_location.call_count == 0
else:
mock_det.unstage()
mock_finished.assert_called_once()
mock_publish_file_location.assert_called_with(done=True, successful=True)
assert mock_det.stopped is stopped
def test_stop(mock_det):
with (
mock.patch.object(mock_det.custom_prepare, "stop_detector") as mock_stop_det,
mock.patch.object(
mock_det.custom_prepare, "stop_detector_backend"
) as mock_stop_detector_backend,
):
mock_det.stop()
mock_stop_det.assert_called_once()
mock_stop_detector_backend.assert_called_once()
assert mock_det.stopped is True
@pytest.mark.parametrize(
"stopped, scaninfo",
[
(False, {"num_points": 500, "frames_per_trigger": 1}),
(True, {"num_points": 500, "frames_per_trigger": 1}),
],
)
def test_finished(mock_det, stopped, scaninfo):
with (
mock.patch.object(mock_det.custom_prepare, "stop_detector") as mock_stop_det,
mock.patch.object(
mock_det.custom_prepare, "stop_detector_backend"
) as mock_stop_file_writer,
):
mock_det.stopped = stopped
mock_det.dxp.current_pixel._read_pv.mock_data = int(
scaninfo["num_points"] * scaninfo["frames_per_trigger"]
)
mock_det.hdf5.array_counter._read_pv.mock_data = int(
scaninfo["num_points"] * scaninfo["frames_per_trigger"]
)
mock_det.scaninfo.frames_per_trigger = scaninfo["frames_per_trigger"]
mock_det.scaninfo.num_points = scaninfo["num_points"]
mock_det.custom_prepare.finished()
assert mock_det.stopped is stopped
mock_stop_det.assert_called_once()
mock_stop_file_writer.assert_called_once()

60
tests/test_fupr_ophyd.py
View File

@ -1,60 +0,0 @@
from unittest import mock
import pytest
from utils import SocketMock
from ophyd_devices.galil.fupr_ophyd import FuprGalilController, FuprGalilMotor
@pytest.fixture
def fsamroy():
FuprGalilController._reset_controller()
fsamroy_motor = FuprGalilMotor(
"A", name="fsamroy", host="mpc2680.psi.ch", port=8081, socket_cls=SocketMock
)
fsamroy_motor.controller.on()
assert isinstance(fsamroy_motor.controller, FuprGalilController)
yield fsamroy_motor
fsamroy_motor.controller.off()
fsamroy_motor.controller._reset_controller()
@pytest.mark.parametrize(
"pos,msg_received,msg_put,sign",
[
(-0.5, b" -12800\n\r", [b"TPA\r", b"MG_BGA\r", b"TPA\r"], 1),
(-0.5, b" 12800\n\r", [b"TPA\r", b"MG_BGA\r", b"TPA\r"], -1),
],
)
def test_axis_get(fsamroy, pos, msg_received, msg_put, sign):
fsamroy.sign = sign
fsamroy.device_manager = mock.MagicMock()
fsamroy.controller.sock.flush_buffer()
fsamroy.controller.sock.buffer_recv = msg_received
val = fsamroy.read()
assert val["fsamroy"]["value"] == pos
assert fsamroy.readback.get() == pos
assert fsamroy.controller.sock.buffer_put == msg_put
@pytest.mark.parametrize(
"target_pos,socket_put_messages,socket_get_messages",
[
(
0,
[b"MG axisref\r", b"PAA=0\r", b"PAA=0\r", b"BGA\r"],
[b"1.00", b"-1", b":", b":", b":", b":", b"-1"],
)
],
)
def test_axis_put(fsamroy, target_pos, socket_put_messages, socket_get_messages):
fsamroy.controller.sock.flush_buffer()
fsamroy.controller.sock.buffer_recv = socket_get_messages
fsamroy.user_setpoint.put(target_pos)
assert fsamroy.controller.sock.buffer_put == socket_put_messages
def test_drive_axis_to_limit(fsamroy):
fsamroy.controller.sock.flush_buffer()
with pytest.raises(NotImplementedError):
fsamroy.controller.drive_axis_to_limit(0, "forward")

149
tests/test_galil.py
View File

@ -1,149 +0,0 @@
from unittest import mock
import pytest
from utils import SocketMock
from ophyd_devices.galil.galil_ophyd import GalilController, GalilMotor
@pytest.fixture(scope="function")
def leyey():
GalilController._reset_controller()
leyey_motor = GalilMotor(
"H", name="leyey", host="mpc2680.psi.ch", port=8081, socket_cls=SocketMock
)
leyey_motor.controller.on()
yield leyey_motor
leyey_motor.controller.off()
leyey_motor.controller._reset_controller()
@pytest.fixture(scope="function")
def leyex():
GalilController._reset_controller()
leyex_motor = GalilMotor(
"A", name="leyey", host="mpc2680.psi.ch", port=8081, socket_cls=SocketMock
)
leyex_motor.controller.on()
yield leyex_motor
leyex_motor.controller.off()
leyex_motor.controller._reset_controller()
@pytest.mark.parametrize("pos,msg,sign", [(1, b" -12800\n\r", 1), (-1, b" -12800\n\r", -1)])
def test_axis_get(leyey, pos, msg, sign):
leyey.sign = sign
leyey.controller.sock.flush_buffer()
leyey.controller.sock.buffer_recv = msg
val = leyey.read()
assert val["leyey"]["value"] == pos
assert leyey.readback.get() == pos
@pytest.mark.parametrize(
"target_pos,socket_put_messages,socket_get_messages",
[
(
0,
[
b"MG allaxref\r",
b"MG_XQ0\r",
b"naxis=7\r",
b"ntarget=0.000\r",
b"movereq=1\r",
b"XQ#NEWPAR\r",
b"MG_XQ0\r",
],
[b"1.00", b"-1", b":", b":", b":", b":", b"-1"],
)
],
)
def test_axis_put(leyey, target_pos, socket_put_messages, socket_get_messages):
leyey.controller.sock.flush_buffer()
leyey.controller.sock.buffer_recv = socket_get_messages
leyey.user_setpoint.put(target_pos)
assert leyey.controller.sock.buffer_put == socket_put_messages
@pytest.mark.parametrize(
"axis_nr,direction,socket_put_messages,socket_get_messages",
[
(
0,
"forward",
[
b"naxis=0\r",
b"ndir=1\r",
b"XQ#NEWPAR\r",
b"XQ#FES\r",
b"MG_BGA\r",
b"MGbcklact[axis]\r",
b"MG_XQ0\r",
b"MG_XQ2\r",
b"MG _LRA, _LFA\r",
],
[b":", b":", b":", b":", b"0", b"0", b"-1", b"-1", b"1.000 0.000"],
),
(
1,
"reverse",
[
b"naxis=1\r",
b"ndir=-1\r",
b"XQ#NEWPAR\r",
b"XQ#FES\r",
b"MG_BGB\r",
b"MGbcklact[axis]\r",
b"MG_XQ0\r",
b"MG_XQ2\r",
b"MG _LRB, _LFB\r",
],
[b":", b":", b":", b":", b"0", b"0", b"-1", b"-1", b"0.000 1.000"],
),
],
)
def test_drive_axis_to_limit(leyex, axis_nr, direction, socket_put_messages, socket_get_messages):
leyex.controller.sock.flush_buffer()
leyex.controller.sock.buffer_recv = socket_get_messages
leyex.controller.drive_axis_to_limit(axis_nr, direction)
assert leyex.controller.sock.buffer_put == socket_put_messages
@pytest.mark.parametrize(
"axis_nr,socket_put_messages,socket_get_messages",
[
(
0,
[
b"naxis=0\r",
b"XQ#NEWPAR\r",
b"XQ#FRM\r",
b"MG_BGA\r",
b"MGbcklact[axis]\r",
b"MG_XQ0\r",
b"MG_XQ2\r",
b"MG axisref[0]\r",
],
[b":", b":", b":", b"0", b"0", b"-1", b"-1", b"1.00"],
),
(
1,
[
b"naxis=1\r",
b"XQ#NEWPAR\r",
b"XQ#FRM\r",
b"MG_BGB\r",
b"MGbcklact[axis]\r",
b"MG_XQ0\r",
b"MG_XQ2\r",
b"MG axisref[1]\r",
],
[b":", b":", b":", b"0", b"0", b"-1", b"-1", b"1.00"],
),
],
)
def test_find_reference(leyex, axis_nr, socket_put_messages, socket_get_messages):
leyex.controller.sock.flush_buffer()
leyex.controller.sock.buffer_recv = socket_get_messages
leyex.controller.find_reference(axis_nr)
assert leyex.controller.sock.buffer_put == socket_put_messages

172
tests/test_galil_flomni.py
View File

@ -1,172 +0,0 @@
from unittest import mock
import pytest
from utils import SocketMock
from ophyd_devices.galil.fgalil_ophyd import FlomniGalilController, FlomniGalilMotor
@pytest.fixture(scope="function")
def leyey():
FlomniGalilController._reset_controller()
leyey_motor = FlomniGalilMotor(
"H", name="leyey", host="mpc2680.psi.ch", port=8081, socket_cls=SocketMock
)
leyey_motor.controller.on()
yield leyey_motor
leyey_motor.controller.off()
leyey_motor.controller._reset_controller()
@pytest.fixture(scope="function")
def leyex():
FlomniGalilController._reset_controller()
leyex_motor = FlomniGalilMotor(
"H", name="leyey", host="mpc2680.psi.ch", port=8081, socket_cls=SocketMock
)
leyex_motor.controller.on()
yield leyex_motor
leyex_motor.controller.off()
leyex_motor.controller._reset_controller()
@pytest.mark.parametrize("pos,msg,sign", [(1, b" -12800\n\r", 1), (-1, b" -12800\n\r", -1)])
def test_axis_get(leyey, pos, msg, sign):
leyey.sign = sign
leyey.controller.sock.flush_buffer()
leyey.controller.sock.buffer_recv = msg
val = leyey.read()
assert val["leyey"]["value"] == pos
assert leyey.readback.get() == pos
@pytest.mark.parametrize(
"target_pos,socket_put_messages,socket_get_messages",
[
(
0,
[
b"MG allaxref\r",
b"MG_XQ0\r",
b"naxis=7\r",
b"ntarget=0.000\r",
b"movereq=1\r",
b"XQ#NEWPAR\r",
b"MG_XQ0\r",
],
[b"1.00", b"-1", b":", b":", b":", b":", b"-1"],
)
],
)
def test_axis_put(leyey, target_pos, socket_put_messages, socket_get_messages):
leyey.controller.sock.flush_buffer()
leyey.controller.sock.buffer_recv = socket_get_messages
leyey.user_setpoint.put(target_pos)
assert leyey.controller.sock.buffer_put == socket_put_messages
@pytest.mark.parametrize(
"axis_nr,direction,socket_put_messages,socket_get_messages",
[
(
0,
"forward",
[
b"naxis=0\r",
b"ndir=1\r",
b"XQ#NEWPAR\r",
b"XQ#FES\r",
b"MG_XQ0\r",
b"MG _MOA\r",
b"MG_XQ0\r",
b"MG _MOA\r",
b"MG _LRA, _LFA\r",
],
[b":", b":", b":", b":", b"0", b"0", b"-1", b"-1", b"1.000 0.000"],
),
(
1,
"reverse",
[
b"naxis=1\r",
b"ndir=-1\r",
b"XQ#NEWPAR\r",
b"XQ#FES\r",
b"MG_XQ0\r",
b"MG _MOB\r",
b"MG_XQ0\r",
b"MG _MOB\r",
b"MG _LRB, _LFB\r",
],
[b":", b":", b":", b":", b"0", b"0", b"-1", b"-1", b"0.000 1.000"],
),
],
)
def test_drive_axis_to_limit(leyex, axis_nr, direction, socket_put_messages, socket_get_messages):
leyex.controller.sock.flush_buffer()
leyex.controller.sock.buffer_recv = socket_get_messages
leyex.controller.drive_axis_to_limit(axis_nr, direction)
assert leyex.controller.sock.buffer_put == socket_put_messages
@pytest.mark.parametrize(
"axis_nr,socket_put_messages,socket_get_messages",
[
(
0,
[
b"naxis=0\r",
b"XQ#NEWPAR\r",
b"XQ#FRM\r",
b"MG_XQ0\r",
b"MG _MOA\r",
b"MG_XQ0\r",
b"MG _MOA\r",
b"MG axisref[0]\r",
],
[b":", b":", b":", b"0", b"0", b"-1", b"-1", b"1.00"],
),
(
1,
[
b"naxis=1\r",
b"XQ#NEWPAR\r",
b"XQ#FRM\r",
b"MG_XQ0\r",
b"MG _MOB\r",
b"MG_XQ0\r",
b"MG _MOB\r",
b"MG axisref[1]\r",
],
[b":", b":", b":", b"0", b"0", b"-1", b"-1", b"1.00"],
),
],
)
def test_find_reference(leyex, axis_nr, socket_put_messages, socket_get_messages):
leyex.controller.sock.flush_buffer()
leyex.controller.sock.buffer_recv = socket_get_messages
leyex.controller.find_reference(axis_nr)
assert leyex.controller.sock.buffer_put == socket_put_messages
@pytest.mark.parametrize(
"axis_Id,socket_put_messages,socket_get_messages,triggered",
[
("A", [b"MG @IN[14]\r"], [b" 1.0000\n"], True),
("B", [b"MG @IN[14]\r"], [b" 0.0000\n"], False),
],
)
def test_fosaz_light_curtain_is_triggered(
axis_Id, socket_put_messages, socket_get_messages, triggered
):
"""test that the light curtain is triggered"""
fosaz = FlomniGalilMotor(
axis_Id, name="fosaz", host="mpc2680.psi.ch", port=8081, socket_cls=SocketMock
)
fosaz.controller.on()
fosaz.controller.sock.flush_buffer()
fosaz.controller.sock.buffer_recv = socket_get_messages
assert fosaz.controller.fosaz_light_curtain_is_triggered() == triggered
assert fosaz.controller.sock.buffer_put == socket_put_messages
fosaz.controller.off()
fosaz.controller._reset_controller()

331
tests/test_mcs_card.py
View File

@ -1,331 +0,0 @@
# pylint: skip-file
import threading
from unittest import mock
import ophyd
import pytest
from bec_lib import MessageEndpoints, messages
from ophyd_devices.epics.devices.mcs_csaxs import (
MCScSAXS,
MCSError,
MCSTimeoutError,
ReadoutMode,
TriggerSource,
)
from tests.utils import DMMock, MockPV
def patch_dual_pvs(device):
for walk in device.walk_signals():
if not hasattr(walk.item, "_read_pv"):
continue
if not hasattr(walk.item, "_write_pv"):
continue
if walk.item._read_pv.pvname.endswith("_RBV"):
walk.item._read_pv = walk.item._write_pv
@pytest.fixture(scope="function")
def mock_det():
name = "mcs"
prefix = "X12SA-MCS:"
sim_mode = False
dm = DMMock()
with mock.patch.object(dm, "connector"):
with (
mock.patch("ophyd_devices.epics.devices.psi_detector_base.FileWriter") as filemixin,
mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.PSIDetectorBase._update_service_config"
) as mock_service_config,
):
with mock.patch.object(ophyd, "cl") as mock_cl:
mock_cl.get_pv = MockPV
mock_cl.thread_class = threading.Thread
with mock.patch.object(MCScSAXS, "_init"):
det = MCScSAXS(name=name, prefix=prefix, device_manager=dm, sim_mode=sim_mode)
patch_dual_pvs(det)
yield det
def test_init():
"""Test the _init function:"""
name = "eiger"
prefix = "X12SA-ES-EIGER9M:"
sim_mode = False
dm = DMMock()
with mock.patch.object(dm, "connector"):
with (
mock.patch("ophyd_devices.epics.devices.psi_detector_base.FileWriter"),
mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.PSIDetectorBase._update_service_config"
),
):
with mock.patch.object(ophyd, "cl") as mock_cl:
mock_cl.get_pv = MockPV
with (
mock.patch(
"ophyd_devices.epics.devices.mcs_csaxs.MCSSetup.initialize_default_parameter"
) as mock_default,
mock.patch(
"ophyd_devices.epics.devices.mcs_csaxs.MCSSetup.initialize_detector"
) as mock_init_det,
mock.patch(
"ophyd_devices.epics.devices.mcs_csaxs.MCSSetup.initialize_detector_backend"
) as mock_init_backend,
):
MCScSAXS(name=name, prefix=prefix, device_manager=dm, sim_mode=sim_mode)
mock_default.assert_called_once()
mock_init_det.assert_called_once()
mock_init_backend.assert_called_once()
@pytest.mark.parametrize(
"trigger_source, channel_advance, channel_source1, pv_channels",
[
(
3,
1,
0,
{
"user_led": 0,
"mux_output": 5,
"input_pol": 0,
"output_pol": 1,
"count_on_start": 0,
"stop_all": 1,
},
)
],
)
def test_initialize_detector(
mock_det, trigger_source, channel_advance, channel_source1, pv_channels
):
"""Test the _init function:
This includes testing the functions:
- initialize_detector
- stop_det
- parent.set_trigger
--> Testing the filewriter is done in test_init_filewriter
Validation upon setting the correct PVs
"""
mock_det.custom_prepare.initialize_detector() # call the method you want to test
assert mock_det.channel_advance.get() == channel_advance
assert mock_det.channel1_source.get() == channel_source1
assert mock_det.user_led.get() == pv_channels["user_led"]
assert mock_det.mux_output.get() == pv_channels["mux_output"]
assert mock_det.input_polarity.get() == pv_channels["input_pol"]
assert mock_det.output_polarity.get() == pv_channels["output_pol"]
assert mock_det.count_on_start.get() == pv_channels["count_on_start"]
assert mock_det.input_mode.get() == trigger_source
def test_trigger(mock_det):
"""Test the trigger function:
Validate that trigger calls the custom_prepare.on_trigger() function
"""
with mock.patch.object(mock_det.custom_prepare, "on_trigger") as mock_on_trigger:
mock_det.trigger()
mock_on_trigger.assert_called_once()
@pytest.mark.parametrize(
"value, num_lines, num_points, done", [(100, 5, 500, False), (500, 5, 500, True)]
)
def test_progress_update(mock_det, value, num_lines, num_points, done):
mock_det.num_lines.set(num_lines)
mock_det.scaninfo.num_points = num_points
calls = mock.call(sub_type="progress", value=value, max_value=num_points, done=done)
with mock.patch.object(mock_det, "_run_subs") as mock_run_subs:
mock_det.custom_prepare._progress_update(value=value)
mock_run_subs.assert_called_once()
assert mock_run_subs.call_args == calls
@pytest.mark.parametrize(
"values, expected_nothing",
[([[100, 120, 140], [200, 220, 240], [300, 320, 340]], False), ([100, 200, 300], True)],
)
def test_on_mca_data(mock_det, values, expected_nothing):
"""Test the on_mca_data function:
Validate that on_mca_data calls the custom_prepare.on_mca_data() function
"""
with mock.patch.object(mock_det.custom_prepare, "_send_data_to_bec") as mock_send_data:
mock_object = mock.MagicMock()
for ii, name in enumerate(mock_det.custom_prepare.mca_names):
mock_object.attr_name = name
mock_det.custom_prepare._on_mca_data(obj=mock_object, value=values[ii])
if not expected_nothing and ii < (len(values) - 1):
assert mock_det.custom_prepare.mca_data[name] == values[ii]
if not expected_nothing:
mock_send_data.assert_called_once()
assert mock_det.custom_prepare.acquisition_done is True
@pytest.mark.parametrize(
"metadata, mca_data",
[
(
{"scan_id": 123},
{"mca1": [100, 120, 140], "mca3": [200, 220, 240], "mca4": [300, 320, 340]},
)
],
)
def test_send_data_to_bec(mock_det, metadata, mca_data):
mock_det.scaninfo.scan_msg = mock.MagicMock()
mock_det.scaninfo.scan_msg.metadata = metadata
mock_det.scaninfo.scan_id = metadata["scan_id"]
mock_det.custom_prepare.mca_data = mca_data
mock_det.custom_prepare._send_data_to_bec()
device_metadata = mock_det.scaninfo.scan_msg.metadata
metadata.update({"async_update": "append", "num_lines": mock_det.num_lines.get()})
data = messages.DeviceMessage(signals=dict(mca_data), metadata=device_metadata)
calls = mock.call(
topic=MessageEndpoints.device_async_readback(
scan_id=metadata["scan_id"], device=mock_det.name
),
msg={"data": data},
expire=1800,
)
assert mock_det.connector.xadd.call_args == calls
@pytest.mark.parametrize(
"scaninfo, triggersource, stopped, expected_exception",
[
(
{"num_points": 500, "frames_per_trigger": 1, "scan_type": "step"},
TriggerSource.MODE3,
False,
False,
),
(
{"num_points": 500, "frames_per_trigger": 1, "scan_type": "fly"},
TriggerSource.MODE3,
False,
False,
),
(
{"num_points": 5001, "frames_per_trigger": 2, "scan_type": "step"},
TriggerSource.MODE3,
False,
True,
),
(
{"num_points": 500, "frames_per_trigger": 2, "scan_type": "random"},
TriggerSource.MODE3,
False,
True,
),
],
)
def test_stage(mock_det, scaninfo, triggersource, stopped, expected_exception):
mock_det.scaninfo.num_points = scaninfo["num_points"]
mock_det.scaninfo.frames_per_trigger = scaninfo["frames_per_trigger"]
mock_det.scaninfo.scan_type = scaninfo["scan_type"]
mock_det.stopped = stopped
with mock.patch.object(mock_det.custom_prepare, "prepare_detector_backend") as mock_prep_fw:
if expected_exception:
with pytest.raises(MCSError):
mock_det.stage()
mock_prep_fw.assert_called_once()
else:
mock_det.stage()
mock_prep_fw.assert_called_once()
# Check set_trigger
mock_det.input_mode.get() == triggersource
if scaninfo["scan_type"] == "step":
assert mock_det.num_use_all.get() == int(scaninfo["frames_per_trigger"]) * int(
scaninfo["num_points"]
)
elif scaninfo["scan_type"] == "fly":
assert mock_det.num_use_all.get() == int(scaninfo["num_points"])
mock_det.preset_real.get() == 0
# # CHeck custom_prepare.arm_acquisition
# assert mock_det.custom_prepare.counter == 0
# assert mock_det.erase_start.get() == 1
# mock_prep_fw.assert_called_once()
# # Check _prep_det
# assert mock_det.cam.num_images.get() == int(
# scaninfo["num_points"] * scaninfo["frames_per_trigger"]
# )
# assert mock_det.cam.num_frames.get() == 1
# mock_publish_file_location.assert_called_with(done=False)
# assert mock_det.cam.acquire.get() == 1
def test_prepare_detector_backend(mock_det):
mock_det.custom_prepare.prepare_detector_backend()
assert mock_det.erase_all.get() == 1
assert mock_det.read_mode.get() == ReadoutMode.EVENT
@pytest.mark.parametrize("stopped, expected_exception", [(False, False), (True, True)])
def test_unstage(mock_det, stopped, expected_exception):
with (
mock.patch.object(mock_det.custom_prepare, "finished") as mock_finished,
mock.patch.object(
mock_det.custom_prepare, "publish_file_location"
) as mock_publish_file_location,
):
mock_det.stopped = stopped
if expected_exception:
mock_det.unstage()
assert mock_det.stopped is True
else:
mock_det.unstage()
mock_finished.assert_called_once()
mock_publish_file_location.assert_called_with(done=True, successful=True)
assert mock_det.stopped is False
def test_stop_detector_backend(mock_det):
mock_det.custom_prepare.stop_detector_backend()
assert mock_det.custom_prepare.acquisition_done is True
def test_stop(mock_det):
with (
mock.patch.object(mock_det.custom_prepare, "stop_detector") as mock_stop_det,
mock.patch.object(
mock_det.custom_prepare, "stop_detector_backend"
) as mock_stop_detector_backend,
):
mock_det.stop()
mock_stop_det.assert_called_once()
mock_stop_detector_backend.assert_called_once()
assert mock_det.stopped is True
@pytest.mark.parametrize(
"stopped, acquisition_done, acquiring_state, expected_exception",
[
(False, True, 0, False),
(False, False, 0, True),
(False, True, 1, True),
(True, True, 0, True),
],
)
def test_finished(mock_det, stopped, acquisition_done, acquiring_state, expected_exception):
mock_det.custom_prepare.acquisition_done = acquisition_done
mock_det.acquiring._read_pv.mock_data = acquiring_state
mock_det.scaninfo.num_points = 500
mock_det.num_lines.put(500)
mock_det.current_channel._read_pv.mock_data = 1
mock_det.stopped = stopped
if expected_exception:
with pytest.raises(MCSTimeoutError):
mock_det.timeout = 0.1
mock_det.custom_prepare.finished()
else:
mock_det.custom_prepare.finished()
if stopped:
assert mock_det.stopped is stopped

141
tests/test_npoint_piezo.py
View File

@ -1,141 +0,0 @@
import pytest
from ophyd_devices.npoint import NPointAxis, NPointController
class SocketMock:
def __init__(self, sock=None):
self.buffer_put = ""
self.buffer_recv = ""
self.is_open = False
if sock is None:
self.open()
else:
self.sock = sock
def connect(self, host, port):
print(f"connecting to {host} port {port}")
# self.sock.create_connection((host, port))
# self.sock.connect((host, port))
def _put(self, msg_bytes):
self.buffer_put = msg_bytes
print(self.buffer_put)
def _recv(self, buffer_length=1024):
print(self.buffer_recv)
return self.buffer_recv
def _initialize_socket(self):
pass
def put(self, msg):
return self._put(msg)
def receive(self, buffer_length=1024):
return self._recv(buffer_length=buffer_length)
def open(self):
self._initialize_socket()
self.is_open = True
def close(self):
self.sock = None
self.is_open = False
@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(pos, msg):
controller = NPointController(SocketMock())
npointx = NPointAxis(controller, 0, "nx")
controller.on()
npointx.set(pos)
assert npointx.controller.socket.buffer_put == msg
@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(pos, msg_in, msg_out):
controller = NPointController(SocketMock())
npointx = NPointAxis(controller, 0, "nx")
controller.on()
npointx.controller.socket.buffer_recv = msg_out
assert pytest.approx(npointx.get(), rel=0.01) == pos
# assert controller.socket.buffer_put == msg_in
@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\x13\x83\x11\x00\x00\x00\x00U"),
(2, b"\xa043\x83\x11U", b"\xa0\x34\x13\x83\x1133\xff\xffU"),
],
)
def test_axis_get_in(axis, msg_in, msg_out):
controller = NPointController(SocketMock())
npointx = NPointAxis(controller, 0, "nx")
controller.on()
controller.socket.buffer_recv = msg_out
controller._get_current_pos(axis)
assert controller.socket.buffer_put == msg_in
def test_axis_out_of_range():
controller = NPointController(SocketMock())
with pytest.raises(ValueError):
npointx = NPointAxis(controller, 3, "nx")
def test_get_axis_out_of_range():
controller = NPointController(SocketMock())
with pytest.raises(ValueError):
controller._get_current_pos(3)
def test_set_axis_out_of_range():
controller = NPointController(SocketMock())
with pytest.raises(ValueError):
controller._set_target_pos(3, 5)
@pytest.mark.parametrize(
"in_buffer, byteorder, signed, val",
[
(["0x0", "0x0", "0xcc", "0xcd"], "big", True, 52429),
(["0xcd", "0xcc", "0x0", "0x0"], "little", True, 52429),
(["cd", "cc", "00", "00"], "little", True, 52429),
],
)
def test_hex_list_to_int(in_buffer, byteorder, signed, val):
assert NPointController._hex_list_to_int(in_buffer, byteorder=byteorder, signed=signed) == val
@pytest.mark.parametrize(
"axis, msg_in, msg_out",
[
(0, b"\xa0x\x10\x83\x11U", b"\xa0\x78\x13\x83\x11\x64\x00\x00\x00U"),
(1, b"\xa0x \x83\x11U", b"\xa0\x78\x13\x83\x11\x64\x00\x00\x00U"),
(2, b"\xa0x0\x83\x11U", b"\xa0\x78\x13\x83\x11\x64\x00\x00\x00U"),
],
)
def test_get_range(axis, msg_in, msg_out):
controller = NPointController(SocketMock())
npointx = NPointAxis(controller, 0, "nx")
controller.on()
controller.socket.buffer_recv = msg_out
val = controller._get_range(axis)
assert controller.socket.buffer_put == msg_in and val == 100

468
tests/test_pilatus_csaxs.py
View File

@ -1,468 +0,0 @@
# pylint: skip-file
import os
import threading
from unittest import mock
import ophyd
import pytest
from bec_lib import MessageEndpoints, messages
from ophyd_devices.epics.devices.pilatus_csaxs import PilatuscSAXS
from tests.utils import DMMock, MockPV
def patch_dual_pvs(device):
for walk in device.walk_signals():
if not hasattr(walk.item, "_read_pv"):
continue
if not hasattr(walk.item, "_write_pv"):
continue
if walk.item._read_pv.pvname.endswith("_RBV"):
walk.item._read_pv = walk.item._write_pv
@pytest.fixture(scope="function")
def mock_det():
name = "pilatus"
prefix = "X12SA-ES-PILATUS300K:"
sim_mode = False
dm = DMMock()
with mock.patch.object(dm, "connector"):
with (
mock.patch("ophyd_devices.epics.devices.psi_detector_base.FileWriter"),
mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.PSIDetectorBase._update_service_config"
),
):
with mock.patch.object(ophyd, "cl") as mock_cl:
mock_cl.get_pv = MockPV
mock_cl.thread_class = threading.Thread
with mock.patch.object(PilatuscSAXS, "_init"):
det = PilatuscSAXS(
name=name, prefix=prefix, device_manager=dm, sim_mode=sim_mode
)
patch_dual_pvs(det)
yield det
@pytest.mark.parametrize("trigger_source, detector_state", [(1, 0)])
# TODO rewrite this one, write test for init_detector, init_filewriter is tested
def test_init_detector(mock_det, trigger_source, detector_state):
"""Test the _init function:
This includes testing the functions:
- _init_detector
- _stop_det
- _set_trigger
--> Testing the filewriter is done in test_init_filewriter
Validation upon setting the correct PVs
"""
mock_det.custom_prepare.initialize_detector() # call the method you want to test
assert mock_det.cam.acquire.get() == detector_state
assert mock_det.cam.trigger_mode.get() == trigger_source
@pytest.mark.parametrize(
"scaninfo, stopped, expected_exception",
[
(
{
"eacc": "e12345",
"num_points": 500,
"frames_per_trigger": 1,
"filepath": "test.h5",
"scan_id": "123",
"mokev": 12.4,
},
False,
False,
),
(
{
"eacc": "e12345",
"num_points": 500,
"frames_per_trigger": 1,
"filepath": "test.h5",
"scan_id": "123",
"mokev": 12.4,
},
True,
False,
),
],
)
def test_stage(mock_det, scaninfo, stopped, expected_exception):
with mock.patch.object(
mock_det.custom_prepare, "publish_file_location"
) as mock_publish_file_location:
mock_det.scaninfo.num_points = scaninfo["num_points"]
mock_det.scaninfo.frames_per_trigger = scaninfo["frames_per_trigger"]
mock_det.filewriter.compile_full_filename.return_value = scaninfo["filepath"]
mock_det.device_manager.add_device("mokev", value=12.4)
mock_det.stopped = stopped
with (
mock.patch.object(
mock_det.custom_prepare, "prepare_detector_backend"
) as mock_data_backend,
mock.patch.object(
mock_det.custom_prepare, "update_readout_time"
) as mock_update_readout_time,
):
mock_det.filepath = scaninfo["filepath"]
if expected_exception:
with pytest.raises(Exception):
mock_det.timeout = 0.1
mock_det.stage()
else:
mock_det.stage()
mock_data_backend.assert_called_once()
mock_update_readout_time.assert_called()
# Check _prep_det
assert mock_det.cam.num_images.get() == int(
scaninfo["num_points"] * scaninfo["frames_per_trigger"]
)
assert mock_det.cam.num_frames.get() == 1
mock_publish_file_location.assert_called_with(done=False)
def test_pre_scan(mock_det):
mock_det.custom_prepare.pre_scan()
assert mock_det.cam.acquire.get() == 1
@pytest.mark.parametrize(
"readout_time, expected_value", [(1e-3, 3e-3), (3e-3, 3e-3), (5e-3, 5e-3), (None, 3e-3)]
)
def test_update_readout_time(mock_det, readout_time, expected_value):
if readout_time is None:
mock_det.custom_prepare.update_readout_time()
assert mock_det.readout_time == expected_value
else:
mock_det.scaninfo.readout_time = readout_time
mock_det.custom_prepare.update_readout_time()
assert mock_det.readout_time == expected_value
@pytest.mark.parametrize(
"scaninfo",
[
(
{
"filepath": "test.h5",
"filepath_raw": "test5_raw.h5",
"successful": True,
"done": False,
"scan_id": "123",
}
),
(
{
"filepath": "test.h5",
"filepath_raw": "test5_raw.h5",
"successful": False,
"done": True,
"scan_id": "123",
}
),
(
{
"filepath": "test.h5",
"filepath_raw": "test5_raw.h5",
"successful": None,
"done": True,
"scan_id": "123",
}
),
],
)
def test_publish_file_location(mock_det, scaninfo):
mock_det.scaninfo.scan_id = scaninfo["scan_id"]
mock_det.filepath = scaninfo["filepath"]
mock_det.filepath_raw = scaninfo["filepath_raw"]
mock_det.custom_prepare.publish_file_location(
done=scaninfo["done"], successful=scaninfo["successful"]
)
if scaninfo["successful"] is None:
msg = messages.FileMessage(
file_path=scaninfo["filepath"],
done=scaninfo["done"],
metadata={"input_path": scaninfo["filepath_raw"]},
)
else:
msg = messages.FileMessage(
file_path=scaninfo["filepath"],
done=scaninfo["done"],
metadata={"input_path": scaninfo["filepath_raw"]},
successful=scaninfo["successful"],
)
expected_calls = [
mock.call(
MessageEndpoints.public_file(scaninfo["scan_id"], mock_det.name),
msg,
pipe=mock_det.connector.pipeline.return_value,
),
mock.call(
MessageEndpoints.file_event(mock_det.name),
msg,
pipe=mock_det.connector.pipeline.return_value,
),
]
assert mock_det.connector.set_and_publish.call_args_list == expected_calls
@pytest.mark.parametrize(
"requests_state, expected_exception, url_delete, url_put",
[
(
True,
False,
"http://x12sa-pd-2:8080/stream/pilatus_2",
"http://xbl-daq-34:8091/pilatus_2/stop",
),
(
False,
False,
"http://x12sa-pd-2:8080/stream/pilatus_2",
"http://xbl-daq-34:8091/pilatus_2/stop",
),
],
)
def test_stop_detector_backend(mock_det, requests_state, expected_exception, url_delete, url_put):
with (
mock.patch.object(
mock_det.custom_prepare, "send_requests_delete"
) as mock_send_requests_delete,
mock.patch.object(mock_det.custom_prepare, "send_requests_put") as mock_send_requests_put,
):
instance_delete = mock_send_requests_delete.return_value
instance_delete.ok = requests_state
instance_put = mock_send_requests_put.return_value
instance_put.ok = requests_state
if expected_exception:
mock_det.custom_prepare.stop_detector_backend()
mock_send_requests_delete.assert_called_once_with(url=url_delete)
mock_send_requests_put.assert_called_once_with(url=url_put)
instance_delete.raise_for_status.called_once()
instance_put.raise_for_status.called_once()
else:
mock_det.custom_prepare.stop_detector_backend()
mock_send_requests_delete.assert_called_once_with(url=url_delete)
mock_send_requests_put.assert_called_once_with(url=url_put)
@pytest.mark.parametrize(
"scaninfo, data_msgs, urls, requests_state, expected_exception",
[
(
{
"filepath_raw": "pilatus_2.h5",
"eacc": "e12345",
"scan_number": 1000,
"scan_directory": "S00000_00999",
"num_points": 500,
"frames_per_trigger": 1,
"headers": {"Content-Type": "application/json", "Accept": "application/json"},
},
[
{
"source": [
{
"searchPath": "/",
"searchPattern": "glob:*.cbf",
"destinationPath": (
"/sls/X12SA/data/e12345/Data10/pilatus_2/S00000_00999"
),
}
]
},
[
"zmqWriter",
"e12345",
{
"addr": "tcp://x12sa-pd-2:8888",
"dst": ["file"],
"numFrm": 500,
"timeout": 2000,
"ifType": "PULL",
"user": "e12345",
},
],
["zmqWriter", "e12345", {"frmCnt": 500, "timeout": 2000}],
],
[
"http://x12sa-pd-2:8080/stream/pilatus_2",
"http://xbl-daq-34:8091/pilatus_2/run",
"http://xbl-daq-34:8091/pilatus_2/wait",
],
True,
False,
),
(
{
"filepath_raw": "pilatus_2.h5",
"eacc": "e12345",
"scan_number": 1000,
"scan_directory": "S00000_00999",
"num_points": 500,
"frames_per_trigger": 1,
"headers": {"Content-Type": "application/json", "Accept": "application/json"},
},
[
{
"source": [
{
"searchPath": "/",
"searchPattern": "glob:*.cbf",
"destinationPath": (
"/sls/X12SA/data/e12345/Data10/pilatus_2/S00000_00999"
),
}
]
},
[
"zmqWriter",
"e12345",
{
"addr": "tcp://x12sa-pd-2:8888",
"dst": ["file"],
"numFrm": 500,
"timeout": 2000,
"ifType": "PULL",
"user": "e12345",
},
],
["zmqWriter", "e12345", {"frmCnt": 500, "timeout": 2000}],
],
[
"http://x12sa-pd-2:8080/stream/pilatus_2",
"http://xbl-daq-34:8091/pilatus_2/run",
"http://xbl-daq-34:8091/pilatus_2/wait",
],
False, # return of res.ok is False!
True,
),
],
)
def test_prep_file_writer(mock_det, scaninfo, data_msgs, urls, requests_state, expected_exception):
with (
mock.patch.object(mock_det.custom_prepare, "close_file_writer") as mock_close_file_writer,
mock.patch.object(mock_det.custom_prepare, "stop_file_writer") as mock_stop_file_writer,
mock.patch.object(mock_det, "filewriter") as mock_filewriter,
mock.patch.object(mock_det.custom_prepare, "create_directory") as mock_create_directory,
mock.patch.object(mock_det.custom_prepare, "send_requests_put") as mock_send_requests_put,
):
mock_det.scaninfo.scan_number = scaninfo["scan_number"]
mock_det.scaninfo.num_points = scaninfo["num_points"]
mock_det.scaninfo.frames_per_trigger = scaninfo["frames_per_trigger"]
mock_det.scaninfo.username = scaninfo["eacc"]
mock_filewriter.compile_full_filename.return_value = scaninfo["filepath_raw"]
mock_filewriter.get_scan_directory.return_value = scaninfo["scan_directory"]
instance = mock_send_requests_put.return_value
instance.ok = requests_state
instance.raise_for_status.side_effect = Exception
if expected_exception:
with pytest.raises(Exception):
mock_det.timeout = 0.1
mock_det.custom_prepare.prepare_data_backend()
mock_close_file_writer.assert_called_once()
mock_stop_file_writer.assert_called_once()
instance.raise_for_status.assert_called_once()
else:
mock_det.custom_prepare.prepare_data_backend()
mock_close_file_writer.assert_called_once()
mock_stop_file_writer.assert_called_once()
# Assert values set on detector
assert mock_det.cam.file_path.get() == "/dev/shm/zmq/"
assert (
mock_det.cam.file_name.get()
== f"{scaninfo['eacc']}_2_{scaninfo['scan_number']:05d}"
)
assert mock_det.cam.auto_increment.get() == 1
assert mock_det.cam.file_number.get() == 0
assert mock_det.cam.file_format.get() == 0
assert mock_det.cam.file_template.get() == "%s%s_%5.5d.cbf"
# Remove last / from destinationPath
mock_create_directory.assert_called_once_with(
os.path.join(data_msgs[0]["source"][0]["destinationPath"])
)
assert mock_send_requests_put.call_count == 3
calls = [
mock.call(url=url, data=data_msg, headers=scaninfo["headers"])
for url, data_msg in zip(urls, data_msgs)
]
for call, mock_call in zip(calls, mock_send_requests_put.call_args_list):
assert call == mock_call
@pytest.mark.parametrize("stopped, expected_exception", [(False, False), (True, True)])
def test_unstage(mock_det, stopped, expected_exception):
with (
mock.patch.object(mock_det.custom_prepare, "finished") as mock_finished,
mock.patch.object(
mock_det.custom_prepare, "publish_file_location"
) as mock_publish_file_location,
):
mock_det.stopped = stopped
if expected_exception:
mock_det.unstage()
assert mock_det.stopped is True
else:
mock_det.unstage()
mock_finished.assert_called_once()
mock_publish_file_location.assert_called_with(done=True, successful=True)
assert mock_det.stopped is False
def test_stop(mock_det):
with (
mock.patch.object(mock_det.custom_prepare, "stop_detector") as mock_stop_det,
mock.patch.object(mock_det.custom_prepare, "stop_file_writer") as mock_stop_file_writer,
mock.patch.object(mock_det.custom_prepare, "close_file_writer") as mock_close_file_writer,
):
mock_det.stop()
mock_stop_det.assert_called_once()
mock_stop_file_writer.assert_called_once()
mock_close_file_writer.assert_called_once()
assert mock_det.stopped is True
@pytest.mark.parametrize(
"stopped, mcs_stage_state, expected_exception",
[
(False, ophyd.Staged.no, False),
(True, ophyd.Staged.no, True),
(False, ophyd.Staged.yes, True),
],
)
def test_finished(mock_det, stopped, mcs_stage_state, expected_exception):
with (
mock.patch.object(mock_det, "device_manager") as mock_dm,
mock.patch.object(mock_det.custom_prepare, "stop_file_writer") as mock_stop_file_friter,
mock.patch.object(mock_det.custom_prepare, "stop_detector") as mock_stop_det,
mock.patch.object(mock_det.custom_prepare, "close_file_writer") as mock_close_file_writer,
):
mock_dm.devices.mcs.obj._staged = mcs_stage_state
mock_det.stopped = stopped
if expected_exception:
with pytest.raises(Exception):
mock_det.timeout = 0.1
mock_det.custom_prepare.finished()
assert mock_det.stopped is stopped
mock_stop_file_friter.assert_called()
mock_stop_det.assert_called_once()
mock_close_file_writer.assert_called_once()
else:
mock_det.custom_prepare.finished()
if stopped:
assert mock_det.stopped is stopped
mock_stop_file_friter.assert_called()
mock_stop_det.assert_called_once()
mock_close_file_writer.assert_called_once()

89
tests/test_rt_flomni.py
View File

@ -1,89 +0,0 @@
from unittest import mock
import pytest
from utils import SocketMock
from ophyd_devices.rt_lamni import RtFlomniController, RtFlomniMotor
from ophyd_devices.rt_lamni.rt_ophyd import RtError
@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()
def test_move_samx_to_scan_region(rt_flomni):
device_manager = rt_flomni.get_device_manager()
device_manager.devices.rtx.user_parameter.get.return_value = 1
rt_flomni.move_samx_to_scan_region(20, 2)
assert mock.call(b"v0\n") not in rt_flomni.sock.put.mock_calls
assert mock.call(b"v1\n") in rt_flomni.sock.put.mock_calls
def test_feedback_enable_without_reset(rt_flomni):
with mock.patch.object(rt_flomni, "set_device_enabled") as set_device_enabled:
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:
rt_flomni.feedback_enable_without_reset()
laser_tracker_on.assert_called_once()
assert mock.call(b"l3\n") in rt_flomni.sock.put.mock_calls
assert mock.call("fsamx", False) in set_device_enabled.mock_calls
assert mock.call("fsamy", False) in set_device_enabled.mock_calls
assert mock.call("foptx", False) in set_device_enabled.mock_calls
assert mock.call("fopty", False) in set_device_enabled.mock_calls
def test_feedback_enable_without_reset_raises(rt_flomni):
with mock.patch.object(rt_flomni, "feedback_is_running", return_value=False):
with mock.patch.object(rt_flomni, "laser_tracker_on") as laser_tracker_on:
with pytest.raises(RtError):
rt_flomni.feedback_enable_without_reset()
laser_tracker_on.assert_called_once()
assert mock.call(b"l3\n") in rt_flomni.sock.put.mock_calls

7
tests/test_simulation.py
View File

@ -7,6 +7,7 @@ from unittest import mock
import h5py
import numpy as np
import pytest
from bec_server.device_server.tests.utils import DMMock
from ophyd import Device, Signal
from ophyd_devices.ophyd_base_devices.bec_protocols import (
@ -20,7 +21,6 @@ from ophyd_devices.sim.sim import SimCamera, SimFlyer, SimMonitor, SimPositioner
from ophyd_devices.sim.sim_frameworks import H5ImageReplayProxy, SlitProxy
from ophyd_devices.sim.sim_signals import ReadOnlySignal
from ophyd_devices.utils.bec_device_base import BECDevice, BECDeviceBase
from tests.utils import DMMock
@pytest.fixture(scope="function")
@ -120,15 +120,18 @@ def test_flyer__init__(flyer):
@pytest.mark.parametrize("center", [-10, 0, 10])
def test_monitor_readback(monitor, center):
"""Test the readback method of SimMonitor."""
motor_pos = 0
monitor.sim.device_manager.add_device("samx", value=motor_pos)
for model_name in monitor.sim.sim_get_models():
monitor.sim.sim_select_model(model_name)
monitor.sim.sim_params["noise_multipler"] = 10
monitor.sim.sim_params["ref_motor"] = "samx"
if "c" in monitor.sim.sim_params:
monitor.sim.sim_params["c"] = center
elif "center" in monitor.sim.sim_params:
monitor.sim.sim_params["center"] = center
assert isinstance(monitor.read()[monitor.name]["value"], monitor.BIT_DEPTH)
expected_value = monitor.sim._model.eval(monitor.sim._model_params, x=0)
expected_value = monitor.sim._model.eval(monitor.sim._model_params, x=motor_pos)
print(expected_value, monitor.read()[monitor.name]["value"])
tolerance = (
monitor.sim.sim_params["noise_multipler"] + 1

217
tests/test_smaract.py
View File

@ -1,217 +0,0 @@
from unittest import mock
import pytest
from utils import SocketMock
from ophyd_devices.smaract import SmaractController
from ophyd_devices.smaract.smaract_controller import SmaractCommunicationMode
from ophyd_devices.smaract.smaract_errors import SmaractCommunicationError, SmaractErrorCode
from ophyd_devices.smaract.smaract_ophyd import SmaractMotor
@pytest.fixture
def controller():
SmaractController._reset_controller()
controller = SmaractController(socket_cls=SocketMock, socket_host="dummy", socket_port=123)
controller.on()
controller.sock.flush_buffer()
yield controller
@pytest.fixture
def lsmarA():
SmaractController._reset_controller()
motor_a = SmaractMotor(
"A", name="lsmarA", host="mpc2680.psi.ch", port=8085, sign=1, socket_cls=SocketMock
)
motor_a.controller.on()
motor_a.controller.sock.flush_buffer()
motor_a.stage()
yield motor_a
@pytest.mark.parametrize(
"axis,position,get_message,return_msg",
[
(0, 50, b":GP0\n", b":P0,50000000\n"),
(1, 0, b":GP1\n", b":P1,0\n"),
(0, -50, b":GP0\n", b":P0,-50000000\n"),
(0, -50.23, b":GP0\n", b":P0,-50230000\n"),
],
)
def test_get_position(controller, axis, position, get_message, return_msg):
controller.sock.buffer_recv = return_msg
val = controller.get_position(axis)
assert val == position
assert controller.sock.buffer_put[0] == get_message
@pytest.mark.parametrize(
"axis,is_referenced,get_message,return_msg,exception",
[
(0, True, b":GPPK0\n", b":PPK0,1\n", None),
(1, True, b":GPPK1\n", b":PPK1,1\n", None),
(0, False, b":GPPK0\n", b":PPK0,0\n", None),
(200, False, b":GPPK0\n", b":PPK0,0\n", ValueError),
],
)
def test_axis_is_referenced(controller, axis, is_referenced, get_message, return_msg, exception):
controller.sock.buffer_recv = return_msg
if exception is not None:
with pytest.raises(exception):
val = controller.axis_is_referenced(axis)
else:
val = controller.axis_is_referenced(axis)
assert val == is_referenced
assert controller.sock.buffer_put[0] == get_message
@pytest.mark.parametrize(
"return_msg,exception,raised",
[
(b"false\n", SmaractCommunicationError, False),
(b":E0,1", SmaractErrorCode, True),
(b":E,1", SmaractCommunicationError, True),
(b":E,-1", SmaractCommunicationError, True),
],
)
def test_socket_put_and_receive_raises_exception(controller, return_msg, exception, raised):
controller.sock.buffer_recv = return_msg
with pytest.raises(exception):
controller.socket_put_and_receive(b"test", raise_if_not_status=True)
controller.sock.flush_buffer()
controller.sock.buffer_recv = return_msg
if raised:
with pytest.raises(exception):
controller.socket_put_and_receive(b"test")
else:
assert controller.socket_put_and_receive(b"test") == return_msg.split(b"\n")[0].decode()
@pytest.mark.parametrize(
"mode,get_message,return_msg", [(0, b":GCM\n", b":CM0\n"), (1, b":GCM\n", b":CM1\n")]
)
def test_communication_mode(controller, mode, get_message, return_msg):
controller.sock.buffer_recv = return_msg
val = controller.get_communication_mode()
assert controller.sock.buffer_put[0] == get_message
assert val == SmaractCommunicationMode(mode)
@pytest.mark.parametrize(
"is_moving,get_message,return_msg",
[
(0, b":GS0\n", b":S0,0\n"),
(1, b":GS0\n", b":S0,1\n"),
(1, b":GS0\n", b":S0,2\n"),
(0, b":GS0\n", b":S0,3\n"),
(1, b":GS0\n", b":S0,4\n"),
(0, b":GS0\n", b":S0,5\n"),
(0, b":GS0\n", b":S0,6\n"),
(1, b":GS0\n", b":S0,7\n"),
(0, b":GS0\n", b":S0,9\n"),
(0, [b":GS0\n", b":GS0\n"], [b":E0,0\n", b":S0,9"]),
],
)
def test_axis_is_moving(controller, is_moving, get_message, return_msg):
controller.sock.buffer_recv = return_msg
val = controller.is_axis_moving(0)
assert val == is_moving
if isinstance(controller.sock.buffer_put, list) and len(controller.sock.buffer_put) == 1:
controller.sock.buffer_put = controller.sock.buffer_put[0]
assert controller.sock.buffer_put == get_message
@pytest.mark.parametrize(
"sensor_id,axis,get_msg,return_msg",
[
(1, 0, b":GST0\n", b":ST0,1\n"),
(6, 0, b":GST0\n", b":ST0,6\n"),
(6, 1, b":GST1\n", b":ST1,6\n"),
],
)
def test_get_sensor_definition(controller, sensor_id, axis, get_msg, return_msg):
controller.sock.buffer_recv = return_msg
sensor = controller.get_sensor_type(axis)
assert sensor.type_code == sensor_id
@pytest.mark.parametrize(
"move_speed,axis,get_msg,return_msg",
[
(50, 0, b":SCLS0,50000000\n", b":E-1,0"),
(0, 0, b":SCLS0,0\n", b":E-1,0"),
(20.23, 1, b":SCLS1,20230000\n", b":E-1,0"),
],
)
def test_set_move_speed(controller, move_speed, axis, get_msg, return_msg):
controller.sock.buffer_recv = return_msg
controller.set_closed_loop_move_speed(axis, move_speed)
assert controller.sock.buffer_put[0] == get_msg
@pytest.mark.parametrize(
"pos,axis,hold_time,get_msg,return_msg",
[
(50, 0, None, b":MPA0,50000000,1000\n", b":E0,0"),
(0, 0, 800, b":MPA0,0,800\n", b":E0,0"),
(20.23, 1, None, b":MPA1,20230000,1000\n", b":E0,0"),
],
)
def test_move_axis_to_absolute_position(controller, pos, axis, hold_time, get_msg, return_msg):
controller.sock.buffer_recv = return_msg
if hold_time is not None:
controller.move_axis_to_absolute_position(axis, pos, hold_time=hold_time)
else:
controller.move_axis_to_absolute_position(axis, pos)
assert controller.sock.buffer_put[0] == get_msg
@pytest.mark.parametrize(
"pos,get_msg,return_msg",
[
(
50,
[b":GPPK0\n", b":MPA0,50000000,1000\n", b":GS0\n", b":GP0\n"],
[b":PPK0,1\n", b":E0,0\n", b":S0,0\n", b":P0,50000000\n"],
),
(
0,
[b":GPPK0\n", b":MPA0,0,1000\n", b":GS0\n", b":GP0\n"],
[b":PPK0,1\n", b":E0,0\n", b":S0,0\n", b":P0,0000000\n"],
),
(
20.23,
[b":GPPK0\n", b":MPA0,20230000,1000\n", b":GS0\n", b":GP0\n"],
[b":PPK0,1\n", b":E0,0\n", b":S0,0\n", b":P0,20230000\n"],
),
(
20.23,
[b":GPPK0\n", b":GPPK0\n", b":MPA0,20230000,1000\n", b":GS0\n", b":GP0\n"],
[b":S0,0\n", b":PPK0,1\n", b":E0,0\n", b":S0,0\n", b":P0,20230000\n"],
),
],
)
def test_move_axis(lsmarA, pos, get_msg, return_msg):
controller = lsmarA.controller
controller.sock.buffer_recv = return_msg
lsmarA.move(pos)
assert controller.sock.buffer_put == get_msg
@pytest.mark.parametrize("num_axes,get_msg,return_msg", [(1, [b":S0\n"], [b":E0,0"])])
def test_stop_axis(lsmarA, num_axes, get_msg, return_msg):
controller = lsmarA.controller
controller.sock.buffer_recv = return_msg
controller.stop_all_axes()
assert controller.sock.buffer_put == get_msg
def test_all_axes_referenced(lsmarA):
controller = lsmarA.controller
with mock.patch.object(controller, "axis_is_referenced", return_value=True) as mock_is_ref:
val = controller.all_axes_referenced()
assert val
mock_is_ref.assert_called_once_with(0)

332
tests/utils.py
View File

@ -1,332 +0,0 @@
from unittest import mock
from bec_lib.devicemanager import DeviceContainer
from bec_lib.tests.utils import ConnectorMock
class SocketMock:
"""Socket Mock. Used for testing"""
def __init__(self, host, port):
self.host = host
self.port = port
self.buffer_put = []
self.buffer_recv = [b""]
self.is_open = False
self.sock = None
self.open()
def connect(self):
"""Mock connect method"""
print(f"connecting to {self.host} port {self.port}")
def _put(self, msg_bytes):
"""Mock put method"""
self.buffer_put.append(msg_bytes)
print(self.buffer_put)
# pylint: disable=unused-argument
def _recv(self, buffer_length=1024):
"""Mock receive method"""
print(self.buffer_recv)
if isinstance(self.buffer_recv, list):
if len(self.buffer_recv) > 0:
ret_val = self.buffer_recv.pop(0)
else:
ret_val = b""
return ret_val
return self.buffer_recv
def _initialize_socket(self):
"""Mock initialize socket method"""
def put(self, msg):
"""Mock put method"""
return self._put(msg)
def receive(self, buffer_length=1024):
"""Mock receive method"""
return self._recv(buffer_length=buffer_length)
def open(self):
"""Mock open method"""
self._initialize_socket()
self.is_open = True
def close(self):
"""Mock close method"""
self.sock = None
self.is_open = False
def flush_buffer(self):
"""Mock flush buffer method"""
self.buffer_put = []
self.buffer_recv = ""
class MockPV:
"""
MockPV class
This class is used for mocking pyepics signals for testing purposes
"""
_fmtsca = "<PV '%(pvname)s', count=%(count)i, type=%(typefull)s, access=%(access)s>"
_fmtarr = "<PV '%(pvname)s', count=%(count)i/%(nelm)i, type=%(typefull)s, access=%(access)s>"
_fields = (
"pvname",
"value",
"char_value",
"status",
"ftype",
"chid",
"host",
"count",
"access",
"write_access",
"read_access",
"severity",
"timestamp",
"posixseconds",
"nanoseconds",
"precision",
"units",
"enum_strs",
"upper_disp_limit",
"lower_disp_limit",
"upper_alarm_limit",
"lower_alarm_limit",
"lower_warning_limit",
"upper_warning_limit",
"upper_ctrl_limit",
"lower_ctrl_limit",
)
def __init__(
self,
pvname,
callback=None,
form="time",
verbose=False,
auto_monitor=None,
count=None,
connection_callback=None,
connection_timeout=None,
access_callback=None,
):
self.pvname = pvname.strip()
self.form = form.lower()
self.verbose = verbose
self._auto_monitor = auto_monitor
self.ftype = None
self.connected = True
self.connection_timeout = connection_timeout
self._user_max_count = count
if self.connection_timeout is None:
self.connection_timeout = 3
self._args = {}.fromkeys(self._fields)
self._args["pvname"] = self.pvname
self._args["count"] = count
self._args["nelm"] = -1
self._args["type"] = "unknown"
self._args["typefull"] = "unknown"
self._args["access"] = "unknown"
self._args["status"] = 0
self.connection_callbacks = []
self.mock_data = 0
if connection_callback is not None:
self.connection_callbacks = [connection_callback]
self.access_callbacks = []
if access_callback is not None:
self.access_callbacks = [access_callback]
self.callbacks = {}
self._put_complete = None
self._monref = None # holder of data returned from create_subscription
self._monref_mask = None
self._conn_started = False
if isinstance(callback, (tuple, list)):
for i, thiscb in enumerate(callback):
if callable(thiscb):
self.callbacks[i] = (thiscb, {})
elif callable(callback):
self.callbacks[0] = (callback, {})
self.chid = None
self.context = mock.MagicMock()
self._cache_key = (pvname, form, self.context)
self._reference_count = 0
for conn_cb in self.connection_callbacks:
conn_cb(pvname=pvname, conn=True, pv=self)
for acc_cb in self.access_callbacks:
acc_cb(True, True, pv=self)
# pylint disable: unused-argument
def wait_for_connection(self, timeout=None):
"""Wait for connection"""
return self.connected
# pylint disable: unused-argument
def get_all_metadata_blocking(self, timeout):
"""Get all metadata blocking"""
md = self._args.copy()
md.pop("value", None)
return md
def get_all_metadata_callback(self, callback, *, timeout):
"""Get all metadata callback"""
def get_metadata_thread(pvname):
md = self.get_all_metadata_blocking(timeout=timeout)
callback(pvname, md)
get_metadata_thread(pvname=self.pvname)
# pylint disable: unused-argument
def put(
self, value, wait=False, timeout=None, use_complete=False, callback=None, callback_data=None
):
"""MOCK PV, put function"""
self.mock_data = value
if callback is not None:
callback(None, None, None)
# pylint: disable=unused-argument
def add_callback(self, callback=None, index=None, run_now=False, with_ctrlvars=True, **kw):
"""Add callback"""
return mock.MagicMock()
# pylint: disable=unused-argument
def get_with_metadata(
self,
count=None,
as_string=False,
as_numpy=True,
timeout=None,
with_ctrlvars=False,
form=None,
use_monitor=True,
as_namespace=False,
):
"""Get MOCKPV data together with metadata"""
return {"value": self.mock_data}
def get(
self,
count=None,
as_string=False,
as_numpy=True,
timeout=None,
with_ctrlvars=False,
use_monitor=True,
):
"""Get value from MOCKPV"""
data = self.get_with_metadata(
count=count,
as_string=as_string,
as_numpy=as_numpy,
timeout=timeout,
with_ctrlvars=with_ctrlvars,
use_monitor=use_monitor,
)
return data["value"] if data is not None else None
class DeviceMock:
"""Device Mock. Used for testing in combination with the DeviceManagerMock
Args:
name (str): name of the device
value (float, optional): initial value of the device. Defaults to 0.0.
Returns:
DeviceMock: DeviceMock object
"""
def __init__(self, name: str, value: float = 0.0):
self.name = name
self.read_buffer = value
self._config = {"deviceConfig": {"limits": [-50, 50]}, "userParameter": None}
self._read_only = False
self._enabled = True
def read(self):
"""Read method for DeviceMock"""
return {self.name: {"value": self.read_buffer}}
def readback(self):
"""Readback method for DeviceMock"""
return self.read_buffer
@property
def read_only(self) -> bool:
"""read only property"""
return self._read_only
@read_only.setter
def read_only(self, val: bool):
"""read only setter"""
self._read_only = val
@property
def enabled(self) -> bool:
"""enabled property"""
return self._enabled
@enabled.setter
def enabled(self, val: bool):
"""enabled setter"""
self._enabled = val
@property
def user_parameter(self):
"""user_parameter property"""
return self._config["userParameter"]
@property
def obj(self):
"""obj property"""
return self
class DMMock:
"""Mock for DeviceManager
The mocked DeviceManager creates a device containert and a connector.
"""
def __init__(self):
self.devices = DeviceContainer()
self.connector = ConnectorMock()
def add_device(self, name: str, value: float = 0.0):
"""Add device to the DeviceManagerMock"""
self.devices[name] = DeviceMock(name, value)
# #TODO check what is the difference to SynSignal!
# class MockSignal(Signal):
# """Can mock an OphydSignal"""
# def __init__(self, read_pv, *, string=False, name=None, parent=None, **kwargs):
# self.read_pv = read_pv
# self._string = bool(string)
# super().__init__(name=name, parent=parent, **kwargs)
# self._waited_for_connection = False
# self._subscriptions = []
# def wait_for_connection(self):
# self._waited_for_connection = True
# def subscribe(self, method, event_type, **kw):
# self._subscriptions.append((method, event_type, kw))
# def describe_configuration(self):
# return {self.name + "_conf": {"source": "SIM:test"}}
# def read_configuration(self):
# return {self.name + "_conf": {"value": 0}}