mirror of
https://github.com/bec-project/bec_widgets.git
synced 2025-07-14 03:31:50 +02:00
127 lines
4.8 KiB
Python
127 lines
4.8 KiB
Python
# pylint: skip-file
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
from bec_lib.messages import BECStatus, ServiceMetricMessage, StatusMessage
|
|
|
|
from bec_widgets.widgets.services.bec_status_box.bec_status_box import (
|
|
BECServiceInfoContainer,
|
|
BECStatusBox,
|
|
)
|
|
|
|
from .client_mocks import mocked_client
|
|
|
|
|
|
@pytest.fixture
|
|
def service_status_fixture():
|
|
yield mock.MagicMock()
|
|
|
|
|
|
@pytest.fixture
|
|
def status_box(qtbot, mocked_client, service_status_fixture):
|
|
widget = BECStatusBox(client=mocked_client, bec_service_status_mixin=service_status_fixture)
|
|
qtbot.addWidget(widget)
|
|
qtbot.waitExposed(widget)
|
|
yield widget
|
|
|
|
|
|
def test_update_top_item(status_box):
|
|
assert status_box.tree.children()[0].children()[0].config.status == "IDLE"
|
|
name = status_box.box_name
|
|
status_box.update_top_item_status(status="RUNNING")
|
|
assert status_box.status_container[name]["info"].status == "RUNNING"
|
|
assert status_box.tree.children()[0].children()[0].config.status == "RUNNING"
|
|
|
|
|
|
def test_create_status_widget(status_box):
|
|
name = "test_service"
|
|
status = BECStatus.IDLE
|
|
info = {"test": "test"}
|
|
metrics = {"metric": "test_metric"}
|
|
item = status_box._create_status_widget(name, status, info, metrics)
|
|
assert item.config.service_name == name
|
|
assert item.config.status == status.name
|
|
assert item.config.info == info
|
|
assert item.config.metrics == metrics
|
|
|
|
|
|
def test_bec_service_container(status_box):
|
|
name = "test_service"
|
|
status = BECStatus.IDLE
|
|
info = {"test": "test"}
|
|
metrics = {"metric": "test_metric"}
|
|
expected_return = BECServiceInfoContainer(
|
|
service_name=name, status=status.name, info=info, metrics=metrics
|
|
)
|
|
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):
|
|
name = "test_service"
|
|
status = BECStatus.IDLE
|
|
info = {"test": "test"}
|
|
metrics = {"metric": "test_metric"}
|
|
assert len(status_box.tree.children()[0].children()) == 1
|
|
status_box.add_tree_item(name, status, info, metrics)
|
|
assert len(status_box.tree.children()[0].children()) == 2
|
|
assert name in status_box.status_container
|
|
|
|
|
|
def test_update_service_status(status_box):
|
|
"""Also checks check redundant tree items"""
|
|
name = "test_service"
|
|
status = BECStatus.IDLE
|
|
info = {"test": "test"}
|
|
metrics = {"metric": "test_metric"}
|
|
status_box.add_tree_item(name, status, info, {})
|
|
not_connected_name = "invalid_service"
|
|
status_box.add_tree_item(not_connected_name, status, info, metrics)
|
|
|
|
services_status = {name: StatusMessage(name=name, status=status, info=info)}
|
|
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.status_container
|
|
status_box.update_service_status(services_status, services_metrics)
|
|
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):
|
|
status_box.CORE_SERVICES = ["test_service"]
|
|
name = "test_service"
|
|
status = BECStatus.RUNNING
|
|
info = {"test": "test"}
|
|
metrics = {"metric": "test_metric"}
|
|
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.tree.children()[0].children()[0].config.status == "RUNNING"
|
|
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.tree.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):
|
|
name = "test_service"
|
|
status = BECStatus.IDLE
|
|
info = {"test": "test"}
|
|
metrics = {"MyData": "This should be shown nicely"}
|
|
status_box.add_tree_item(name, status, info, metrics)
|
|
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.tree.itemDoubleClicked.emit(item, 0)
|
|
assert mock_show_popup.call_count == 1
|