mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-08-10 04:00:36 +02:00
509 lines
22 KiB
Python
509 lines
22 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
|
|
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",
|
|
},
|
|
)
|
|
)
|
|
# emit through the timer: _proc_update verifies its sender and skips plain calls
|
|
log_panel._model.log_queue._update_timer.timeout.emit()
|
|
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 _feed(log_panel: LogPanel, messages: list[LogMessage]):
|
|
queue = log_panel._model.log_queue
|
|
queue._incoming.extend(messages)
|
|
# emit through the timer so _proc_update's verify_sender check passes (a plain
|
|
# method call is skipped with "Sender is None")
|
|
queue._update_timer.timeout.emit()
|
|
|
|
|
|
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(LogLevel.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_detaches_from_queue(qtbot, log_panel: LogPanel):
|
|
queue = log_panel._model.log_queue
|
|
log_panel.close()
|
|
_feed(log_panel, [make_log_msg(0)])
|
|
assert log_panel._model.rowCount() == 3 # closed panel no longer receives updates
|
|
assert len(queue) == 4 # the shared history still ingests
|
|
|
|
|
|
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_default_and_preset(qtbot, log_panel: LogPanel):
|
|
box = log_panel._toolbar.filter_level_dropdown
|
|
assert box.currentIndex() == 0
|
|
assert box.currentText() == "All levels"
|
|
assert box.itemData(0) is None
|
|
assert "CONSOLE_LOG" not in [box.itemText(i) for i in range(box.count())]
|
|
log_panel._toolbar.set_level(LogLevel.SUCCESS)
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 1, timeout=200)
|
|
log_panel._toolbar.set_level(None)
|
|
qtbot.waitUntil(lambda: log_panel._proxy.rowCount() == 3, timeout=200)
|
|
|
|
|
|
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_set_level_with_unlisted_level_keeps_filter(qtbot, log_panel: LogPanel):
|
|
log_panel._proxy.update_level_filter(LogLevel.CONSOLE_LOG)
|
|
log_panel._toolbar.set_level(LogLevel.CONSOLE_LOG) # not in the dropdown
|
|
assert log_panel._proxy._level_num == LogLevel.CONSOLE_LOG.value # filter not wiped
|
|
assert log_panel._toolbar.filter_level_dropdown.currentIndex() == 0
|
|
|
|
|
|
|
|
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)
|