diff --git a/bec_widgets/applications/views/device_manager_view/device_manager_dialogs/device_form_dialog.py b/bec_widgets/applications/views/device_manager_view/device_manager_dialogs/device_form_dialog.py index 08bb81f0..9d57daf0 100644 --- a/bec_widgets/applications/views/device_manager_view/device_manager_dialogs/device_form_dialog.py +++ b/bec_widgets/applications/views/device_manager_view/device_manager_dialogs/device_form_dialog.py @@ -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." diff --git a/bec_widgets/applications/views/device_manager_view/device_manager_dialogs/upload_redis_dialog.py b/bec_widgets/applications/views/device_manager_view/device_manager_dialogs/upload_redis_dialog.py index cb6f52b3..e4e746df 100644 --- a/bec_widgets/applications/views/device_manager_view/device_manager_dialogs/upload_redis_dialog.py +++ b/bec_widgets/applications/views/device_manager_view/device_manager_dialogs/upload_redis_dialog.py @@ -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_()) diff --git a/bec_widgets/applications/views/device_manager_view/device_manager_display_widget.py b/bec_widgets/applications/views/device_manager_view/device_manager_display_widget.py index c7690f2e..8c388664 100644 --- a/bec_widgets/applications/views/device_manager_view/device_manager_display_widget.py +++ b/bec_widgets/applications/views/device_manager_view/device_manager_display_widget.py @@ -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.""" diff --git a/bec_widgets/tests/utils.py b/bec_widgets/tests/utils.py index bf4cdf0e..d9fd7d43 100644 --- a/bec_widgets/tests/utils.py +++ b/bec_widgets/tests/utils.py @@ -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), diff --git a/bec_widgets/widgets/control/device_manager/components/device_table/device_table.py b/bec_widgets/widgets/control/device_manager/components/device_table/device_table.py index 343c6438..7f2edb32 100644 --- a/bec_widgets/widgets/control/device_manager/components/device_table/device_table.py +++ b/bec_widgets/widgets/control/device_manager/components/device_table/device_table.py @@ -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 = [] diff --git a/bec_widgets/widgets/control/device_manager/components/ophyd_validation/ophyd_validation.py b/bec_widgets/widgets/control/device_manager/components/ophyd_validation/ophyd_validation.py index af2c3058..5ff9d978 100644 --- a/bec_widgets/widgets/control/device_manager/components/ophyd_validation/ophyd_validation.py +++ b/bec_widgets/widgets/control/device_manager/components/ophyd_validation/ophyd_validation.py @@ -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: diff --git a/tests/unit_tests/test_device_manager_components.py b/tests/unit_tests/test_device_manager_components.py index a690d059..80c31b6b 100644 --- a/tests/unit_tests/test_device_manager_components.py +++ b/tests/unit_tests/test_device_manager_components.py @@ -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 = [ diff --git a/tests/unit_tests/test_device_manager_view.py b/tests/unit_tests/test_device_manager_view.py index e822d1e9..214855e2 100644 --- a/tests/unit_tests/test_device_manager_view.py +++ b/tests/unit_tests/test_device_manager_view.py @@ -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):