bec/bec_widgets
0
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2025-07-13 11:11:49 +02:00
Code Activity

test(scan-history): fix tests after message update

Browse Source
This commit is contained in:
appel_c
2025-07-10 08:13:08 +02:00
parent 2ae1c0e71c
commit e02c34eb87
1 changed files with 61 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
tests/unit_tests/test_scan_history_browser.py
View File

@ -1,7 +1,7 @@
from unittest import mock
import pytest
from bec_lib.messages import ScanHistoryMessage
from bec_lib.messages import ScanHistoryMessage, _StoredDataInfo
from pytestqt import qtbot
from qtpy import QtCore
@ -32,7 +32,14 @@ def scan_history_msg():
exit_status="closed",
num_points=10,
request_inputs={"some_input": "value"},
device_data_info={"device1": (10,), "device2": (20,), "device3": (1,)},
stored_data_info={
"device2": {
"device2_signal1": _StoredDataInfo(shape=(10,)),
"device2_signal2": _StoredDataInfo(shape=(20,)),
"device2_signal3": _StoredDataInfo(shape=(25,)),
},
"device3": {"device3_signal1": _StoredDataInfo(shape=(1,))},
},
)
@ -50,7 +57,21 @@ def scan_history_msg_2():
exit_status="closed",
num_points=5,
request_inputs={"some_input": "new_value"},
device_data_info={"device0": (15,), "device5": (25,), "device2313": (3,), "device2": (20,)},
stored_data_info={
"device0": {
"device0_signal1": _StoredDataInfo(shape=(15,)),
"device0_signal2": _StoredDataInfo(shape=(25,)),
"device0_signal3": _StoredDataInfo(shape=(3,)),
"device0_signal4": _StoredDataInfo(shape=(20,)),
},
"device2": {
"device2_signal1": _StoredDataInfo(shape=(10,)),
"device2_signal2": _StoredDataInfo(shape=(20,)),
"device2_signal3": _StoredDataInfo(shape=(25,)),
"device2_signal4": _StoredDataInfo(shape=(30,)),
},
"device1": {"device1_signal1": _StoredDataInfo(shape=(25,))},
},
)
@ -93,38 +114,40 @@ def test_scan_history_device_viewer_receive_msg(
"""Test updating devices from scan history."""
# Update with first scan history message
assert scan_history_device_viewer.scan_history_msg is None
assert scan_history_device_viewer.device_model.devices == []
assert scan_history_device_viewer.device_model.rowCount() == 0
assert scan_history_device_viewer.signal_model.signals == []
assert scan_history_device_viewer.signal_model.rowCount() == 0
scan_history_device_viewer.update_devices_from_scan_history(scan_history_msg)
assert scan_history_device_viewer.scan_history_msg == scan_history_msg
assert scan_history_device_viewer.device_model.devices == [
("device2", (20,)),
("device1", (10,)),
("device3", (1,)),
assert scan_history_device_viewer.device_combo.currentText() == "device2"
assert scan_history_device_viewer.signal_model.signals == [
("device2_signal3", _StoredDataInfo(shape=(25,))),
("device2_signal2", _StoredDataInfo(shape=(20,))),
("device2_signal1", _StoredDataInfo(shape=(10,))),
]
current_index = scan_history_device_viewer.device_combo.currentIndex()
current_index = scan_history_device_viewer.signal_combo.currentIndex()
assert current_index == 0
device_name = scan_history_device_viewer.device_combo.model().data(
scan_history_device_viewer.device_combo.model().index(current_index, 0), QtCore.Qt.UserRole
signal_name = scan_history_device_viewer.signal_combo.model().data(
scan_history_device_viewer.signal_combo.model().index(current_index, 0), QtCore.Qt.UserRole
)
assert device_name == "device2"
assert signal_name == "device2_signal3"
## Update of second message should not change the device if still available
new_msg = scan_history_msg_2
scan_history_device_viewer.update_devices_from_scan_history(new_msg)
assert scan_history_device_viewer.scan_history_msg == new_msg
assert scan_history_device_viewer.device_model.devices == [
("device5", (25,)),
("device2", (20,)),
("device0", (15,)),
("device2313", (3,)),
assert scan_history_device_viewer.signal_model.signals == [
("device2_signal4", _StoredDataInfo(shape=(30,))),
("device2_signal3", _StoredDataInfo(shape=(25,))),
("device2_signal2", _StoredDataInfo(shape=(20,))),
("device2_signal1", _StoredDataInfo(shape=(10,))),
]
current_index = scan_history_device_viewer.device_combo.currentIndex()
assert scan_history_device_viewer.device_combo.currentText() == "device2"
current_index = scan_history_device_viewer.signal_combo.currentIndex()
assert current_index == 1
device_name = scan_history_device_viewer.device_combo.model().data(
scan_history_device_viewer.device_combo.model().index(current_index, 0), QtCore.Qt.UserRole
signal_name = scan_history_device_viewer.signal_combo.model().data(
scan_history_device_viewer.signal_combo.model().index(current_index, 0), QtCore.Qt.UserRole
)
assert device_name == "device2"
assert signal_name == "device2_signal3"
def test_scan_history_device_viewer_clear_view(qtbot, scan_history_device_viewer, scan_history_msg):
@ -142,9 +165,10 @@ def test_scan_history_device_viewer_on_request_plotting_clicked(
"""Test the request plotting button click."""
scan_history_device_viewer.update_devices_from_scan_history(scan_history_msg)
def plotting_callback(device_name, msg):
def plotting_callback(device_name, signal_name, msg):
"""Callback to check if the request plotting signal is emitted."""
assert device_name == "device2"
assert msg == scan_history_msg
scan_history_device_viewer.request_history_plot.connect(plotting_callback)
@ -263,12 +287,12 @@ def test_scan_history_browser(qtbot, scan_history_browser, scan_history_msg, sca
qtbot.waitUntil(
lambda: scan_history_browser.scan_history_metadata_viewer.scan_history_msg
== scan_history_msg_2,
timeout=1000,
timeout=5000,
)
qtbot.waitUntil(
lambda: scan_history_browser.scan_history_device_viewer.scan_history_msg
== scan_history_msg_2,
timeout=1000,
timeout=5000,
)
# Click on second scan item history to select it
@ -284,20 +308,20 @@ def test_scan_history_browser(qtbot, scan_history_browser, scan_history_msg, sca
qtbot.waitUntil(
lambda: scan_history_browser.scan_history_metadata_viewer.scan_history_msg
== scan_history_msg,
timeout=1000,
timeout=5000,
)
qtbot.waitUntil(
lambda: scan_history_browser.scan_history_device_viewer.scan_history_msg
== scan_history_msg,
timeout=1000,
timeout=5000,
)
callback_args = []
def plotting_callback(device_name, msg):
def plotting_callback(device_name, signal_name, msg):
"""Callback to check if the request plotting signal is emitted."""
# device_name should be the first device
callback_args.append((device_name, msg))
callback_args.append((device_name, signal_name, msg))
scan_history_browser.scan_history_device_viewer.request_history_plot.connect(plotting_callback)
# Test emit plotting request
@ -305,9 +329,12 @@ def test_scan_history_browser(qtbot, scan_history_browser, scan_history_msg, sca
scan_history_browser.scan_history_device_viewer.request_plotting_button,
QtCore.Qt.LeftButton,
)
qtbot.waitUntil(lambda: len(callback_args) > 0, timeout=1000)
assert callback_args[0][1] == scan_history_msg
assert callback_args[0][0] in callback_args[0][1].device_data_info.keys()
qtbot.waitUntil(lambda: len(callback_args) > 0, timeout=5000)
assert callback_args[0][2] == scan_history_msg
device_name = callback_args[0][0]
signal_name = callback_args[0][1]
assert device_name in scan_history_msg.stored_data_info.keys()
assert signal_name in scan_history_msg.stored_data_info[device_name].keys()
# Test clearing the view, removing both scans
scan_history_browser.scan_history_view.remove_scan(-1)
@ -318,9 +345,9 @@ def test_scan_history_browser(qtbot, scan_history_browser, scan_history_msg, sca
qtbot.waitUntil(
lambda: scan_history_browser.scan_history_metadata_viewer.scan_history_msg is None,
timeout=1000,
timeout=5000,
)
qtbot.waitUntil(
lambda: scan_history_browser.scan_history_device_viewer.scan_history_msg is None,
timeout=1000,
timeout=5000,
)