bec/ophyd_devices
0
0
Fork 0
mirror of https://github.com/bec-project/ophyd_devices.git synced 2026-01-16 15:29:20 +01:00
Code Issues Packages Projects Releases Wiki Activity

refactor: renamed scanID to scan_id

Browse Source
This commit is contained in:
wakonig_k
2024-03-13 21:09:12 +01:00
parent 46063b905b
commit 1c7737ccda
15 changed files with 226 additions and 588 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
tests/test_delay_generator_csaxs.py
View File

@@ -1,7 +1,8 @@
# pylint: skip-file
import pytest
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
@@ -26,7 +27,7 @@ def mock_DDGSetup():
@pytest.fixture(
params=[
{
"scanID": "1234",
"scan_id": "1234",
"scan_type": "step",
"num_points": 500,
"frames_per_trigger": 1,
@@ -34,7 +35,7 @@ def mock_DDGSetup():
"readout_time": 0.1,
},
{
"scanID": "1234",
"scan_id": "1234",
"scan_type": "step",
"num_points": 500,
"frames_per_trigger": 5,
@@ -42,7 +43,7 @@ def mock_DDGSetup():
"readout_time": 0,
},
{
"scanID": "1234",
"scan_id": "1234",
"scan_type": "fly",
"num_points": 500,
"frames_per_trigger": 1,
@@ -50,7 +51,7 @@ def mock_DDGSetup():
"readout_time": 0.2,
},
{
"scanID": "1234",
"scan_id": "1234",
"scan_type": "fly",
"num_points": 500,
"frames_per_trigger": 5,
@@ -135,8 +136,8 @@ def channel_pairs(request):
return request.param
def test_check_scanID(mock_DDGSetup, scaninfo, ddg_config_defaults, ddg_config_scan):
"""Test the check_scanID method."""
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)
@@ -145,12 +146,12 @@ def test_check_scanID(mock_DDGSetup, scaninfo, ddg_config_defaults, ddg_config_s
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_scanID()
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_scanID method."""
"""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)
@@ -185,11 +186,7 @@ def test_on_trigger(mock_DDGSetup, scaninfo, ddg_config_defaults, ddg_config_sca
def test_initialize_default_parameter(
mock_DDGSetup,
scaninfo,
ddg_config_defaults,
ddg_config_scan,
channel_pairs,
mock_DDGSetup, scaninfo, ddg_config_defaults, ddg_config_scan, channel_pairs
):
"""Test the initialize_default_parameter method."""
# Set first attributes of parent class
@@ -229,13 +226,7 @@ def test_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,
):
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():

136
tests/test_eiger9m_csaxs.py
View File

@@ -1,13 +1,12 @@
# pylint: skip-file
import pytest
import threading
from unittest import mock
import ophyd
import pytest
from bec_lib import MessageEndpoints, messages
from bec_lib import messages, MessageEndpoints
from ophyd_devices.epics.devices.eiger9m_csaxs import Eiger9McSAXS
from tests.utils import DMMock, MockPV
@@ -72,26 +71,9 @@ def test_init():
@pytest.mark.parametrize(
"trigger_source, detector_state, expected_exception",
[
(
2,
1,
True,
),
(
2,
0,
False,
),
],
"trigger_source, detector_state, expected_exception", [(2, 1, True), (2, 0, False)]
)
def test_initialize_detector(
mock_det,
trigger_source,
detector_state,
expected_exception,
):
def test_initialize_detector(mock_det, trigger_source, detector_state, expected_exception):
"""Test the _init function:
This includes testing the functions:
@@ -125,13 +107,7 @@ def test_trigger(mock_det):
@pytest.mark.parametrize(
"readout_time, expected_value",
[
(1e-3, 3e-3),
(3e-3, 3e-3),
(5e-3, 5e-3),
(None, 3e-3),
],
"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:
@@ -146,34 +122,10 @@ def test_update_readout_time(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,
),
("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(
@@ -215,7 +167,7 @@ def test_initialize_detector_backend(
"num_points": 500,
"frames_per_trigger": 1,
"filepath": "test.h5",
"scanID": "123",
"scan_id": "123",
"mokev": 12.4,
},
{"state": "READY"},
@@ -230,7 +182,7 @@ def test_initialize_detector_backend(
"num_points": 500,
"frames_per_trigger": 1,
"filepath": "test.h5",
"scanID": "123",
"scan_id": "123",
"mokev": 12.4,
},
{"state": "BUSY"},
@@ -245,7 +197,7 @@ def test_initialize_detector_backend(
"num_points": 500,
"frames_per_trigger": 1,
"filepath": "test.h5",
"scanID": "123",
"scan_id": "123",
"mokev": 18.4,
},
{"state": "READY"},
@@ -257,13 +209,7 @@ def test_initialize_detector_backend(
],
)
def test_stage(
mock_det,
scaninfo,
daq_status,
daq_cfg,
detector_state,
stopped,
expected_exception,
mock_det, scaninfo, daq_status, daq_cfg, detector_state, stopped, expected_exception
):
with mock.patch.object(
mock_det.custom_prepare, "std_client"
@@ -309,7 +255,7 @@ def test_stage(
"num_points": 500,
"frames_per_trigger": 1,
"filepath": "test.h5",
"scanID": "123",
"scan_id": "123",
},
{"state": "BUSY", "acquisition": {"state": "WAITING_IMAGES"}},
False,
@@ -320,7 +266,7 @@ def test_stage(
"num_points": 500,
"frames_per_trigger": 1,
"filepath": "test.h5",
"scanID": "123",
"scan_id": "123",
},
{"state": "BUSY", "acquisition": {"state": "WAITING_IMAGES"}},
False,
@@ -331,7 +277,7 @@ def test_stage(
"num_points": 500,
"frames_per_trigger": 1,
"filepath": "test.h5",
"scanID": "123",
"scan_id": "123",
},
{"state": "BUSY", "acquisition": {"state": "ERROR"}},
True,
@@ -373,24 +319,8 @@ def test_prepare_detector_backend(mock_det, scaninfo, daq_status, expected_excep
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,
):
@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:
@@ -416,13 +346,13 @@ def test_stop_detector_backend(mock_det):
@pytest.mark.parametrize(
"scaninfo",
[
({"filepath": "test.h5", "successful": True, "done": False, "scanID": "123"}),
({"filepath": "test.h5", "successful": False, "done": True, "scanID": "123"}),
({"filepath": "test.h5", "successful": None, "done": True, "scanID": "123"}),
({"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.scanID = scaninfo["scanID"]
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"]
@@ -435,7 +365,7 @@ def test_publish_file_location(mock_det, scaninfo):
)
expected_calls = [
mock.call(
MessageEndpoints.public_file(scaninfo["scanID"], mock_det.name),
MessageEndpoints.public_file(scaninfo["scan_id"], mock_det.name),
msg,
pipe=mock_det.connector.pipeline.return_value,
),
@@ -465,40 +395,28 @@ def test_stop(mock_det):
[
(
False,
{
"num_points": 500,
"frames_per_trigger": 4,
},
{"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,
},
{"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,
},
{"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,
},
{"num_points": 500, "frames_per_trigger": 1},
0,
{"acquisition": {"state": "FINISHED", "stats": {"n_write_completed": 500}}},
False,

65
tests/test_falcon_csaxs.py
View File

@@ -1,14 +1,14 @@
# pylint: skip-file
import pytest
from unittest import mock
import threading
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
from bec_lib import messages, MessageEndpoints
def patch_dual_pvs(device):
@@ -47,10 +47,7 @@ def mock_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),
],
[(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(
@@ -94,13 +91,7 @@ def test_init_detector(
@pytest.mark.parametrize(
"readout_time, expected_value",
[
(1e-3, 3e-3),
(3e-3, 3e-3),
(5e-3, 5e-3),
(None, 3e-3),
],
"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:
@@ -131,10 +122,10 @@ def test_initialize_default_parameter(mock_det):
"frames_per_trigger": 1,
"exp_time": 0.1,
"filepath": "test.h5",
"scanID": "123",
"scan_id": "123",
"mokev": 12.4,
}
),
)
],
)
def test_stage(mock_det, scaninfo):
@@ -203,13 +194,13 @@ def test_prepare_data_backend(mock_det, scaninfo):
@pytest.mark.parametrize(
"scaninfo",
[
({"filepath": "test.h5", "successful": True, "done": False, "scanID": "123"}),
({"filepath": "test.h5", "successful": False, "done": True, "scanID": "123"}),
({"filepath": "test.h5", "successful": None, "done": True, "scanID": "123"}),
({"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.scanID = scaninfo["scanID"]
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"]
@@ -222,7 +213,7 @@ def test_publish_file_location(mock_det, scaninfo):
)
expected_calls = [
mock.call(
MessageEndpoints.public_file(scaninfo["scanID"], mock_det.name),
MessageEndpoints.public_file(scaninfo["scan_id"], mock_det.name),
msg,
pipe=mock_det.connector.pipeline.return_value,
),
@@ -235,13 +226,7 @@ def test_publish_file_location(mock_det, scaninfo):
assert mock_det.connector.set_and_publish.call_args_list == expected_calls
@pytest.mark.parametrize(
"detector_state, expected_exception",
[
(1, False),
(0, True),
],
)
@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
@@ -261,24 +246,8 @@ def test_trigger(mock_det):
mock_on_trigger.assert_called_once()
@pytest.mark.parametrize(
"stopped, expected_abort",
[
(
False,
False,
),
(
True,
True,
),
],
)
def test_unstage(
mock_det,
stopped,
expected_abort,
):
@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:

29
tests/test_grashopper_tomcat.py
View File

@@ -1,20 +1,21 @@
# pylint: skip-file
import pytest
from unittest import mock
import pytest
from ophyd_devices.epics.devices.grashopper_tomcat import (
GrashopperTOMCATSetup,
AutoMode,
ImageMode,
DetectorState,
ImageBinning,
VideoMode,
PixelFormat,
COLORMODE,
TriggerSource,
MemoryPolling,
AutoMode,
DetectorState,
GrashopperError,
GrashopperTimeoutError,
GrashopperTOMCATSetup,
ImageBinning,
ImageMode,
MemoryPolling,
PixelFormat,
TriggerSource,
VideoMode,
)
@@ -38,7 +39,7 @@ def mock_GrashopperSetup():
@pytest.fixture(
params=[
{
"scanID": "1234",
"scan_id": "1234",
"scan_type": "step",
"num_points": 500,
"frames_per_trigger": 1,
@@ -46,7 +47,7 @@ def mock_GrashopperSetup():
"readout_time": 0.1,
},
{
"scanID": "1234",
"scan_id": "1234",
"scan_type": "step",
"num_points": 500,
"frames_per_trigger": 5,
@@ -54,7 +55,7 @@ def mock_GrashopperSetup():
"readout_time": 0,
},
{
"scanID": "1234",
"scan_id": "1234",
"scan_type": "fly",
"num_points": 500,
"frames_per_trigger": 1,
@@ -62,7 +63,7 @@ def mock_GrashopperSetup():
"readout_time": 0.2,
},
{
"scanID": "1234",
"scan_id": "1234",
"scan_type": "fly",
"num_points": 500,
"frames_per_trigger": 5,

109
tests/test_mcs_card.py
View File

@@ -1,8 +1,10 @@
# pylint: skip-file
import pytest
from unittest import mock
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,
@@ -11,9 +13,7 @@ from ophyd_devices.epics.devices.mcs_csaxs import (
ReadoutMode,
TriggerSource,
)
from tests.utils import DMMock, MockPV
from bec_lib import messages, MessageEndpoints
def patch_dual_pvs(device):
@@ -89,15 +89,11 @@ def test_init():
"count_on_start": 0,
"stop_all": 1,
},
),
)
],
)
def test_initialize_detector(
mock_det,
trigger_source,
channel_advance,
channel_source1,
pv_channels,
mock_det, trigger_source, channel_advance, channel_source1, pv_channels
):
"""Test the _init function:
@@ -145,10 +141,7 @@ def test_progress_update(mock_det, value, num_lines, num_points, done):
@pytest.mark.parametrize(
"values, expected_nothing",
[
([[100, 120, 140], [200, 220, 240], [300, 320, 340]], False),
([100, 200, 300], True),
],
[([[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:
@@ -171,15 +164,15 @@ def test_on_mca_data(mock_det, values, expected_nothing):
"metadata, mca_data",
[
(
{"scanID": 123},
{"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.scanID = metadata["scanID"]
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
@@ -187,7 +180,7 @@ def test_send_data_to_bec(mock_det, metadata, mca_data):
data = messages.DeviceMessage(signals=dict(mca_data), metadata=device_metadata)
calls = mock.call(
topic=MessageEndpoints.device_async_readback(
scanID=metadata["scanID"], device=mock_det.name
scan_id=metadata["scan_id"], device=mock_det.name
),
msg={"data": data},
expire=1800,
@@ -200,54 +193,32 @@ def test_send_data_to_bec(mock_det, metadata, mca_data):
"scaninfo, triggersource, stopped, expected_exception",
[
(
{
"num_points": 500,
"frames_per_trigger": 1,
"scan_type": "step",
},
{"num_points": 500, "frames_per_trigger": 1, "scan_type": "step"},
TriggerSource.MODE3,
False,
False,
),
(
{
"num_points": 500,
"frames_per_trigger": 1,
"scan_type": "fly",
},
{"num_points": 500, "frames_per_trigger": 1, "scan_type": "fly"},
TriggerSource.MODE3,
False,
False,
),
(
{
"num_points": 5001,
"frames_per_trigger": 2,
"scan_type": "step",
},
{"num_points": 5001, "frames_per_trigger": 2, "scan_type": "step"},
TriggerSource.MODE3,
False,
True,
),
(
{
"num_points": 500,
"frames_per_trigger": 2,
"scan_type": "random",
},
{"num_points": 500, "frames_per_trigger": 2, "scan_type": "random"},
TriggerSource.MODE3,
False,
True,
),
],
)
def test_stage(
mock_det,
scaninfo,
triggersource,
stopped,
expected_exception,
):
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"]
@@ -290,24 +261,8 @@ def test_prepare_detector_backend(mock_det):
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,
):
@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:
@@ -342,30 +297,10 @@ def test_stop(mock_det):
@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,
),
(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):

103
tests/test_pilatus_csaxs.py
View File

@@ -1,14 +1,13 @@
# pylint: skip-file
import os
import pytest
import threading
from unittest import mock
import ophyd
import pytest
from bec_lib import MessageEndpoints, messages
from bec_lib import messages, MessageEndpoints
from ophyd_devices.epics.devices.pilatus_csaxs import PilatuscSAXS
from tests.utils import DMMock, MockPV
@@ -45,18 +44,9 @@ def mock_det():
yield det
@pytest.mark.parametrize(
"trigger_source, detector_state",
[
(1, 0),
],
)
@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,
):
def test_init_detector(mock_det, trigger_source, detector_state):
"""Test the _init function:
This includes testing the functions:
@@ -82,7 +72,7 @@ def test_init_detector(
"num_points": 500,
"frames_per_trigger": 1,
"filepath": "test.h5",
"scanID": "123",
"scan_id": "123",
"mokev": 12.4,
},
False,
@@ -94,7 +84,7 @@ def test_init_detector(
"num_points": 500,
"frames_per_trigger": 1,
"filepath": "test.h5",
"scanID": "123",
"scan_id": "123",
"mokev": 12.4,
},
True,
@@ -102,12 +92,7 @@ def test_init_detector(
),
],
)
def test_stage(
mock_det,
scaninfo,
stopped,
expected_exception,
):
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:
@@ -145,13 +130,7 @@ def test_pre_scan(mock_det):
@pytest.mark.parametrize(
"readout_time, expected_value",
[
(1e-3, 3e-3),
(3e-3, 3e-3),
(5e-3, 5e-3),
(None, 3e-3),
],
"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:
@@ -172,7 +151,7 @@ def test_update_readout_time(mock_det, readout_time, expected_value):
"filepath_raw": "test5_raw.h5",
"successful": True,
"done": False,
"scanID": "123",
"scan_id": "123",
}
),
(
@@ -181,7 +160,7 @@ def test_update_readout_time(mock_det, readout_time, expected_value):
"filepath_raw": "test5_raw.h5",
"successful": False,
"done": True,
"scanID": "123",
"scan_id": "123",
}
),
(
@@ -190,13 +169,13 @@ def test_update_readout_time(mock_det, readout_time, expected_value):
"filepath_raw": "test5_raw.h5",
"successful": None,
"done": True,
"scanID": "123",
"scan_id": "123",
}
),
],
)
def test_publish_file_location(mock_det, scaninfo):
mock_det.scaninfo.scanID = scaninfo["scanID"]
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(
@@ -217,7 +196,7 @@ def test_publish_file_location(mock_det, scaninfo):
)
expected_calls = [
mock.call(
MessageEndpoints.public_file(scaninfo["scanID"], mock_det.name),
MessageEndpoints.public_file(scaninfo["scan_id"], mock_det.name),
msg,
pipe=mock_det.connector.pipeline.return_value,
),
@@ -306,14 +285,7 @@ def test_stop_detector_backend(mock_det, requests_state, expected_exception, url
"user": "e12345",
},
],
[
"zmqWriter",
"e12345",
{
"frmCnt": 500,
"timeout": 2000,
},
],
["zmqWriter", "e12345", {"frmCnt": 500, "timeout": 2000}],
],
[
"http://x12sa-pd-2:8080/stream/pilatus_2",
@@ -357,14 +329,7 @@ def test_stop_detector_backend(mock_det, requests_state, expected_exception, url
"user": "e12345",
},
],
[
"zmqWriter",
"e12345",
{
"frmCnt": 500,
"timeout": 2000,
},
],
["zmqWriter", "e12345", {"frmCnt": 500, "timeout": 2000}],
],
[
"http://x12sa-pd-2:8080/stream/pilatus_2",
@@ -435,24 +400,8 @@ def test_prep_file_writer(mock_det, scaninfo, data_msgs, urls, requests_state, e
assert call == mock_call
@pytest.mark.parametrize(
"stopped, expected_exception",
[
(
False,
False,
),
(
True,
True,
),
],
)
def test_unstage(
mock_det,
stopped,
expected_exception,
):
@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:
@@ -485,21 +434,9 @@ def test_stop(mock_det):
@pytest.mark.parametrize(
"stopped, mcs_stage_state, expected_exception",
[
(
False,
ophyd.Staged.no,
False,
),
(
True,
ophyd.Staged.no,
True,
),
(
False,
ophyd.Staged.yes,
True,
),
(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):