mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-03-05 00:12:49 +01:00
test(device-manager): use mocked client for tests
This commit is contained in:
Jan Wyzula
@@ -344,6 +344,7 @@ class DeviceFormDialog(QtWidgets.QDialog):
|
||||
# Config unchanged, we can reuse previous connection status. Only do this if the new
|
||||
# connection status is UNKNOWN as the current validation should not test the connection.
|
||||
connection_status = self._validation_result[2]
|
||||
validation_msg = self._validation_result[3]
|
||||
except Exception:
|
||||
logger.debug(
|
||||
f"Device config validation changed for config: {device_config} compared to previous validation. Using status from recent validation."
|
||||
|
||||
@@ -4,7 +4,7 @@ from __future__ import annotations
|
||||
|
||||
from enum import IntEnum
|
||||
from functools import partial
|
||||
from typing import TYPE_CHECKING, Dict, List, Tuple
|
||||
from typing import TYPE_CHECKING, List, Tuple
|
||||
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_qthemes import apply_theme, material_icon
|
||||
@@ -12,16 +12,17 @@ from qtpy import QtCore, QtGui, QtWidgets
|
||||
|
||||
from bec_widgets.utils.colors import get_accent_colors
|
||||
from bec_widgets.utils.error_popups import SafeSlot
|
||||
from bec_widgets.widgets.control.device_manager.components import OphydValidation
|
||||
from bec_widgets.widgets.control.device_manager.components.ophyd_validation import (
|
||||
ConfigStatus,
|
||||
ConnectionStatus,
|
||||
get_validation_icons,
|
||||
)
|
||||
from bec_widgets.widgets.progress.bec_progressbar.bec_progressbar import BECProgressBar
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from bec_widgets.utils.colors import AccentColor
|
||||
from bec_widgets.widgets.control.device_manager.components.device_table.device_table import (
|
||||
_ValidationResultIter,
|
||||
)
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
@@ -234,22 +235,18 @@ class UploadRedisDialog(QtWidgets.QDialog):
|
||||
class UploadAction(IntEnum):
|
||||
"""Enum for upload actions."""
|
||||
|
||||
CANCEL = QtWidgets.QDialog.Rejected
|
||||
OK = QtWidgets.QDialog.Accepted
|
||||
CANCEL = QtWidgets.QDialog.DialogCode.Rejected
|
||||
OK = QtWidgets.QDialog.DialogCode.Accepted
|
||||
CONNECTION_TEST_REQUESTED = 999
|
||||
|
||||
# Signal to trigger upload after confirmation
|
||||
upload_confirmed = QtCore.Signal(int)
|
||||
# Request ophyd validation for all untested device connections
|
||||
# list of device configs, added: bool, connect: bool
|
||||
request_ophyd_validation = QtCore.Signal(list, bool, bool)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
parent,
|
||||
ophyd_test_widget: OphydValidation,
|
||||
device_configs: dict[str, Tuple[dict, int, int]] | None = None,
|
||||
):
|
||||
def __init__(self, parent, device_configs: dict[str, Tuple[dict, int, int]] | None = None):
|
||||
super().__init__(parent=parent)
|
||||
|
||||
self.device_configs: dict[str, Tuple[dict, int, int]] = device_configs or {}
|
||||
self.ophyd_test_widget = ophyd_test_widget
|
||||
self._transparent_button_style = "background-color: transparent; border: none;"
|
||||
|
||||
self.colors = get_accent_colors()
|
||||
@@ -267,14 +264,9 @@ class UploadRedisDialog(QtWidgets.QDialog):
|
||||
self.has_invalid_configs: int = 0
|
||||
self.has_untested_connections: int = 0
|
||||
self.has_cannot_connect: int = 0
|
||||
self._current_progress: int | None = None
|
||||
|
||||
self._setup_ui()
|
||||
self._update_ui()
|
||||
# Disable validation features if no ophyd test widget provided, else connect validation
|
||||
self._validation_connection = self.ophyd_test_widget.validation_completed.connect(
|
||||
self._update_from_ophyd_device_tests
|
||||
)
|
||||
|
||||
def set_device_config(self, device_configs: dict[str, Tuple[dict, int, int]]):
|
||||
"""
|
||||
@@ -288,18 +280,6 @@ class UploadRedisDialog(QtWidgets.QDialog):
|
||||
self.device_configs = device_configs
|
||||
self._update_ui()
|
||||
|
||||
def accept(self):
|
||||
self.cleanup()
|
||||
return super().accept()
|
||||
|
||||
def reject(self):
|
||||
self.cleanup()
|
||||
return super().reject()
|
||||
|
||||
def cleanup(self):
|
||||
"""Cleanup on dialog finish."""
|
||||
self.ophyd_test_widget.validation_completed.disconnect(self._validation_connection)
|
||||
|
||||
def _setup_ui(self):
|
||||
"""Setup the main UI for the dialog."""
|
||||
self.setWindowTitle("Upload Configuration to BEC Server")
|
||||
@@ -347,11 +327,6 @@ class UploadRedisDialog(QtWidgets.QDialog):
|
||||
button_layout.addWidget(self.validate_connections_btn)
|
||||
button_layout.addStretch()
|
||||
button_layout.addSpacing(16)
|
||||
|
||||
# Progress bar
|
||||
self._progress_bar = BECProgressBar(self)
|
||||
self._progress_bar.setVisible(False)
|
||||
button_layout.addWidget(self._progress_bar)
|
||||
action_layout.addLayout(button_layout)
|
||||
|
||||
# Status indicator
|
||||
@@ -498,7 +473,7 @@ class UploadRedisDialog(QtWidgets.QDialog):
|
||||
|
||||
@SafeSlot()
|
||||
def _validate_connections(self):
|
||||
"""Request validation of all untested connections."""
|
||||
"""Request validation of all untested connections. This will close the dialog."""
|
||||
testable_devices: List[dict] = []
|
||||
for _, (config, _, connection_status) in self.device_configs.items():
|
||||
if connection_status == ConnectionStatus.UNKNOWN.value:
|
||||
@@ -507,13 +482,8 @@ class UploadRedisDialog(QtWidgets.QDialog):
|
||||
testable_devices.append(config)
|
||||
|
||||
if len(testable_devices) > 0:
|
||||
self.validate_connections_btn.setEnabled(False)
|
||||
self._progress_bar.setVisible(True)
|
||||
self._progress_bar.maximum = len(testable_devices)
|
||||
self._progress_bar.minimum = 0
|
||||
self._progress_bar.set_value(0)
|
||||
self._current_progress = 0
|
||||
self.ophyd_test_widget.change_device_configs(testable_devices, added=True, connect=True)
|
||||
self.request_ophyd_validation.emit(testable_devices, True, True)
|
||||
self.done(self.UploadAction.CONNECTION_TEST_REQUESTED)
|
||||
|
||||
@SafeSlot()
|
||||
def _handle_upload(self):
|
||||
@@ -611,35 +581,40 @@ class UploadRedisDialog(QtWidgets.QDialog):
|
||||
return
|
||||
self.update_device_status(device_config, config_status, connection_status)
|
||||
|
||||
@SafeSlot(list)
|
||||
def _multiple_updates_from_ophyd_device_tests(self, validation_results: _ValidationResultIter):
|
||||
"""
|
||||
Callback slot for receiving multiple validation result updates from the ophyd test widget.
|
||||
|
||||
Args:
|
||||
validation_results (list): List of tuples containing (device_config, config_status, connection_status, validation_msg).
|
||||
"""
|
||||
for cfg, cfg_status, conn_status, val_msg in validation_results:
|
||||
self.update_device_status(cfg, cfg_status, conn_status)
|
||||
self._update_ui()
|
||||
|
||||
@SafeSlot(dict, int, int)
|
||||
def update_device_status(self, device_config: dict, config_status: int, connection_status: int):
|
||||
"""Update the status of a specific device."""
|
||||
# Update device config status
|
||||
self._update_device_configs(device_config, config_status, connection_status, "")
|
||||
# Recalculate summaries and UI state
|
||||
self._update_ui()
|
||||
|
||||
def _update_device_configs(
|
||||
self,
|
||||
device_config: dict[str, Any],
|
||||
config_status: int,
|
||||
connection_status: int,
|
||||
validation_msg: str,
|
||||
):
|
||||
device_name = device_config.get("name", "")
|
||||
old_config, _, _ = self.device_configs.get(device_name, (None, None, None))
|
||||
if old_config is not None:
|
||||
self.device_configs[device_name] = (device_config, config_status, connection_status)
|
||||
if self._current_progress is not None:
|
||||
self._current_progress += 1
|
||||
self._progress_bar.set_value(self._current_progress)
|
||||
if self._current_progress >= self._progress_bar.maximum:
|
||||
self._progress_bar.setVisible(False)
|
||||
self._progress_bar.set_value(0)
|
||||
self._current_progress = None
|
||||
self.validation_completed()
|
||||
self._update_ui()
|
||||
return
|
||||
|
||||
# Update UI sections
|
||||
self.config_section.add_device(device_config, config_status, connection_status)
|
||||
|
||||
# Recalculate summaries and UI state
|
||||
self._update_ui()
|
||||
|
||||
def validation_completed(self):
|
||||
"""Called when connection validation is completed."""
|
||||
self.validate_connections_btn.setEnabled(True)
|
||||
self._update_ui()
|
||||
else:
|
||||
# If device not found, add it
|
||||
self.config_section.add_device(device_config, config_status, connection_status)
|
||||
|
||||
|
||||
def main(): # pragma: no cover
|
||||
@@ -705,12 +680,7 @@ def main(): # pragma: no cover
|
||||
]
|
||||
configs = {cfg[0]["name"]: cfg for cfg in sample_configs}
|
||||
apply_theme("dark")
|
||||
from unittest import mock
|
||||
|
||||
ophyd_test_widget = mock.MagicMock(spec=OphydValidation)
|
||||
dialog = UploadRedisDialog(
|
||||
parent=None, device_configs=configs, ophyd_test_widget=ophyd_test_widget
|
||||
)
|
||||
dialog = UploadRedisDialog(parent=None, device_configs=configs)
|
||||
dialog.show()
|
||||
|
||||
sys.exit(app.exec_())
|
||||
|
||||
@@ -35,6 +35,7 @@ from bec_widgets.widgets.control.device_manager.components import (
|
||||
)
|
||||
from bec_widgets.widgets.control.device_manager.components._util import SharedSelectionSignal
|
||||
from bec_widgets.widgets.control.device_manager.components.ophyd_validation.ophyd_validation_utils import (
|
||||
ConfigStatus,
|
||||
ConnectionStatus,
|
||||
)
|
||||
from bec_widgets.widgets.services.device_browser.device_item.config_communicator import (
|
||||
@@ -57,7 +58,7 @@ class DeviceManagerDisplayWidget(DockAreaWidget):
|
||||
|
||||
request_ophyd_validation = Signal(list, bool, bool)
|
||||
|
||||
def __init__(self, parent=None, client=None, *args, **kwargs):
|
||||
def __init__(self, parent=None, *args, **kwargs):
|
||||
super().__init__(parent=parent, variant="compact", *args, **kwargs)
|
||||
|
||||
# Push to Redis dialog
|
||||
@@ -312,6 +313,13 @@ class DeviceManagerDisplayWidget(DockAreaWidget):
|
||||
configs = list(self.device_table_view.get_selected_device_configs())
|
||||
if not configs:
|
||||
configs = self.device_table_view.get_device_config()
|
||||
# Adjust the state of the icons in the device table view
|
||||
self.device_table_view.update_multiple_device_validations(
|
||||
[
|
||||
(cfg, ConfigStatus.UNKNOWN.value, ConnectionStatus.UNKNOWN.value, "")
|
||||
for cfg in configs
|
||||
]
|
||||
)
|
||||
self.request_ophyd_validation.emit(configs, True, connect)
|
||||
|
||||
def _update_config_enabled_button(self, enabled: bool):
|
||||
@@ -474,7 +482,10 @@ class DeviceManagerDisplayWidget(DockAreaWidget):
|
||||
validation_results = self.device_table_view.get_validation_results()
|
||||
# Create and show upload dialog
|
||||
self._upload_redis_dialog = UploadRedisDialog(
|
||||
parent=self, device_configs=validation_results, ophyd_test_widget=self.ophyd_test_view
|
||||
parent=self, device_configs=validation_results
|
||||
)
|
||||
self._upload_redis_dialog.request_ophyd_validation.connect(
|
||||
self.request_ophyd_validation.emit
|
||||
)
|
||||
|
||||
# Show dialog
|
||||
@@ -484,6 +495,10 @@ class DeviceManagerDisplayWidget(DockAreaWidget):
|
||||
self._push_composition_to_redis(action="set")
|
||||
elif reply == UploadRedisDialog.UploadAction.CANCEL:
|
||||
self.ophyd_test_view.cancel_all_validations()
|
||||
elif reply == UploadRedisDialog.UploadAction.CONNECTION_TEST_REQUESTED:
|
||||
return QMessageBox.information(
|
||||
self, "Connection Test Requested", "Running connection test on untested devices."
|
||||
)
|
||||
|
||||
def _push_composition_to_redis(self, action: ConfigAction):
|
||||
"""Push the current device composition to Redis."""
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
# pylint: skip-file
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from bec_lib.device import Device as BECDevice
|
||||
@@ -255,6 +256,13 @@ class DMMock:
|
||||
signals.append((device_name, signal_name, signal_info))
|
||||
return signals
|
||||
|
||||
def _get_redis_device_config(self) -> list[dict]:
|
||||
"""Mock method to emulate DeviceManager._get_redis_device_config."""
|
||||
configs = []
|
||||
for device in self.devices.values():
|
||||
configs.append(device._config)
|
||||
return configs
|
||||
|
||||
|
||||
DEVICES = [
|
||||
FakePositioner("samx", limits=[-10, 10], read_value=2.0),
|
||||
|
||||
@@ -210,8 +210,8 @@ class DeviceTable(BECWidget, QtWidgets.QWidget):
|
||||
|
||||
_auto_size_request = QtCore.Signal()
|
||||
|
||||
def __init__(self, parent: QtWidgets.QWidget | None = None):
|
||||
super().__init__(parent=parent)
|
||||
def __init__(self, parent: QtWidgets.QWidget | None = None, client=None):
|
||||
super().__init__(parent=parent, client=client)
|
||||
self.headers_key_map: dict[str, str] = {
|
||||
"Valid": "valid",
|
||||
"Connect": "connect",
|
||||
@@ -823,13 +823,14 @@ class DeviceTable(BECWidget, QtWidgets.QWidget):
|
||||
# Public API to be called via signals/slots
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
@SafeSlot(list)
|
||||
@SafeSlot(list, bool)
|
||||
def set_device_config(self, device_configs: _DeviceCfgIter, skip_validation: bool = False):
|
||||
"""
|
||||
Set the device config. This will clear any existing configs.
|
||||
|
||||
Args:
|
||||
device_configs (Iterable[dict[str, Any]]): The device configs to set.
|
||||
skip_validation (bool): Whether to skip validation for the set devices.
|
||||
"""
|
||||
self.set_busy(True, text="Loading device configurations...")
|
||||
with self.table_sort_on_hold:
|
||||
@@ -857,7 +858,7 @@ class DeviceTable(BECWidget, QtWidgets.QWidget):
|
||||
self.device_config_in_sync_with_redis.emit(in_sync_with_redis)
|
||||
self.set_busy(False, text="")
|
||||
|
||||
@SafeSlot(list)
|
||||
@SafeSlot(list, bool)
|
||||
def add_device_configs(self, device_configs: _DeviceCfgIter, skip_validation: bool = False):
|
||||
"""
|
||||
Add devices to the config. If a device already exists, it will be replaced. If the validation is
|
||||
@@ -866,6 +867,7 @@ class DeviceTable(BECWidget, QtWidgets.QWidget):
|
||||
|
||||
Args:
|
||||
device_configs (Iterable[dict[str, Any]]): The device configs to add.
|
||||
skip_validation (bool): Whether to skip validation for the added devices.
|
||||
"""
|
||||
self.set_busy(True, text="Adding device configurations...")
|
||||
already_in_table = []
|
||||
@@ -894,13 +896,14 @@ class DeviceTable(BECWidget, QtWidgets.QWidget):
|
||||
self.device_config_in_sync_with_redis.emit(in_sync_with_redis)
|
||||
self.set_busy(False, text="")
|
||||
|
||||
@SafeSlot(list)
|
||||
@SafeSlot(list, bool)
|
||||
def update_device_configs(self, device_configs: _DeviceCfgIter, skip_validation: bool = False):
|
||||
"""
|
||||
Update devices in the config. If a device does not exist, it will be added.
|
||||
|
||||
Args:
|
||||
device_configs (Iterable[dict[str, Any]]): The device configs to update.
|
||||
skip_validation (bool): Whether to skip validation for the updated devices.
|
||||
"""
|
||||
self.set_busy(True, text="Loading device configurations...")
|
||||
cfgs_updated = []
|
||||
|
||||
@@ -494,7 +494,14 @@ class OphydValidation(BECWidget, QtWidgets.QWidget):
|
||||
def device_table_config_changed(
|
||||
self, device_configs: list[dict[str, Any]], added: bool, skip_validation: bool
|
||||
) -> None:
|
||||
"""Slot to handle device config changes in the device table."""
|
||||
"""
|
||||
Slot to handle device config changes in the device table.
|
||||
|
||||
Args:
|
||||
device_configs (list[dict[str, Any]]): List of device configurations.
|
||||
added (bool): Whether the devices are added to the existing list.
|
||||
skip_validation (bool): Whether to skip validation for the added devices.
|
||||
"""
|
||||
self.change_device_configs(
|
||||
device_configs=device_configs, added=added, skip_validation=skip_validation
|
||||
)
|
||||
@@ -528,7 +535,6 @@ class OphydValidation(BECWidget, QtWidgets.QWidget):
|
||||
force_connect (bool, optional): Whether to force connection during validation. Defaults to False.
|
||||
timeout (float, optional): Timeout for connection attempt. Defaults to 5.0.
|
||||
skip_validation (bool, optional): Whether to skip validation for the added devices. Defaults to False.
|
||||
keep_device_item_in_list (bool, optional): Whether to keep the device item in the list after validation in success case.
|
||||
"""
|
||||
if not READY_TO_TEST:
|
||||
logger.error("Cannot change device configs: dependencies not available.")
|
||||
@@ -652,7 +658,7 @@ class OphydValidation(BECWidget, QtWidgets.QWidget):
|
||||
)
|
||||
widget.request_rerun_validation.connect(self._on_request_rerun_validation)
|
||||
self.list_widget.add_widget_item(device_name, widget)
|
||||
if skip_validation is False:
|
||||
if not skip_validation:
|
||||
self.__delayed_submit_test(widget, connect, force_connect, timeout)
|
||||
|
||||
def _remove_device(self, device_name: str) -> None:
|
||||
|
||||
@@ -57,6 +57,8 @@ from bec_widgets.widgets.control.device_manager.components.ophyd_validation.vali
|
||||
)
|
||||
from bec_widgets.widgets.utility.toggle.toggle import ToggleSwitch
|
||||
|
||||
from .client_mocks import mocked_client
|
||||
|
||||
|
||||
class TestConstants:
|
||||
"""Test class for constants and configuration values."""
|
||||
@@ -296,9 +298,9 @@ class TestDeviceTable:
|
||||
"""Test class for DeviceTable component."""
|
||||
|
||||
@pytest.fixture
|
||||
def device_table(self, qtbot) -> Generator[DeviceTable, None, None]:
|
||||
def device_table(self, qtbot, mocked_client) -> Generator[DeviceTable, None, None]:
|
||||
"""Fixture to create a DeviceTable instance."""
|
||||
table = DeviceTable()
|
||||
table = DeviceTable(client=mocked_client)
|
||||
qtbot.addWidget(table)
|
||||
qtbot.waitExposed(table)
|
||||
yield table
|
||||
@@ -997,7 +999,7 @@ class TestOphydValidation:
|
||||
assert label.text() == "Connect Legend:"
|
||||
|
||||
@pytest.fixture
|
||||
def ophyd_test(self, qtbot):
|
||||
def ophyd_test(self, qtbot, mocked_client):
|
||||
"""Fixture to create an OphydValidation instance. We patch the method that starts the polling loop to avoid side effects."""
|
||||
with (
|
||||
mock.patch(
|
||||
@@ -1009,7 +1011,7 @@ class TestOphydValidation:
|
||||
return_value=False,
|
||||
),
|
||||
):
|
||||
widget = OphydValidation()
|
||||
widget = OphydValidation(client=mocked_client)
|
||||
qtbot.addWidget(widget)
|
||||
qtbot.waitExposed(widget)
|
||||
yield widget
|
||||
@@ -1034,6 +1036,47 @@ class TestOphydValidation:
|
||||
qtbot.mouseClick(ophyd_test._stop_validation_button, QtCore.Qt.LeftButton)
|
||||
assert click_event.is_set()
|
||||
|
||||
def test_ophyd_test_keep_visible_after_validation(self, ophyd_test: OphydValidation, qtbot):
|
||||
"""Test the keep visible after validation logic."""
|
||||
# Initially false
|
||||
assert len(ophyd_test._keep_visible_after_validation) == 0
|
||||
|
||||
# Add device to keep visible
|
||||
ophyd_test.add_device_to_keep_visible_after_validation("device_1")
|
||||
assert "device_1" in ophyd_test._keep_visible_after_validation
|
||||
# Add second device
|
||||
ophyd_test.add_device_to_keep_visible_after_validation("device_2")
|
||||
assert "device_2" in ophyd_test._keep_visible_after_validation
|
||||
assert len(ophyd_test._keep_visible_after_validation) == 2
|
||||
|
||||
# Remove device
|
||||
ophyd_test.remove_device_to_keep_visible_after_validation("device_1")
|
||||
assert "device_1" not in ophyd_test._keep_visible_after_validation
|
||||
assert "device_2" in ophyd_test._keep_visible_after_validation
|
||||
|
||||
# Change config with skip validation and device in keep visible list
|
||||
with (
|
||||
mock.patch.object(
|
||||
ophyd_test, "_is_device_in_redis_session", return_value=True
|
||||
) as mock_is_device_in_redis_session,
|
||||
mock.patch.object(ophyd_test, "_add_device_config") as mock_add_device_config,
|
||||
mock.patch.object(
|
||||
ophyd_test, "_on_device_test_completed"
|
||||
) as mock_on_device_test_completed,
|
||||
):
|
||||
ophyd_test.change_device_configs(
|
||||
[{"name": "device_2", "deviceClass": "TestClass"}],
|
||||
added=True,
|
||||
skip_validation=False,
|
||||
)
|
||||
mock_add_device_config.assert_called_once()
|
||||
mock_on_device_test_completed.assert_called_once_with(
|
||||
{"name": "device_2", "deviceClass": "TestClass"},
|
||||
ConfigStatus.VALID.value,
|
||||
ConnectionStatus.CONNECTED.value,
|
||||
"Device already in session.",
|
||||
)
|
||||
|
||||
def test_ophyd_test_adding_devices(self, ophyd_test: OphydValidation, qtbot):
|
||||
"""Test adding devices to OphydValidation widget."""
|
||||
sample_devices = [
|
||||
|
||||
@@ -41,6 +41,8 @@ from bec_widgets.widgets.control.device_manager.components.ophyd_validation.ophy
|
||||
OphydValidation,
|
||||
)
|
||||
|
||||
from .client_mocks import mocked_client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def device_config() -> dict:
|
||||
@@ -164,6 +166,67 @@ class TestDeviceManagerViewDialogs:
|
||||
connection_settings_layout.count() == fields_in_config * 2
|
||||
) # Each field has a label and a widget
|
||||
|
||||
def test_device_form_dialog_help_methods(
|
||||
self, device_form_dialog: DeviceFormDialog, device_config, qtbot
|
||||
):
|
||||
"""Test help methods in DeviceFormDialog."""
|
||||
# Test handle devices already in session results
|
||||
dialog = device_form_dialog
|
||||
|
||||
# Test _handle_devices_already_in_session_results
|
||||
with mock.patch.object(dialog, "_handle_validation_result") as mock_handle_validation:
|
||||
dialog._handle_devices_already_in_session_results([(device_config, 0, 0, "")])
|
||||
mock_handle_validation.assert_called_once_with(device_config, 0, 0, "")
|
||||
mock_handle_validation.reset_mock()
|
||||
dialog._handle_devices_already_in_session_results([])
|
||||
mock_handle_validation.assert_not_called()
|
||||
mock_handle_validation.reset_mock()
|
||||
dialog._handle_devices_already_in_session_results(
|
||||
[(device_config, 1, 0, ""), (device_config, 0, 0, "")]
|
||||
)
|
||||
mock_handle_validation.assert_called_once_with(
|
||||
device_config, 1, 0, ""
|
||||
) # Should be called with first
|
||||
|
||||
# Test _handle_validation_result
|
||||
# I. No wait dialog present
|
||||
dialog._handle_validation_result(device_config, 1, 3, "All good")
|
||||
assert dialog._validation_result == (device_config, 1, 3, "All good")
|
||||
|
||||
# II. No previous validation, but wait dialog present
|
||||
with mock.patch.object(dialog, "_wait_dialog") as mock_wait_dialog:
|
||||
dialog._handle_validation_result(device_config, 1, 3, "All good")
|
||||
assert dialog.config_validation_result == (device_config, 1, 3, "All good")
|
||||
mock_wait_dialog.accept.assert_called_once()
|
||||
mock_wait_dialog.close.assert_called_once()
|
||||
mock_wait_dialog.deleteLater.assert_called_once()
|
||||
|
||||
mock_wait_dialog.reset_mock()
|
||||
assert dialog._wait_dialog is None
|
||||
|
||||
# III. Previous validation present and the same config, wait dialog present
|
||||
with mock.patch.object(dialog, "_wait_dialog") as mock_wait_dialog:
|
||||
dialog._validation_result = (device_config, 1, 1, "Previous bad")
|
||||
dialog._handle_validation_result(device_config, 1, 3, "All good")
|
||||
assert dialog.config_validation_result == (device_config, 1, 1, "Previous bad")
|
||||
mock_wait_dialog.accept.assert_called_once()
|
||||
mock_wait_dialog.close.assert_called_once()
|
||||
mock_wait_dialog.deleteLater.assert_called_once()
|
||||
|
||||
mock_wait_dialog.reset_mock()
|
||||
assert dialog._wait_dialog is None
|
||||
|
||||
# IV. Previous validation present but different config, wait dialog present
|
||||
with mock.patch.object(dialog, "_wait_dialog") as mock_wait_dialog:
|
||||
different_config = device_config.copy()
|
||||
different_config["deviceClass"] = "DifferentClass"
|
||||
dialog._validation_result = (different_config, 1, 1, "Previous bad")
|
||||
dialog._handle_validation_result(device_config, 1, 3, "All good")
|
||||
assert dialog.config_validation_result == (device_config, 1, 3, "All good")
|
||||
mock_wait_dialog.accept.assert_called_once()
|
||||
mock_wait_dialog.close.assert_called_once()
|
||||
mock_wait_dialog.deleteLater.assert_called_once()
|
||||
|
||||
def test_set_device_config(self, device_form_dialog: DeviceFormDialog, qtbot):
|
||||
"""Test setting device configuration in DeviceFormDialog."""
|
||||
dialog = device_form_dialog
|
||||
@@ -330,9 +393,7 @@ class TestDeviceManagerViewDialogs:
|
||||
@pytest.fixture
|
||||
def upload_redis_dialog(self, qtbot):
|
||||
"""Fixture for UploadRedisDialog."""
|
||||
dialog = UploadRedisDialog(
|
||||
parent=None, ophyd_test_widget=mock.MagicMock(spec=OphydValidation), device_configs={}
|
||||
)
|
||||
dialog = UploadRedisDialog(parent=None, device_configs={})
|
||||
try:
|
||||
qtbot.addWidget(dialog)
|
||||
qtbot.waitExposed(dialog)
|
||||
@@ -460,37 +521,16 @@ class TestDeviceManagerViewDialogs:
|
||||
|
||||
assert dialog.config_section.summary_label.text() == expected_text
|
||||
|
||||
def test_upload_redis_validate_connections(self, device_configs_invalid, qtbot):
|
||||
"""Test the validate connections method in UploadRedisDialog."""
|
||||
configs = device_configs_invalid
|
||||
ophyd_test_mock = mock.MagicMock(spec=OphydValidation)
|
||||
try:
|
||||
dialog = UploadRedisDialog(
|
||||
parent=None, ophyd_test_widget=ophyd_test_mock, device_configs=configs
|
||||
)
|
||||
qtbot.addWidget(dialog)
|
||||
qtbot.waitExposed(dialog)
|
||||
|
||||
with mock.patch.object(
|
||||
dialog.ophyd_test_widget, "change_device_configs"
|
||||
) as mock_change:
|
||||
dialog._validate_connections()
|
||||
mock_change.assert_called_once_with(
|
||||
[cfg for k, (cfg, _, _) in configs.items() if k in ["Device_0", "Device_3"]],
|
||||
added=True,
|
||||
connect=True,
|
||||
)
|
||||
finally:
|
||||
dialog.close()
|
||||
|
||||
|
||||
class TestDeviceManagerView:
|
||||
"""Test class for DeviceManagerView functionality."""
|
||||
|
||||
@pytest.fixture
|
||||
def dm_view(self, qtbot):
|
||||
def dm_view(self, qtbot, mocked_client):
|
||||
"""Fixture for DeviceManagerView."""
|
||||
widget = DeviceManagerView()
|
||||
# Assign the mocked client
|
||||
widget.device_manager_widget.client = mocked_client
|
||||
qtbot.addWidget(widget)
|
||||
qtbot.waitExposed(widget)
|
||||
yield widget
|
||||
@@ -513,7 +553,6 @@ class TestDeviceManagerView:
|
||||
# Reset for test loading current config
|
||||
dm_widget._initialized = False
|
||||
dm_widget.stacked_layout.setCurrentWidget(dm_widget._overlay_widget)
|
||||
dm_widget.client.device_manager = mock.MagicMock()
|
||||
|
||||
with mock.patch.object(
|
||||
dm_widget.client.device_manager, "_get_redis_device_config"
|
||||
@@ -532,12 +571,26 @@ class TestDeviceManagerView:
|
||||
mock_set.assert_called_once_with([])
|
||||
|
||||
@pytest.fixture
|
||||
def device_manager_display_widget(self, qtbot):
|
||||
"""Fixture for DeviceManagerDisplayWidget within DeviceManagerView."""
|
||||
widget = DeviceManagerDisplayWidget()
|
||||
qtbot.addWidget(widget)
|
||||
qtbot.waitExposed(widget)
|
||||
yield widget
|
||||
def device_manager_display_widget(self, qtbot, mocked_client):
|
||||
"""Fixture for DeviceManagerDisplayWidget within DeviceManagerView.
|
||||
We will patch the OphydValidation _thread_pool_poll_loop to avoid starting threads during tests,
|
||||
and the _is_device_in_redis_session method to avoid Redis dependencies
|
||||
"""
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"bec_widgets.widgets.control.device_manager.components.ophyd_validation.ophyd_validation.OphydValidation._thread_pool_poll_loop",
|
||||
return_value=None,
|
||||
),
|
||||
mock.patch(
|
||||
"bec_widgets.widgets.control.device_manager.components.ophyd_validation.ophyd_validation.OphydValidation._is_device_in_redis_session",
|
||||
return_value=False,
|
||||
),
|
||||
):
|
||||
widget = DeviceManagerDisplayWidget(client=mocked_client)
|
||||
qtbot.addWidget(widget)
|
||||
qtbot.waitExposed(widget)
|
||||
yield widget
|
||||
|
||||
@pytest.fixture
|
||||
def device_configs(self, device_config: dict):
|
||||
|
||||
Reference in New Issue
Block a user