loop-uh
1586042ef1
Some checks failed
Windows project source guards / test (push) Has been cancelled
- release_windows_gate.ps1: тесты и сборка под общим замком тяжёлой работы машины (%ProgramData%\ZapretBuild\heavy.lock, тот же берёт сборка Zapret 2 GUI); ожидание до 40 минут, ssh-срок 7200 с. - Сторож зависаний: в покое — три попытки (ложная тревога видна в каждой, пауза нагрузки — нет); с блокировкой — ровно одна запись о самой блокировке, прочие короче неё. - Раскладка таблицы серверов: ждём итоговую ширину до 5 с, а не 50 мс. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
80 lines
2.8 KiB
Python
80 lines
2.8 KiB
Python
"""GUI-stall watchdog: ловит блокировку GUI-потока и снимает её стек.
|
||
|
||
Keep the ``test_app_*`` prefix: widget test modules must sort before
|
||
``tests/test_engine_process_stop.py`` which creates a bare QCoreApplication
|
||
at import time (see tests/test_app_nodes_page_view.py).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import time
|
||
import unittest
|
||
|
||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||
|
||
from PyQt6.QtCore import QEventLoop, QTimer
|
||
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.diagnostics.gui_stall_watchdog import GuiStallWatchdog
|
||
|
||
|
||
def _spin(ms: int) -> None:
|
||
loop = QEventLoop()
|
||
QTimer.singleShot(ms, loop.quit)
|
||
loop.exec()
|
||
|
||
|
||
def _blocking_culprit(seconds: float) -> None:
|
||
time.sleep(seconds)
|
||
|
||
|
||
class GuiStallWatchdogTests(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
# Хвосты прошлых тестов (deleteLater) не должны попасть в замер.
|
||
app.sendPostedEvents(None, 0)
|
||
_spin(50)
|
||
self.watchdog = GuiStallWatchdog(threshold_ms=50)
|
||
self.watchdog.start()
|
||
_spin(40)
|
||
self.watchdog.reset()
|
||
|
||
def tearDown(self) -> None:
|
||
self.watchdog.stop()
|
||
|
||
def test_idle_loop_has_no_stalls(self) -> None:
|
||
# Ложная тревога сторожа видна в каждой попытке; разовая пауза потока,
|
||
# когда машину грузит соседняя сборка, — нет.
|
||
for _attempt in range(3):
|
||
self.watchdog.reset()
|
||
_spin(300)
|
||
if not self.watchdog.records:
|
||
break
|
||
self.assertEqual(self.watchdog.records, [])
|
||
|
||
def test_blocking_call_is_measured_with_stack(self) -> None:
|
||
QTimer.singleShot(0, lambda: _blocking_culprit(0.2))
|
||
deadline = time.monotonic() + 5.0
|
||
culprits = []
|
||
while time.monotonic() < deadline:
|
||
_spin(50)
|
||
culprits = [r for r in self.watchdog.records if "_blocking_culprit" in r.stack]
|
||
if culprits:
|
||
break
|
||
# Ровно одна запись о самой блокировке; паузы нагрузки — короче неё.
|
||
self.assertEqual(len(culprits), 1)
|
||
record = culprits[0]
|
||
self.assertGreaterEqual(record.duration_ms, 180)
|
||
for other in self.watchdog.records:
|
||
if other is not record:
|
||
self.assertLess(other.duration_ms, record.duration_ms)
|
||
self.assertGreaterEqual(self.watchdog.max_stall_ms, 180)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|