bec/csaxs_bec
0
0
Fork 2
Code Issues 17 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

test: fix instruction message for lamni scan

Browse Source
This commit is contained in:
appel_c
2026-04-21 16:11:48 +02:00
parent 3a249557e9
commit 5917cc2b1d
3 changed files with 53 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/tests_devices/conftest.py
+3
View File
@@ -0,0 +1,3 @@
import os
os.environ.setdefault("OPHYD_CONTROL_LAYER", "dummy")
tests/tests_devices/test_eiger.py
+1 -1
View File
@@ -1,7 +1,7 @@
# pylint: skip-file
import os
import threading
from time import time
import time
from typing import TYPE_CHECKING, Generator
from unittest import mock
tests/tests_scans/test_lamni_fermat_scan.py
+49 -114
View File
@@ -52,16 +52,7 @@ def device_manager_mock():
metadata={"RID": "1234"},
),
[
messages.DeviceInstructionMessage(
metadata={
"readout_priority": "monitored",
"RID": "1234",
"device_instr_id": "diid",
},
device=["rtx", "rty"],
action="read",
parameter={},
),
None,
messages.DeviceInstructionMessage(
metadata={
"readout_priority": "monitored",
@@ -261,6 +252,12 @@ def device_manager_mock():
"kwargs": {},
},
),
messages.DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "RID": "1234"},
device=None,
action="scan_report_instruction",
parameter={"device_progress": ["rt_positions"]},
),
messages.DeviceInstructionMessage(
metadata={"readout_priority": "monitored", "RID": "1234"},
device=None,
@@ -297,26 +294,6 @@ def device_manager_mock():
action="read",
parameter={},
),
messages.DeviceInstructionMessage(
metadata={
"readout_priority": "monitored",
"RID": "1234",
"device_instr_id": "diid",
},
device="rtx",
action="set",
parameter={"value": 1.3681828686580249},
),
messages.DeviceInstructionMessage(
metadata={
"readout_priority": "monitored",
"RID": "1234",
"device_instr_id": "diid",
},
device="rty",
action="set",
parameter={"value": 2.1508313829565293},
),
messages.DeviceInstructionMessage(
metadata={
"readout_priority": "monitored",
@@ -333,19 +310,9 @@ def device_manager_mock():
"RID": "1234",
"device_instr_id": "diid",
},
device="rtx",
action="set",
parameter={"value": 1.3681828686580249},
),
messages.DeviceInstructionMessage(
metadata={
"readout_priority": "monitored",
"RID": "1234",
"device_instr_id": "diid",
},
device="rty",
action="set",
parameter={"value": 2.1508313829565293},
device="rt_positions",
action="kickoff",
parameter={"configure": {}},
),
None,
messages.DeviceInstructionMessage(
@@ -359,48 +326,7 @@ def device_manager_mock():
action="read",
parameter={"group": "monitored"},
),
messages.DeviceInstructionMessage(
metadata={
"readout_priority": "monitored",
"RID": "1234",
"device_instr_id": "diid",
},
device="rtx",
action="set",
parameter={"value": -0.7700589354581364},
),
messages.DeviceInstructionMessage(
metadata={
"readout_priority": "monitored",
"RID": "1234",
"device_instr_id": "diid",
},
device="rty",
action="set",
parameter={"value": -0.8406005210092851},
),
None,
messages.DeviceInstructionMessage(
metadata={
"readout_priority": "monitored",
"RID": "1234",
"point_id": 1,
"device_instr_id": "diid",
},
device=["bpm4i"],
action="read",
parameter={"group": "monitored"},
),
messages.DeviceInstructionMessage(
metadata={
"readout_priority": "monitored",
"RID": "1234",
"device_instr_id": "diid",
},
device=["bpm4i", "lsamx", "lsamy", "samx", "samy"],
action="complete",
parameter={},
),
messages.DeviceInstructionMessage(
metadata={"device_instr_id": "diid"},
device=["bpm4i", "lsamx", "lsamy", "samx", "samy"],
@@ -417,7 +343,7 @@ def device_manager_mock():
)
],
)
def test_LamNIFermatScan(scan_msg, reference_scan_list, scan_assembler):
def test_LamNIFermatScan(scan_msg, reference_scan_list, scan_assembler, ScanStubStatusMock):
scan = scan_assembler(
LamNIFermatScan,
parameter=scan_msg.content.get("parameter"),
@@ -425,38 +351,47 @@ def test_LamNIFermatScan(scan_msg, reference_scan_list, scan_assembler):
**scan_msg.content["parameter"]["kwargs"],
)
def fake_done():
yield False
yield True
def fake_complete(*args, **kwargs):
yield None
return ScanStubStatusMock(done_func=fake_done)
with mock.patch.object(scan.stubs, "_get_result_from_status", return_value=0):
with mock.patch.object(scan, "_check_min_positions") as check_min_pos:
scan_instructions = list(scan.run())
check_min_pos.assert_called_once()
with mock.patch.object(scan.stubs, "complete", side_effect=fake_complete):
scan_instructions = list(scan.run())
check_min_pos.assert_called_once()
for ii, instr in enumerate(scan_instructions):
if instr is None:
continue
if instr.metadata.get("scan_id") is not None:
instr.metadata["scan_id"] = "scan_id"
if instr.metadata.get("RID") is not None:
instr.metadata["RID"] = scan.metadata.get("RID")
if instr.metadata.get("device_instr_id") is not None:
instr.metadata["device_instr_id"] = "diid"
if instr.content["action"] == "rpc":
instr.content["parameter"]["rpc_id"] = "rpc_id"
if instr.content["parameter"].get("value"):
assert np.isclose(
instr.content["parameter"].get("value"),
reference_scan_list[ii].content["parameter"].get("value"),
)
instr.content["parameter"]["value"] = reference_scan_list[ii].content["parameter"][
"value"
]
if instr.content["parameter"].get("positions"):
assert np.isclose(
instr.content["parameter"].get("positions"),
reference_scan_list[ii].content["parameter"].get("positions"),
).all()
instr.content["parameter"]["positions"] = reference_scan_list[ii].content[
"parameter"
]["positions"]
for ii, instr in enumerate(scan_instructions):
if instr is None:
continue
if instr.metadata.get("scan_id") is not None:
instr.metadata["scan_id"] = "scan_id"
if instr.metadata.get("RID") is not None:
instr.metadata["RID"] = scan.metadata.get("RID")
if instr.metadata.get("device_instr_id") is not None:
instr.metadata["device_instr_id"] = "diid"
if instr.content["action"] == "rpc":
instr.content["parameter"]["rpc_id"] = "rpc_id"
if instr.content["parameter"].get("value"):
assert np.isclose(
instr.content["parameter"].get("value"),
reference_scan_list[ii].content["parameter"].get("value"),
)
instr.content["parameter"]["value"] = reference_scan_list[ii].content[
"parameter"
]["value"]
if instr.content["parameter"].get("positions"):
assert np.isclose(
instr.content["parameter"].get("positions"),
reference_scan_list[ii].content["parameter"].get("positions"),
).all()
instr.content["parameter"]["positions"] = reference_scan_list[ii].content[
"parameter"
]["positions"]
assert scan_instructions == reference_scan_list