mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-06-09 06:38:57 +02:00
feat: migrate logpanel to table model/view
This commit is contained in:
+18
-5
@@ -1,3 +1,5 @@
|
||||
import traceback
|
||||
|
||||
import pytest
|
||||
import qtpy.QtCore
|
||||
from pytestqt.exceptions import TimeoutError as QtBotTimeoutError
|
||||
@@ -5,12 +7,14 @@ from qtpy.QtCore import QTimer
|
||||
|
||||
|
||||
class TestableQTimer(QTimer):
|
||||
_instances: list[tuple[QTimer, str]] = []
|
||||
_instances: list[tuple[QTimer, str, str]] = []
|
||||
_current_test_name: str = ""
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
TestableQTimer._instances.append((self, TestableQTimer._current_test_name))
|
||||
tb = traceback.format_stack()
|
||||
init_line = list(filter(lambda msg: "QTimer(" in msg, tb))[-1]
|
||||
TestableQTimer._instances.append((self, TestableQTimer._current_test_name, init_line))
|
||||
|
||||
@classmethod
|
||||
def check_all_stopped(cls, qtbot):
|
||||
@@ -20,12 +24,21 @@ class TestableQTimer(QTimer):
|
||||
except RuntimeError as e:
|
||||
return "already deleted" in e.args[0]
|
||||
|
||||
def _format_timers(timers: list[tuple[QTimer, str, str]]):
|
||||
return "\n".join(
|
||||
f"Timer: {t[0]}\n in test: {t[1]}\n created at:{t[2]}" for t in timers
|
||||
)
|
||||
|
||||
try:
|
||||
qtbot.waitUntil(lambda: all(_is_done_or_deleted(timer) for timer, _ in cls._instances))
|
||||
qtbot.waitUntil(
|
||||
lambda: all(_is_done_or_deleted(timer) for timer, _, _ in cls._instances)
|
||||
)
|
||||
except QtBotTimeoutError as exc:
|
||||
active_timers = list(filter(lambda t: t[0].isActive(), cls._instances))
|
||||
(t.stop() for t, _ in cls._instances)
|
||||
raise TimeoutError(f"Failed to stop all timers: {active_timers}") from exc
|
||||
(t.stop() for t, _, _ in cls._instances)
|
||||
raise TimeoutError(
|
||||
f"Failed to stop all timers:\n{_format_timers(active_timers)}"
|
||||
) from exc
|
||||
cls._instances = []
|
||||
|
||||
|
||||
|
||||
@@ -363,23 +363,6 @@ def test_widgets_e2e_image(qtbot, connected_client_gui_obj, random_generator_fro
|
||||
maybe_remove_dock_area(qtbot, gui=gui, random_int_gen=random_generator_from_seed)
|
||||
|
||||
|
||||
# TODO re-enable when issue is resolved #560
|
||||
# @pytest.mark.timeout(PYTEST_TIMEOUT)
|
||||
# def test_widgets_e2e_log_panel(qtbot, connected_client_gui_obj, random_generator_from_seed):
|
||||
# """Test the LogPanel widget."""
|
||||
# gui = connected_client_gui_obj
|
||||
# bec = gui._client
|
||||
# # Create dock_area, dock, widget
|
||||
# dock, widget = create_widget(qtbot, gui, gui.available_widgets.LogPanel)
|
||||
# dock: client.BECDock
|
||||
# widget: client.LogPanel
|
||||
|
||||
# # No rpc calls to check so far
|
||||
|
||||
# # Test removing the widget, or leaving it open for the next test
|
||||
# maybe_remove_dock_area(qtbot, gui=gui, random_int_gen=random_generator_from_seed)
|
||||
|
||||
|
||||
@pytest.mark.timeout(PYTEST_TIMEOUT)
|
||||
def test_widgets_e2e_minesweeper(qtbot, connected_client_gui_obj, random_generator_from_seed):
|
||||
"""Test the MineSweeper widget."""
|
||||
|
||||
@@ -2095,7 +2095,6 @@ class TestFlatToolbarActions:
|
||||
"flat_progress_bar",
|
||||
"flat_terminal",
|
||||
"flat_bec_shell",
|
||||
"flat_log_panel",
|
||||
"flat_sbb_monitor",
|
||||
]
|
||||
|
||||
@@ -2155,11 +2154,6 @@ class TestFlatToolbarActions:
|
||||
action.trigger()
|
||||
mock_new.assert_called_once_with(widget_type)
|
||||
|
||||
def test_flat_log_panel_action_disabled(self, advanced_dock_area):
|
||||
"""Test that flat log panel action is disabled."""
|
||||
action = advanced_dock_area.toolbar.components.get_action("flat_log_panel").action
|
||||
assert not action.isEnabled()
|
||||
|
||||
|
||||
class TestModeTransitions:
|
||||
"""Test mode transitions and state consistency."""
|
||||
|
||||
+106
-146
@@ -7,163 +7,123 @@ from collections import deque
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from bec_lib.logger import LogLevel
|
||||
from bec_lib.messages import LogMessage
|
||||
from bec_lib.redis_connector import StreamMessage
|
||||
from qtpy.QtCore import QDateTime
|
||||
|
||||
from bec_widgets.widgets.utility.logpanel._util import (
|
||||
log_time,
|
||||
replace_escapes,
|
||||
simple_color_format,
|
||||
)
|
||||
from bec_widgets.widgets.utility.logpanel.logpanel import DEFAULT_LOG_COLORS, LogPanel
|
||||
from bec_widgets.widgets.utility.logpanel.logpanel import LogPanel, TimestampUpdate
|
||||
|
||||
from .client_mocks import mocked_client
|
||||
|
||||
TEST_TABLE_STRING = "2025-01-15 15:57:18 | bec_server.scan_server.scan_queue | [INFO] | \n \x1b[3m primary queue / ScanQueueStatus.RUNNING \x1b[0m\n┏━━━━━━━━━━━━━┳━━━━━━━━━┳━━━━━━━━━┳━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━┓\n┃\x1b[1m \x1b[0m\x1b[1m queue_id \x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mscan_id\x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mis_scan\x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mtype\x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mscan_numb…\x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mIQ status\x1b[0m\x1b[1m \x1b[0m┃\n┡━━━━━━━━━━━━━╇━━━━━━━━━╇━━━━━━━━━╇━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━┩\n│ bbe50c82-6… │ None │ False │ mv │ None │ PENDING │\n└─────────────┴─────────┴─────────┴──────┴────────────┴───────────┘\n\n"
|
||||
|
||||
TEST_LOG_MESSAGES = [
|
||||
LogMessage(
|
||||
metadata={},
|
||||
log_type="debug",
|
||||
log_msg={
|
||||
"text": "datetime | debug | test log message",
|
||||
"record": {"time": {"timestamp": 123456789.000}},
|
||||
"service_name": "ScanServer",
|
||||
},
|
||||
),
|
||||
LogMessage(
|
||||
metadata={},
|
||||
log_type="info",
|
||||
log_msg={
|
||||
"text": "datetime | info | test log message",
|
||||
"record": {"time": {"timestamp": 123456789.007}},
|
||||
"service_name": "ScanServer",
|
||||
},
|
||||
),
|
||||
LogMessage(
|
||||
metadata={},
|
||||
log_type="success",
|
||||
log_msg={
|
||||
"text": "datetime | success | test log message",
|
||||
"record": {"time": {"timestamp": 123456789.012}},
|
||||
"service_name": "ScanServer",
|
||||
},
|
||||
),
|
||||
]
|
||||
|
||||
TEST_COMBINED_PLAINTEXT = "datetime | debug | test log message\ndatetime | info | test log message\ndatetime | success | test log message\n"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def raw_queue():
|
||||
yield deque(TEST_LOG_MESSAGES, maxlen=100)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def log_panel(qtbot, mocked_client: MagicMock):
|
||||
widget = LogPanel(client=mocked_client, service_status=MagicMock())
|
||||
qtbot.addWidget(widget)
|
||||
qtbot.waitExposed(widget)
|
||||
yield widget
|
||||
|
||||
|
||||
def test_log_panel_init(log_panel: LogPanel):
|
||||
assert log_panel.plain_text == ""
|
||||
|
||||
|
||||
def test_table_string_processing():
|
||||
assert "\x1b" in TEST_TABLE_STRING
|
||||
sanitized = replace_escapes(TEST_TABLE_STRING)
|
||||
assert "\x1b" not in sanitized
|
||||
assert " " not in sanitized
|
||||
assert "\n" not in sanitized
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
["msg", "color"], zip(TEST_LOG_MESSAGES, ["#0000CC", "#FFFFFF", "#00FF00"])
|
||||
)
|
||||
def test_color_format(msg: LogMessage, color: str):
|
||||
assert color in simple_color_format(msg, DEFAULT_LOG_COLORS)
|
||||
|
||||
|
||||
def test_logpanel_output(qtbot, log_panel: LogPanel):
|
||||
log_panel._log_manager._data = deque(TEST_LOG_MESSAGES)
|
||||
log_panel._on_redraw()
|
||||
assert log_panel.plain_text == TEST_COMBINED_PLAINTEXT
|
||||
|
||||
def display_queue_empty():
|
||||
print(log_panel._log_manager._display_queue)
|
||||
return len(log_panel._log_manager._display_queue) == 0
|
||||
|
||||
next_text = "datetime | error | test log message"
|
||||
msg = LogMessage(
|
||||
metadata={},
|
||||
log_type="error",
|
||||
log_msg={"text": next_text, "record": {}, "service_name": "ScanServer"},
|
||||
)
|
||||
log_panel._log_manager._process_incoming_log_msg(
|
||||
msg.content, msg.metadata, _override_slot_params={"verify_sender": False}
|
||||
)
|
||||
|
||||
qtbot.waitUntil(display_queue_empty, timeout=5000)
|
||||
assert log_panel.plain_text == TEST_COMBINED_PLAINTEXT + next_text + "\n"
|
||||
|
||||
|
||||
def test_level_filter(log_panel: LogPanel):
|
||||
log_panel._log_manager._data = deque(TEST_LOG_MESSAGES)
|
||||
log_panel._log_manager.update_level_filter("INFO")
|
||||
log_panel._on_redraw()
|
||||
assert (
|
||||
log_panel.plain_text
|
||||
== "datetime | info | test log message\ndatetime | success | test log message\n"
|
||||
)
|
||||
|
||||
|
||||
def test_clear_button(log_panel: LogPanel):
|
||||
log_panel._log_manager._data = deque(TEST_LOG_MESSAGES)
|
||||
log_panel.toolbar.clear_button.click()
|
||||
assert log_panel._log_manager._data == deque([])
|
||||
|
||||
|
||||
def test_timestamp_filter(log_panel: LogPanel):
|
||||
log_panel._log_manager._timestamp_start = QDateTime(1973, 11, 29, 21, 33, 9, 5, 1)
|
||||
pytest.approx(log_panel._log_manager._timestamp_start.toMSecsSinceEpoch() / 1000, 123456789.005)
|
||||
log_panel._log_manager._timestamp_end = QDateTime(1973, 11, 29, 21, 33, 9, 10, 1)
|
||||
pytest.approx(log_panel._log_manager._timestamp_end.toMSecsSinceEpoch() / 1000, 123456789.010)
|
||||
filter_ = log_panel._log_manager._create_timestamp_filter()
|
||||
assert not filter_(TEST_LOG_MESSAGES[0])
|
||||
assert filter_(TEST_LOG_MESSAGES[1])
|
||||
assert not filter_(TEST_LOG_MESSAGES[2])
|
||||
|
||||
|
||||
def test_error_handling_in_callback(log_panel: LogPanel):
|
||||
log_panel._log_manager.new_message = MagicMock()
|
||||
|
||||
with patch("bec_widgets.widgets.utility.logpanel.logpanel.logger") as logger:
|
||||
# generally errors should be logged
|
||||
log_panel._log_manager.new_message.emit = MagicMock(
|
||||
side_effect=ValueError("Something went wrong")
|
||||
)
|
||||
msg = LogMessage(
|
||||
{"data": msg}
|
||||
for msg in [
|
||||
LogMessage(
|
||||
metadata={},
|
||||
log_type="debug",
|
||||
log_msg={
|
||||
"text": "datetime | debug | test log message",
|
||||
"record": {"time": {"timestamp": 123456789.000}},
|
||||
"record": {
|
||||
"time": {"timestamp": 123456789.000, "repr": "2025-01-01 00:00:01"},
|
||||
"message": "test debug message abcd",
|
||||
"function": "_debug",
|
||||
},
|
||||
"service_name": "ScanServer",
|
||||
},
|
||||
),
|
||||
LogMessage(
|
||||
metadata={},
|
||||
log_type="info",
|
||||
log_msg={
|
||||
"text": "datetime | info | test info log message",
|
||||
"record": {
|
||||
"time": {"timestamp": 123456789.007, "repr": "2025-01-01 00:00:02"},
|
||||
"message": "test info message efgh",
|
||||
"function": "_info",
|
||||
},
|
||||
"service_name": "DeviceServer",
|
||||
},
|
||||
),
|
||||
LogMessage(
|
||||
metadata={},
|
||||
log_type="success",
|
||||
log_msg={
|
||||
"text": "datetime | success | test log message",
|
||||
"record": {
|
||||
"time": {"timestamp": 123456789.012, "repr": "2025-01-01 00:00:03"},
|
||||
"message": "test success message ijkl",
|
||||
"function": "_success",
|
||||
},
|
||||
"service_name": "ScanServer",
|
||||
},
|
||||
),
|
||||
]
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def log_panel(qtbot, mocked_client):
|
||||
mocked_client.connector.xread = lambda *_, **__: TEST_LOG_MESSAGES
|
||||
widget = LogPanel()
|
||||
qtbot.addWidget(widget)
|
||||
qtbot.waitExposed(widget)
|
||||
yield widget
|
||||
widget._model.log_queue.cleanup()
|
||||
widget.close()
|
||||
widget.deleteLater()
|
||||
qtbot.wait(100)
|
||||
|
||||
|
||||
def test_log_panel_init(qtbot, log_panel: LogPanel):
|
||||
assert log_panel
|
||||
|
||||
|
||||
def test_log_panel_filters(qtbot, log_panel: LogPanel):
|
||||
assert log_panel._proxy.rowCount() == 3
|
||||
# Service filter
|
||||
log_panel._update_service_filter({"DeviceServer"})
|
||||
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 1, timeout=200)
|
||||
log_panel._update_service_filter(set())
|
||||
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 3, timeout=200)
|
||||
# Text filter
|
||||
log_panel._proxy.update_filter_text("efgh")
|
||||
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 1, timeout=200)
|
||||
log_panel._proxy.update_filter_text("")
|
||||
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 3, timeout=200)
|
||||
# Time filter
|
||||
log_panel._proxy.update_timestamp(
|
||||
TimestampUpdate(value=QDateTime.fromMSecsSinceEpoch(123456789004), update_type="start")
|
||||
)
|
||||
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 2, timeout=200)
|
||||
log_panel._proxy.update_timestamp(
|
||||
TimestampUpdate(value=QDateTime.fromMSecsSinceEpoch(123456789009), update_type="end")
|
||||
)
|
||||
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 1, timeout=200)
|
||||
log_panel._proxy.update_timestamp(TimestampUpdate(value=None, update_type="start"))
|
||||
log_panel._proxy.update_timestamp(TimestampUpdate(value=None, update_type="end"))
|
||||
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 3, timeout=200)
|
||||
# Level filter
|
||||
log_panel._proxy.update_level_filter(LogLevel.SUCCESS)
|
||||
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 1, timeout=200)
|
||||
log_panel._proxy.update_level_filter(None)
|
||||
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 3, timeout=200)
|
||||
|
||||
|
||||
def test_log_panel_update(qtbot, log_panel: LogPanel):
|
||||
log_panel._model.log_queue._incoming.append(
|
||||
LogMessage(
|
||||
metadata={},
|
||||
log_type="error",
|
||||
log_msg={
|
||||
"text": "datetime | error | test log message",
|
||||
"record": {
|
||||
"time": {"timestamp": 123456789.015, "repr": "2025-01-01 00:00:03"},
|
||||
"message": "test error message xyz",
|
||||
"function": "_error",
|
||||
},
|
||||
"service_name": "ScanServer",
|
||||
},
|
||||
)
|
||||
log_panel._log_manager._process_incoming_log_msg(
|
||||
msg.content, msg.metadata, _override_slot_params={"verify_sender": False}
|
||||
)
|
||||
logger.warning.assert_called_once()
|
||||
|
||||
# this specific error should be ignored and not relogged
|
||||
log_panel._log_manager.new_message.emit = MagicMock(
|
||||
side_effect=RuntimeError("Internal C++ object (BecLogsQueue) already deleted.")
|
||||
)
|
||||
log_panel._log_manager._process_incoming_log_msg(
|
||||
msg.content, msg.metadata, _override_slot_params={"verify_sender": False}
|
||||
)
|
||||
logger.warning.assert_called_once()
|
||||
)
|
||||
log_panel._model.log_queue._proc_update()
|
||||
qtbot.waitUntil(lambda: log_panel._model.rowCount() == 4, timeout=500)
|
||||
|
||||
Reference in New Issue
Block a user