From 5d435bd5eefee6bb2c4de6c01f91b27fa78efe57 Mon Sep 17 00:00:00 2001 From: appel_c Date: Thu, 27 Jun 2024 13:26:02 +0200 Subject: [PATCH] refactor: simplify logic in bec_status_box --- .../widgets/bec_status_box/bec_status_box.py | 430 ++++++++---------- .../widgets/bec_status_box/status_item.py | 75 +-- tests/unit_tests/test_bec_status_box.py | 38 +- 3 files changed, 239 insertions(+), 304 deletions(-) diff --git a/bec_widgets/widgets/bec_status_box/bec_status_box.py b/bec_widgets/widgets/bec_status_box/bec_status_box.py index 0b9d42fa..d1c0aaa5 100644 --- a/bec_widgets/widgets/bec_status_box/bec_status_box.py +++ b/bec_widgets/widgets/bec_status_box/bec_status_box.py @@ -5,15 +5,16 @@ The widget automatically updates the status of all running BEC services, and dis from __future__ import annotations import sys +from collections import defaultdict +from dataclasses import dataclass from typing import TYPE_CHECKING import qdarktheme from bec_lib.utils.import_utils import lazy_import_from -from pydantic import BaseModel, Field, field_validator from qtpy.QtCore import QObject, QTimer, Signal, Slot from qtpy.QtWidgets import QTreeWidget, QTreeWidgetItem -from bec_widgets.utils.bec_connector import BECConnector, ConnectionConfig +from bec_widgets.utils.bec_connector import BECConnector from bec_widgets.widgets.bec_status_box.status_item import StatusItem if TYPE_CHECKING: @@ -23,45 +24,18 @@ if TYPE_CHECKING: BECStatus = lazy_import_from("bec_lib.messages", ("BECStatus",)) -class BECStatusBoxConfig(ConnectionConfig): - pass - - -class BECServiceInfoContainer(BaseModel): +@dataclass +class BECServiceInfoContainer: """Container to store information about the BEC services.""" service_name: str - status: BECStatus | str = Field( - default="NOTCONNECTED", - description="The status of the service. Can be any of the BECStatus names, or NOTCONNECTED.", - ) + status: str info: dict metrics: dict | None - model_config: dict = {"validate_assignment": True} - - @field_validator("status") - @classmethod - def validate_status(cls, v): - """Validate input for status. Accept BECStatus and NOTCONNECTED. - - Args: - v (BECStatus | str): The input value. - - Returns: - str: The validated status. - """ - if v in list(BECStatus.__members__.values()): - return v.name - if v in list(BECStatus.__members__.keys()) or v == "NOTCONNECTED": - return v - raise ValueError( - f"Status must be one of {BECStatus.__members__.values()} or 'NOTCONNECTED'. Input {v}" - ) class BECServiceStatusMixin(QObject): - """A mixin class to update the service status, and metrics. - It emits a signal 'services_update' when the service status is updated. + """Mixin to receive the latest service status from the BEC server and emit it via services_update signal. Args: client (BECClient): The client object to connect to the BEC server. @@ -77,21 +51,18 @@ class BECServiceStatusMixin(QObject): self._service_update_timer.start(1000) def _get_service_status(self): - """Pull latest service and metrics updates from REDIS for all services, and emit both via 'services_update' signal.""" + """Get the latest service status from the BEC server.""" # pylint: disable=protected-access self.client._update_existing_services() self.services_update.emit(self.client._services_info, self.client._services_metric) class BECStatusBox(BECConnector, QTreeWidget): - """A widget to display the status of different BEC services. - This widget automatically updates the status of all running BEC services, and displays their status. - Information about the individual services is collapsible, and double clicking on - the individual service will display the metrics about the service. + """An autonomous widget to display the status of BEC services. Args: parent Optional : The parent widget for the BECStatusBox. Defaults to None. - service_name Optional(str): The name of the top service label. Defaults to "BEC Server". + box_name Optional(str): The name of the top service label. Defaults to "BEC Server". client Optional(BECClient): The client object to connect to the BEC server. Defaults to None config Optional(BECStatusBoxConfig | dict): The configuration for the status box. Defaults to None. gui_id Optional(str): The unique id for the widget. Defaults to None. @@ -99,216 +70,61 @@ class BECStatusBox(BECConnector, QTreeWidget): CORE_SERVICES = ["DeviceServer", "ScanServer", "SciHub", "ScanBundler", "FileWriterManager"] - service_update = Signal(dict) + service_update = Signal(BECServiceInfoContainer) bec_core_state = Signal(str) + _initialized = False + _bec_status_box = None + def __init__( self, parent=None, - service_name: str = "BEC Server", + box_name: str = "BEC Server", client: BECClient = None, - config: BECStatusBoxConfig | dict = None, bec_service_status_mixin: BECServiceStatusMixin = None, gui_id: str = None, ): - if config is None: - config = BECStatusBoxConfig(widget_class=self.__class__.__name__) - else: - if isinstance(config, dict): - config = BECStatusBoxConfig(**config) - super().__init__(client=client, config=config, gui_id=gui_id) + if self._initialized == True: + return + super().__init__(client=client, gui_id=gui_id) QTreeWidget.__init__(self, parent=parent) - self.service_name = service_name - self.config = config - - self.bec_service_info_container = {} - self.tree_items = {} - self.tree_top_item = None + self.box_name = box_name + self.status_container = defaultdict(lambda: {"info": None, "item": None, "widget": None}) + self._initialized = False if not bec_service_status_mixin: bec_service_status_mixin = BECServiceStatusMixin(client=self.client) self.bec_service_status = bec_service_status_mixin - self.init_ui() - self.bec_service_status.services_update.connect(self.update_service_status) - self.bec_core_state.connect(self.update_top_item_status) - self.itemDoubleClicked.connect(self.on_tree_item_double_clicked) + if not self._initialized: + self.init_ui() + self.bec_service_status.services_update.connect(self.update_service_status) + self.bec_core_state.connect(self.update_top_item_status) + self.itemDoubleClicked.connect(self.on_tree_item_double_clicked) + + def __new__(cls, *args, forced: bool = False, **kwargs): + if forced: + cls._initialized = False + cls._bec_status_box = super(BECStatusBox, cls).__new__(cls) + return cls._bec_status_box + if cls._bec_status_box is not None and cls._initialized is True: + return cls._bec_status_box + cls._bec_status_box = super(BECStatusBox, cls).__new__(cls) + return cls._bec_status_box def init_ui(self) -> None: - """Initialize the UI for the status box, and add QTreeWidget as the basis for the status box.""" + """Init the UI for the BECStatusBox widget, should only take place once.""" self.init_ui_tree_widget() - top_label = self._create_status_widget(self.service_name, status=BECStatus.IDLE) - self.tree_top_item = QTreeWidgetItem() - self.tree_top_item.setExpanded(True) - self.tree_top_item.setDisabled(True) - self.addTopLevelItem(self.tree_top_item) - self.setItemWidget(self.tree_top_item, 0, top_label) + top_label = self._create_status_widget(self.box_name, status=BECStatus.IDLE) + tree_item = QTreeWidgetItem() + tree_item.setExpanded(True) + tree_item.setDisabled(True) + self.status_container[self.box_name].update({"item": tree_item, "widget": top_label}) + self.addTopLevelItem(tree_item) + self.setItemWidget(tree_item, 0, top_label) self.service_update.connect(top_label.update_config) - - def _create_status_widget( - self, service_name: str, status=BECStatus, info: dict = None, metrics: dict = None - ) -> StatusItem: - """Creates a StatusItem (QWidget) for the given service, and stores all relevant - information about the service in the bec_service_info_container. - - Args: - service_name (str): The name of the service. - status (BECStatus): The status of the service. - info Optional(dict): The information about the service. Default is {} - metric Optional(dict): Metrics for the respective service. Default is None - - Returns: - StatusItem: The status item widget. - """ - if info is None: - info = {} - self._update_bec_service_container(service_name, status, info, metrics) - item = StatusItem( - parent=self, - config={ - "service_name": service_name, - "status": status.name, - "info": info, - "metrics": metrics, - }, - ) - return item - - @Slot(str) - def update_top_item_status(self, status: BECStatus) -> None: - """Method to update the status of the top item in the tree widget. - Gets the status from the Signal 'bec_core_state' and updates the StatusItem via the signal 'service_update'. - - Args: - status (BECStatus): The state of the core services. - """ - self.bec_service_info_container[self.service_name].status = status - self.service_update.emit(self.bec_service_info_container[self.service_name].model_dump()) - - def _update_bec_service_container( - self, service_name: str, status: BECStatus, info: dict, metrics: dict = None - ) -> None: - """Update the bec_service_info_container with the newest status and metrics for the BEC service. - If information about the service already exists, it will create a new entry. - - Args: - service_name (str): The name of the service. - service_info (StatusMessage): A class containing the service status. - service_metric (ServiceMetricMessage): A class containing the service metrics. - """ - container = self.bec_service_info_container.get(service_name, None) - if container: - container.status = status - container.info = info - container.metrics = metrics - return - service_info_item = BECServiceInfoContainer( - service_name=service_name, status=status, info=info, metrics=metrics - ) - self.bec_service_info_container.update({service_name: service_info_item}) - - @Slot(dict, dict) - def update_service_status(self, services_info: dict, services_metric: dict) -> None: - """Callback function services_metric from BECServiceStatusMixin. - It updates the status of all services. - - Args: - services_info (dict): A dictionary containing the service status for all running BEC services. - services_metric (dict): A dictionary containing the service metrics for all running BEC services. - """ - checked = [] - services_info = self.update_core_services(services_info, services_metric) - checked.extend(self.CORE_SERVICES) - - for service_name, msg in sorted(services_info.items()): - checked.append(service_name) - metric_msg = services_metric.get(service_name, None) - metrics = metric_msg.metrics if metric_msg else None - if service_name in self.tree_items: - self._update_bec_service_container( - service_name=service_name, status=msg.status, info=msg.info, metrics=metrics - ) - self.service_update.emit(self.bec_service_info_container[service_name].model_dump()) - continue - - item_widget = self._create_status_widget( - service_name=service_name, status=msg.status, info=msg.info, metrics=metrics - ) - item = QTreeWidgetItem() - item.setDisabled(True) - self.service_update.connect(item_widget.update_config) - self.tree_top_item.addChild(item) - self.setItemWidget(item, 0, item_widget) - self.tree_items.update({service_name: (item, item_widget)}) - - self.check_redundant_tree_items(checked) - - def update_core_services(self, services_info: dict, services_metric: dict) -> dict: - """Method to process status and metrics updates of core services (stored in CORE_SERVICES). - If a core services is not connected, it should not be removed from the status widget - - Args: - services_info (dict): A dictionary containing the service status of different services. - services_metric (dict): A dictionary containing the service metrics of different services. - - Returns: - dict: The services_info dictionary after removing the info updates related to the CORE_SERVICES - """ - bec_core_state = "RUNNING" - for service_name in sorted(self.CORE_SERVICES): - metric_msg = services_metric.get(service_name, None) - metrics = metric_msg.metrics if metric_msg else None - if service_name not in services_info: - self.bec_service_info_container[service_name].status = "NOTCONNECTED" - bec_core_state = "ERROR" - else: - msg = services_info.pop(service_name) - self._update_bec_service_container( - service_name=service_name, status=msg.status, info=msg.info, metrics=metrics - ) - bec_core_state = ( - "RUNNING" if (msg.status.value > 1 and bec_core_state == "RUNNING") else "ERROR" - ) - - if service_name in self.tree_items: - self.service_update.emit(self.bec_service_info_container[service_name].model_dump()) - continue - self.add_tree_item(service_name, msg.status, msg.info, metrics) - - self.bec_core_state.emit(bec_core_state) - return services_info - - def check_redundant_tree_items(self, checked: list) -> None: - """Utility method to check and remove redundant objects from the BECStatusBox. - - Args: - checked (list): A list of services that are currently running. - """ - to_be_deleted = [key for key in self.tree_items if key not in checked] - - for key in to_be_deleted: - item, _ = self.tree_items.pop(key) - self.tree_top_item.removeChild(item) - - def add_tree_item( - self, service_name: str, status: BECStatus, info: dict = None, metrics: dict = None - ) -> None: - """Method to add a new QTreeWidgetItem together with a StatusItem to the tree widget. - - Args: - service_name (str): The name of the service. - service_status_msg (StatusMessage): The status of the service. - metrics (dict): The metrics of the service. - """ - item_widget = self._create_status_widget( - service_name=service_name, status=status, info=info, metrics=metrics - ) - item = QTreeWidgetItem() - self.service_update.connect(item_widget.update_config) - self.tree_top_item.addChild(item) - self.setItemWidget(item, 0, item_widget) - self.tree_items.update({service_name: (item, item_widget)}) + self._initialized = True def init_ui_tree_widget(self) -> None: """Initialise the tree widget for the status box.""" @@ -323,6 +139,151 @@ class BECStatusBox(BECConnector, QTreeWidget): "QTreeWidget::item:selected {}" ) + def _create_status_widget( + self, service_name: str, status=BECStatus, info: dict = None, metrics: dict = None + ) -> StatusItem: + """Creates a StatusItem (QWidget) for the given service, and stores all relevant + information about the service in the status_container. + + Args: + service_name (str): The name of the service. + status (BECStatus): The status of the service. + info Optional(dict): The information about the service. Default is {} + metric Optional(dict): Metrics for the respective service. Default is None + + Returns: + StatusItem: The status item widget. + """ + if info is None: + info = {} + self._update_status_container(service_name, status, info, metrics) + item = StatusItem(parent=self, config=self.status_container[service_name]["info"]) + return item + + @Slot(str) + def update_top_item_status(self, status: BECStatus) -> None: + """Method to update the status of the top item in the tree widget. + Gets the status from the Signal 'bec_core_state' and updates the StatusItem via the signal 'service_update'. + + Args: + status (BECStatus): The state of the core services. + """ + self.status_container[self.box_name]["info"].status = status + self.service_update.emit(self.status_container[self.box_name]["info"]) + + def _update_status_container( + self, service_name: str, status: BECStatus, info: dict, metrics: dict = None + ) -> None: + """Update the status_container with the newest status and metrics for the BEC service. + If information about the service already exists, it will create a new entry. + + Args: + service_name (str): The name of the service. + status (BECStatus): The status of the service. + info (dict): The information about the service. + metrics (dict): The metrics of the service. + """ + container = self.status_container[service_name].get("info", None) + + if container: + container.status = status.name + container.info = info + container.metrics = metrics + return + service_info_item = BECServiceInfoContainer( + service_name=service_name, status=status.name, info=info, metrics=metrics + ) + self.status_container[service_name].update({"info": service_info_item}) + + @Slot(dict, dict) + def update_service_status(self, services_info: dict, services_metric: dict) -> None: + """Callback function services_metric from BECServiceStatusMixin. + It updates the status of all services. + + Args: + services_info (dict): A dictionary containing the service status for all running BEC services. + services_metric (dict): A dictionary containing the service metrics for all running BEC services. + """ + checked = [self.box_name] + services_info = self.update_core_services(services_info, services_metric) + checked.extend(self.CORE_SERVICES) + + for service_name, msg in sorted(services_info.items()): + checked.append(service_name) + metric_msg = services_metric.get(service_name, None) + metrics = metric_msg.metrics if metric_msg else None + if service_name in self.status_container: + self._update_status_container(service_name, msg.status, msg.info, metrics) + self.service_update.emit(self.status_container[service_name]["info"]) + continue + + self.add_tree_item(service_name, msg.status, msg.info, metrics) + self.check_redundant_tree_items(checked) + + def update_core_services(self, services_info: dict, services_metric: dict) -> dict: + """Update the core services of BEC, and emit the updated status to the BECStatusBox. + + Args: + services_info (dict): A dictionary containing the service status of different services. + services_metric (dict): A dictionary containing the service metrics of different services. + + Returns: + dict: The services_info dictionary after removing the info updates related to the CORE_SERVICES + """ + core_state = BECStatus.RUNNING + for service_name in sorted(self.CORE_SERVICES): + metric_msg = services_metric.get(service_name, None) + metrics = metric_msg.metrics if metric_msg else None + msg = services_info.pop(service_name, None) + if service_name not in self.status_container: + self.add_tree_item(service_name, msg.status, msg.info, metrics) + continue + if not msg: + self.status_container[service_name]["info"].status = "NOTCONNECTED" + core_state = None + else: + self._update_status_container(service_name, msg.status, msg.info, metrics) + if core_state: + core_state = msg.status if msg.status.value < core_state.value else core_state + + self.service_update.emit(self.status_container[service_name]["info"]) + + # self.add_tree_item(service_name, msg.status, msg.info, metrics) + + self.bec_core_state.emit(core_state.name if core_state else "NOTCONNECTED") + return services_info + + def check_redundant_tree_items(self, checked: list) -> None: + """Utility method to check and remove redundant objects from the BECStatusBox. + + Args: + checked (list): A list of services that are currently running. + """ + to_be_deleted = [key for key in self.status_container if key not in checked] + + for key in to_be_deleted: + obj = self.status_container.pop(key) + item = obj["item"] + self.status_container[self.box_name]["item"].removeChild(item) + + def add_tree_item( + self, service_name: str, status: BECStatus, info: dict = None, metrics: dict = None + ) -> None: + """Method to add a new QTreeWidgetItem together with a StatusItem to the tree widget. + + Args: + service_name (str): The name of the service. + status (BECStatus): The status of the service. + info (dict): The information about the service. + metrics (dict): The metrics of the service. + """ + item_widget = self._create_status_widget(service_name, status, info, metrics) + item = QTreeWidgetItem() # setDisabled=True + self.service_update.connect(item_widget.update_config) + self.status_container[self.box_name]["item"].addChild(item) + self.setItemWidget(item, 0, item_widget) + self.status_container[service_name].update({"item": item, "widget": item_widget}) + @Slot(QTreeWidgetItem, int) def on_tree_item_double_clicked(self, item: QTreeWidgetItem, column: int) -> None: """Callback function for double clicks on individual QTreeWidgetItems in the collapsed section. @@ -331,11 +292,16 @@ class BECStatusBox(BECConnector, QTreeWidget): item (QTreeWidgetItem): The item that was double clicked. column (int): The column that was double clicked. """ - for _, (tree_item, status_widget) in self.tree_items.items(): - if tree_item == item: - status_widget.show_popup() + for _, objects in self.status_container.items(): + if objects["item"] == item: + objects["widget"].show_popup() def closeEvent(self, event): + """Upon closing the widget, clean up the BECStatusBox and the QTreeWidget. + + Args: + event: The close event. + """ super().cleanup() QTreeWidget().closeEvent(event) diff --git a/bec_widgets/widgets/bec_status_box/status_item.py b/bec_widgets/widgets/bec_status_box/status_item.py index f4f9afcf..f0c9628b 100644 --- a/bec_widgets/widgets/bec_status_box/status_item.py +++ b/bec_widgets/widgets/bec_status_box/status_item.py @@ -2,17 +2,12 @@ The widget is bound to be used with the BECStatusBox widget.""" import enum -import sys from datetime import datetime -import qdarktheme from bec_lib.utils.import_utils import lazy_import_from -from pydantic import Field from qtpy.QtCore import Qt, Slot from qtpy.QtWidgets import QDialog, QHBoxLayout, QLabel, QStyle, QVBoxLayout, QWidget -from bec_widgets.utils.bec_connector import ConnectionConfig - # TODO : Put normal imports back when Pydantic gets faster BECStatus = lazy_import_from("bec_lib.messages", ("BECStatus",)) @@ -27,39 +22,29 @@ class IconsEnum(enum.Enum): NOTCONNECTED = "SP_TitleBarContextHelpButton" -class StatusWidgetConfig(ConnectionConfig): - """Configuration class for the status item widget.""" - - service_name: str - status: str - info: dict - metrics: dict | None - icon_size: tuple = Field(default=(24, 24), description="The size of the icon in the widget.") - font_size: int = Field(16, description="The font size of the text in the widget.") - - class StatusItem(QWidget): """A widget to display the status of a service. Args: parent: The parent widget. - config (dict): The configuration for the service. + config (dict): The configuration for the service, must be a BECServiceInfoContainer. """ - def __init__(self, parent=None, config: dict = None): - if config is None: - config = StatusWidgetConfig(widget_class=self.__class__.__name__) - else: - if isinstance(config, dict): - config = StatusWidgetConfig(**config) - self.config = config + def __init__(self, parent: QWidget = None, config=None): QWidget.__init__(self, parent=parent) + if config is None: + # needed because we need parent to be the first argument for QT Designer + raise ValueError( + "Please initialize the StatusItem with a BECServiceInfoContainer for config, received None." + ) + self.config = config self.parent = parent self.layout = None - self.config = config - self._popup_label_ref = {} self._label = None self._icon = None + self.icon_size = (24, 24) + + self._popup_label_ref = {} self.init_ui() def init_ui(self) -> None: @@ -74,23 +59,21 @@ class StatusItem(QWidget): self.update_ui() @Slot(dict) - def update_config(self, config: dict) -> None: - """Update the configuration of the status item widget. - This method is invoked from the parent widget. - The UI values are later updated based on the new configuration. + def update_config(self, config) -> None: + """Update the config of the status item widget. Args: - config (dict): Config updates from parent widget. + config (dict): Config updates from parent widget, must be a BECServiceInfoContainer. """ - if config["service_name"] != self.config.service_name: + if self.config is None or config.service_name != self.config.service_name: return - self.config.status = config["status"] - self.config.info = config["info"] - self.config.metrics = config["metrics"] + self.config = config self.update_ui() def update_ui(self) -> None: """Update the UI of the labels, and popup dialog.""" + if self.config is None: + return self.set_text() self.set_status() self._set_popup_text() @@ -99,8 +82,8 @@ class StatusItem(QWidget): """Set the text of the QLabel basae on the config.""" service = self.config.service_name status = self.config.status - if "BECClient" in service.split("/"): - service = service.split("/")[0] + "/..." + service.split("/")[1][-4:] + if len(service.split("/")) > 1 and service.split("/")[0].startswith("BEC"): + service = service.split("/", maxsplit=1)[0] + "/..." + service.split("/")[1][-4:] if status == "NOTCONNECTED": status = "NOT CONNECTED" text = f"{service} is {status}" @@ -110,7 +93,7 @@ class StatusItem(QWidget): """Set the status icon for the status item widget.""" icon_name = IconsEnum[self.config.status].value icon = self.style().standardIcon(getattr(QStyle.StandardPixmap, icon_name)) - self._icon.setPixmap(icon.pixmap(*self.config.icon_size)) + self._icon.setPixmap(icon.pixmap(*self.icon_size)) self._icon.setAlignment(Qt.AlignmentFlag.AlignRight) def show_popup(self) -> None: @@ -153,19 +136,3 @@ class StatusItem(QWidget): def _cleanup_popup_label(self) -> None: """Cleanup the popup label.""" self._popup_label_ref.clear() - - -def main(): - """Run the status item widget.""" - # pylint: disable=import-outside-toplevel - from qtpy.QtWidgets import QApplication - - app = QApplication(sys.argv) - qdarktheme.setup_theme("auto") - main_window = StatusItem() - main_window.show() - sys.exit(app.exec()) - - -if __name__ == "__main__": - main() diff --git a/tests/unit_tests/test_bec_status_box.py b/tests/unit_tests/test_bec_status_box.py index bf91dc29..c58a8904 100644 --- a/tests/unit_tests/test_bec_status_box.py +++ b/tests/unit_tests/test_bec_status_box.py @@ -1,3 +1,4 @@ +# pylint: skip-file from unittest import mock import pytest @@ -15,9 +16,7 @@ def service_status_fixture(): @pytest.fixture def status_box(qtbot, mocked_client, service_status_fixture): - widget = BECStatusBox( - client=mocked_client, service_name="test", bec_service_status_mixin=service_status_fixture - ) + widget = BECStatusBox(client=mocked_client, bec_service_status_mixin=service_status_fixture) qtbot.addWidget(widget) qtbot.waitExposed(widget) yield widget @@ -25,8 +24,9 @@ def status_box(qtbot, mocked_client, service_status_fixture): def test_update_top_item(status_box): assert status_box.children()[0].children()[0].config.status == "IDLE" + name = status_box.box_name status_box.update_top_item_status(status="RUNNING") - assert status_box.bec_service_info_container["test"].status == "RUNNING" + assert status_box.status_container[name]["info"].status == "RUNNING" assert status_box.children()[0].children()[0].config.status == "RUNNING" @@ -48,13 +48,13 @@ def test_bec_service_container(status_box): info = {"test": "test"} metrics = {"metric": "test_metric"} expected_return = BECServiceInfoContainer( - service_name=name, status=status, info=info, metrics=metrics + service_name=name, status=status.name, info=info, metrics=metrics ) - assert status_box.service_name in status_box.bec_service_info_container - assert len(status_box.bec_service_info_container) == 1 - status_box._update_bec_service_container(name, status, info, metrics) - assert len(status_box.bec_service_info_container) == 2 - assert status_box.bec_service_info_container[name] == expected_return + assert status_box.box_name in status_box.status_container + assert len(status_box.status_container) == 1 + status_box._update_status_container(name, status, info, metrics) + assert len(status_box.status_container) == 2 + assert status_box.status_container[name]["info"] == expected_return def test_add_tree_item(status_box): @@ -65,7 +65,7 @@ def test_add_tree_item(status_box): assert len(status_box.children()[0].children()) == 1 status_box.add_tree_item(name, status, info, metrics) assert len(status_box.children()[0].children()) == 2 - assert name in status_box.tree_items + assert name in status_box.status_container def test_update_service_status(status_box): @@ -82,10 +82,10 @@ def test_update_service_status(status_box): services_metrics = {name: ServiceMetricMessage(name=name, metrics=metrics)} with mock.patch.object(status_box, "update_core_services", return_value=services_status): - assert not_connected_name in status_box.tree_items + assert not_connected_name in status_box.status_container status_box.update_service_status(services_status, services_metrics) - assert status_box.tree_items[name][1].config.metrics == metrics - assert not_connected_name not in status_box.tree_items + assert status_box.status_container[name]["widget"].config.metrics == metrics + assert not_connected_name not in status_box.status_container def test_update_core_services(status_box): @@ -99,14 +99,14 @@ def test_update_core_services(status_box): status_box.update_core_services(services_status, services_metrics) assert status_box.children()[0].children()[0].config.status == "RUNNING" - assert status_box.tree_items[name][1].config.metrics == metrics + assert status_box.status_container[name]["widget"].config.metrics == metrics status = BECStatus.IDLE services_status = {name: StatusMessage(name=name, status=status, info=info)} services_metrics = {name: ServiceMetricMessage(name=name, metrics=metrics)} status_box.update_core_services(services_status, services_metrics) - assert status_box.children()[0].children()[0].config.status == "ERROR" - assert status_box.tree_items[name][1].config.metrics == metrics + assert status_box.children()[0].children()[0].config.status == status.name + assert status_box.status_container[name]["widget"].config.metrics == metrics def test_double_click_item(status_box): @@ -115,7 +115,9 @@ def test_double_click_item(status_box): info = {"test": "test"} metrics = {"MyData": "This should be shown nicely"} status_box.add_tree_item(name, status, info, metrics) - item, status_item = status_box.tree_items[name] + container = status_box.status_container[name] + item = container["item"] + status_item = container["widget"] with mock.patch.object(status_item, "show_popup") as mock_show_popup: status_box.itemDoubleClicked.emit(item, 0) assert mock_show_popup.call_count == 1