mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-07-10 21:40:59 +02:00
852 lines
34 KiB
Python
852 lines
34 KiB
Python
import shiboken6
|
|
from bec_lib import bl_states, messages
|
|
from qtpy.QtCore import QCoreApplication, QEvent, QRect, Qt
|
|
from qtpy.QtGui import QPainter, QPixmap
|
|
from qtpy.QtWidgets import QMessageBox, QStyleOptionViewItem
|
|
|
|
from bec_widgets.utils.toolbars.toolbar import ModularToolBar
|
|
from bec_widgets.utils.widget_io import WidgetIO
|
|
from bec_widgets.widgets.services.beamline_states import beamline_state_manager as manager_module
|
|
from bec_widgets.widgets.services.beamline_states import beamline_state_pill as pill_module
|
|
from bec_widgets.widgets.services.beamline_states.beamline_state_manager import BeamlineStateManager
|
|
from bec_widgets.widgets.services.beamline_states.beamline_state_pill import BeamlineStatePill
|
|
from bec_widgets.widgets.services.beamline_states.dialogs import AddBeamlineStateDialog
|
|
|
|
from .client_mocks import mocked_client
|
|
from .conftest import create_widget
|
|
|
|
|
|
def _state(name: str, state_type: str, parameters: dict | None = None):
|
|
return messages.BeamlineStateConfig(
|
|
name=name, state_type=state_type, parameters=parameters or {}
|
|
)
|
|
|
|
|
|
def _wire_state(
|
|
state_class: type[bl_states.BeamlineState], config: bl_states.BeamlineStateConfig
|
|
) -> messages.BeamlineStateConfig:
|
|
return messages.BeamlineStateConfig(
|
|
name=config.name,
|
|
state_type=state_class.__name__,
|
|
parameters=config.model_dump(exclude={"name"}),
|
|
)
|
|
|
|
|
|
def _limits_state(name: str = "limits", **overrides) -> messages.BeamlineStateConfig:
|
|
values = {
|
|
"device": "samx",
|
|
"signal": "samx",
|
|
"low_limit": 0.0,
|
|
"high_limit": 10.0,
|
|
"tolerance": 0.1,
|
|
}
|
|
values.update(overrides)
|
|
config = bl_states.DeviceWithinLimitsState.CONFIG_CLASS(name=name, **values)
|
|
return _wire_state(bl_states.DeviceWithinLimitsState, config)
|
|
|
|
|
|
def _shutter_state(
|
|
name: str = "shutter_open", device: str = "samy"
|
|
) -> messages.BeamlineStateConfig:
|
|
config = bl_states.ShutterState.CONFIG_CLASS(name=name, device=device)
|
|
return _wire_state(bl_states.ShutterState, config)
|
|
|
|
|
|
class _FakeScanInterlock:
|
|
def __init__(self, enabled: bool = False, states_watched: dict[str, str] | None = None):
|
|
self.enabled = enabled
|
|
self._states = dict(states_watched or {})
|
|
self.added: list[tuple[str, str]] = []
|
|
self.removed: list[str] = []
|
|
|
|
@property
|
|
def states_watched(self) -> dict[str, str]:
|
|
return dict(self._states)
|
|
|
|
def add_state_to_interlock(self, state_name: str, required_value: str = "valid") -> None:
|
|
self.added.append((state_name, required_value))
|
|
self._states[state_name] = required_value
|
|
|
|
def remove_state_from_interlock(self, state_name: str) -> None:
|
|
self.removed.append(state_name)
|
|
self._states.pop(state_name, None)
|
|
|
|
|
|
def _install_fake_scan_interlock(
|
|
manager: BeamlineStateManager, fake_interlock: _FakeScanInterlock
|
|
) -> None:
|
|
manager._scan_interlock = fake_interlock
|
|
manager._refresh_scan_interlock()
|
|
|
|
|
|
def test_beamline_state_pill_updates_from_message(qtbot, mocked_client):
|
|
pill = create_widget(qtbot, BeamlineStatePill, state_name="shutter_open", client=mocked_client)
|
|
pill.update_state({"name": "shutter_open", "status": "valid", "label": "Shutter is open."}, {})
|
|
|
|
assert pill._state_name == "shutter_open"
|
|
assert pill._name_label.text() == "shutter_open"
|
|
assert pill._status_label.text() == "VALID"
|
|
assert pill._detail_label.text() == "Shutter is open."
|
|
assert not pill._icon_label.pixmap().isNull()
|
|
assert pill.toolTip() == "Shutter is open."
|
|
|
|
|
|
def test_beamline_state_pill_ignores_other_states(qtbot, mocked_client):
|
|
pill = create_widget(qtbot, BeamlineStatePill, state_name="shutter_open", client=mocked_client)
|
|
pill.update_state(
|
|
{"name": "other_state", "status": "invalid", "label": "Should be ignored."}, {}
|
|
)
|
|
|
|
assert pill._status_label.text() == "UNKNOWN"
|
|
assert pill.toolTip() == "No state information available."
|
|
|
|
|
|
def test_beamline_state_pill_expands_and_emits_updated_limits(qtbot, mocked_client):
|
|
limits_pill = create_widget(qtbot, BeamlineStatePill, state_name="limits", client=mocked_client)
|
|
limits_pill.set_state_config(_limits_state())
|
|
|
|
assert limits_pill._settings.isHidden()
|
|
assert limits_pill._config_form is None
|
|
assert not limits_pill._update_button.isEnabled()
|
|
assert not limits_pill._revert_button.isEnabled()
|
|
|
|
qtbot.mouseClick(limits_pill._header, Qt.MouseButton.LeftButton)
|
|
assert limits_pill._config_form is not None
|
|
high_limit = limits_pill._config_form.input_widget("high_limit")
|
|
high_limit.setValue(20.0)
|
|
|
|
assert not limits_pill._settings.isHidden()
|
|
assert limits_pill._update_button.isEnabled()
|
|
assert limits_pill._revert_button.isEnabled()
|
|
assert (
|
|
limits_pill._config_form.field_widget("high_limit").property("beamlineStateDirty") is True
|
|
)
|
|
assert limits_pill._config_form.get_data()["device"] == "samx"
|
|
assert limits_pill.edited_config().high_limit == 20.0
|
|
|
|
with qtbot.waitSignal(limits_pill.update_requested) as signal:
|
|
limits_pill._update_button.click()
|
|
|
|
assert signal.args[0] == "limits"
|
|
assert isinstance(signal.args[1], bl_states.DeviceWithinLimitsState.CONFIG_CLASS)
|
|
assert signal.args[1].device == "samx"
|
|
assert signal.args[1].signal == "samx"
|
|
assert signal.args[1].low_limit == 0.0
|
|
assert signal.args[1].high_limit == 20.0
|
|
assert signal.args[1].tolerance == 0.1
|
|
assert not limits_pill._settings.isHidden()
|
|
|
|
|
|
def test_beamline_state_pill_first_expand_uses_config_class_without_rebuild(
|
|
qtbot, mocked_client, monkeypatch
|
|
):
|
|
limits_pill = create_widget(qtbot, BeamlineStatePill, state_name="limits", client=mocked_client)
|
|
set_model_calls = []
|
|
original_set_model = pill_module.PydanticWidgetForm.set_model
|
|
|
|
def set_model_spy(self, model, data=None):
|
|
set_model_calls.append(model)
|
|
return original_set_model(self, model, data=data)
|
|
|
|
monkeypatch.setattr(pill_module.PydanticWidgetForm, "set_model", set_model_spy)
|
|
limits_pill.set_state_config(_limits_state())
|
|
|
|
limits_pill.set_expanded(True)
|
|
assert limits_pill._config_form is not None
|
|
assert set_model_calls == []
|
|
|
|
|
|
def test_beamline_state_pill_reverts_changed_settings(qtbot, mocked_client):
|
|
limits_pill = create_widget(qtbot, BeamlineStatePill, state_name="limits", client=mocked_client)
|
|
limits_pill.set_state_config(_limits_state())
|
|
|
|
limits_pill.set_expanded(True)
|
|
assert limits_pill._config_form is not None
|
|
low_limit = limits_pill._config_form.input_widget("low_limit")
|
|
low_limit.setValue(-5.0)
|
|
|
|
assert limits_pill._update_button.isEnabled()
|
|
assert limits_pill._config_form.field_widget("low_limit").property("beamlineStateDirty") is True
|
|
|
|
limits_pill._revert_button.click()
|
|
|
|
assert low_limit.value() == 0.0
|
|
assert not limits_pill._update_button.isEnabled()
|
|
assert not limits_pill._revert_button.isEnabled()
|
|
assert (
|
|
limits_pill._config_form.field_widget("low_limit").property("beamlineStateDirty") is False
|
|
)
|
|
|
|
|
|
def test_beamline_state_pill_releases_form_on_collapse(qtbot, mocked_client):
|
|
limits_pill = create_widget(qtbot, BeamlineStatePill, state_name="limits", client=mocked_client)
|
|
limits_pill.set_state_config(_limits_state())
|
|
|
|
limits_pill.set_expanded(True)
|
|
device_widget = limits_pill._config_form.input_widget("device")
|
|
|
|
limits_pill.set_expanded(False)
|
|
QCoreApplication.sendPostedEvents(None, QEvent.Type.DeferredDelete)
|
|
|
|
assert limits_pill._config_form is None
|
|
assert not shiboken6.isValid(device_widget)
|
|
|
|
limits_pill.set_expanded(True)
|
|
|
|
assert limits_pill._config_form is not None
|
|
assert limits_pill._config_form.input_widget("high_limit").value() == 10.0
|
|
assert not limits_pill._update_button.isEnabled()
|
|
|
|
|
|
def test_beamline_state_pill_collapse_discards_unsaved_edits(qtbot, mocked_client):
|
|
limits_pill = create_widget(qtbot, BeamlineStatePill, state_name="limits", client=mocked_client)
|
|
limits_pill.set_state_config(_limits_state())
|
|
|
|
limits_pill.set_expanded(True)
|
|
limits_pill._config_form.input_widget("high_limit").setValue(20.0)
|
|
assert limits_pill._update_button.isEnabled()
|
|
|
|
limits_pill.set_expanded(False)
|
|
|
|
assert limits_pill._config_form is None
|
|
|
|
limits_pill.set_expanded(True)
|
|
|
|
assert limits_pill._config_form.input_widget("high_limit").value() == 10.0
|
|
assert not limits_pill._update_button.isEnabled()
|
|
|
|
|
|
def test_beamline_state_pill_does_not_override_themed_input_controls(qtbot, mocked_client):
|
|
limits_pill = create_widget(qtbot, BeamlineStatePill, state_name="limits", client=mocked_client)
|
|
limits_pill.set_state_config(_limits_state())
|
|
limits_pill.set_expanded(True)
|
|
|
|
stylesheet = limits_pill.styleSheet()
|
|
|
|
assert "QAbstractSpinBox" not in stylesheet
|
|
assert "QComboBox" not in stylesheet
|
|
assert "QCheckBox::indicator" not in stylesheet
|
|
|
|
|
|
def test_beamline_state_manager_adds_and_removes_pills(qtbot, mocked_client):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
beamline_state_manager.update_available_states(
|
|
{
|
|
"states": [
|
|
_state("shutter_open", "ShutterState"),
|
|
_state("limits", "DeviceWithinLimitsState"),
|
|
]
|
|
},
|
|
{},
|
|
)
|
|
|
|
assert sorted(beamline_state_manager._state_pills) == ["limits", "shutter_open"]
|
|
assert beamline_state_manager._model.rowCount() == 2
|
|
assert beamline_state_manager._state_pills["shutter_open"]._name_label.text() == "shutter_open"
|
|
assert not beamline_state_manager._empty_label.isVisible()
|
|
|
|
beamline_state_manager._state_pills["limits"].update_state(
|
|
{"name": "limits", "status": "valid", "label": "Within limits."}, {}
|
|
)
|
|
summary = beamline_state_manager.state_summary()
|
|
assert summary["limits"] == {"status": "valid", "label": "Within limits."}
|
|
assert summary["shutter_open"]["status"] == "unknown"
|
|
|
|
beamline_state_manager.update_available_states(
|
|
{"states": [_state("limits", "DeviceWithinLimitsState")]}, {}
|
|
)
|
|
|
|
assert sorted(beamline_state_manager._state_pills) == ["limits"]
|
|
assert beamline_state_manager._model.rowCount() == 1
|
|
|
|
|
|
def test_beamline_state_manager_ignores_unchanged_available_states(qtbot, mocked_client):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
content = {"states": [_limits_state()]}
|
|
|
|
beamline_state_manager.update_available_states(content, {})
|
|
pill = beamline_state_manager._state_pills["limits"]
|
|
|
|
beamline_state_manager.update_available_states(content, {})
|
|
|
|
assert beamline_state_manager._state_pills["limits"] is pill
|
|
assert pill._config_form is None
|
|
|
|
|
|
def test_beamline_state_manager_adds_state_without_recreating_existing_pills(qtbot, mocked_client):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
limits_state = _limits_state()
|
|
shutter_state = _state("shutter_open", "ShutterState")
|
|
|
|
beamline_state_manager.update_available_states({"states": [limits_state]}, {})
|
|
pill = beamline_state_manager._state_pills["limits"]
|
|
pill.set_expanded(True)
|
|
config_form = pill._config_form
|
|
|
|
beamline_state_manager.update_available_states({"states": [limits_state, shutter_state]}, {})
|
|
|
|
assert beamline_state_manager._state_pills["limits"] is pill
|
|
assert pill._config_form is config_form
|
|
assert pill.is_expanded()
|
|
assert sorted(beamline_state_manager._state_pills) == ["limits", "shutter_open"]
|
|
|
|
|
|
def test_beamline_state_manager_header_click_expands_pill_once(qtbot, mocked_client):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
beamline_state_manager.update_available_states(
|
|
{"states": [_state("limits", "DeviceWithinLimitsState", {"device": "samx"})]}, {}
|
|
)
|
|
|
|
pill = beamline_state_manager._state_pills["limits"]
|
|
assert pill._settings.isHidden()
|
|
|
|
qtbot.mouseClick(pill._header, Qt.MouseButton.LeftButton)
|
|
|
|
assert not pill._settings.isHidden()
|
|
|
|
|
|
def test_beamline_state_manager_preserves_expanded_pill_on_refresh(qtbot, mocked_client):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
state = _state("limits", "DeviceWithinLimitsState", {"device": "samx", "high_limit": 10.0})
|
|
beamline_state_manager.update_available_states({"states": [state]}, {})
|
|
|
|
beamline_state_manager._state_pills["limits"].set_expanded(True)
|
|
beamline_state_manager.update_available_states({"states": [state]}, {})
|
|
|
|
assert beamline_state_manager._state_pills["limits"].is_expanded()
|
|
assert not beamline_state_manager._state_pills["limits"]._settings.isHidden()
|
|
|
|
|
|
def test_beamline_state_manager_filters_status(qtbot, mocked_client):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
beamline_state_manager.update_available_states(
|
|
{"states": [_shutter_state(), _limits_state()]}, {}
|
|
)
|
|
|
|
assert isinstance(beamline_state_manager._toolbar, ModularToolBar)
|
|
assert not beamline_state_manager._toolbar.components.exists("refresh")
|
|
|
|
beamline_state_manager._state_pills["limits"].update_state(
|
|
{"name": "limits", "status": "valid", "label": "Within limits."}, {}
|
|
)
|
|
beamline_state_manager._state_pills["shutter_open"].update_state(
|
|
{"name": "shutter_open", "status": "invalid", "label": "Closed."}, {}
|
|
)
|
|
beamline_state_manager._selected_statuses = {"valid"}
|
|
beamline_state_manager._apply_filters()
|
|
|
|
assert not beamline_state_manager._hidden_summary.isHidden()
|
|
assert "1 state is hidden" in beamline_state_manager._hidden_summary.text()
|
|
assert not beamline_state_manager._view.isRowHidden(
|
|
beamline_state_manager._model.index_for_name("limits").row()
|
|
)
|
|
assert beamline_state_manager._view.isRowHidden(
|
|
beamline_state_manager._model.index_for_name("shutter_open").row()
|
|
)
|
|
|
|
beamline_state_manager._hidden_summary.click()
|
|
|
|
assert not beamline_state_manager._view.isRowHidden(
|
|
beamline_state_manager._model.index_for_name("shutter_open").row()
|
|
)
|
|
assert shiboken6.isValid(beamline_state_manager._state_pills["shutter_open"])
|
|
|
|
beamline_state_manager._hidden_summary.click()
|
|
|
|
assert beamline_state_manager._view.isRowHidden(
|
|
beamline_state_manager._model.index_for_name("shutter_open").row()
|
|
)
|
|
assert shiboken6.isValid(beamline_state_manager._state_pills["shutter_open"])
|
|
|
|
|
|
def test_beamline_state_manager_status_filter_reacts_to_state_changes(qtbot, mocked_client):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
beamline_state_manager.update_available_states(
|
|
{"states": [_state("limits", "DeviceWithinLimitsState", {"device": "samx"})]}, {}
|
|
)
|
|
|
|
beamline_state_manager._selected_statuses = {"valid"}
|
|
beamline_state_manager._state_pills["limits"].update_state(
|
|
{"name": "limits", "status": "valid", "label": "Within limits."}, {}
|
|
)
|
|
|
|
assert beamline_state_manager._hidden_summary.isHidden()
|
|
|
|
beamline_state_manager._state_pills["limits"].update_state(
|
|
{"name": "limits", "status": "invalid", "label": "Out of limits."}, {}
|
|
)
|
|
|
|
assert not beamline_state_manager._hidden_summary.isHidden()
|
|
assert beamline_state_manager._view.isRowHidden(
|
|
beamline_state_manager._model.index_for_name("limits").row()
|
|
)
|
|
|
|
|
|
def test_beamline_state_manager_filters_devices(qtbot, mocked_client, monkeypatch):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
beamline_state_manager.update_available_states(
|
|
{
|
|
"states": [
|
|
_limits_state(name="samx_limits"),
|
|
_limits_state(name="samy_limits", device="samy"),
|
|
]
|
|
},
|
|
{},
|
|
)
|
|
|
|
beamline_state_manager._device_filter_text = "samx"
|
|
beamline_state_manager._apply_filters()
|
|
|
|
assert not beamline_state_manager._hidden_summary.isHidden()
|
|
assert "1 state is hidden" in beamline_state_manager._hidden_summary.text()
|
|
|
|
captured = {}
|
|
|
|
class FakeDeviceFilterDialog:
|
|
def __init__(self, devices, selected_devices, device_filter_text, parent):
|
|
captured["devices"] = devices
|
|
captured["selected_devices"] = selected_devices
|
|
captured["device_filter_text"] = device_filter_text
|
|
captured["parent"] = parent
|
|
|
|
def exec(self):
|
|
return 0
|
|
|
|
monkeypatch.setattr(manager_module, "DeviceFilterDialog", FakeDeviceFilterDialog)
|
|
|
|
beamline_state_manager.open_device_filter_dialog()
|
|
|
|
assert captured["devices"] == ["samx", "samy"]
|
|
assert captured["device_filter_text"] == "samx"
|
|
assert captured["parent"] is beamline_state_manager
|
|
|
|
|
|
def test_beamline_state_manager_collapse_all(qtbot, mocked_client):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
beamline_state_manager.update_available_states(
|
|
{"states": [_limits_state(), _state("shutter_open", "ShutterState", {"device": "samy"})]},
|
|
{},
|
|
)
|
|
|
|
for pill in beamline_state_manager._state_pills.values():
|
|
pill.set_expanded(True)
|
|
assert all(pill.is_expanded() for pill in beamline_state_manager._state_pills.values())
|
|
|
|
collapse_action = beamline_state_manager._toolbar.components.get_action("collapse_all")
|
|
collapse_action.action.trigger()
|
|
|
|
assert not any(pill.is_expanded() for pill in beamline_state_manager._state_pills.values())
|
|
|
|
|
|
def test_beamline_state_manager_backend_echo_repopulates_expanded_pill(qtbot, mocked_client):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
beamline_state_manager.update_available_states({"states": [_limits_state()]}, {})
|
|
|
|
pill = beamline_state_manager._state_pills["limits"]
|
|
pill.set_expanded(True)
|
|
high_limit = pill._config_form.input_widget("high_limit")
|
|
high_limit.setValue(20.0)
|
|
|
|
assert pill._update_button.isEnabled()
|
|
|
|
beamline_state_manager.update_available_states({"states": [_limits_state(high_limit=20.0)]}, {})
|
|
|
|
assert pill.is_expanded()
|
|
assert high_limit.value() == 20.0
|
|
assert not pill._update_button.isEnabled()
|
|
assert pill._config_form.dirty_fields() == set()
|
|
|
|
|
|
def test_beamline_state_manager_updates_state_parameters(qtbot, mocked_client):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
beamline_state_manager.update_available_states({"states": [_limits_state()]}, {})
|
|
|
|
class StateClient:
|
|
def __init__(self):
|
|
self.parameters = None
|
|
|
|
def update_parameters(self, **kwargs):
|
|
self.parameters = kwargs
|
|
|
|
class StateManager:
|
|
def __init__(self):
|
|
self.limits = StateClient()
|
|
|
|
mocked_client.beamline_states = StateManager()
|
|
pill = beamline_state_manager._state_pills["limits"]
|
|
pill.set_expanded(True)
|
|
high_limit = pill._config_form.input_widget("high_limit")
|
|
high_limit.setValue(20.0)
|
|
|
|
assert pill._update_button.isEnabled()
|
|
|
|
beamline_state_manager._update_state_parameters("limits", pill.edited_config())
|
|
|
|
assert mocked_client.beamline_states.limits.parameters == {
|
|
"device": "samx",
|
|
"signal": "samx",
|
|
"low_limit": 0.0,
|
|
"high_limit": 20.0,
|
|
"tolerance": 0.1,
|
|
}
|
|
assert not pill._update_button.isEnabled()
|
|
assert pill._config_form.field_widget("high_limit").property("beamlineStateDirty") is False
|
|
|
|
|
|
def test_beamline_state_manager_removes_state(qtbot, mocked_client, monkeypatch):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
|
|
class StateManager:
|
|
def __init__(self):
|
|
self.deleted = None
|
|
|
|
def delete(self, state_name):
|
|
self.deleted = state_name
|
|
|
|
mocked_client.beamline_states = StateManager()
|
|
monkeypatch.setattr(
|
|
QMessageBox, "question", lambda *args, **kwargs: QMessageBox.StandardButton.Yes
|
|
)
|
|
|
|
beamline_state_manager._remove_state_requested("limits")
|
|
|
|
assert mocked_client.beamline_states.deleted == "limits"
|
|
|
|
|
|
def test_beamline_state_pill_emits_interlock_toggle_request(qtbot, mocked_client):
|
|
pill = create_widget(qtbot, BeamlineStatePill, state_name="limits", client=mocked_client)
|
|
|
|
with qtbot.waitSignal(pill.scan_interlock_toggle_requested) as include_signal:
|
|
pill._interlock_button.click()
|
|
|
|
assert include_signal.args == ["limits", True]
|
|
|
|
pill.set_scan_interlock("valid", False)
|
|
|
|
with qtbot.waitSignal(pill.scan_interlock_toggle_requested) as exclude_signal:
|
|
pill._interlock_button.click()
|
|
|
|
assert exclude_signal.args == ["limits", False]
|
|
|
|
|
|
def test_beamline_state_pill_included_state_forces_card_background(qtbot, mocked_client):
|
|
pill = create_widget(qtbot, BeamlineStatePill, state_name="limits", client=mocked_client)
|
|
colors = BeamlineStatePill._state_colors("unknown")
|
|
|
|
assert f"border: 1px solid {colors['card_border']}" not in pill.styleSheet().split(":hover")[0]
|
|
assert not pill._shadow.isEnabled()
|
|
|
|
pill.set_scan_interlock("valid", False)
|
|
|
|
assert f"border: 1px solid {colors['card_border']}" in pill.styleSheet()
|
|
assert pill._shadow.isEnabled()
|
|
assert "Watched by the scan interlock" in pill._interlock_button.toolTip()
|
|
|
|
pill.set_scan_interlock(None, False)
|
|
|
|
assert not pill._shadow.isEnabled()
|
|
assert "Not watched by the scan interlock" in pill._interlock_button.toolTip()
|
|
|
|
|
|
def test_beamline_state_pill_triggered_interlock_animates(qtbot, mocked_client):
|
|
pill = create_widget(qtbot, BeamlineStatePill, state_name="limits", client=mocked_client)
|
|
|
|
pill.set_scan_interlock("valid", True)
|
|
|
|
assert pill._interlock_animation.state() == pill._interlock_animation.State.Running
|
|
pill.interlock_pulse = 0.5
|
|
|
|
stylesheet = pill.styleSheet()
|
|
assert "border: 2px solid" in stylesheet
|
|
assert "qlineargradient" in stylesheet
|
|
|
|
pill.set_scan_interlock("valid", False)
|
|
|
|
assert pill._interlock_animation.state() == pill._interlock_animation.State.Stopped
|
|
assert pill._interlock_pulse == 0.0
|
|
assert "border: 2px solid" not in pill.styleSheet()
|
|
assert "border: 1px solid" in pill.styleSheet()
|
|
|
|
|
|
def test_beamline_state_pill_traveling_gradient_keeps_stops_sorted():
|
|
for phase in (0.0, 0.2, 0.5, 0.8, 1.0):
|
|
gradient = BeamlineStatePill._traveling_gradient("#101010", "#ff0000", phase)
|
|
positions = [
|
|
float(stop.split()[0].removeprefix("stop:"))
|
|
for stop in gradient.split(", ")
|
|
if stop.startswith("stop:")
|
|
]
|
|
assert positions == sorted(positions)
|
|
assert positions[0] == 0.0
|
|
assert positions[-1] == 1.0
|
|
|
|
|
|
def test_beamline_state_manager_toolbar_scan_interlock_on_right(qtbot, mocked_client):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
|
|
bundle = beamline_state_manager._toolbar.bundles["beamline_state_manager"]
|
|
assert list(bundle.bundle_actions)[-3:] == [
|
|
"scan_interlock_spacer",
|
|
"separator",
|
|
"scan_interlock",
|
|
]
|
|
|
|
spacer_action = beamline_state_manager._toolbar.components.get_action("scan_interlock_spacer")
|
|
assert spacer_action.container.sizePolicy().horizontalPolicy() == (
|
|
spacer_action.container.sizePolicy().Policy.Expanding
|
|
)
|
|
|
|
interlock_action = beamline_state_manager._toolbar.components.get_action("scan_interlock")
|
|
assert interlock_action.action.isCheckable()
|
|
assert not interlock_action.action.isChecked()
|
|
assert "disabled" in interlock_action.action.toolTip()
|
|
|
|
|
|
def test_beamline_state_manager_groups_states_under_headers(qtbot, mocked_client):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
beamline_state_manager.update_available_states(
|
|
{"states": [_limits_state(), _shutter_state()]}, {}
|
|
)
|
|
model = beamline_state_manager._model
|
|
|
|
assert model.rowCount() == 2
|
|
assert model.data(model.index(0, 0), model.HeaderRole) is None
|
|
|
|
_install_fake_scan_interlock(
|
|
beamline_state_manager, _FakeScanInterlock(states_watched={"shutter_open": "valid"})
|
|
)
|
|
|
|
assert model.rowCount() == 4
|
|
assert model.data(model.index(0, 0), model.HeaderRole) == model.INTERLOCK_HEADER
|
|
assert model.data(model.index(0, 0), Qt.ItemDataRole.DisplayRole) == "Scan interlock states"
|
|
assert model.data(model.index(1, 0), model.NameRole) == "shutter_open"
|
|
assert model.data(model.index(2, 0), model.HeaderRole) == model.OTHERS_HEADER
|
|
assert (
|
|
model.data(model.index(2, 0), Qt.ItemDataRole.DisplayRole)
|
|
== "Not included in scan interlock"
|
|
)
|
|
assert model.data(model.index(3, 0), model.NameRole) == "limits"
|
|
assert sorted(beamline_state_manager._state_pills) == ["limits", "shutter_open"]
|
|
|
|
_install_fake_scan_interlock(
|
|
beamline_state_manager,
|
|
_FakeScanInterlock(states_watched={"shutter_open": "valid", "limits": "valid"}),
|
|
)
|
|
|
|
assert model.rowCount() == 3
|
|
assert model.data(model.index(0, 0), model.HeaderRole) == model.INTERLOCK_HEADER
|
|
assert model.data(model.index(1, 0), model.NameRole) == "limits"
|
|
assert model.data(model.index(2, 0), model.NameRole) == "shutter_open"
|
|
|
|
_install_fake_scan_interlock(beamline_state_manager, _FakeScanInterlock())
|
|
|
|
assert model.rowCount() == 2
|
|
assert model.data(model.index(0, 0), model.HeaderRole) is None
|
|
|
|
|
|
def test_beamline_state_manager_header_visibility_follows_filters(qtbot, mocked_client):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
beamline_state_manager.update_available_states(
|
|
{"states": [_limits_state(), _shutter_state()]}, {}
|
|
)
|
|
_install_fake_scan_interlock(
|
|
beamline_state_manager, _FakeScanInterlock(states_watched={"shutter_open": "valid"})
|
|
)
|
|
|
|
beamline_state_manager._state_pills["limits"].update_state(
|
|
{"name": "limits", "status": "valid", "label": "Within limits."}, {}
|
|
)
|
|
beamline_state_manager._state_pills["shutter_open"].update_state(
|
|
{"name": "shutter_open", "status": "invalid", "label": "Closed."}, {}
|
|
)
|
|
beamline_state_manager._selected_statuses = {"valid"}
|
|
beamline_state_manager._apply_filters()
|
|
|
|
model = beamline_state_manager._model
|
|
assert beamline_state_manager._view.isRowHidden(0)
|
|
assert beamline_state_manager._view.isRowHidden(model.index_for_name("shutter_open").row())
|
|
assert not beamline_state_manager._view.isRowHidden(2)
|
|
assert not beamline_state_manager._view.isRowHidden(model.index_for_name("limits").row())
|
|
|
|
beamline_state_manager.clear_filters()
|
|
|
|
assert not beamline_state_manager._view.isRowHidden(0)
|
|
|
|
|
|
def test_beamline_state_manager_paints_section_headers(qtbot, mocked_client):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
beamline_state_manager.update_available_states(
|
|
{"states": [_limits_state(), _shutter_state()]}, {}
|
|
)
|
|
_install_fake_scan_interlock(
|
|
beamline_state_manager,
|
|
_FakeScanInterlock(enabled=True, states_watched={"shutter_open": "valid"}),
|
|
)
|
|
|
|
model = beamline_state_manager._model
|
|
delegate = beamline_state_manager._delegate
|
|
header_index = model.index(0, 0)
|
|
|
|
assert delegate.sizeHint(QStyleOptionViewItem(), header_index).height() == (
|
|
delegate.HEADER_HEIGHT
|
|
)
|
|
assert model.flags(header_index) == Qt.ItemFlag.NoItemFlags
|
|
|
|
target = QPixmap(400, delegate.HEADER_HEIGHT)
|
|
target.fill(Qt.GlobalColor.transparent)
|
|
painter = QPainter(target)
|
|
option = QStyleOptionViewItem()
|
|
option.rect = QRect(0, 0, 400, delegate.HEADER_HEIGHT)
|
|
delegate.paint(painter, option, header_index)
|
|
delegate.paint(painter, option, model.index(2, 0))
|
|
painter.end()
|
|
|
|
|
|
def test_beamline_state_manager_marks_triggered_pills(qtbot, mocked_client):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
beamline_state_manager.update_available_states({"states": [_shutter_state()]}, {})
|
|
fake_interlock = _FakeScanInterlock(enabled=True, states_watched={"shutter_open": "valid"})
|
|
_install_fake_scan_interlock(beamline_state_manager, fake_interlock)
|
|
|
|
pill = beamline_state_manager._state_pills["shutter_open"]
|
|
assert pill._interlock_required_status == "valid"
|
|
assert pill._interlock_triggered
|
|
|
|
pill.update_state({"name": "shutter_open", "status": "valid", "label": "Open."}, {})
|
|
|
|
assert not pill._interlock_triggered
|
|
|
|
pill.update_state({"name": "shutter_open", "status": "invalid", "label": "Closed."}, {})
|
|
|
|
assert pill._interlock_triggered
|
|
|
|
fake_interlock.enabled = False
|
|
beamline_state_manager._refresh_scan_interlock()
|
|
|
|
assert not pill._interlock_triggered
|
|
assert pill._interlock_required_status == "valid"
|
|
|
|
|
|
def test_beamline_state_manager_toolbar_toggle_writes_backend(qtbot, mocked_client):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
fake_interlock = _FakeScanInterlock()
|
|
_install_fake_scan_interlock(beamline_state_manager, fake_interlock)
|
|
interlock_action = beamline_state_manager._toolbar.components.get_action("scan_interlock")
|
|
|
|
interlock_action.action.setChecked(True)
|
|
|
|
assert fake_interlock.enabled is True
|
|
assert beamline_state_manager._interlock_enabled is True
|
|
assert "armed" in interlock_action.action.toolTip()
|
|
|
|
interlock_action.action.setChecked(False)
|
|
|
|
assert fake_interlock.enabled is False
|
|
assert "disabled" in interlock_action.action.toolTip()
|
|
|
|
|
|
def test_beamline_state_manager_pill_toggle_calls_backend(qtbot, mocked_client):
|
|
beamline_state_manager = create_widget(qtbot, BeamlineStateManager, client=mocked_client)
|
|
beamline_state_manager.update_available_states({"states": [_limits_state()]}, {})
|
|
fake_interlock = _FakeScanInterlock()
|
|
_install_fake_scan_interlock(beamline_state_manager, fake_interlock)
|
|
|
|
beamline_state_manager._state_pills["limits"]._interlock_button.click()
|
|
|
|
assert fake_interlock.added == [("limits", "valid")]
|
|
# The toggle refreshes immediately, so the state is reflected without a backend notification.
|
|
assert beamline_state_manager._interlock_states == {"limits": "valid"}
|
|
|
|
beamline_state_manager._state_pills["limits"]._interlock_button.click()
|
|
|
|
assert fake_interlock.removed == ["limits"]
|
|
assert beamline_state_manager._interlock_states == {}
|
|
|
|
|
|
def test_add_beamline_state_dialog_uses_generated_widgets_and_normalizes_name(qtbot, mocked_client):
|
|
add_state_dialog = create_widget(qtbot, AddBeamlineStateDialog, client=mocked_client)
|
|
limits_index = add_state_dialog._type_combo.findText(bl_states.DeviceWithinLimitsState.__name__)
|
|
assert limits_index >= 0
|
|
add_state_dialog._type_combo.setCurrentIndex(limits_index)
|
|
|
|
assert add_state_dialog._config_form.model is bl_states.DeviceWithinLimitsState.CONFIG_CLASS
|
|
|
|
name = add_state_dialog._config_form.input_widget("name")
|
|
device = add_state_dialog._config_form.input_widget("device")
|
|
signal = add_state_dialog._config_form.input_widget("signal")
|
|
low_limit = add_state_dialog._config_form.field_widget("low_limit")
|
|
high_limit = add_state_dialog._config_form.field_widget("high_limit")
|
|
|
|
name.setText("samx-limits")
|
|
WidgetIO.set_value(device, "samx")
|
|
WidgetIO.set_value(signal, "samx")
|
|
low_limit.checkbox.setChecked(True)
|
|
high_limit.checkbox.setChecked(True)
|
|
high_limit.value_widget.setValue(15.0)
|
|
|
|
config = add_state_dialog.config()
|
|
|
|
assert config.name == "samx_limits"
|
|
assert config.device == "samx"
|
|
assert config.signal == "samx"
|
|
assert config.low_limit == 0.0
|
|
assert config.high_limit == 15.0
|
|
|
|
|
|
def test_add_beamline_state_dialog_generates_name_only_after_valid_device_selection(
|
|
qtbot, mocked_client
|
|
):
|
|
add_state_dialog = create_widget(qtbot, AddBeamlineStateDialog, client=mocked_client)
|
|
name = add_state_dialog._config_form.input_widget("name")
|
|
device = add_state_dialog._config_form.input_widget("device")
|
|
|
|
device.setCurrentText("s")
|
|
|
|
assert name.text() == ""
|
|
|
|
device.set_device("samx")
|
|
|
|
assert name.text() == "samx_device_within_limits_state"
|
|
|
|
|
|
def test_add_beamline_state_dialog_switches_state_type_without_collapsing(qtbot, mocked_client):
|
|
add_state_dialog = create_widget(qtbot, AddBeamlineStateDialog, client=mocked_client)
|
|
initial_height = add_state_dialog.height()
|
|
limits_index = add_state_dialog._type_combo.findText("DeviceWithinLimitsState")
|
|
assert limits_index >= 0
|
|
shutter_index = add_state_dialog._type_combo.findText("ShutterState")
|
|
assert shutter_index >= 0
|
|
|
|
add_state_dialog._type_combo.setCurrentIndex(shutter_index)
|
|
qtbot.wait(0)
|
|
|
|
assert add_state_dialog._config_form.model is bl_states.DeviceStateConfig
|
|
assert add_state_dialog._config_form_host.count() == 1
|
|
assert not add_state_dialog._config_form.isHidden()
|
|
assert not add_state_dialog._buttons.isHidden()
|
|
assert add_state_dialog.sizeHint().height() > add_state_dialog._buttons.sizeHint().height()
|
|
assert add_state_dialog.minimumHeight() == add_state_dialog.maximumHeight()
|
|
|
|
add_state_dialog._type_combo.setCurrentIndex(limits_index)
|
|
qtbot.wait(0)
|
|
|
|
assert add_state_dialog._config_form.model is bl_states.DeviceWithinLimitsState.CONFIG_CLASS
|
|
assert add_state_dialog.height() >= initial_height
|
|
assert add_state_dialog.minimumHeight() == add_state_dialog.maximumHeight()
|
|
|
|
|
|
def test_add_beamline_state_dialog_cleanup_deletes_device_widgets(qtbot, mocked_client):
|
|
add_state_dialog = create_widget(qtbot, AddBeamlineStateDialog, client=mocked_client)
|
|
device = add_state_dialog._config_form.input_widget("device")
|
|
signal = add_state_dialog._config_form.input_widget("signal")
|
|
|
|
add_state_dialog.reject()
|
|
assert shiboken6.isValid(device)
|
|
assert shiboken6.isValid(signal)
|
|
|
|
add_state_dialog.cleanup()
|
|
QCoreApplication.sendPostedEvents(None, QEvent.Type.DeferredDelete)
|
|
|
|
assert not shiboken6.isValid(device)
|
|
assert not shiboken6.isValid(signal)
|