1
0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-04-09 10:10:55 +02:00

Compare commits

...

5 Commits

Author SHA1 Message Date
e2f074b1aa fix: make main() function working, to be able to test out of Qt Designer 2024-07-03 14:10:04 +02:00
011103fde3 refactor: put QTreeWidget in a container, rather than inheriting from it, and avoid multiple inheritance of BECConnector
Inheriting from QTreeWidget causes havoc with display (see #245),
also it is important to parent items OR to give them a label (weird)
otherwise it also has display glitches.

Most of the time, composition has to be preferred over inheritance ;
inheritance is a question of behaviour - is the behaviour the same ?
Here, a widget is really not a BECConnector, but uses a BECConnector
(at least this is my understanding). Because a Widget and BECConnector
do not behave the same. It is easier, lighter to deal with single
inheritance.

The singleton usage is superfluous, since the underlying client is already
a singleton. Multiple BECConnector objects can be created, there won't
be more connections.

BECServiceStatusMixin has been removed in favor of the widget's own timer,
since it was causing "QObject::killTimer: Timers cannot be stopped from
another thread" errors (at least on my computer).

Also removes "redundant items check" ; where do those would come from?
2024-07-03 14:09:55 +02:00
f90bc00c18 fix: make error StatusMessage in case service info msg is None
Makes handling of status easier, no need for special cases
2024-07-03 14:09:55 +02:00
63a0056388 fix: add designer plugin classes 2024-07-03 14:09:55 +02:00
5d435bd5ee refactor: simplify logic in bec_status_box 2024-07-03 14:05:45 +02:00
6 changed files with 236 additions and 264 deletions

View File

@@ -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 qtpy.QtCore import Signal, Slot
from qtpy.QtWidgets import QTreeWidget, QTreeWidgetItem, QVBoxLayout, QWidget
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:
@@ -21,77 +22,25 @@ if TYPE_CHECKING:
# TODO : Put normal imports back when Pydantic gets faster
BECStatus = lazy_import_from("bec_lib.messages", ("BECStatus",))
StatusMessage = lazy_import_from("bec_lib.messages", ("StatusMessage",))
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.
Args:
client (BECClient): The client object to connect to the BEC server.
"""
services_update = Signal(dict, dict)
def __init__(self, client: BECClient):
super().__init__()
self.client = client
self._service_update_timer = QTimer()
self._service_update_timer.timeout.connect(self._get_service_status)
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."""
# 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.
class BECStatusBox(QWidget):
"""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,58 +48,68 @@ class BECStatusBox(BECConnector, QTreeWidget):
CORE_SERVICES = ["DeviceServer", "ScanServer", "SciHub", "ScanBundler", "FileWriterManager"]
service_update = Signal(dict)
service_update = Signal(BECServiceInfoContainer)
bec_core_state = Signal(str)
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)
QTreeWidget.__init__(self, parent=parent)
QWidget.__init__(self, parent=parent)
self.setLayout(QVBoxLayout(self))
self.tree = QTreeWidget(self)
self.layout().addWidget(self.tree)
self.tree.setHeaderHidden(True)
self.tree.setStyleSheet(
"QTreeWidget::item:!selected "
"{ "
"border: 1px solid gainsboro; "
"border-left: none; "
"border-top: none; "
"}"
"QTreeWidget::item:selected {}"
)
self.box_name = box_name
self.status_container = defaultdict(lambda: {"info": None, "item": None, "widget": None})
self.service_name = service_name
self.config = config
self.bec_service_info_container = {}
self.tree_items = {}
self.tree_top_item = None
if not bec_service_status_mixin:
bec_service_status_mixin = BECServiceStatusMixin(client=self.client)
self.bec_service_status = bec_service_status_mixin
self.connector = BECConnector(client=client, gui_id=gui_id)
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)
self.tree.itemDoubleClicked.connect(self.on_tree_item_double_clicked)
self.startTimer(
1000
) # use qobject's own timer instead of creating one, which may be stopped from another thread(?)
def timerEvent(self, event):
"""Get the latest service status from the BEC server."""
# pylint: disable=protected-access
self.connector.client._update_existing_services()
self.update_service_status(
self.connector.client._services_info, self.connector.client._services_metric
)
def init_ui(self) -> None:
"""Initialize the UI for the status box, and add QTreeWidget as the basis for the status box."""
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)
"""Init the UI for the BECStatusBox widget"""
top_label = self._create_status_widget(self.box_name, status=BECStatus.IDLE)
tree_item = QTreeWidgetItem(self.tree)
tree_item.setExpanded(True)
tree_item.setDisabled(True)
self.status_container[self.box_name].update({"item": tree_item, "widget": top_label})
self.tree.setItemWidget(tree_item, 0, top_label)
self.tree.addTopLevelItem(tree_item)
self.service_update.connect(top_label.update_config)
self._initialized = True
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.
information about the service in the status_container.
Args:
service_name (str): The name of the service.
@@ -163,16 +122,8 @@ class BECStatusBox(BECConnector, QTreeWidget):
"""
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,
},
)
self._update_status_container(service_name, status, info, metrics)
item = StatusItem(parent=self.tree, config=self.status_container[service_name]["info"])
return item
@Slot(str)
@@ -183,30 +134,32 @@ class BECStatusBox(BECConnector, QTreeWidget):
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())
self.status_container[self.box_name]["info"].status = status
self.service_update.emit(self.status_container[self.box_name]["info"])
def _update_bec_service_container(
def _update_status_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.
"""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.
service_info (StatusMessage): A class containing the service status.
service_metric (ServiceMetricMessage): A class containing the service metrics.
status (BECStatus): The status of the service.
info (dict): The information about the service.
metrics (dict): The metrics of the service.
"""
container = self.bec_service_info_container.get(service_name, None)
container = self.status_container[service_name].get("info", None)
if container:
container.status = status
container.status = status.name
container.info = info
container.metrics = metrics
return
service_info_item = BECServiceInfoContainer(
service_name=service_name, status=status, info=info, metrics=metrics
service_name=service_name, status=status.name, info=info, metrics=metrics
)
self.bec_service_info_container.update({service_name: service_info_item})
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:
@@ -217,7 +170,7 @@ class BECStatusBox(BECConnector, QTreeWidget):
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 = []
checked = [self.box_name]
services_info = self.update_core_services(services_info, services_metric)
checked.extend(self.CORE_SERVICES)
@@ -225,28 +178,13 @@ class BECStatusBox(BECConnector, QTreeWidget):
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)
if service_name not in self.status_container:
self.add_tree_item(service_name, msg.status, msg.info, metrics)
self._update_status_container(service_name, msg.status, msg.info, metrics)
self.service_update.emit(self.status_container[service_name]["info"])
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
"""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.
@@ -255,42 +193,26 @@ class BECStatusBox(BECConnector, QTreeWidget):
Returns:
dict: The services_info dictionary after removing the info updates related to the CORE_SERVICES
"""
bec_core_state = "RUNNING"
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
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"
)
msg = services_info.pop(service_name, None)
if msg is None:
msg = StatusMessage(name=service_name, status=BECStatus.ERROR, info={})
if service_name not in self.status_container:
self.add_tree_item(service_name, msg.status, msg.info, metrics)
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._update_status_container(service_name, msg.status, msg.info, metrics)
core_state = msg.status if msg.status.value < core_state.value else core_state
self.bec_core_state.emit(bec_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.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:
@@ -298,30 +220,18 @@ class BECStatusBox(BECConnector, QTreeWidget):
Args:
service_name (str): The name of the service.
service_status_msg (StatusMessage): The status 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=service_name, status=status, info=info, metrics=metrics
)
item = QTreeWidgetItem()
item_widget = self._create_status_widget(service_name, status, info, metrics)
toplevel_item = self.status_container[self.box_name]["item"]
item = QTreeWidgetItem(toplevel_item) # setDisabled=True
toplevel_item.addChild(item)
self.tree.setItemWidget(item, 0, item_widget)
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)})
def init_ui_tree_widget(self) -> None:
"""Initialise the tree widget for the status box."""
self.setHeaderHidden(True)
self.setStyleSheet(
"QTreeWidget::item:!selected "
"{ "
"border: 1px solid gainsboro; "
"border-left: none; "
"border-top: none; "
"}"
"QTreeWidget::item:selected {}"
)
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:
@@ -331,24 +241,47 @@ 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):
super().cleanup()
QTreeWidget().closeEvent(event)
self.connector.cleanup()
def main():
"""Main method to run the BECStatusBox widget."""
# pylint: disable=import-outside-toplevel
from bec_lib.client import BECClient
from bec_lib.logger import bec_logger
from bec_lib.service_config import ServiceConfig
from qtpy.QtWidgets import QApplication
from bec_widgets.utils.bec_dispatcher import QtRedisConnector
# logging has to be configured before create QApplication,
# otherwise it ends badly with segfault...
# (seems to be a threading issue with loguru and probably Redis connector,
# which has to be a QtRedisConnector for Qt apps... Otherwise it is not
# thread-safe somehow ; didn't want to debug all this now)
logger = bec_logger.logger
service_config = ServiceConfig()
bec_logger.configure(
service_config.redis,
QtRedisConnector,
service_name="test_status_box",
service_config=service_config.service_config,
)
app = QApplication(sys.argv)
qdarktheme.setup_theme("auto")
main_window = BECStatusBox()
main_window.show()
client = BECClient()
status = BECStatusBox(parent=None, client=client, gui_id="test")
status.show()
sys.exit(app.exec())

View File

@@ -0,0 +1 @@
{'files': ['bec_status_box.py']}

View File

@@ -0,0 +1,54 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtGui import QIcon
from bec_widgets.widgets.bec_status_box.bec_status_box import BECStatusBox
DOM_XML = """
<ui language='c++'>
<widget class='BECStatusBox' name='bec_status_box'>
</widget>
</ui>
"""
class BECStatusBoxPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
t = BECStatusBox(parent)
return t
def domXml(self):
return DOM_XML
def group(self):
return ""
def icon(self):
return QIcon()
def includeFile(self):
return "bec_status_box"
def initialize(self, form_editor):
self._form_editor = form_editor
def isContainer(self):
return False
def isInitialized(self):
return self._form_editor is not None
def name(self):
return "BECStatusBox"
def toolTip(self):
return "Widget to display the BECStatus from all active services."
def whatsThis(self):
return self.toolTip()

View File

@@ -0,0 +1,15 @@
def main(): # pragma: no cover
from qtpy import PYSIDE6
if not PYSIDE6:
print("PYSIDE6 is not available in the environment. Cannot patch designer.")
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from bec_widgets.widgets.bec_status_box.bec_status_box_plugin import BECStatusBoxPlugin
QPyDesignerCustomWidgetCollection.addCustomWidget(BECStatusBoxPlugin())
if __name__ == "__main__": # pragma: no cover
main()

View File

@@ -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()

View File

@@ -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