loop-uh
fe37d0e7a3
Some checks failed
Windows project source guards / test (push) Failing after 39s
Замер на win10: у «шумного» ядра каждая строка лога с новым временем, id соединения или портом становилась отдельной записью журнала, а весь журнал копировался в GUI на каждую строку и переформатировался — CPU и память GUI росли с аптаймом (при 15 тыс. записей +0,19 ядра, +33 МБ). - RuntimeErrorJournal: записи группируются без учёта чисел (последний исходный текст сохраняется), не больше 300 записей — вытесняется та, что дольше всех не повторялась; порядок снимка прежний. - Снимок журнала уходит в GUI не чаще раза в секунду; вкладка «Ошибки ядер» форматирует записи только когда открыта. - Сбор трафика по процессам: вместо набора всех id соединений за сессию — владелец живых соединений (удаляется при закрытии) и счётчик на процесс.
56 lines
1.8 KiB
Python
56 lines
1.8 KiB
Python
"""Снимок журнала ошибок уходит в GUI не чаще раза в секунду.
|
|
|
|
Keep the ``test_app_*`` prefix (QApplication must exist before
|
|
tests/test_engine_process_stop.py creates a bare QCoreApplication).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import unittest
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
from PyQt6.QtCore import QEventLoop, QObject, QTimer, pyqtSignal
|
|
from PyQt6.QtWidgets import QApplication
|
|
|
|
_existing = QApplication.instance()
|
|
if _existing is not None and not isinstance(_existing, QApplication):
|
|
raise RuntimeError("widget tests need a QApplication")
|
|
app = _existing or QApplication([])
|
|
|
|
from xray_fluent.application.controller import AppController
|
|
from xray_fluent.diagnostics.runtime_errors import RuntimeErrorJournal, core_failure
|
|
|
|
|
|
def _spin(ms: int) -> None:
|
|
loop = QEventLoop()
|
|
QTimer.singleShot(ms, loop.quit)
|
|
loop.exec()
|
|
|
|
|
|
class _Host(QObject):
|
|
runtime_errors_changed = pyqtSignal(object)
|
|
_schedule_runtime_errors_emit = AppController._schedule_runtime_errors_emit
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.runtime_errors = RuntimeErrorJournal()
|
|
|
|
|
|
class RuntimeErrorsThrottleTests(unittest.TestCase):
|
|
def test_burst_of_errors_emits_one_snapshot(self) -> None:
|
|
host = _Host()
|
|
snapshots = []
|
|
host.runtime_errors_changed.connect(snapshots.append)
|
|
for i in range(200):
|
|
host.runtime_errors.record(core_failure("xray", "runtime", f"[Warning] failed to dial port {i}"))
|
|
host._schedule_runtime_errors_emit()
|
|
self.assertEqual(snapshots, []) # не на каждую строку
|
|
_spin(1200)
|
|
self.assertEqual(len(snapshots), 1)
|
|
self.assertEqual(snapshots[0][0].occurrences, 200)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|