72 lines
2.2 KiB
Python
72 lines
2.2 KiB
Python
from unittest import mock
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from bec_lib import messages
|
|
from bec_server.scan_server.tests.fixtures import *
|
|
|
|
from csaxs_bec.scans.owis_grid import OwisGrid
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"scan_msg",
|
|
[
|
|
messages.ScanQueueMessage(
|
|
scan_type="owis_grid",
|
|
parameter={
|
|
"args": {
|
|
"start_y": 0,
|
|
"end_y": 1,
|
|
"interval_y": 10,
|
|
"start_x": 0,
|
|
"end_x": 1,
|
|
"interval_x": 5,
|
|
},
|
|
"kwargs": {"exp_time": 0.1, "readout_time": 3e-3},
|
|
},
|
|
queue="primary",
|
|
metadata={"RID": "1234"},
|
|
)
|
|
],
|
|
)
|
|
def test_owis_grid(scan_msg, scan_assembler, ScanStubStatusMock):
|
|
request = scan_assembler(OwisGrid, *scan_msg.content["parameter"]["args"].values())
|
|
request.high_velocity = 10
|
|
request.high_acc_time = 0.2
|
|
request.base_velocity = 0.0625
|
|
|
|
def fake_done():
|
|
yield False
|
|
yield True
|
|
|
|
def fake_set(*args, **kwargs):
|
|
yield "fake_set"
|
|
return ScanStubStatusMock(done_func=fake_done)
|
|
|
|
with (
|
|
mock.patch.object(request.stubs, "set", side_effect=fake_set),
|
|
mock.patch.object(request.stubs, "_get_result_from_status"),
|
|
mock.patch.object(
|
|
request, "get_initial_motor_properties"
|
|
) as mock_get_init_motor_properties,
|
|
):
|
|
scan_instructions = list(request.run())
|
|
|
|
mock_get_init_motor_properties.assert_called_once()
|
|
assert request.point_id == scan_msg.content["parameter"]["args"]["interval_x"]
|
|
assert np.isclose(
|
|
request.target_velocity,
|
|
(
|
|
(
|
|
scan_msg.content["parameter"]["args"]["end_y"]
|
|
- scan_msg.content["parameter"]["args"]["start_y"]
|
|
)
|
|
/ scan_msg.content["parameter"]["args"]["interval_y"]
|
|
)
|
|
/ (
|
|
scan_msg.content["parameter"]["kwargs"]["exp_time"]
|
|
+ scan_msg.content["parameter"]["kwargs"]["readout_time"]
|
|
),
|
|
rtol=1e-2,
|
|
)
|