37678bf66b
Наблюдатель рывков показал, что интерфейс дёргается не от конкуренции за GIL, а от синхронной работы прямо в момент действия пользователя. Два адреса из лога закрываются здесь. Первый клик по странице строил её целиком и добавлял в стек в GUI-потоке: `show_page → ensure_page → stacked_widget.addWidget` давало рывки на 64–70 мс для «Мои пресеты», «Оформление» и Premium. Теперь эти страницы готовятся в паузе после старта, как уже делались профиль, пресеты и Telegram Proxy. Греются по одной с разносом по времени: одновременная сборка снова заняла бы GUI-поток одним куском. Первое обращение к Telegram Proxy тянуло asyncio вместе с asyncio.windows_events — рывок на 64 мс посреди работы. Прогрев импортов теперь общий и переживает неудачу любого модуля, поэтому туда добавлен asyncio рядом с qtawesome. Тесты: test_secondary_page_warmup — состав плана, разнос по времени, единственный запуск каждой страницы, устойчивость к падению одной из них и к умершему хосту; test_import_warmup — фоновый демон-поток, продолжение после неудачного импорта, наличие asyncio в списке. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
101 lines
3.7 KiB
Python
101 lines
3.7 KiB
Python
from __future__ import annotations
|
||
|
||
import unittest
|
||
from types import SimpleNamespace
|
||
from unittest.mock import Mock, patch
|
||
|
||
from app.page_names import PageName
|
||
from main.post_startup_secondary_page_warmup import (
|
||
SECONDARY_PAGE_WARMUP_PLAN,
|
||
install_secondary_page_warmup,
|
||
)
|
||
|
||
|
||
def _startup_host(ensure_page=None) -> SimpleNamespace:
|
||
return SimpleNamespace(
|
||
startup_interactive_ready=SimpleNamespace(connect=Mock()),
|
||
startup_state=SimpleNamespace(interactive_logged=True),
|
||
ensure_page=ensure_page or Mock(return_value=object()),
|
||
is_alive=Mock(return_value=True),
|
||
)
|
||
|
||
|
||
def _immediate_startup_gate():
|
||
"""Гейт старта планирует запуск через QTimer, которого в тестах нет."""
|
||
return patch(
|
||
"main.post_startup_gate.QTimer.singleShot",
|
||
side_effect=lambda _delay, callback: callback(),
|
||
)
|
||
|
||
|
||
class SecondaryPageWarmupTests(unittest.TestCase):
|
||
"""Страницы греются в паузе, чтобы первый клик не строил их в GUI-потоке."""
|
||
|
||
def test_plan_covers_pages_that_stuttered_on_first_click(self) -> None:
|
||
pages = {page for page, _delay in SECONDARY_PAGE_WARMUP_PLAN}
|
||
|
||
self.assertIn(PageName.ZAPRET2_USER_PRESETS, pages)
|
||
self.assertIn(PageName.APPEARANCE, pages)
|
||
self.assertIn(PageName.PREMIUM, pages)
|
||
|
||
def test_pages_are_spread_out_in_time(self) -> None:
|
||
delays = [delay for _page, delay in SECONDARY_PAGE_WARMUP_PLAN]
|
||
|
||
self.assertEqual(delays, sorted(delays))
|
||
self.assertEqual(len(set(delays)), len(delays))
|
||
# Не пересекаемся с прогревом Telegram Proxy (3000 мс).
|
||
self.assertTrue(all(delay > 3_000 for delay in delays))
|
||
|
||
def test_each_page_is_scheduled_once(self) -> None:
|
||
scheduled: list[tuple[int, object]] = []
|
||
host = _startup_host()
|
||
|
||
with patch(
|
||
"main.post_startup_secondary_page_warmup.schedule_after",
|
||
side_effect=lambda delay, callback: scheduled.append((delay, callback)),
|
||
), _immediate_startup_gate():
|
||
install_secondary_page_warmup(host, log_startup_metric=Mock())
|
||
|
||
self.assertEqual(len(scheduled), len(SECONDARY_PAGE_WARMUP_PLAN))
|
||
|
||
for _delay, callback in scheduled:
|
||
callback()
|
||
|
||
warmed = [call.args[0] for call in host.ensure_page.call_args_list]
|
||
self.assertEqual(warmed, [page for page, _delay in SECONDARY_PAGE_WARMUP_PLAN])
|
||
|
||
def test_failing_page_does_not_break_the_rest(self) -> None:
|
||
scheduled: list[object] = []
|
||
host = _startup_host(ensure_page=Mock(side_effect=RuntimeError("сломалась")))
|
||
|
||
with patch(
|
||
"main.post_startup_secondary_page_warmup.schedule_after",
|
||
side_effect=lambda delay, callback: scheduled.append(callback),
|
||
), _immediate_startup_gate():
|
||
install_secondary_page_warmup(host, log_startup_metric=Mock())
|
||
|
||
for callback in scheduled:
|
||
callback()
|
||
|
||
self.assertEqual(host.ensure_page.call_count, len(SECONDARY_PAGE_WARMUP_PLAN))
|
||
|
||
def test_dead_host_stops_warmup(self) -> None:
|
||
scheduled: list[object] = []
|
||
host = _startup_host()
|
||
|
||
with patch(
|
||
"main.post_startup_secondary_page_warmup.schedule_after",
|
||
side_effect=lambda delay, callback: scheduled.append(callback),
|
||
), patch(
|
||
"main.post_startup_secondary_page_warmup.is_startup_host_alive",
|
||
return_value=False,
|
||
), _immediate_startup_gate():
|
||
install_secondary_page_warmup(host, log_startup_metric=Mock())
|
||
for callback in scheduled:
|
||
callback()
|
||
|
||
host.ensure_page.assert_not_called()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|