bec/bec_widgets
1
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-08-05 01:32:34 +02:00
Code Activity

test(device-manager): use mocked client for tests

Browse Source
This commit is contained in:
appel_c
2026-03-05 18:12:31 +01:00
committed by wyzula_j
parent 89d5c5abdb
commit 836fedd50e
8 changed files with 218 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bec_widgets/applications/views/device_manager_view/device_manager_dialogs/device_form_dialog.py
+1
View File
@@ -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."
bec_widgets/applications/views/device_manager_view/device_manager_dialogs/upload_redis_dialog.py
+41 -71
View File
@@ -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_())
bec_widgets/applications/views/device_manager_view/device_manager_display_widget.py
+17 -2
View File
@@ -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."""
bec_widgets/tests/utils.py
+8
View File
@@ -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),
bec_widgets/widgets/control/device_manager/components/device_table/device_table.py
+8 -5
View File
@@ -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 = []
bec_widgets/widgets/control/device_manager/components/ophyd_validation/ophyd_validation.py
+9 -3
View File
@@ -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: