usov_i-org/bec
1
0
Fork 0
mirror of https://github.com/ivan-usov-org/bec.git synced 2025-04-22 02:20:02 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity

test: add and fixed tests for scan_to_csv and scan_export_cm

Browse Source
This commit is contained in:
appel_c 2024-01-18 08:18:00 +01:00
parent 07654ec043
commit 260ff38d62
2 changed files with 44 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
bec_lib/tests/test_core_utils.py
View File

@ -6,6 +6,7 @@ from unittest import mock
import pytest import pytest
from bec_lib import messages from bec_lib import messages
from bec_lib.scan_items import ScanItem
from bec_lib.scan_report import ScanReport from bec_lib.scan_report import ScanReport
from bec_lib.utils import _write_csv, scan_to_dict, scan_to_csv from bec_lib.utils import _write_csv, scan_to_dict, scan_to_csv
@ -27,36 +28,34 @@ def test__write_csv():
def test_scan_to_dict(): def test_scan_to_dict():
"""Test scan_to_dict function.""" """Test scan_to_dict function."""
input_dict = input_dict = create_scan_report() # input_dict = input_dict = create_scan_report()
scan_manager_mock = mock.MagicMock()
scan_item = ScanItem(
scan_manager=scan_manager_mock, queueID="test", scan_number=[1], scanID=["tmp"], status="OK"
)
for scan_msg in create_scan_messages().values():
scan_item.data.set(scan_msg.content["point_id"], scan_msg)
output_dict = { output_dict = {
"timestamp": defaultdict( "timestamp": {
list, "bpm4i": [1689340694.3702202, 1689340694.3702202, 1689340694.3702202],
{ "samx": [1689957983.5892477, 1689957983.6680737, 1689957983.747301],
"bpm4i": [1689340694.3702202, 1689340694.3702202, 1689340694.3702202], "samx_setpoint": [1689957983.5038617, 1689957983.607795, 1689957983.6842027],
"samx": [1689957983.5892477, 1689957983.6680737, 1689957983.747301], "samx_motor_is_moving": [1689957983.5894659, 1689957983.6683168, 1689957983.7475493],
"samx_setpoint": [1689957983.5038617, 1689957983.607795, 1689957983.6842027], },
"samx_motor_is_moving": [ "value": {
1689957983.5894659, "bpm4i": [0.8140756084557457, 0.980531184880259, 0.8360317021600272],
1689957983.6683168, "samx": [-2.997623536163973, 2.0203129364662855, 7.01317320459588],
1689957983.7475493, "samx_setpoint": [-2.9878629905097327, 2.0121370094902673, 7.012137009490267],
], "samx_motor_is_moving": [0, 0, 0],
}, },
),
"value": defaultdict(
list,
{
"bpm4i": [0.8140756084557457, 0.980531184880259, 0.8360317021600272],
"samx": [-2.997623536163973, 2.0203129364662855, 7.01317320459588],
"samx_setpoint": [-2.9878629905097327, 2.0121370094902673, 7.012137009490267],
"samx_motor_is_moving": [0, 0, 0],
},
),
} }
scanreport_mock = mock.MagicMock() scanreport_mock = mock.MagicMock()
scanreport_mock.scan.data.__getitem__ = input_dict.__getitem__ # scanreport_mock.scan.return_value = scan_item
scanreport_mock.scan.data.values.return_value = input_dict.values() scanreport_mock.scan = scan_item
# scanreport_mock.scan.data.__getitem__ = scan_item.data.__getitem__
# scanreport_mock.scan.data.values.return_value = scan_item.data.values()
return_dict = scan_to_dict(scanreport_mock, flat=True) return_dict = scan_to_dict(scanreport_mock, flat=True)
assert return_dict == output_dict assert return_dict == output_dict
@ -106,7 +105,7 @@ def test_scan_to_csv():
) )
def create_scan_report(): def create_scan_messages():
return { return {
0: messages.ScanMessage( 0: messages.ScanMessage(
**{ **{

20
bec_lib/tests/test_scan_context.py
View File

@ -3,7 +3,7 @@ from unittest import mock
import pytest import pytest
from bec_lib.device import DeviceBase from bec_lib.device import DeviceBase
from bec_lib.scans import DatasetIdOnHold, HideReport, Metadata, ScanDef, ScanGroup from bec_lib.scans import DatasetIdOnHold, HideReport, Metadata, ScanDef, ScanExport, ScanGroup
from bec_lib.tests.utils import bec_client from bec_lib.tests.utils import bec_client
# pylint: disable=no-member # pylint: disable=no-member
@ -89,6 +89,24 @@ def test_scan_group_cm(bec_client):
assert client.scans._scan_group is None assert client.scans._scan_group is None
@pytest.mark.parametrize("abort_on_ctrl_c", [True, False])
def test_scan_export_cm(abort_on_ctrl_c):
scan_export = ScanExport("temp")
scan_export._get_client = mock.MagicMock()
scan_export._get_client.return_value = mock_client = mock.MagicMock()
mock_client._service_config = mock_abort = mock.PropertyMock()
mock_abort.abort_on_ctrl_c = abort_on_ctrl_c
scan_export._export_to_csv = mock_to_csv = mock.MagicMock()
if not abort_on_ctrl_c:
with pytest.raises(RuntimeError):
with scan_export:
... # Do nothing
else:
with scan_export:
... # Do nothgin
assert mock_to_csv.call_count == 1
def test_parameter_bundler(bec_client): def test_parameter_bundler(bec_client):
client = bec_client client = bec_client
dev = client.device_manager.devices dev = client.device_manager.devices