mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-10-13 17:50:34 +02:00
wakonig_k
efc65bdf07
Keep receiver lifetime checks in dispatcher invalidation and relay delivery. Deliver subscription context through metadata so waveform async updates and log ingestion work without QObject.sender(). Reject stale scan updates. Add regression coverage for queued delivery, subscription context isolation, and waveform and logpanel updates through the dispatcher relay.
606 lines
26 KiB
Python
606 lines
26 KiB
Python
# pylint: disable=no-member
|
|
# pylint: disable=missing-function-docstring
|
|
# pylint: disable=redefined-outer-name
|
|
# pylint: disable=protected-access
|
|
|
|
from collections import deque
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from bec_lib.logger import LogLevel
|
|
from bec_lib.messages import LogMessage
|
|
from qtpy.QtCore import QDateTime, Qt
|
|
|
|
from bec_widgets.utils.colors import apply_theme, get_accent_colors, get_theme_name
|
|
from bec_widgets.widgets.utility.logpanel.logpanel import LogPanel, TimestampUpdate
|
|
|
|
from .client_mocks import mocked_client
|
|
|
|
TEST_LOG_MESSAGES = [
|
|
{"data": msg}
|
|
for msg in [
|
|
LogMessage(
|
|
metadata={},
|
|
log_type="debug",
|
|
log_msg={
|
|
"text": "datetime | debug | test log message",
|
|
"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, monkeypatch):
|
|
monkeypatch.setattr(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()) # empty include-list shows nothing
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 0, timeout=200)
|
|
log_panel._update_service_filter(None) # no service filter shows everything
|
|
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 (include-list of level names)
|
|
log_panel._proxy.update_level_filter({"SUCCESS"})
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 1, timeout=200)
|
|
log_panel._proxy.update_level_filter({"SUCCESS", "DEBUG"})
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 2, timeout=200)
|
|
log_panel._proxy.update_level_filter(set()) # empty include-list shows nothing
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 0, 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._model.log_queue._proc_update()
|
|
qtbot.waitUntil(lambda: log_panel._model.rowCount() == 4, timeout=500)
|
|
|
|
|
|
def make_log_msg(i: int, log_type: str = "info", service: str = "ScanServer") -> LogMessage:
|
|
return LogMessage(
|
|
metadata={},
|
|
log_type=log_type,
|
|
log_msg={
|
|
"text": f"datetime | {log_type} | m{i}",
|
|
"record": {
|
|
"time": {"timestamp": 123456789.100 + i, "repr": "2025-01-01 00:00:04"},
|
|
"message": f"m{i}",
|
|
"function": "_test",
|
|
},
|
|
"service_name": service,
|
|
},
|
|
)
|
|
|
|
|
|
def test_log_panel_ingests_queued_dispatcher_message(qtbot, log_panel: LogPanel):
|
|
"""Log ingestion works through the relay without a Qt sender context."""
|
|
import threading
|
|
|
|
queue = log_panel._model.log_queue
|
|
queue._update_timer.stop()
|
|
wrapper = next(
|
|
slot
|
|
for slot in queue.bec_dispatcher._registered_slots.values()
|
|
if slot.cb == queue._process_incoming_log_msg
|
|
)
|
|
thread = threading.Thread(target=wrapper, args=(make_log_msg(1).model_dump(), {}))
|
|
thread.start()
|
|
thread.join(timeout=5)
|
|
assert not thread.is_alive()
|
|
assert len(queue._incoming) == 0
|
|
|
|
qtbot.waitUntil(lambda: len(queue._incoming) == 1)
|
|
queue._proc_update()
|
|
assert log_panel._model.rowCount() == 4
|
|
|
|
|
|
def _feed(log_panel: LogPanel, messages: list[LogMessage]):
|
|
queue = log_panel._model.log_queue
|
|
queue._incoming.extend(messages)
|
|
queue._proc_update()
|
|
|
|
|
|
def _patched_const(monkeypatch, **overrides):
|
|
import bec_widgets.widgets.utility.logpanel.logpanel as lp
|
|
|
|
values = {
|
|
"FUZZ_THRESHOLD": lp._CONST.FUZZ_THRESHOLD,
|
|
"UPDATE_INTERVAL_MS": lp._CONST.UPDATE_INTERVAL_MS,
|
|
"TRIM_CHUNK": lp._CONST.TRIM_CHUNK,
|
|
"headers": lp._CONST.headers,
|
|
}
|
|
values.update(overrides)
|
|
monkeypatch.setattr(lp, "_CONST", SimpleNamespace(**values))
|
|
|
|
|
|
def test_log_panel_appends_incrementally_and_filters_new_rows(qtbot, log_panel: LogPanel):
|
|
log_panel._proxy.update_level_filter({"WARNING"})
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 0, timeout=200)
|
|
with qtbot.waitSignal(log_panel._model.rowsInserted, timeout=500) as blocker:
|
|
_feed(log_panel, [make_log_msg(0, "warning"), make_log_msg(1, "debug")])
|
|
assert blocker.args[1:] == [3, 4] # one contiguous append of both rows
|
|
assert log_panel._model.rowCount() == 5
|
|
# the proxy evaluated only the new rows: exactly the warning one is shown
|
|
assert log_panel._proxy.rowCount() == 1
|
|
assert log_panel._proxy.index(0, 3).data() == "m0"
|
|
|
|
|
|
def test_log_panel_trims_in_chunks(qtbot, log_panel: LogPanel, monkeypatch):
|
|
_patched_const(monkeypatch, TRIM_CHUNK=3)
|
|
monkeypatch.setattr(log_panel._model, "_max_length", 6)
|
|
_feed(log_panel, [make_log_msg(i) for i in range(4)])
|
|
# overflow 1 < TRIM_CHUNK: buffer transiently exceeds max_length
|
|
assert log_panel._model.rowCount() == 7
|
|
with qtbot.waitSignal(log_panel._model.rowsRemoved, timeout=500) as blocker:
|
|
_feed(log_panel, [make_log_msg(i) for i in range(4, 8)])
|
|
assert blocker.args[1:] == [0, 4] # overflow of 5 trimmed from the top
|
|
assert log_panel._model.rowCount() == 6
|
|
assert log_panel._model.record(0).message == "m2"
|
|
assert log_panel._model.record(5).message == "m7"
|
|
|
|
|
|
def test_log_panel_huge_batch_resets_to_tail(qtbot, log_panel: LogPanel, monkeypatch):
|
|
monkeypatch.setattr(log_panel._model, "_max_length", 4)
|
|
with qtbot.waitSignal(log_panel._model.modelReset, timeout=500):
|
|
_feed(log_panel, [make_log_msg(i) for i in range(6)])
|
|
assert log_panel._model.rowCount() == 4
|
|
assert log_panel._model.record(0).message == "m2"
|
|
assert log_panel._model.record(3).message == "m5"
|
|
|
|
|
|
def test_log_panel_trim_anchors_scroll_position(qtbot, log_panel: LogPanel, monkeypatch):
|
|
_patched_const(monkeypatch, TRIM_CHUNK=5)
|
|
monkeypatch.setattr(log_panel._model, "_max_length", 20)
|
|
log_panel.resize(600, 300)
|
|
log_panel.show()
|
|
qtbot.waitExposed(log_panel)
|
|
_feed(log_panel, [make_log_msg(i) for i in range(17)])
|
|
assert log_panel._model.rowCount() == 20
|
|
scrollbar = log_panel._table.verticalScrollBar()
|
|
qtbot.waitUntil(lambda: scrollbar.maximum() >= 5, timeout=500)
|
|
scrollbar.setValue(5)
|
|
_feed(log_panel, [make_log_msg(i) for i in range(17, 22)]) # overflow 5 -> trim 5
|
|
assert scrollbar.value() == 0 # shifted by the removed count, content stays anchored
|
|
|
|
|
|
def test_log_panel_follows_tail_when_pinned_to_bottom(qtbot, log_panel: LogPanel, monkeypatch):
|
|
_patched_const(monkeypatch, TRIM_CHUNK=5)
|
|
monkeypatch.setattr(log_panel._model, "_max_length", 20)
|
|
log_panel.resize(600, 300)
|
|
log_panel.show()
|
|
qtbot.waitExposed(log_panel)
|
|
_feed(log_panel, [make_log_msg(i) for i in range(17)])
|
|
scrollbar = log_panel._table.verticalScrollBar()
|
|
qtbot.waitUntil(lambda: scrollbar.maximum() >= 5, timeout=500)
|
|
log_panel._table.scrollToBottom()
|
|
for start in range(17, 37, 5): # several trim cycles at steady state
|
|
_feed(log_panel, [make_log_msg(i) for i in range(start, start + 5)])
|
|
assert scrollbar.value() == scrollbar.maximum() # still tailing the newest logs
|
|
last_visible = log_panel._proxy.index(log_panel._proxy.rowCount() - 1, 3).data()
|
|
assert last_visible == "m36"
|
|
|
|
|
|
def test_log_panel_survives_malformed_messages(qtbot, log_panel: LogPanel):
|
|
# shapes that pass LogMessage validation (log_msg is `dict | str`, no inner schema)
|
|
# but used to break record flattening, filtering, or painting
|
|
poison = [
|
|
LogMessage(metadata={}, log_type="info", log_msg={"record": "oops"}),
|
|
LogMessage(metadata={}, log_type="info", log_msg={"record": {"message": 5, "time": 1.0}}),
|
|
LogMessage(metadata={}, log_type="console_log", log_msg="plain string payload"),
|
|
LogMessage(
|
|
metadata={},
|
|
log_type="info",
|
|
log_msg={
|
|
"service_name": ["not", "a", "string"],
|
|
"record": {"message": {"nested": 1}, "time": {"timestamp": "abc", "repr": 7}},
|
|
},
|
|
),
|
|
]
|
|
# activate every filter type so filterAcceptsRow runs all comparisons on the new rows
|
|
log_panel._proxy.update_timestamp(
|
|
TimestampUpdate(value=QDateTime.fromMSecsSinceEpoch(0), update_type="start")
|
|
)
|
|
log_panel._proxy.update_service_filter({"ScanServer"})
|
|
log_panel._proxy.update_filter_text("payload")
|
|
_feed(log_panel, poison)
|
|
assert log_panel._model.rowCount() == 7 # the whole batch landed, nothing raised
|
|
# a panel constructed after the poison entered the shared history must still build
|
|
second_panel = LogPanel()
|
|
qtbot.addWidget(second_panel)
|
|
assert second_panel._model.rowCount() == 7
|
|
second_panel.close()
|
|
|
|
|
|
def test_log_panel_close_tears_down_queue(qtbot, log_panel: LogPanel):
|
|
from bec_widgets.widgets.utility.logpanel.logpanel import BecLogsQueue
|
|
|
|
queue = log_panel._model.log_queue
|
|
assert not queue.rpc_register.object_is_registered(queue) # never an RPC connection
|
|
log_panel.close()
|
|
# the last panel's close fully tears the queue down - nothing lingers after it
|
|
assert BecLogsQueue._instance is None
|
|
assert not queue._update_timer.isActive()
|
|
_feed(log_panel, [make_log_msg(0)])
|
|
assert log_panel._model.rowCount() == 3 # closed panel no longer receives updates
|
|
|
|
|
|
def test_log_panel_queue_survives_until_last_panel_closes(qtbot, mocked_client, monkeypatch):
|
|
from bec_widgets.widgets.utility.logpanel.logpanel import BecLogsQueue
|
|
|
|
monkeypatch.setattr(mocked_client.connector, "xread", lambda *_, **__: TEST_LOG_MESSAGES)
|
|
first = LogPanel()
|
|
qtbot.addWidget(first)
|
|
second = LogPanel()
|
|
qtbot.addWidget(second)
|
|
queue = first._model.log_queue
|
|
assert second._model.log_queue is queue
|
|
second.close()
|
|
assert BecLogsQueue._instance is queue # one panel still open
|
|
assert queue._update_timer.isActive()
|
|
first.close()
|
|
assert BecLogsQueue._instance is None
|
|
assert not queue._update_timer.isActive()
|
|
# a new panel re-creates the queue and backfills its history from Redis
|
|
reopened = LogPanel()
|
|
qtbot.addWidget(reopened)
|
|
assert reopened._model.log_queue is not queue
|
|
assert reopened._model.rowCount() == 3
|
|
reopened.close()
|
|
|
|
|
|
def test_direct_queue_construction_registers_singleton(qtbot, mocked_client, monkeypatch):
|
|
from bec_widgets.widgets.utility.logpanel.logpanel import BecLogsQueue
|
|
|
|
monkeypatch.setattr(mocked_client.connector, "xread", lambda *_, **__: TEST_LOG_MESSAGES)
|
|
queue = BecLogsQueue(None, client=mocked_client)
|
|
try:
|
|
assert BecLogsQueue._instance is queue
|
|
assert BecLogsQueue.instance() is queue
|
|
with pytest.raises(RuntimeError):
|
|
BecLogsQueue(None, client=mocked_client)
|
|
finally:
|
|
queue.cleanup()
|
|
|
|
|
|
def test_log_panel_copy_selection(qtbot, log_panel: LogPanel):
|
|
from qtpy.QtWidgets import QApplication
|
|
|
|
log_panel._table.selectAll()
|
|
log_panel._copy_selection()
|
|
copied = QApplication.clipboard().text().splitlines()
|
|
assert len(copied) == 3
|
|
assert "test debug message abcd" in copied[0]
|
|
assert "[DEBUG]" in copied[0] and "ScanServer" in copied[0]
|
|
|
|
|
|
def test_log_panel_detail_pane(qtbot, log_panel: LogPanel):
|
|
assert log_panel._detail.isHidden()
|
|
log_panel._table.setCurrentIndex(log_panel._proxy.index(1, 0))
|
|
log_panel._show_details()
|
|
assert not log_panel._detail.isHidden()
|
|
assert log_panel._detail_text.toPlainText() == "test info message efgh"
|
|
assert "INFO" in log_panel._detail_header.text()
|
|
# selection changes update the open pane (sync is deferred by one event-loop turn)
|
|
log_panel._table.setCurrentIndex(log_panel._proxy.index(2, 0))
|
|
qtbot.waitUntil(
|
|
lambda: log_panel._detail_text.toPlainText() == "test success message ijkl", timeout=500
|
|
)
|
|
log_panel._hide_details()
|
|
assert log_panel._detail.isHidden()
|
|
|
|
|
|
def test_log_panel_clear_view_keeps_history(qtbot, log_panel: LogPanel):
|
|
queue = log_panel._model.log_queue
|
|
log_panel._clear_view()
|
|
assert log_panel._model.rowCount() == 0
|
|
assert len(queue) == 3 # shared history untouched
|
|
_feed(log_panel, [make_log_msg(0)])
|
|
assert log_panel._model.rowCount() == 1 # new logs keep arriving after a clear
|
|
|
|
|
|
def test_log_panel_toolbar_service_selection(qtbot, log_panel: LogPanel):
|
|
toolbar = log_panel._toolbar
|
|
toolbar._known_services = {"ScanServer", "DeviceServer"}
|
|
toolbar.set_service_selection({"DeviceServer"})
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 1, timeout=200)
|
|
assert toolbar.service_button.text() == "DeviceServer"
|
|
toolbar.hide_service("DeviceServer")
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 0, timeout=200)
|
|
assert toolbar.service_button.text() == "No services"
|
|
# re-checking every known service through the menu collapses to the unfiltered state
|
|
toolbar.set_service_selection({"ScanServer"})
|
|
toolbar._on_service_toggled("DeviceServer", True)
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 3, timeout=200)
|
|
assert toolbar.service_button.text() == "All services"
|
|
assert toolbar._checked_services is None
|
|
|
|
|
|
def test_log_panel_toolbar_level_selection(qtbot, log_panel: LogPanel):
|
|
toolbar = log_panel._toolbar
|
|
assert toolbar._checked_levels is None
|
|
assert toolbar.level_button.text() == "All levels"
|
|
assert "CONSOLE_LOG" not in toolbar._level_actions # internal levels stay unlisted
|
|
toolbar.set_level_selection({"SUCCESS"})
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 1, timeout=200)
|
|
assert toolbar.level_button.text() == "SUCCESS"
|
|
# unchecking a level through the menu removes it from the include-list
|
|
toolbar._sync_level_menu()
|
|
toolbar._level_actions["SUCCESS"].setChecked(False)
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 0, timeout=200)
|
|
assert toolbar.level_button.text() == "No levels"
|
|
# re-checking every listed level collapses to the unfiltered state
|
|
for action in toolbar._level_actions.values():
|
|
action.setChecked(True)
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 3, timeout=200)
|
|
assert toolbar._checked_levels is None
|
|
assert toolbar.level_button.text() == "All levels"
|
|
|
|
|
|
def test_log_panel_constructor_level_filter_expands_to_threshold(qtbot, mocked_client, monkeypatch):
|
|
monkeypatch.setattr(mocked_client.connector, "xread", lambda *_, **__: TEST_LOG_MESSAGES)
|
|
widget = LogPanel(level_filter=LogLevel.SUCCESS)
|
|
qtbot.addWidget(widget)
|
|
try:
|
|
# threshold semantics preserved: SUCCESS and above (the debug/info rows hide)
|
|
assert widget._proxy.rowCount() == 1
|
|
assert widget._toolbar._checked_levels == {"SUCCESS", "WARNING", "ERROR", "CRITICAL"}
|
|
assert widget._toolbar.level_button.text() == "4 levels"
|
|
finally:
|
|
widget._model.log_queue.cleanup()
|
|
|
|
|
|
def test_log_panel_search_debounce_and_match_count(qtbot, log_panel: LogPanel):
|
|
toolbar = log_panel._toolbar
|
|
toolbar.search_textbox.setText("efgh")
|
|
assert log_panel._proxy.rowCount() == 3 # not yet applied - debounced
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 1, timeout=1000)
|
|
assert toolbar.match_label.text() == "1 / 3"
|
|
toolbar.search_textbox.clear() # clearing applies instantly
|
|
assert log_panel._proxy.rowCount() == 3
|
|
assert toolbar.match_label.text() == "3"
|
|
|
|
|
|
def test_log_panel_jump_button_counts_new_rows(qtbot, log_panel: LogPanel, monkeypatch):
|
|
monkeypatch.setattr(log_panel._model, "_max_length", 50)
|
|
log_panel.resize(600, 300)
|
|
log_panel.show()
|
|
qtbot.waitExposed(log_panel)
|
|
_feed(log_panel, [make_log_msg(i) for i in range(30)])
|
|
table = log_panel._table
|
|
qtbot.waitUntil(lambda: table.verticalScrollBar().maximum() > 0, timeout=500)
|
|
table.verticalScrollBar().setValue(0) # scroll up to read history
|
|
_feed(log_panel, [make_log_msg(i) for i in range(30, 35)])
|
|
assert table._jump_button.isVisible()
|
|
assert table._jump_button.text() == "5 new"
|
|
table._jump_button.click()
|
|
qtbot.waitUntil(lambda: not table._jump_button.isVisible(), timeout=500)
|
|
scrollbar = table.verticalScrollBar()
|
|
assert scrollbar.value() == scrollbar.maximum()
|
|
|
|
|
|
def test_log_panel_constructor_service_filter_stays_applied(qtbot, mocked_client, monkeypatch):
|
|
monkeypatch.setattr(mocked_client.connector, "xread", lambda *_, **__: TEST_LOG_MESSAGES)
|
|
widget = LogPanel(service_filter={"DeviceServer"})
|
|
qtbot.addWidget(widget)
|
|
try:
|
|
assert widget._proxy.rowCount() == 1 # only the DeviceServer row
|
|
assert widget._toolbar.service_button.text() == "DeviceServer"
|
|
finally:
|
|
widget._model.log_queue.cleanup()
|
|
|
|
|
|
def test_log_panel_context_service_filters_on_fresh_panel(qtbot, log_panel: LogPanel):
|
|
# neither action may depend on the service menu having been opened before
|
|
log_panel._filter_service_only("DeviceServer")
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 1, timeout=200)
|
|
log_panel._toolbar.set_service_selection(None)
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 3, timeout=200)
|
|
log_panel._filter_service_hide("ScanServer")
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 1, timeout=200)
|
|
assert log_panel._proxy.index(0, 2).data() == "DeviceServer"
|
|
|
|
|
|
def test_log_panel_filter_change_does_not_show_jump_pill(qtbot, log_panel: LogPanel, monkeypatch):
|
|
monkeypatch.setattr(log_panel._model, "_max_length", 100)
|
|
log_panel.resize(600, 300)
|
|
log_panel.show()
|
|
qtbot.waitExposed(log_panel)
|
|
_feed(log_panel, [make_log_msg(i) for i in range(40)])
|
|
table = log_panel._table
|
|
qtbot.waitUntil(lambda: table.verticalScrollBar().maximum() > 0, timeout=500)
|
|
table.verticalScrollBar().setValue(0)
|
|
log_panel._proxy.update_filter_text("m1") # tighten, then relax: no new logs arrived
|
|
log_panel._proxy.update_filter_text("")
|
|
assert not table._jump_button.isVisible()
|
|
assert table._new_below == 0
|
|
|
|
|
|
def test_log_panel_filter_tighten_does_not_leak_scroll_latch(qtbot, log_panel, monkeypatch):
|
|
monkeypatch.setattr(log_panel._model, "_max_length", 100)
|
|
log_panel.resize(600, 300)
|
|
log_panel.show()
|
|
qtbot.waitExposed(log_panel)
|
|
_feed(log_panel, [make_log_msg(i) for i in range(40)])
|
|
table = log_panel._table
|
|
qtbot.waitUntil(lambda: table.verticalScrollBar().maximum() > 0, timeout=500)
|
|
table.scrollToBottom()
|
|
log_panel._proxy.update_filter_text("m") # removal-only cycle while pinned to bottom
|
|
table.verticalScrollBar().setValue(0) # then scroll up to read
|
|
_feed(log_panel, [make_log_msg(99)]) # a new matching log must not yank the view
|
|
assert table.verticalScrollBar().value() == 0
|
|
|
|
|
|
def test_log_panel_detail_pane_freezes_when_record_trimmed(qtbot, log_panel, monkeypatch):
|
|
_patched_const(monkeypatch, TRIM_CHUNK=3)
|
|
monkeypatch.setattr(log_panel._model, "_max_length", 8)
|
|
log_panel._table.setCurrentIndex(log_panel._proxy.index(0, 0))
|
|
log_panel._show_details()
|
|
assert log_panel._detail_text.toPlainText() == "test debug message abcd"
|
|
_feed(log_panel, [make_log_msg(i) for i in range(4)]) # append-only, 7 rows
|
|
_feed(log_panel, [make_log_msg(i) for i in range(4, 8)]) # trims the shown record away
|
|
qtbot.wait(100) # let the deferred selection sync run
|
|
assert log_panel._detail_text.toPlainText() == "test debug message abcd"
|
|
assert "(no longer in buffer)" in log_panel._detail_header.text()
|
|
# an explicit click follows the selection again, out of the freeze
|
|
log_panel._table.setCurrentIndex(log_panel._proxy.index(0, 0))
|
|
log_panel._on_row_clicked()
|
|
assert log_panel._detail_frozen is False
|
|
assert "(no longer in buffer)" not in log_panel._detail_header.text()
|
|
|
|
|
|
def test_log_panel_unlisted_levels_follow_selection(qtbot, log_panel: LogPanel):
|
|
# CONSOLE_LOG rows are visible while unfiltered, hidden by any narrowed selection,
|
|
# and still selectable explicitly through the proxy API
|
|
_feed(log_panel, [make_log_msg(0, "console_log")])
|
|
assert log_panel._proxy.rowCount() == 4
|
|
log_panel._toolbar.set_level_selection({"DEBUG", "INFO", "SUCCESS"})
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 3, timeout=200)
|
|
log_panel._proxy.update_level_filter({"CONSOLE_LOG"})
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 1, timeout=200)
|
|
|
|
|
|
def test_log_panel_font_size_property(qtbot, log_panel: LogPanel):
|
|
from qtpy.QtGui import QFontMetrics
|
|
|
|
assert log_panel.font_size == 0 # default: system font size
|
|
default_point_size = log_panel._table.font().pointSize()
|
|
default_level_width = log_panel._table.columnWidth(0)
|
|
|
|
log_panel.font_size = 16
|
|
assert log_panel.font_size == 16
|
|
assert log_panel._table.font().pointSize() == 16
|
|
assert log_panel._detail_text.font().pointSize() == 16
|
|
metrics = QFontMetrics(log_panel._table.font())
|
|
assert log_panel._table.verticalHeader().defaultSectionSize() == metrics.height() + 4
|
|
assert log_panel._table.columnWidth(0) > default_level_width # widths recomputed
|
|
|
|
log_panel.font_size = 0 # back to system default
|
|
assert log_panel._table.font().pointSize() == default_point_size
|
|
assert log_panel._table.columnWidth(0) == default_level_width
|
|
|
|
|
|
def test_log_panel_custom_range_dialog_prefills_active_bounds(qtbot, log_panel: LogPanel):
|
|
toolbar = log_panel._toolbar
|
|
toolbar._apply_time_preset("Last 5 min", 300)
|
|
qtbot.waitUntil(lambda: log_panel._proxy._ts_start is not None, timeout=200)
|
|
dialog, bounds = toolbar._build_custom_range_dialog()
|
|
try:
|
|
start_enable, start_edit = bounds["start"]
|
|
end_enable, end_edit = bounds["end"]
|
|
# prefilled from the active preset instead of stale defaults
|
|
assert start_enable.isChecked()
|
|
assert start_edit.dateTime() == toolbar._active_start
|
|
assert not end_enable.isChecked()
|
|
# applying with an added end bound routes both bounds to the proxy
|
|
end_enable.setChecked(True)
|
|
end_edit.setDateTime(toolbar._active_start.addSecs(60))
|
|
toolbar._apply_custom_range(bounds)
|
|
qtbot.waitUntil(lambda: log_panel._proxy._ts_end is not None, timeout=200)
|
|
assert "-" in toolbar.timerange_button.text()
|
|
# and 'All time' clears everything again
|
|
toolbar._apply_time_preset("All time", None)
|
|
qtbot.waitUntil(lambda: log_panel._proxy._ts_start is None, timeout=200)
|
|
assert log_panel._proxy._ts_end is None
|
|
assert toolbar.timerange_button.text() == "All time"
|
|
finally:
|
|
dialog.deleteLater()
|
|
|
|
|
|
def test_log_panel_colors_follow_theme(qtbot, log_panel: LogPanel):
|
|
info_index = log_panel._model.index(1, 0)
|
|
success_index = log_panel._model.index(2, 0)
|
|
|
|
# INFO rows have no explicit color so the view falls back to the palette text color
|
|
assert log_panel._model.data(info_index, Qt.ItemDataRole.ForegroundRole) is None
|
|
|
|
original_theme = get_theme_name()
|
|
other_theme = "dark" if original_theme == "light" else "light"
|
|
try:
|
|
for theme in (other_theme, original_theme):
|
|
apply_theme(theme)
|
|
success_color = log_panel._model.data(success_index, Qt.ItemDataRole.ForegroundRole)
|
|
assert success_color == get_accent_colors().success
|
|
finally:
|
|
apply_theme(original_theme)
|