mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-08-10 12:10:37 +02:00
1810 lines
72 KiB
Python
1810 lines
72 KiB
Python
# pylint: disable = no-name-in-module,missing-class-docstring, missing-module-docstring
|
|
from threading import Event, get_ident
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, call, patch
|
|
|
|
import pytest
|
|
from bec_lib.endpoints import MessageEndpoints
|
|
from bec_lib.messages import AvailableResourceMessage, ScanHistoryMessage
|
|
from bec_lib.scan_history import ScanHistory
|
|
from qtpy.QtCore import QModelIndex, QPoint, Qt
|
|
from qtpy.QtWidgets import QCheckBox, QDialog, QStyle
|
|
|
|
from bec_widgets.utils.forms_from_types.items import StrFormItem
|
|
from bec_widgets.utils.widget_io import WidgetIO
|
|
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
|
|
from bec_widgets.widgets.control.scan_control import ScanControl
|
|
from bec_widgets.widgets.control.scan_control import scan_control as scan_control_module
|
|
from bec_widgets.widgets.control.scan_control.scan_control import ScanControlConfig
|
|
from bec_widgets.widgets.control.scan_control.scan_info_adapter import ScanInfoAdapter
|
|
from bec_widgets.widgets.control.scan_control.scan_selection_dialog import ScanSelectionDialog
|
|
|
|
from .client_mocks import mocked_client
|
|
|
|
# pylint: disable=no-member
|
|
# pylint: disable=missing-function-docstring
|
|
# pylint: disable=redefined-outer-name
|
|
# pylint: disable=protected-access
|
|
|
|
available_scans_message = AvailableResourceMessage(
|
|
resource={
|
|
"line_scan": {
|
|
"class": "LineScan",
|
|
"base_class": "ScanBase",
|
|
"doc": (
|
|
"Run a line scan.\n\n"
|
|
"Args:\n"
|
|
" device (DeviceBase | str): Device to move.\n\n"
|
|
"Examples:\n"
|
|
" >>> scans.line_scan(samx, 0, 1)"
|
|
),
|
|
"arg_input": {"device": "device", "start": "float", "stop": "float"},
|
|
"gui_config": {
|
|
"scan_class_name": "LineScan",
|
|
"arg_group": {
|
|
"name": "Scan Arguments",
|
|
"bundle": 3,
|
|
"arg_inputs": {"device": "device", "start": "float", "stop": "float"},
|
|
"inputs": [
|
|
{
|
|
"arg": True,
|
|
"name": "device",
|
|
"type": "device",
|
|
"display_name": "Device",
|
|
"tooltip": None,
|
|
"default": None,
|
|
"expert": False,
|
|
},
|
|
{
|
|
"arg": True,
|
|
"name": "start",
|
|
"type": "float",
|
|
"display_name": "Start",
|
|
"tooltip": None,
|
|
"default": None,
|
|
"expert": False,
|
|
},
|
|
{
|
|
"arg": True,
|
|
"name": "stop",
|
|
"type": "float",
|
|
"display_name": "Stop",
|
|
"tooltip": None,
|
|
"default": None,
|
|
"expert": False,
|
|
},
|
|
],
|
|
"min": 1,
|
|
"max": None,
|
|
},
|
|
"kwarg_groups": [
|
|
{
|
|
"name": "Movement Parameters",
|
|
"inputs": [
|
|
{
|
|
"arg": False,
|
|
"name": "steps",
|
|
"type": "int",
|
|
"display_name": "Steps",
|
|
"tooltip": "Number of steps",
|
|
"default": None,
|
|
"expert": False,
|
|
},
|
|
{
|
|
"arg": False,
|
|
"name": "relative",
|
|
"type": "bool",
|
|
"display_name": "Relative",
|
|
"tooltip": "If True, the start and end positions are relative to the current position",
|
|
"default": False,
|
|
"expert": False,
|
|
},
|
|
],
|
|
},
|
|
{
|
|
"name": "Acquisition Parameters",
|
|
"inputs": [
|
|
{
|
|
"arg": False,
|
|
"name": "exp_time",
|
|
"type": "float",
|
|
"display_name": "Exp Time",
|
|
"tooltip": "Exposure time in s",
|
|
"default": 0,
|
|
"expert": False,
|
|
},
|
|
{
|
|
"arg": False,
|
|
"name": "burst_at_each_point",
|
|
"type": "int",
|
|
"display_name": "Burst At Each Point",
|
|
"tooltip": "Number of acquisition per point",
|
|
"default": 1,
|
|
"expert": False,
|
|
},
|
|
],
|
|
},
|
|
],
|
|
},
|
|
"required_kwargs": ["steps", "relative"],
|
|
"arg_bundle_size": {"bundle": 3, "min": 1, "max": None},
|
|
},
|
|
"grid_scan": {
|
|
"class": "Scan",
|
|
"base_class": "ScanBase",
|
|
"doc": "Run a grid scan over one or more devices.",
|
|
"arg_input": {"device": "device", "start": "float", "stop": "float", "steps": "int"},
|
|
"gui_config": {
|
|
"scan_class_name": "Scan",
|
|
"arg_group": {
|
|
"name": "Scan Arguments",
|
|
"bundle": 4,
|
|
"arg_inputs": {
|
|
"device": "device",
|
|
"start": "float",
|
|
"stop": "float",
|
|
"steps": "int",
|
|
},
|
|
"inputs": [
|
|
{
|
|
"arg": True,
|
|
"name": "device",
|
|
"type": "device",
|
|
"display_name": "Device",
|
|
"tooltip": None,
|
|
"default": None,
|
|
"expert": False,
|
|
},
|
|
{
|
|
"arg": True,
|
|
"name": "start",
|
|
"type": "float",
|
|
"display_name": "Start",
|
|
"tooltip": None,
|
|
"default": None,
|
|
"expert": False,
|
|
},
|
|
{
|
|
"arg": True,
|
|
"name": "stop",
|
|
"type": "float",
|
|
"display_name": "Stop",
|
|
"tooltip": None,
|
|
"default": None,
|
|
"expert": False,
|
|
},
|
|
{
|
|
"arg": True,
|
|
"name": "steps",
|
|
"type": "int",
|
|
"display_name": "Steps",
|
|
"tooltip": None,
|
|
"default": None,
|
|
"expert": False,
|
|
},
|
|
],
|
|
"min": 2,
|
|
"max": None,
|
|
},
|
|
"kwarg_groups": [
|
|
{
|
|
"name": "Scan Parameters",
|
|
"inputs": [
|
|
{
|
|
"arg": False,
|
|
"name": "exp_time",
|
|
"type": "float",
|
|
"display_name": "Exp Time",
|
|
"tooltip": "Exposure time in seconds",
|
|
"default": 0,
|
|
"expert": False,
|
|
},
|
|
{
|
|
"arg": False,
|
|
"name": "settling_time",
|
|
"type": "float",
|
|
"display_name": "Settling Time",
|
|
"tooltip": "Settling time in seconds",
|
|
"default": 0,
|
|
"expert": False,
|
|
},
|
|
{
|
|
"arg": False,
|
|
"name": "burst_at_each_point",
|
|
"type": "int",
|
|
"display_name": "Burst At Each Point",
|
|
"tooltip": "Number of exposures at each point",
|
|
"default": 1,
|
|
"expert": False,
|
|
},
|
|
{
|
|
"arg": False,
|
|
"name": "relative",
|
|
"type": "bool",
|
|
"display_name": "Relative",
|
|
"tooltip": "If True, the motors will be moved relative to their current position",
|
|
"default": False,
|
|
"expert": False,
|
|
},
|
|
{
|
|
"arg": False,
|
|
"name": "optim_trajectory",
|
|
"type": {"Literal": ("option1", "option2", "option3", None)},
|
|
"display_name": "Optim Trajectory",
|
|
"tooltip": None,
|
|
"default": None,
|
|
"expert": False,
|
|
},
|
|
],
|
|
}
|
|
],
|
|
},
|
|
"required_kwargs": ["relative"],
|
|
"arg_bundle_size": {"bundle": 4, "min": 2, "max": None},
|
|
},
|
|
"not_supported_scan_class": {"base_class": "NotSupportedScanClass"},
|
|
}
|
|
)
|
|
|
|
scan_history = ScanHistoryMessage(
|
|
metadata={},
|
|
scan_id="79cbef20-9ebe-45bb-a44c-f518be27a25c",
|
|
scan_number=1,
|
|
dataset_number=1,
|
|
file_path="/somepath/scan_1.h5",
|
|
exit_status="closed",
|
|
start_time=1750618470.936856,
|
|
end_time=1750618473.668227,
|
|
scan_name="line_scan",
|
|
num_points=100,
|
|
request_inputs={
|
|
"arg_bundle": ["samx", 0.0, 2.0],
|
|
"inputs": {},
|
|
"kwargs": {
|
|
"steps": 10,
|
|
"exp_time": 2,
|
|
"relative": False,
|
|
"system_config": {"file_suffix": None, "file_directory": None},
|
|
},
|
|
},
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def scan_control(qtbot, mocked_client): # , mock_dev):
|
|
mocked_client.connector.set_and_publish(
|
|
MessageEndpoints.available_scans(), available_scans_message
|
|
)
|
|
mocked_client.connector.xadd(
|
|
topic=MessageEndpoints.scan_history(), msg_dict={"data": scan_history}
|
|
)
|
|
widget = ScanControl(client=mocked_client)
|
|
qtbot.addWidget(widget)
|
|
qtbot.waitExposed(widget)
|
|
yield widget
|
|
|
|
|
|
def test_populate_scans(scan_control, mocked_client):
|
|
expected_scans = ["line_scan", "grid_scan"]
|
|
items = [
|
|
scan_control.comboBox_scan_selection.itemText(i)
|
|
for i in range(scan_control.comboBox_scan_selection.count())
|
|
]
|
|
|
|
assert scan_control.comboBox_scan_selection.count() == 2
|
|
assert sorted(items) == sorted(expected_scans)
|
|
|
|
|
|
def test_scan_selector_items_and_combo_show_doc_tooltips(scan_control):
|
|
line_index = scan_control.comboBox_scan_selection.findText("line_scan")
|
|
line_tooltip = scan_control.comboBox_scan_selection.itemData(
|
|
line_index, Qt.ItemDataRole.ToolTipRole
|
|
)
|
|
|
|
assert "line_scan" in line_tooltip
|
|
assert "Run a line scan." in line_tooltip
|
|
assert "Parameters:" in line_tooltip
|
|
assert "device: DeviceBase | str" in line_tooltip
|
|
assert scan_control.comboBox_scan_selection.toolTip() == line_tooltip
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText("grid_scan")
|
|
|
|
assert "grid_scan" in scan_control.comboBox_scan_selection.toolTip()
|
|
assert "Run a grid scan" in scan_control.comboBox_scan_selection.toolTip()
|
|
|
|
|
|
def test_scan_info_button_shows_styled_selected_scan_docstring(scan_control, qtbot):
|
|
assert not scan_control.scan_info_button.icon().isNull()
|
|
assert scan_control.scan_info_button.accessibleName() == "Scan information"
|
|
|
|
with patch.object(scan_control.client.connector, "get") as connector_get:
|
|
qtbot.mouseClick(scan_control.scan_info_button, Qt.MouseButton.LeftButton)
|
|
connector_get.assert_not_called()
|
|
|
|
assert scan_control._scan_info_dialog.isVisible()
|
|
assert not scan_control._scan_info_dialog.isModal()
|
|
assert scan_control._scan_info_dialog.windowTitle() == "Scan information: line_scan"
|
|
plain_text = scan_control._scan_info_dialog.text_browser.toPlainText()
|
|
assert "line_scan" in plain_text
|
|
assert "Arguments" in plain_text
|
|
assert "device" in plain_text
|
|
assert "Examples" in plain_text
|
|
assert ">>> scans.line_scan" in plain_text
|
|
style = scan_control._scan_info_dialog.text_browser.document().defaultStyleSheet()
|
|
assert "h1" in style
|
|
assert "pre" in style
|
|
|
|
|
|
def test_scan_info_button_handles_missing_docstring(scan_control, qtbot):
|
|
scan_control.available_scans["line_scan"].pop("doc")
|
|
|
|
qtbot.mouseClick(scan_control.scan_info_button, Qt.MouseButton.LeftButton)
|
|
|
|
assert "No documentation is available for this scan." in (
|
|
scan_control._scan_info_dialog.text_browser.toPlainText()
|
|
)
|
|
|
|
|
|
def test_allowed_scans_property_filters_selector(scan_control):
|
|
scan_control.comboBox_scan_selection.setCurrentText("grid_scan")
|
|
|
|
scan_control.allowed_scans = ["line_scan", "unknown_scan", "line_scan"]
|
|
|
|
# The configured filter is kept verbatim (deduplicated) so that scans that are not
|
|
# available right now reappear once the scan server publishes them.
|
|
assert scan_control.allowed_scans == ["line_scan", "unknown_scan"]
|
|
assert scan_control.config.allowed_scans == ["line_scan", "unknown_scan"]
|
|
assert scan_control.comboBox_scan_selection.count() == 1
|
|
assert scan_control.current_scan == "line_scan"
|
|
|
|
|
|
def test_allowed_scans_none_clears_filter(scan_control):
|
|
scan_control.allowed_scans = ["line_scan"]
|
|
assert scan_control.comboBox_scan_selection.count() == 1
|
|
|
|
scan_control.allowed_scans = None
|
|
|
|
# the filter is unset internally, while the property reports what the selector shows
|
|
assert scan_control.config.allowed_scans is None
|
|
assert scan_control.allowed_scans == ["line_scan", "grid_scan"]
|
|
assert scan_control.comboBox_scan_selection.count() == 2
|
|
|
|
|
|
def test_allowed_scans_override_support_filter(scan_control):
|
|
scan_control.allowed_scans = ["not_supported_scan_class", "line_scan"]
|
|
|
|
items = [
|
|
scan_control.comboBox_scan_selection.itemText(i)
|
|
for i in range(scan_control.comboBox_scan_selection.count())
|
|
]
|
|
|
|
assert items == ["not_supported_scan_class", "line_scan"]
|
|
|
|
|
|
def test_filter_change_saves_current_scan_parameters(scan_control):
|
|
assert scan_control.current_scan == "line_scan"
|
|
|
|
scan_control.allowed_scans = ["grid_scan"]
|
|
|
|
assert scan_control.current_scan == "grid_scan"
|
|
assert "line_scan" in scan_control.config.scans
|
|
|
|
|
|
def test_allowed_scans_matching_nothing_disable_scan_info_and_run(scan_control):
|
|
# a filter whose entries are all unavailable leaves the selector empty
|
|
scan_control.allowed_scans = ["only_scan_that_is_not_available"]
|
|
|
|
assert scan_control.comboBox_scan_selection.count() == 0
|
|
assert scan_control.comboBox_scan_selection.toolTip() == ""
|
|
assert not scan_control.scan_info_button.isEnabled()
|
|
assert not scan_control.button_run_scan.isEnabled()
|
|
# run_scan must not raise even if triggered without a selected scan
|
|
scan_control.run_scan()
|
|
|
|
|
|
def test_configured_allowed_scans_are_preserved(qtbot, mocked_client):
|
|
mocked_client.connector.set_and_publish(
|
|
MessageEndpoints.available_scans(), available_scans_message
|
|
)
|
|
config = ScanControlConfig(widget_class="ScanControl", allowed_scans=["grid_scan"])
|
|
|
|
widget = ScanControl(client=mocked_client, config=config)
|
|
qtbot.addWidget(widget)
|
|
|
|
assert widget.allowed_scans == ["grid_scan"]
|
|
assert widget.comboBox_scan_selection.count() == 1
|
|
assert widget.comboBox_scan_selection.currentText() == "grid_scan"
|
|
|
|
|
|
def test_configured_default_scan_is_preserved_and_applied(qtbot, mocked_client):
|
|
mocked_client.connector.set_and_publish(
|
|
MessageEndpoints.available_scans(), available_scans_message
|
|
)
|
|
|
|
widget = ScanControl(client=mocked_client, default_scan="grid_scan")
|
|
qtbot.addWidget(widget)
|
|
assert widget.comboBox_scan_selection.currentText() == "grid_scan"
|
|
|
|
# A default_scan from a passed-in config must not be wiped by the ctor default None
|
|
config = ScanControlConfig(widget_class="ScanControl", default_scan="grid_scan")
|
|
widget_from_config = ScanControl(client=mocked_client, config=config)
|
|
qtbot.addWidget(widget_from_config)
|
|
assert widget_from_config.config.default_scan == "grid_scan"
|
|
assert widget_from_config.comboBox_scan_selection.currentText() == "grid_scan"
|
|
|
|
|
|
def test_scan_selector_settings_dialog_applies_checked_scans(scan_control, monkeypatch, qtbot):
|
|
def select_line_scan(dialog):
|
|
labels = [dialog.checkbox_for_scan(name).text() for name in ("line_scan", "grid_scan")]
|
|
assert labels == ["line_scan", "grid_scan"]
|
|
checkbox = dialog.checkbox_for_scan("grid_scan")
|
|
assert isinstance(checkbox, QCheckBox)
|
|
checkbox.setChecked(False)
|
|
return QDialog.DialogCode.Accepted
|
|
|
|
monkeypatch.setattr(ScanSelectionDialog, "exec", select_line_scan)
|
|
|
|
qtbot.mouseClick(scan_control.scan_selector_settings_button, Qt.MouseButton.LeftButton)
|
|
|
|
assert scan_control.allowed_scans == ["line_scan"]
|
|
assert scan_control.comboBox_scan_selection.count() == 1
|
|
assert scan_control.comboBox_scan_selection.currentText() == "line_scan"
|
|
|
|
|
|
def test_scan_selector_settings_dialog_all_checked_clears_filter(scan_control, monkeypatch, qtbot):
|
|
scan_control.allowed_scans = ["line_scan"]
|
|
|
|
def check_everything(dialog):
|
|
dialog.checkbox_for_scan("grid_scan").setChecked(True)
|
|
return QDialog.DialogCode.Accepted
|
|
|
|
monkeypatch.setattr(ScanSelectionDialog, "exec", check_everything)
|
|
|
|
qtbot.mouseClick(scan_control.scan_selector_settings_button, Qt.MouseButton.LeftButton)
|
|
|
|
assert scan_control.config.allowed_scans is None
|
|
assert scan_control.comboBox_scan_selection.count() == 2
|
|
|
|
|
|
def test_scan_selector_settings_dialog_is_released_after_use(scan_control, monkeypatch, qtbot):
|
|
monkeypatch.setattr(ScanSelectionDialog, "exec", lambda dialog: QDialog.DialogCode.Rejected)
|
|
with patch.object(ScanSelectionDialog, "deleteLater") as delete_later:
|
|
qtbot.mouseClick(scan_control.scan_selector_settings_button, Qt.MouseButton.LeftButton)
|
|
delete_later.assert_called_once()
|
|
|
|
|
|
def test_scan_selector_dialog_select_all_and_clear_buttons(qtbot):
|
|
dialog = ScanSelectionDialog(
|
|
scan_names=["line_scan", "grid_scan"], selected_scans=["line_scan"]
|
|
)
|
|
qtbot.addWidget(dialog)
|
|
dialog.show()
|
|
qtbot.waitExposed(dialog)
|
|
|
|
qtbot.mouseClick(dialog.select_all_button, Qt.MouseButton.LeftButton)
|
|
assert dialog.selected_scans() == ["line_scan", "grid_scan"]
|
|
|
|
qtbot.mouseClick(dialog.clear_button, Qt.MouseButton.LeftButton)
|
|
assert dialog.selected_scans() == []
|
|
|
|
|
|
def test_scan_selector_dialog_select_all_clears_the_filter(scan_control, monkeypatch, qtbot):
|
|
scan_control.allowed_scans = ["line_scan"]
|
|
|
|
def select_all(dialog):
|
|
dialog.set_all_checked(True)
|
|
return QDialog.DialogCode.Accepted
|
|
|
|
monkeypatch.setattr(ScanSelectionDialog, "exec", select_all)
|
|
|
|
qtbot.mouseClick(scan_control.scan_selector_settings_button, Qt.MouseButton.LeftButton)
|
|
|
|
assert scan_control.config.allowed_scans is None
|
|
assert scan_control.comboBox_scan_selection.count() == 2
|
|
|
|
|
|
def test_scan_selector_dialog_whole_row_click_toggles_checkbox(qtbot):
|
|
dialog = ScanSelectionDialog(
|
|
scan_names=["line_scan", "grid_scan"], selected_scans=["line_scan"]
|
|
)
|
|
qtbot.addWidget(dialog)
|
|
dialog.show()
|
|
qtbot.waitExposed(dialog)
|
|
|
|
checkbox = dialog.checkbox_for_scan("line_scan")
|
|
assert isinstance(checkbox, QCheckBox)
|
|
assert checkbox.isChecked()
|
|
|
|
indicator_width = checkbox.style().pixelMetric(
|
|
QStyle.PixelMetric.PM_IndicatorWidth, widget=checkbox
|
|
)
|
|
label_right = indicator_width + 8 + checkbox.fontMetrics().horizontalAdvance(checkbox.text())
|
|
empty_row_x = checkbox.rect().right() - 8
|
|
assert empty_row_x > label_right
|
|
qtbot.mouseClick(
|
|
checkbox, Qt.MouseButton.LeftButton, pos=QPoint(empty_row_x, checkbox.rect().center().y())
|
|
)
|
|
|
|
assert not checkbox.isChecked()
|
|
assert dialog.selected_scans() == []
|
|
|
|
|
|
def test_scan_selector_dialog_info_button_opens_docs_without_toggling(qtbot):
|
|
dialog = ScanSelectionDialog(
|
|
scan_names=["line_scan"],
|
|
selected_scans=["line_scan"],
|
|
scan_docs={"line_scan": available_scans_message.resource["line_scan"]["doc"]},
|
|
)
|
|
qtbot.addWidget(dialog)
|
|
dialog.setModal(True)
|
|
dialog.show()
|
|
qtbot.waitExposed(dialog)
|
|
|
|
checkbox = dialog.checkbox_for_scan("line_scan")
|
|
info_button = dialog.info_button_for_scan("line_scan")
|
|
qtbot.mouseClick(info_button, Qt.MouseButton.LeftButton)
|
|
|
|
assert checkbox.isChecked()
|
|
assert dialog._scan_info_dialog.parent() is dialog
|
|
assert dialog._scan_info_dialog.isVisible()
|
|
assert dialog._scan_info_dialog.windowTitle() == "Scan information: line_scan"
|
|
assert "Arguments" in dialog._scan_info_dialog.text_browser.toPlainText()
|
|
|
|
|
|
def test_scan_selector_settings_properties_are_profile_safe(scan_control):
|
|
exported = scan_control.export_settings()
|
|
|
|
# "No filter" is exported as the full scan list, which the setter maps back onto an
|
|
# unset filter, so future scans keep appearing after a profile round trip.
|
|
assert exported["allowed_scans"] == ["line_scan", "grid_scan"]
|
|
assert exported["hide_scan_selector_settings_button"] is False
|
|
|
|
scan_control.load_settings(exported)
|
|
assert scan_control.config.allowed_scans is None
|
|
|
|
scan_control.load_settings(
|
|
{"allowed_scans": ["grid_scan"], "hide_scan_selector_settings_button": True}
|
|
)
|
|
|
|
assert scan_control.allowed_scans == ["grid_scan"]
|
|
assert scan_control.comboBox_scan_selection.currentText() == "grid_scan"
|
|
assert scan_control.hide_scan_selector_settings_button is True
|
|
assert scan_control.scan_selector_settings_button.isHidden()
|
|
|
|
scan_control.load_settings({"allowed_scans": None})
|
|
|
|
assert scan_control.config.allowed_scans is None
|
|
assert scan_control.comboBox_scan_selection.count() == 2
|
|
|
|
|
|
def test_scan_control_uses_gui_visibility_and_signature(qtbot, mocked_client):
|
|
scan_info = {
|
|
"class": "AnnotatedScan",
|
|
"base_class": "ScanBase",
|
|
"arg_input": {
|
|
"device": "DeviceBase",
|
|
"start": {
|
|
"Annotated": {
|
|
"type": "float",
|
|
"metadata": {
|
|
"ScanArgument": {
|
|
"display_name": "Start Position",
|
|
"description": "Start position",
|
|
"tooltip": "Custom start tooltip",
|
|
"expert": False,
|
|
"alternative_group": None,
|
|
"units": None,
|
|
"reference_units": "device",
|
|
}
|
|
},
|
|
}
|
|
},
|
|
"stop": {
|
|
"Annotated": {
|
|
"type": "float",
|
|
"metadata": {
|
|
"ScanArgument": {
|
|
"display_name": None,
|
|
"description": "Stop position",
|
|
"tooltip": None,
|
|
"expert": False,
|
|
"alternative_group": None,
|
|
"units": None,
|
|
"reference_units": "device",
|
|
}
|
|
},
|
|
}
|
|
},
|
|
},
|
|
"arg_bundle_size": {"bundle": 3, "min": 1, "max": None},
|
|
"gui_visibility": {
|
|
"Movement Parameters": ["steps", "step_size"],
|
|
"Acquisition Parameters": ["exp_time", "relative"],
|
|
},
|
|
"required_kwargs": [],
|
|
"signature": [
|
|
{"name": "args", "kind": "VAR_POSITIONAL", "default": "_empty", "annotation": "_empty"},
|
|
{"name": "steps", "kind": "KEYWORD_ONLY", "default": 10, "annotation": "int"},
|
|
{
|
|
"name": "step_size",
|
|
"kind": "KEYWORD_ONLY",
|
|
"default": None,
|
|
"annotation": {
|
|
"Annotated": {
|
|
"type": "float",
|
|
"metadata": {
|
|
"ScanArgument": {
|
|
"display_name": "Step Size Custom",
|
|
"description": "Step size",
|
|
"tooltip": "Custom step tooltip",
|
|
"expert": False,
|
|
"alternative_group": "scan_resolution",
|
|
"units": "mm",
|
|
"reference_units": None,
|
|
}
|
|
},
|
|
}
|
|
},
|
|
},
|
|
{
|
|
"name": "exp_time",
|
|
"kind": "KEYWORD_ONLY",
|
|
"default": 0,
|
|
"annotation": {
|
|
"Annotated": {
|
|
"type": "float",
|
|
"metadata": {
|
|
"ScanArgument": {
|
|
"display_name": None,
|
|
"description": None,
|
|
"tooltip": "Exposure time",
|
|
"expert": False,
|
|
"alternative_group": None,
|
|
"units": "s",
|
|
"reference_units": None,
|
|
}
|
|
},
|
|
}
|
|
},
|
|
},
|
|
{"name": "relative", "kind": "KEYWORD_ONLY", "default": False, "annotation": "bool"},
|
|
{"name": "kwargs", "kind": "VAR_KEYWORD", "default": "_empty", "annotation": "_empty"},
|
|
],
|
|
}
|
|
mocked_client.connector.set_and_publish(
|
|
MessageEndpoints.available_scans(),
|
|
AvailableResourceMessage(resource={"annotated_scan": scan_info}),
|
|
)
|
|
|
|
widget = ScanControl(client=mocked_client)
|
|
qtbot.addWidget(widget)
|
|
qtbot.waitExposed(widget)
|
|
widget.comboBox_scan_selection.setCurrentText("annotated_scan")
|
|
|
|
assert widget.comboBox_scan_selection.count() == 1
|
|
assert widget.arg_box.label_for_widget(widget.arg_box.widgets[1]).text() == "Start Position"
|
|
assert "Custom start tooltip\nUnits from: device" in widget.arg_box.widgets[1].toolTip()
|
|
with patch.object(mocked_client.device_manager.devices.samx, "egu", return_value="mm"):
|
|
WidgetIO.set_value(widget.arg_box.widgets[0], "samx")
|
|
assert widget.arg_box.label_for_widget(widget.arg_box.widgets[1]).text() == "Start Position"
|
|
assert widget.arg_box.widgets[1].suffix() == " mm"
|
|
assert "Custom start tooltip\nUnits: mm" in widget.arg_box.widgets[1].toolTip()
|
|
widget.arg_box.widgets[0].setCurrentText("not_a_device")
|
|
assert widget.arg_box.label_for_widget(widget.arg_box.widgets[1]).text() == "Start Position"
|
|
assert widget.arg_box.widgets[1].suffix() == ""
|
|
assert "Custom start tooltip\nUnits from: device" in widget.arg_box.widgets[1].toolTip()
|
|
assert [box.title() for box in widget.kwarg_boxes] == [
|
|
"Movement Parameters",
|
|
"Acquisition Parameters",
|
|
]
|
|
assert widget.kwarg_boxes[0].label_for_widget(widget.kwarg_boxes[0].widgets[1]).text() == (
|
|
"Step Size Custom"
|
|
)
|
|
assert widget.kwarg_boxes[0].widgets[1].suffix() == " mm"
|
|
assert "Custom step tooltip\nUnits: mm" in widget.kwarg_boxes[0].widgets[1].toolTip()
|
|
assert widget.kwarg_boxes[1].label_for_widget(widget.kwarg_boxes[1].widgets[0]).text() == (
|
|
"Exp Time"
|
|
)
|
|
assert "Exposure time\nUnits: s" in widget.kwarg_boxes[1].widgets[0].toolTip()
|
|
|
|
|
|
def test_scan_info_adapter_skips_duplicate_visible_kwargs():
|
|
scan_info = {
|
|
"class": "DuplicateScan",
|
|
"base_class": "ScanBaseV4",
|
|
"arg_input": {},
|
|
"arg_bundle_size": {"bundle": 0, "min": None, "max": None},
|
|
"gui_visibility": {
|
|
"Scan Parameters": ["relative", "burst_at_each_point"],
|
|
"Acquisition Parameters": ["exp_time", "burst_at_each_point"],
|
|
},
|
|
"signature": [
|
|
{"name": "relative", "kind": "KEYWORD_ONLY", "default": False, "annotation": "bool"},
|
|
{
|
|
"name": "burst_at_each_point",
|
|
"kind": "KEYWORD_ONLY",
|
|
"default": 1,
|
|
"annotation": "int",
|
|
},
|
|
{"name": "exp_time", "kind": "KEYWORD_ONLY", "default": 0, "annotation": "float"},
|
|
],
|
|
}
|
|
|
|
gui_config = ScanInfoAdapter().build_scan_ui_config(scan_info)
|
|
groups = {
|
|
group["name"]: [input_spec["name"] for input_spec in group["inputs"]]
|
|
for group in gui_config["kwarg_groups"]
|
|
}
|
|
|
|
assert groups == {
|
|
"Scan Parameters": ["relative", "burst_at_each_point"],
|
|
"Acquisition Parameters": ["exp_time"],
|
|
}
|
|
|
|
|
|
def test_scan_info_adapter_rejects_unsupported_visible_inputs():
|
|
scan_info = {
|
|
"class": "UnsupportedScan",
|
|
"base_class": "ScanBaseV4",
|
|
"arg_input": {},
|
|
"arg_bundle_size": {"bundle": 0, "min": None, "max": None},
|
|
"gui_visibility": {"Regions": ["regions"]},
|
|
"signature": [
|
|
{
|
|
"name": "regions",
|
|
"kind": "KEYWORD_ONLY",
|
|
"default": "_empty",
|
|
"annotation": {
|
|
"Generic": {
|
|
"origin": "list",
|
|
"args": [
|
|
{"Generic": {"origin": "tuple", "args": ["float", "float", "int"]}}
|
|
],
|
|
}
|
|
},
|
|
}
|
|
],
|
|
}
|
|
|
|
gui_config = ScanInfoAdapter().build_scan_ui_config(scan_info)
|
|
unsupported_inputs = ScanInfoAdapter.unsupported_inputs(gui_config)
|
|
|
|
assert [input_spec["name"] for input_spec in unsupported_inputs] == ["regions"]
|
|
assert ScanInfoAdapter.has_scan_ui_config(scan_info) is False
|
|
|
|
|
|
def test_scan_info_adapter_skips_hidden_visible_kwargs():
|
|
scan_info = {
|
|
"class": "HiddenScan",
|
|
"base_class": "ScanBaseV4",
|
|
"arg_input": {},
|
|
"arg_bundle_size": {"bundle": 0, "min": None, "max": None},
|
|
"gui_visibility": {"Acquisition": ["exp_time", "internal_token"]},
|
|
"signature": [
|
|
{"name": "exp_time", "kind": "KEYWORD_ONLY", "default": 0, "annotation": "float"},
|
|
{
|
|
"name": "internal_token",
|
|
"kind": "KEYWORD_ONLY",
|
|
"default": None,
|
|
"annotation": {
|
|
"Annotated": {
|
|
"type": "str",
|
|
"metadata": {
|
|
"ScanArgument": {"display_name": "Internal Token", "hidden": True}
|
|
},
|
|
}
|
|
},
|
|
},
|
|
],
|
|
}
|
|
|
|
gui_config = ScanInfoAdapter().build_scan_ui_config(scan_info)
|
|
|
|
assert [input_spec["name"] for input_spec in gui_config["kwarg_groups"][0]["inputs"]] == [
|
|
"exp_time"
|
|
]
|
|
|
|
|
|
def test_scan_control_propagates_reference_units_across_kwarg_groups(qtbot, mocked_client):
|
|
scan_info = {
|
|
"class": "RoundScan",
|
|
"base_class": "ScanBaseV4",
|
|
"arg_input": {},
|
|
"arg_bundle_size": {"bundle": 0, "min": None, "max": None},
|
|
"gui_visibility": {
|
|
"Motors": ["motor_1", "motor_2"],
|
|
"Ring Parameters": ["inner_radius", "outer_radius", "center_1", "center_2"],
|
|
},
|
|
"required_kwargs": [],
|
|
"signature": [
|
|
{
|
|
"name": "motor_1",
|
|
"kind": "POSITIONAL_OR_KEYWORD",
|
|
"default": "_empty",
|
|
"annotation": "DeviceBase",
|
|
},
|
|
{
|
|
"name": "motor_2",
|
|
"kind": "POSITIONAL_OR_KEYWORD",
|
|
"default": "_empty",
|
|
"annotation": "DeviceBase",
|
|
},
|
|
{
|
|
"name": "inner_radius",
|
|
"kind": "POSITIONAL_OR_KEYWORD",
|
|
"default": "_empty",
|
|
"annotation": {
|
|
"Annotated": {
|
|
"type": "float",
|
|
"metadata": {
|
|
"ScanArgument": {
|
|
"display_name": "Inner Radius",
|
|
"units": None,
|
|
"reference_units": "motor_1",
|
|
"ge": 0,
|
|
}
|
|
},
|
|
}
|
|
},
|
|
},
|
|
{
|
|
"name": "outer_radius",
|
|
"kind": "POSITIONAL_OR_KEYWORD",
|
|
"default": "_empty",
|
|
"annotation": {
|
|
"Annotated": {
|
|
"type": "float",
|
|
"metadata": {
|
|
"ScanArgument": {
|
|
"display_name": "Outer Radius",
|
|
"units": None,
|
|
"reference_units": "motor_1",
|
|
"ge": 0,
|
|
}
|
|
},
|
|
}
|
|
},
|
|
},
|
|
{
|
|
"name": "center_1",
|
|
"kind": "KEYWORD_ONLY",
|
|
"default": 0,
|
|
"annotation": {
|
|
"Annotated": {
|
|
"type": "float",
|
|
"metadata": {
|
|
"ScanArgument": {
|
|
"display_name": "Center Motor 1",
|
|
"units": None,
|
|
"reference_units": "motor_1",
|
|
}
|
|
},
|
|
}
|
|
},
|
|
},
|
|
{
|
|
"name": "center_2",
|
|
"kind": "KEYWORD_ONLY",
|
|
"default": 0,
|
|
"annotation": {
|
|
"Annotated": {
|
|
"type": "float",
|
|
"metadata": {
|
|
"ScanArgument": {
|
|
"display_name": "Center Motor 2",
|
|
"units": None,
|
|
"reference_units": "motor_2",
|
|
}
|
|
},
|
|
}
|
|
},
|
|
},
|
|
],
|
|
}
|
|
mocked_client.connector.set_and_publish(
|
|
MessageEndpoints.available_scans(),
|
|
AvailableResourceMessage(resource={"round_scan": scan_info}),
|
|
)
|
|
|
|
widget = ScanControl(client=mocked_client)
|
|
qtbot.addWidget(widget)
|
|
qtbot.waitExposed(widget)
|
|
widget.comboBox_scan_selection.setCurrentText("round_scan")
|
|
|
|
motor_box = widget.kwarg_boxes[0]
|
|
ring_box = widget.kwarg_boxes[1]
|
|
|
|
assert "Units from: motor_1" in ring_box.widgets[0].toolTip()
|
|
assert ring_box.widgets[0].suffix() == ""
|
|
|
|
with patch.object(mocked_client.device_manager.devices.samx, "egu", return_value="mm"):
|
|
WidgetIO.set_value(motor_box.widgets[0], "samx")
|
|
|
|
assert ring_box.widgets[0].suffix() == " mm"
|
|
assert ring_box.widgets[1].suffix() == " mm"
|
|
assert ring_box.widgets[2].suffix() == " mm"
|
|
assert ring_box.widgets[3].suffix() == ""
|
|
assert "Units: mm" in ring_box.widgets[0].toolTip()
|
|
|
|
motor_box.widgets[0].setCurrentText("not_a_device")
|
|
|
|
assert ring_box.widgets[0].suffix() == ""
|
|
assert ring_box.widgets[1].suffix() == ""
|
|
assert ring_box.widgets[2].suffix() == ""
|
|
assert "Units from: motor_1" in ring_box.widgets[0].toolTip()
|
|
|
|
|
|
def test_current_scan(scan_control, mocked_client):
|
|
current_scan = scan_control.current_scan
|
|
wrong_scan = "error_scan"
|
|
scan_control.current_scan = wrong_scan
|
|
assert scan_control.current_scan == current_scan
|
|
new_scan = "grid_scan" if current_scan == "line_scan" else "line_scan"
|
|
scan_control.current_scan = new_scan
|
|
assert scan_control.current_scan == new_scan
|
|
|
|
|
|
def test_scan_switch_runs_cleanup_on_previous_inputs(scan_control):
|
|
"""Switching scans tears down the old group boxes; the BECWidget inputs inside
|
|
(device comboboxes) must go through close() so their cleanup runs, instead of
|
|
being destroyed by deleteLater() without cleanup."""
|
|
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
|
|
old_inputs = [w for w in scan_control.arg_box.widgets if hasattr(w, "_destroyed")]
|
|
assert old_inputs, "line_scan arg box should contain at least one BECWidget input"
|
|
assert all(not w._destroyed for w in old_inputs)
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText("grid_scan")
|
|
|
|
# closeEvent ran cleanup and flagged each old input as destroyed
|
|
assert all(w._destroyed for w in old_inputs)
|
|
register = scan_control.rpc_register
|
|
assert all(not register.object_is_registered(w) for w in old_inputs)
|
|
|
|
|
|
@pytest.mark.parametrize("scan_name", ["line_scan", "grid_scan"])
|
|
def test_on_scan_selected(scan_control, scan_name):
|
|
expected_scan_info = available_scans_message.resource[scan_name]
|
|
scan_control.comboBox_scan_selection.setCurrentText(scan_name)
|
|
|
|
# Check arg_box labels and widgets
|
|
inputs_per_bundle = len(expected_scan_info["arg_input"])
|
|
for index, (arg_key, arg_value) in enumerate(expected_scan_info["arg_input"].items()):
|
|
assert scan_control.arg_box.label_texts()[index].lower() == arg_key
|
|
|
|
for row in range(expected_scan_info["arg_bundle_size"]["min"]):
|
|
widget = scan_control.arg_box.get_bundle_widgets(row)[index]
|
|
expected_widget_type = scan_control.arg_box.WIDGET_HANDLER.get(arg_value, None)
|
|
assert isinstance(widget, expected_widget_type) # Confirm the widget type matches
|
|
if isinstance(widget, DeviceComboBox):
|
|
assert widget.currentText() == ""
|
|
assert widget.autocomplete is True
|
|
assert widget.include_signals_with_write_access is True
|
|
assert "samx" in widget.devices
|
|
assert (
|
|
"async_device" in widget.devices
|
|
) # async device should also be present in the device list
|
|
assert len(scan_control.arg_box.widgets) == (
|
|
inputs_per_bundle * expected_scan_info["arg_bundle_size"]["min"]
|
|
)
|
|
|
|
# Check kwargs boxes
|
|
kwargs_group = [param for param in expected_scan_info["gui_config"]["kwarg_groups"]]
|
|
print(kwargs_group)
|
|
|
|
for kwarg_box, kwarg_group in zip(scan_control.kwarg_boxes, kwargs_group):
|
|
assert kwarg_box.title() == kwarg_group["name"]
|
|
for index, kwarg_info in enumerate(kwarg_group["inputs"]):
|
|
widget = kwarg_box.widgets[index]
|
|
assert kwarg_box.label_for_widget(widget).text() == kwarg_info["display_name"]
|
|
if isinstance(kwarg_info["type"], dict) and "Literal" in kwarg_info["type"]:
|
|
expected_widget_type = kwarg_box.WIDGET_HANDLER.get("dict", None)
|
|
else:
|
|
expected_widget_type = kwarg_box.WIDGET_HANDLER.get(kwarg_info["type"], None)
|
|
assert isinstance(widget, expected_widget_type)
|
|
|
|
|
|
@pytest.mark.parametrize("scan_name", ["line_scan", "grid_scan"])
|
|
def test_add_remove_bundle(scan_control, scan_name, qtbot):
|
|
expected_scan_info = available_scans_message.resource[scan_name]
|
|
scan_control.comboBox_scan_selection.setCurrentText(scan_name)
|
|
|
|
# Initial number of args row
|
|
initial_num_of_rows = scan_control.arg_box.count_arg_rows()
|
|
|
|
assert initial_num_of_rows == expected_scan_info["arg_bundle_size"]["min"]
|
|
|
|
scan_control.arg_box.button_add_bundle.click()
|
|
scan_control.arg_box.button_add_bundle.click()
|
|
|
|
if expected_scan_info["arg_bundle_size"]["max"] is None:
|
|
assert scan_control.arg_box.count_arg_rows() == initial_num_of_rows + 2
|
|
|
|
# Remove one bundle
|
|
scan_control.arg_box.button_remove_bundle.click()
|
|
qtbot.wait(200)
|
|
|
|
assert scan_control.arg_box.count_arg_rows() == initial_num_of_rows + 1
|
|
|
|
|
|
def test_run_line_scan_with_parameters(scan_control, mocked_client):
|
|
scan_name = "line_scan"
|
|
kwargs = {"exp_time": 0.1, "steps": 10, "relative": True, "burst_at_each_point": 1}
|
|
args = {"device": "samx", "start": -5, "stop": 5}
|
|
mock_slot = MagicMock()
|
|
scan_control.scan_args.connect(mock_slot)
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText(scan_name)
|
|
|
|
# Set kwargs in the UI
|
|
for kwarg_box in scan_control.kwarg_boxes:
|
|
for widget in kwarg_box.widgets:
|
|
if widget.arg_name in kwargs:
|
|
WidgetIO.set_value(widget, kwargs[widget.arg_name])
|
|
|
|
# Set args in the UI
|
|
for widget in scan_control.arg_box.widgets:
|
|
if widget.arg_name in args:
|
|
WidgetIO.set_value(widget, args[widget.arg_name])
|
|
|
|
# Mock the scan function
|
|
mocked_scan_function = MagicMock()
|
|
setattr(mocked_client.scans, scan_name, mocked_scan_function)
|
|
|
|
# Run the scan
|
|
scan_control.button_run_scan.click()
|
|
|
|
# Retrieve the actual arguments passed to the mock
|
|
called_args, called_kwargs = mocked_scan_function.call_args
|
|
|
|
# Check if the scan function was called correctly
|
|
expected_device = mocked_client.device_manager.devices.samx
|
|
expected_args_list = [expected_device, args["start"], args["stop"]]
|
|
assert called_args == tuple(expected_args_list)
|
|
assert called_kwargs == kwargs | {
|
|
"metadata": {"sample_name": "", "comment": "", "scan_name": "line_scan"}
|
|
}
|
|
|
|
# Check the emitted signal
|
|
mock_slot.assert_called_once()
|
|
emitted_args_list = mock_slot.call_args[0][0]
|
|
assert len(emitted_args_list) == 3 # Expected 3 arguments for line_scan
|
|
assert emitted_args_list == [expected_device, -5.0, 5.0]
|
|
|
|
|
|
def test_run_grid_scan_with_parameters(scan_control, mocked_client):
|
|
scan_name = "grid_scan"
|
|
kwargs = {"exp_time": 0.2, "settling_time": 0.1, "relative": False, "burst_at_each_point": 2}
|
|
args_row1 = {"device": "samx", "start": -10, "stop": 10, "steps": 20}
|
|
args_row2 = {"device": "samy", "start": -5, "stop": 5, "steps": 10}
|
|
mock_slot = MagicMock()
|
|
scan_control.scan_args.connect(mock_slot)
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText(scan_name)
|
|
|
|
# Ensure there are two rows in the arg_box
|
|
current_rows = scan_control.arg_box.count_arg_rows()
|
|
required_rows = 2
|
|
while current_rows < required_rows:
|
|
scan_control.arg_box.add_widget_bundle()
|
|
current_rows += 1
|
|
|
|
# Set kwargs in the UI
|
|
for kwarg_box in scan_control.kwarg_boxes:
|
|
for widget in kwarg_box.widgets:
|
|
if widget.arg_name in kwargs:
|
|
WidgetIO.set_value(widget, kwargs[widget.arg_name])
|
|
|
|
# Set args in the UI for both rows
|
|
arg_widgets = scan_control.arg_box.widgets # This is a flat list of widgets
|
|
num_columns = len(scan_control.arg_box.inputs)
|
|
num_rows = int(len(arg_widgets) / num_columns)
|
|
assert num_rows == required_rows # We expect 2 rows for grid_scan
|
|
|
|
# Set values for first row
|
|
for i in range(num_columns):
|
|
widget = arg_widgets[i]
|
|
arg_name = widget.arg_name
|
|
if arg_name in args_row1:
|
|
WidgetIO.set_value(widget, args_row1[arg_name])
|
|
|
|
# Set values for second row
|
|
for i in range(num_columns):
|
|
widget = arg_widgets[num_columns + i] # Next row
|
|
arg_name = widget.arg_name
|
|
if arg_name in args_row2:
|
|
WidgetIO.set_value(widget, args_row2[arg_name])
|
|
|
|
# Mock the scan function
|
|
mocked_scan_function = MagicMock()
|
|
setattr(mocked_client.scans, scan_name, mocked_scan_function)
|
|
|
|
# Run the scan
|
|
scan_control.button_run_scan.click()
|
|
|
|
# Retrieve the actual arguments passed to the mock
|
|
called_args, called_kwargs = mocked_scan_function.call_args
|
|
|
|
# Check if the scan function was called correctly
|
|
expected_device1 = mocked_client.device_manager.devices.samx
|
|
expected_device2 = mocked_client.device_manager.devices.samy
|
|
expected_args_list = [
|
|
expected_device1,
|
|
args_row1["start"],
|
|
args_row1["stop"],
|
|
args_row1["steps"],
|
|
expected_device2,
|
|
args_row2["start"],
|
|
args_row2["stop"],
|
|
args_row2["steps"],
|
|
]
|
|
assert called_args == tuple(expected_args_list)
|
|
assert called_kwargs == kwargs | {
|
|
"metadata": {"sample_name": "", "comment": "", "scan_name": "grid_scan"},
|
|
"optim_trajectory": None,
|
|
}
|
|
|
|
# Check the emitted signal
|
|
mock_slot.assert_called_once()
|
|
emitted_args_list = mock_slot.call_args[0][0]
|
|
assert len(emitted_args_list) == 8 # Expected 8 arguments for grid_scan
|
|
assert emitted_args_list == expected_args_list
|
|
|
|
|
|
def test_changing_scans_remember_parameters(scan_control, mocked_client):
|
|
scan_name = "line_scan"
|
|
kwargs = {"exp_time": 0.1, "steps": 10, "relative": True, "burst_at_each_point": 1}
|
|
args = {"device": "samx", "start": -5, "stop": 5}
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText(scan_name)
|
|
|
|
# Set kwargs in the UI
|
|
for kwarg_box in scan_control.kwarg_boxes:
|
|
for widget in kwarg_box.widgets:
|
|
for key, value in kwargs.items():
|
|
if widget.arg_name == key:
|
|
WidgetIO.set_value(widget, value)
|
|
break
|
|
# Set args in the UI
|
|
for widget in scan_control.arg_box.widgets:
|
|
for key, value in args.items():
|
|
if widget.arg_name == key:
|
|
WidgetIO.set_value(widget, value)
|
|
break
|
|
|
|
scan_control.save_current_scan_parameters()
|
|
|
|
# Change the scan
|
|
new_scan_name = "grid_scan"
|
|
scan_control.comboBox_scan_selection.setCurrentText(new_scan_name)
|
|
|
|
# Check if kwargs are same as in the line_scan
|
|
grid_args, grid_kwargs = scan_control.get_scan_parameters(bec_object=False)
|
|
assert grid_kwargs["exp_time"] == kwargs["exp_time"]
|
|
assert grid_kwargs["relative"] == kwargs["relative"]
|
|
assert grid_kwargs["burst_at_each_point"] == kwargs["burst_at_each_point"]
|
|
|
|
|
|
def test_scan_selection_does_not_fetch_last_scan_parameters(
|
|
scan_control, mocked_client, monkeypatch
|
|
):
|
|
get_last = MagicMock(wraps=mocked_client.connector.get_last)
|
|
xrange = MagicMock(wraps=mocked_client.connector.xrange)
|
|
monkeypatch.setattr(mocked_client.connector, "get_last", get_last)
|
|
monkeypatch.setattr(mocked_client.connector, "xrange", xrange)
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
|
|
assert scan_control.comboBox_scan_selection.currentText() == "line_scan"
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText("grid_scan")
|
|
|
|
get_last.assert_not_called()
|
|
xrange.assert_not_called()
|
|
|
|
|
|
def test_restore_last_scan_parameters_button_fetches_on_demand(
|
|
scan_control, mocked_client, monkeypatch, qtbot
|
|
):
|
|
get_last = MagicMock(wraps=mocked_client.connector.get_last)
|
|
xrange = MagicMock(wraps=mocked_client.connector.xrange)
|
|
monkeypatch.setattr(mocked_client.connector, "get_last", get_last)
|
|
monkeypatch.setattr(mocked_client.connector, "xrange", xrange)
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText("grid_scan")
|
|
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
|
|
get_last.assert_not_called()
|
|
xrange.assert_not_called()
|
|
|
|
scan_control.last_scan_button.click()
|
|
|
|
qtbot.waitUntil(lambda: get_last.call_count == 2)
|
|
# a one-entry tip probe (memo validity check), then the recent window
|
|
assert get_last.call_args_list == [
|
|
call(MessageEndpoints.scan_history(), count=1),
|
|
call(MessageEndpoints.scan_history(), count=scan_control.RECENT_SCAN_HISTORY_COUNT),
|
|
]
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
xrange.assert_not_called()
|
|
args, kwargs = scan_control.get_scan_parameters(bec_object=False)
|
|
assert args == ["samx", 0.0, 2.0]
|
|
assert kwargs["steps"] == 10
|
|
assert kwargs["relative"] is False
|
|
assert kwargs["exp_time"] == 2
|
|
|
|
|
|
def test_restore_last_scan_parameters_does_not_block_gui(
|
|
scan_control, mocked_client, monkeypatch, qtbot
|
|
):
|
|
original_get_last = mocked_client.connector.get_last
|
|
fetch_started = Event()
|
|
allow_fetch_to_finish = Event()
|
|
fetch_thread_ids = []
|
|
apply_thread_ids = []
|
|
|
|
original_set_parameters = scan_control.arg_box.set_parameters
|
|
|
|
def tracking_set_parameters(*args, **kwargs):
|
|
apply_thread_ids.append(get_ident())
|
|
return original_set_parameters(*args, **kwargs)
|
|
|
|
def blocking_get_last(*args, **kwargs):
|
|
fetch_thread_ids.append(get_ident())
|
|
fetch_started.set()
|
|
allow_fetch_to_finish.wait(timeout=5)
|
|
return original_get_last(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(mocked_client.connector, "get_last", blocking_get_last)
|
|
monkeypatch.setattr(scan_control.arg_box, "set_parameters", tracking_set_parameters)
|
|
gui_thread_id = get_ident()
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
|
|
scan_control.last_scan_button.click()
|
|
|
|
try:
|
|
qtbot.waitUntil(fetch_started.is_set)
|
|
assert len(fetch_thread_ids) == 1
|
|
assert fetch_thread_ids[0] != gui_thread_id
|
|
assert not scan_control.last_scan_button.isEnabled()
|
|
finally:
|
|
allow_fetch_to_finish.set()
|
|
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
assert apply_thread_ids == [gui_thread_id]
|
|
args, kwargs = scan_control.get_scan_parameters(bec_object=False)
|
|
assert args == ["samx", 0.0, 2.0]
|
|
assert kwargs["steps"] == 10
|
|
|
|
|
|
def test_restore_last_scan_parameters_reenables_button_after_fetch_failure(
|
|
scan_control, mocked_client, monkeypatch, qtbot
|
|
):
|
|
get_last = MagicMock(side_effect=RuntimeError("history unavailable"))
|
|
monkeypatch.setattr(mocked_client.connector, "get_last", get_last)
|
|
|
|
scan_control.last_scan_button.click()
|
|
|
|
qtbot.waitUntil(lambda: get_last.call_count == 1)
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
|
|
|
|
def test_restore_last_scan_parameters_reenables_button_when_worker_task_raises(
|
|
scan_control, monkeypatch, qtbot
|
|
):
|
|
# bypass the fetch's own error handling: the worker emits ``failed`` instead of
|
|
# ``completed``, so the button must recover from that signal, not from the watchdog
|
|
monkeypatch.setattr(
|
|
scan_control._last_scan_fetch_watchdog, "start", MagicMock(name="watchdog_disabled")
|
|
)
|
|
monkeypatch.setattr(
|
|
scan_control,
|
|
"_fetch_last_executed_scan_parameters",
|
|
MagicMock(side_effect=RuntimeError("worker exploded")),
|
|
)
|
|
|
|
scan_control.last_scan_button.click()
|
|
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
|
|
|
|
def test_restore_last_scan_parameters_escalates_to_bounded_history_window(
|
|
scan_control, mocked_client, monkeypatch, qtbot
|
|
):
|
|
recent_scan = scan_history.model_copy(update={"scan_name": "grid_scan"})
|
|
# a full window means older entries may exist, so the deeper - but still capped - read runs
|
|
full_window = [{"data": recent_scan}] * scan_control.RECENT_SCAN_HISTORY_COUNT
|
|
# responses for: tip probe, recent window, deep window
|
|
get_last = MagicMock(
|
|
side_effect=[[{"data": recent_scan}], full_window, [{"data": scan_history}]]
|
|
)
|
|
xrange = MagicMock(wraps=mocked_client.connector.xrange)
|
|
monkeypatch.setattr(mocked_client.connector, "get_last", get_last)
|
|
monkeypatch.setattr(mocked_client.connector, "xrange", xrange)
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
|
|
scan_control.last_scan_button.click()
|
|
|
|
qtbot.waitUntil(lambda: get_last.call_count == 3)
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
assert get_last.call_args_list == [
|
|
call(MessageEndpoints.scan_history(), count=1),
|
|
call(MessageEndpoints.scan_history(), count=scan_control.RECENT_SCAN_HISTORY_COUNT),
|
|
call(MessageEndpoints.scan_history(), count=scan_control.MAX_HISTORY_LOOKBACK),
|
|
]
|
|
# the unbounded full-stream read is gone: it stalled the event loop from any thread
|
|
xrange.assert_not_called()
|
|
args, kwargs = scan_control.get_scan_parameters(bec_object=False)
|
|
assert args == ["samx", 0.0, 2.0]
|
|
assert kwargs["steps"] == 10
|
|
|
|
|
|
def test_restore_last_scan_parameters_never_reads_unbounded_stream(
|
|
scan_control, mocked_client, monkeypatch, qtbot
|
|
):
|
|
recent_scan = scan_history.model_copy(update={"scan_name": "grid_scan"})
|
|
# every window comes back full, so the lookup can never conclude the stream is exhausted
|
|
get_last = MagicMock(
|
|
side_effect=lambda _endpoint, count: [{"data": recent_scan}] * count # noqa: ARG005
|
|
)
|
|
xrange = MagicMock(wraps=mocked_client.connector.xrange)
|
|
monkeypatch.setattr(mocked_client.connector, "get_last", get_last)
|
|
monkeypatch.setattr(mocked_client.connector, "xrange", xrange)
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
|
|
scan_control.last_scan_button.click()
|
|
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
xrange.assert_not_called()
|
|
assert (
|
|
max(call_args.kwargs["count"] for call_args in get_last.call_args_list)
|
|
== scan_control.MAX_HISTORY_LOOKBACK
|
|
)
|
|
|
|
|
|
def test_restore_last_scan_parameters_logs_when_lookback_exhausted(
|
|
scan_control, mocked_client, monkeypatch, qtbot
|
|
):
|
|
recent_scan = scan_history.model_copy(update={"scan_name": "grid_scan"})
|
|
get_last = MagicMock(side_effect=lambda _endpoint, count: [{"data": recent_scan}] * count)
|
|
monkeypatch.setattr(mocked_client.connector, "get_last", get_last)
|
|
# bec_logger is loguru-based, so caplog does not see it
|
|
warning = MagicMock()
|
|
monkeypatch.setattr(scan_control_module.logger, "warning", warning)
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
|
|
scan_control.last_scan_button.click()
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
|
|
assert any("older history is not searched" in c.args[0] for c in warning.call_args_list)
|
|
|
|
|
|
def test_restore_last_scan_parameters_skips_deeper_read_when_stream_exhausted(
|
|
scan_control, mocked_client, monkeypatch, qtbot
|
|
):
|
|
recent_scan = scan_history.model_copy(update={"scan_name": "grid_scan"})
|
|
# a partial window proves the whole stream was searched; no deeper read is allowed
|
|
get_last = MagicMock(return_value=[{"data": recent_scan}])
|
|
xrange = MagicMock(wraps=mocked_client.connector.xrange)
|
|
monkeypatch.setattr(mocked_client.connector, "get_last", get_last)
|
|
monkeypatch.setattr(mocked_client.connector, "xrange", xrange)
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
|
|
scan_control.last_scan_button.click()
|
|
|
|
qtbot.waitUntil(lambda: get_last.call_count == 2)
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
assert get_last.call_args_list == [
|
|
call(MessageEndpoints.scan_history(), count=1),
|
|
call(MessageEndpoints.scan_history(), count=scan_control.RECENT_SCAN_HISTORY_COUNT),
|
|
]
|
|
xrange.assert_not_called()
|
|
|
|
|
|
def test_restore_last_scan_parameters_memoizes_missing_scan_lookups(
|
|
scan_control, mocked_client, monkeypatch, qtbot
|
|
):
|
|
recent_scan = scan_history.model_copy(update={"scan_name": "grid_scan"})
|
|
get_last = MagicMock(side_effect=lambda _endpoint, count: [{"data": recent_scan}] * count)
|
|
monkeypatch.setattr(mocked_client.connector, "get_last", get_last)
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
|
|
scan_control.last_scan_button.click()
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
calls_after_first = get_last.call_count # tip probe + recent window + deep window
|
|
|
|
scan_control.last_scan_button.click()
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
|
|
# the repeat click for a never-measured scan costs one tip probe, no window reads
|
|
assert get_last.call_count == calls_after_first + 1
|
|
assert get_last.call_args_list[-1] == call(MessageEndpoints.scan_history(), count=1)
|
|
|
|
|
|
def test_restore_last_scan_parameters_memo_reapplies_and_invalidates(
|
|
scan_control, mocked_client, monkeypatch, qtbot
|
|
):
|
|
get_last = MagicMock(wraps=mocked_client.connector.get_last)
|
|
monkeypatch.setattr(mocked_client.connector, "get_last", get_last)
|
|
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
|
|
|
|
scan_control.last_scan_button.click()
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
calls_after_first = get_last.call_count # tip probe + recent window (hit)
|
|
|
|
# user edits the form; restoring again is answered from the memo with one probe read
|
|
scan_control.arg_box.set_parameters(["samx", 1.0, 5.0])
|
|
scan_control.last_scan_button.click()
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
args, _ = scan_control.get_scan_parameters(bec_object=False)
|
|
assert args == ["samx", 0.0, 2.0]
|
|
assert get_last.call_count == calls_after_first + 1
|
|
assert get_last.call_args_list[-1] == call(MessageEndpoints.scan_history(), count=1)
|
|
|
|
# a new history entry moves the stream tip and invalidates the memo
|
|
new_msg = scan_history.model_copy(update={"scan_name": "grid_scan", "scan_id": "memo_tip"})
|
|
mocked_client.connector.xadd(topic=MessageEndpoints.scan_history(), msg_dict={"data": new_msg})
|
|
scan_control.last_scan_button.click()
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
# tip probe plus a fresh recent-window read
|
|
assert get_last.call_count == calls_after_first + 3
|
|
|
|
|
|
def test_restore_last_scan_parameters_handles_single_entry_window(
|
|
scan_control, mocked_client, monkeypatch, qtbot
|
|
):
|
|
# get_last returns a bare dict (not a list) for count == 1
|
|
monkeypatch.setattr(scan_control, "RECENT_SCAN_HISTORY_COUNT", 1)
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
|
|
scan_control.last_scan_button.click()
|
|
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
args, kwargs = scan_control.get_scan_parameters(bec_object=False)
|
|
assert args == ["samx", 0.0, 2.0]
|
|
assert kwargs["steps"] == 10
|
|
|
|
|
|
def test_restore_last_scan_parameters_uses_history_cache(
|
|
scan_control, mocked_client, monkeypatch, qtbot
|
|
):
|
|
history = ScanHistory(mocked_client, load_threaded=False)
|
|
history._scan_data[scan_history.scan_id] = scan_history
|
|
history._scan_ids.append(scan_history.scan_id)
|
|
monkeypatch.setattr(mocked_client, "history", history, raising=False)
|
|
get_last = MagicMock(wraps=mocked_client.connector.get_last)
|
|
monkeypatch.setattr(mocked_client.connector, "get_last", get_last)
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
|
|
scan_control.last_scan_button.click()
|
|
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
# only the one-entry tip probe validating the cache; no window reads
|
|
get_last.assert_called_once_with(MessageEndpoints.scan_history(), count=1)
|
|
args, kwargs = scan_control.get_scan_parameters(bec_object=False)
|
|
assert args == ["samx", 0.0, 2.0]
|
|
assert kwargs["steps"] == 10
|
|
|
|
|
|
def test_restore_last_scan_parameters_skips_stale_history_cache(
|
|
scan_control, mocked_client, monkeypatch, qtbot
|
|
):
|
|
"""A cache whose newest entry lags the stream tip (e.g. the newest scan's file is not
|
|
readable yet) must not answer the lookup with an older execution."""
|
|
stale = scan_history.model_copy(
|
|
update={"scan_id": "stale_entry", "request_inputs": {"arg_bundle": ["samx", 5.0, 9.0]}}
|
|
)
|
|
history = ScanHistory(mocked_client, load_threaded=False)
|
|
history._scan_data.clear()
|
|
history._scan_ids.clear()
|
|
history._scan_data[stale.scan_id] = stale
|
|
history._scan_ids.append(stale.scan_id)
|
|
monkeypatch.setattr(mocked_client, "history", history, raising=False)
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
|
|
scan_control.last_scan_button.click()
|
|
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
# the stream entry (steps=10), not the stale cache entry, wins
|
|
args, kwargs = scan_control.get_scan_parameters(bec_object=False)
|
|
assert args == ["samx", 0.0, 2.0]
|
|
assert kwargs["steps"] == 10
|
|
|
|
|
|
def test_restore_last_scan_parameters_discards_result_on_scan_switch(
|
|
scan_control, mocked_client, monkeypatch, qtbot
|
|
):
|
|
original_get_last = mocked_client.connector.get_last
|
|
fetch_started = Event()
|
|
allow_fetch_to_finish = Event()
|
|
|
|
def blocking_get_last(*args, **kwargs):
|
|
fetch_started.set()
|
|
allow_fetch_to_finish.wait(timeout=5)
|
|
return original_get_last(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(mocked_client.connector, "get_last", blocking_get_last)
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
|
|
scan_control.last_scan_button.click()
|
|
qtbot.waitUntil(fetch_started.is_set)
|
|
scan_control.comboBox_scan_selection.setCurrentText("grid_scan")
|
|
allow_fetch_to_finish.set()
|
|
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
# the fetched line_scan parameters must not leak into the grid_scan form
|
|
args, _ = scan_control.get_scan_parameters(bec_object=False)
|
|
assert args != ["samx", 0.0, 2.0]
|
|
|
|
|
|
def test_restore_last_scan_parameters_watchdog_recovers_from_hanging_fetch(
|
|
scan_control, mocked_client, monkeypatch, qtbot
|
|
):
|
|
original_get_last = mocked_client.connector.get_last
|
|
fetch_started = Event()
|
|
allow_fetch_to_finish = Event()
|
|
|
|
def blocking_get_last(*args, **kwargs):
|
|
fetch_started.set()
|
|
allow_fetch_to_finish.wait(timeout=5)
|
|
return original_get_last(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(mocked_client.connector, "get_last", blocking_get_last)
|
|
scan_control._last_scan_fetch_watchdog.setInterval(50)
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
|
|
original_args, _ = scan_control.get_scan_parameters(bec_object=False)
|
|
scan_control.last_scan_button.click()
|
|
qtbot.waitUntil(fetch_started.is_set)
|
|
|
|
# the watchdog re-enables the button while the fetch is still hanging
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
allow_fetch_to_finish.set()
|
|
|
|
# the late result must be discarded once the watchdog has given up on the fetch
|
|
qtbot.wait(200)
|
|
args, _ = scan_control.get_scan_parameters(bec_object=False)
|
|
assert args == original_args
|
|
|
|
|
|
def test_get_scan_parameters_from_redis(scan_control, qtbot):
|
|
scan_name = "line_scan"
|
|
scan_control.comboBox_scan_selection.setCurrentText(scan_name)
|
|
|
|
scan_control.last_scan_button.click()
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
|
|
args, kwargs = scan_control.get_scan_parameters(bec_object=False)
|
|
|
|
assert args == ["samx", 0.0, 2.0]
|
|
assert kwargs == {
|
|
"steps": 10,
|
|
"relative": False,
|
|
"exp_time": 2.0,
|
|
"burst_at_each_point": 1,
|
|
"metadata": {"comment": "", "sample_name": "", "scan_name": "line_scan"},
|
|
}
|
|
|
|
|
|
TEST_MD = {
|
|
"comment": "",
|
|
"sample_name": "Test Sample",
|
|
"scan_name": "grid_scan",
|
|
"test key 1": "test value 1",
|
|
"test key 2": "test value 2",
|
|
}
|
|
TEST_TABLE_ENTRY = [["test key 1", "test value 1"], ["test key 2", "test value 2"]]
|
|
|
|
|
|
def test_scan_metadata_is_updated_even_without_default_form_changes(
|
|
scan_control: ScanControl, qtbot
|
|
):
|
|
assert scan_control._metadata_form._scan_name == "line_scan"
|
|
scan_control.comboBox_scan_selection.setCurrentText("grid_scan")
|
|
assert scan_control._metadata_form._scan_name == "grid_scan"
|
|
scan_control._metadata_form._additional_metadata._add_button.click()
|
|
qtbot.wait(100)
|
|
table_model = scan_control._metadata_form._additional_metadata._table_model
|
|
model_key = table_model.index(0, 0, QModelIndex())
|
|
table_model.setData(model_key, "test key 1", Qt.EditRole)
|
|
model_value = model_key.siblingAtColumn(1)
|
|
table_model.setData(model_value, "test value 1", Qt.EditRole)
|
|
assert scan_control._metadata_form._additional_metadata.dump_dict() == {
|
|
"test key 1": "test value 1"
|
|
}
|
|
assert scan_control._scan_metadata == {
|
|
"comment": "",
|
|
"sample_name": "",
|
|
"scan_name": "grid_scan",
|
|
"test key 1": "test value 1",
|
|
}
|
|
|
|
|
|
def test_scan_metadata_is_connected(scan_control):
|
|
assert scan_control._metadata_form._scan_name == "line_scan"
|
|
scan_control.comboBox_scan_selection.setCurrentText("grid_scan")
|
|
assert scan_control._metadata_form._scan_name == "grid_scan"
|
|
sample_name = scan_control._metadata_form._form_grid.layout().itemAtPosition(2, 1).widget()
|
|
assert isinstance(sample_name, StrFormItem)
|
|
sample_name._main_widget.setText("Test Sample")
|
|
|
|
scan_control._metadata_form._additional_metadata._table_model._data = TEST_TABLE_ENTRY
|
|
scan_control._metadata_form.validate_form()
|
|
assert scan_control._scan_metadata == TEST_MD
|
|
|
|
|
|
def test_scan_metadata_is_passed_to_scan_function(scan_control: ScanControl):
|
|
scan_control.comboBox_scan_selection.setCurrentText("grid_scan")
|
|
|
|
sample_name = scan_control._metadata_form._form_grid.layout().itemAtPosition(2, 1).widget()
|
|
sample_name._main_widget.setText("Test Sample")
|
|
scan_control._metadata_form._additional_metadata._table_model._data = TEST_TABLE_ENTRY
|
|
scan_control._metadata_form.validate_form()
|
|
|
|
assert scan_control._scan_metadata == TEST_MD
|
|
|
|
scans = SimpleNamespace(grid_scan=MagicMock())
|
|
with (
|
|
patch.object(scan_control, "scans", scans),
|
|
patch.object(scan_control, "get_scan_parameters", lambda: ((), {"metadata": TEST_MD})),
|
|
):
|
|
scan_control.run_scan()
|
|
scans.grid_scan.assert_called_once_with(metadata=TEST_MD)
|
|
|
|
|
|
def test_restore_parameters_with_fewer_arg_bundles(scan_control, qtbot):
|
|
"""
|
|
Ensure that when more argument bundles are present than exist in the
|
|
stored history, restoring parameters regenerates the arg box to the
|
|
correct (smaller) size and sets the values properly.
|
|
This is a check for the previous infinite loop bug.
|
|
"""
|
|
# Select the scan type that has history with only one arg bundle
|
|
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
|
|
|
|
# Manually add bundles so we end up with three rows
|
|
while scan_control.arg_box.count_arg_rows() < 3:
|
|
scan_control.arg_box.add_widget_bundle()
|
|
assert scan_control.arg_box.count_arg_rows() == 3
|
|
|
|
# Trigger restore of parameters from history
|
|
scan_control.last_scan_button.click()
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
|
|
# After restore, arg_box should have only one bundle (the history size)
|
|
assert scan_control.arg_box.count_arg_rows() == 1
|
|
|
|
# Verify that the restored parameter values match the history
|
|
args, kwargs = scan_control.get_scan_parameters(bec_object=False)
|
|
assert args == ["samx", 0.0, 2.0]
|
|
assert kwargs["steps"] == 10
|
|
|
|
|
|
def test_allowed_scans_property_reports_selector_contents_on_init(scan_control):
|
|
"""A freshly added widget must report the scans it shows, not an empty list.
|
|
|
|
``allowed_scans`` is a list-typed Qt property, and Qt cannot convert ``None`` to a
|
|
list: reporting the unset filter as ``None`` surfaced as ``[]`` through the property
|
|
system, which a profile then stored and restored as "allow nothing".
|
|
"""
|
|
visible_scans = [
|
|
scan_control.comboBox_scan_selection.itemText(index)
|
|
for index in range(scan_control.comboBox_scan_selection.count())
|
|
]
|
|
|
|
assert scan_control.config.allowed_scans is None
|
|
assert scan_control.allowed_scans == visible_scans
|
|
assert scan_control.property("allowed_scans") == visible_scans
|
|
|
|
|
|
def test_allowed_scans_property_round_trip_keeps_filter_unset(scan_control):
|
|
"""Restoring the property in the same session must not turn it into a filter."""
|
|
scan_control.load_settings({"allowed_scans": scan_control.property("allowed_scans")})
|
|
|
|
assert scan_control.config.allowed_scans is None
|
|
assert scan_control.comboBox_scan_selection.count() == 2
|
|
|
|
# a scan published later still shows up, i.e. the filter really is unset
|
|
scan_control.available_scans["extra_scan"] = scan_control.available_scans["line_scan"]
|
|
scan_control._update_scan_selector()
|
|
|
|
assert scan_control.comboBox_scan_selection.count() == 3
|
|
|
|
|
|
def test_allowed_scans_empty_list_restores_unfiltered_selector(scan_control):
|
|
"""Profiles written before the fix hold an empty list; it must not blank the selector."""
|
|
scan_control.allowed_scans = []
|
|
|
|
assert scan_control.config.allowed_scans is None
|
|
assert scan_control.comboBox_scan_selection.count() == 2
|
|
assert scan_control.comboBox_scan_selection.currentText() == "line_scan"
|
|
assert scan_control.button_run_scan.isEnabled()
|
|
|
|
|
|
def test_allowed_scans_restored_after_new_scan_keeps_stored_selection(scan_control):
|
|
"""A profile stored before a new scan appeared is restored as an explicit filter.
|
|
|
|
"No filter" and "every scan selected" serialize to the same plain list, so the stored
|
|
selection wins and the newer scan stays hidden until the filter is cleared again. This
|
|
pins the documented limitation of the property rather than an intended feature.
|
|
"""
|
|
saved = scan_control.property("allowed_scans")
|
|
scan_control.available_scans["extra_scan"] = scan_control.available_scans["line_scan"]
|
|
|
|
scan_control.load_settings({"allowed_scans": saved})
|
|
|
|
assert scan_control.config.allowed_scans == saved
|
|
assert scan_control.comboBox_scan_selection.count() == 2
|
|
|
|
# clearing the filter picks the newer scan up again
|
|
scan_control.allowed_scans = None
|
|
|
|
assert scan_control.comboBox_scan_selection.count() == 3
|
|
|
|
|
|
def test_allowed_scans_empty_at_construction_shows_all_scans(qtbot, mocked_client):
|
|
mocked_client.connector.set_and_publish(
|
|
MessageEndpoints.available_scans(), available_scans_message
|
|
)
|
|
widget = ScanControl(client=mocked_client, allowed_scans=[])
|
|
qtbot.addWidget(widget)
|
|
qtbot.waitExposed(widget)
|
|
|
|
assert widget.config.allowed_scans is None
|
|
assert widget.comboBox_scan_selection.count() == 2
|
|
assert widget.button_run_scan.isEnabled() or widget._scan_metadata is None
|
|
|
|
|
|
def test_allowed_scans_reordered_full_list_keeps_order(scan_control):
|
|
supported = scan_control._supported_scan_names()
|
|
reordered = list(reversed(supported))
|
|
|
|
scan_control.allowed_scans = reordered
|
|
|
|
# a reordered full list is an explicit filter that preserves the caller's order
|
|
assert scan_control.config.allowed_scans == reordered
|
|
items = [
|
|
scan_control.comboBox_scan_selection.itemText(i)
|
|
for i in range(scan_control.comboBox_scan_selection.count())
|
|
]
|
|
assert items == reordered
|
|
|
|
# the same list in the supported order clears the filter
|
|
scan_control.allowed_scans = supported
|
|
assert scan_control.config.allowed_scans is None
|
|
|
|
|
|
def test_restore_last_scan_parameters_memo_survives_overlapping_workers(
|
|
scan_control, mocked_client, monkeypatch, qtbot
|
|
):
|
|
"""The watchdog re-enables the button while a slow worker is still running, so two
|
|
fetch workers can touch the memo at once. Both must finish and leave a usable memo."""
|
|
original_get_last = mocked_client.connector.get_last
|
|
first_fetch_running = Event()
|
|
release_first_fetch = Event()
|
|
worker_threads = set()
|
|
|
|
def slow_get_last(*args, **kwargs):
|
|
worker_threads.add(get_ident())
|
|
if not first_fetch_running.is_set():
|
|
first_fetch_running.set()
|
|
release_first_fetch.wait(timeout=5)
|
|
return original_get_last(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(mocked_client.connector, "get_last", slow_get_last)
|
|
scan_control._last_scan_fetch_watchdog.setInterval(50)
|
|
|
|
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
|
|
scan_control.last_scan_button.click()
|
|
qtbot.waitUntil(first_fetch_running.is_set)
|
|
|
|
# the watchdog gives the button back while the first worker is still blocked
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
scan_control.last_scan_button.click()
|
|
qtbot.waitUntil(lambda: len(worker_threads) == 2)
|
|
release_first_fetch.set()
|
|
|
|
qtbot.waitUntil(scan_control.last_scan_button.isEnabled)
|
|
qtbot.waitUntil(lambda: "line_scan" in scan_control._last_scan_lookup_memo)
|
|
assert get_ident() not in worker_threads
|
|
args, kwargs = scan_control.get_scan_parameters(bec_object=False)
|
|
assert args == ["samx", 0.0, 2.0]
|
|
assert kwargs["steps"] == 10
|