zapretgui/tests/test_backend_page_data_warmup.py

from __future__ import annotations

import unittest
from types import SimpleNamespace
from unittest.mock import Mock, patch


class BackendPageDataWarmupTests(unittest.TestCase):
    def test_installer_warms_backend_caches_without_page_host(self) -> None:
        from main import post_startup_backend_warmup
        from main.post_startup_backend_warmup import install_backend_page_data_warmup

        class Signal:
            def __init__(self) -> None:
                self._callback = None

            def connect(self, callback) -> None:
                self._callback = callback

            def emit(self, value: str = "") -> None:
                self._callback(value)

        signal = Signal()
        startup_host = SimpleNamespace(
            startup_interactive_ready=signal,
            startup_state=SimpleNamespace(interactive_logged=False),
            is_alive=Mock(return_value=True),
        )
        premium_feature = SimpleNamespace(warm_page_data_cache=Mock())
        logs_feature = SimpleNamespace(warm_page_data_cache=Mock())
        metric = Mock()
        delays: list[int] = []
        warmup_flags: list[bool] = []
        queued_tasks: list[tuple[str, str]] = []

        with (
            patch.object(
                post_startup_backend_warmup,
                "schedule_after",
                side_effect=lambda delay_ms, callback: delays.append(delay_ms) or callback(),
            ),
            patch.object(
                post_startup_backend_warmup,
                "enqueue_subsystem_task",
                side_effect=lambda queue, name, target, warmup=False: (
                    warmup_flags.append(warmup) or queued_tasks.append((queue, name)) or target()
                ),
            ),
            patch.object(post_startup_backend_warmup.appearance_settings, "warm_page_initial_state_cache") as warm_appearance,
        ):
            install_backend_page_data_warmup(
                startup_host,
                premium_feature=premium_feature,
                logs_feature=logs_feature,
                log_startup_metric=metric,
            )
            signal.emit("interactive")

        # Подготовка данных страницы заранее — прогрев: она пропускает вперёд
        # всё, что нужно программе для работы.
        self.assertTrue(warmup_flags and all(warmup_flags))
        self.assertEqual(delays, [8000, 18000])
        self.assertEqual(
            queued_tasks,
            [
                ("pages", "BackendPageDataWarmup-Appearance"),
                ("pages", "BackendPageDataWarmup-Logs"),
                ("premium", "BackendPageDataWarmup-Premium"),
            ],
        )
        premium_feature.warm_page_data_cache.assert_called_once_with()
        logs_feature.warm_page_data_cache.assert_called_once_with()
        warm_appearance.assert_called_once_with()
        self.assertFalse(hasattr(startup_host, "warm_page"))
        metric.assert_any_call(
            "StartupBackendPageDataWarmupQueued",
            "8000ms after interactive; premium 18000ms after interactive",
        )
        metric.assert_any_call("StartupBackendPageDataWarmupStarted", "appearance, logs")
        metric.assert_any_call("StartupBackendPageDataWarmupStarted", "premium")

    def test_hosts_page_data_is_warmed_in_background_without_page_host(self) -> None:
        from main import post_startup_hosts_warmup
        from main.post_startup_hosts_warmup import install_hosts_page_warmup

        class Signal:
            def __init__(self) -> None:
                self._callback = None

            def connect(self, callback) -> None:
                self._callback = callback

            def emit(self, value: str = "") -> None:
                self._callback(value)

        signal = Signal()
        startup_host = SimpleNamespace(
            startup_interactive_ready=signal,
            startup_state=SimpleNamespace(interactive_logged=False),
            is_alive=Mock(return_value=True),
        )
        hosts_feature = SimpleNamespace(warm_page_data_cache=Mock(return_value=True))
        metric = Mock()
        delays: list[int] = []
        warmup_flags: list[bool] = []
        queued_tasks: list[tuple[str, str]] = []

        with (
            patch.object(
                post_startup_hosts_warmup,
                "schedule_after",
                side_effect=lambda delay_ms, callback: delays.append(delay_ms) or callback(),
            ),
            patch.object(
                post_startup_hosts_warmup,
                "enqueue_subsystem_task",
                side_effect=lambda queue, name, target, warmup=False: (
                    warmup_flags.append(warmup) or queued_tasks.append((queue, name)) or target()
                ),
            ),
        ):
            install_hosts_page_warmup(
                startup_host,
                hosts_feature=hosts_feature,
                log_startup_metric=metric,
            )
            signal.emit("interactive")

        # Подготовка данных страницы заранее — прогрев: она пропускает вперёд
        # всё, что нужно программе для работы.
        self.assertTrue(warmup_flags and all(warmup_flags))
        self.assertEqual(delays, [0])
        self.assertEqual(queued_tasks, [("hosts", "HostsPageDataWarmup")])
        hosts_feature.warm_page_data_cache.assert_called_once_with()
        self.assertFalse(hasattr(startup_host, "ensure_page"))
        metric.assert_any_call("StartupHostsPageWarmupQueued", "0ms after interactive")
        metric.assert_any_call("StartupHostsPageWarmupStarted", "backend_cache")
        metric.assert_any_call("StartupHostsPageWarmupFinished", "backend_cache")

    def test_hosts_feature_warms_page_snapshot_for_first_open(self) -> None:
        from app.feature_facades.hosts import build_hosts_feature

        snapshot = object()
        load_snapshot = Mock(return_value=snapshot)
        metric_stages: list[str] = []

        with (
            patch("hosts.public.load_page_snapshot", load_snapshot),
            patch(
                "app.feature_facades.hosts.log_ui_timing_since",
                side_effect=lambda _scope, _name, stage, *_args, **_kwargs: metric_stages.append(stage),
            ),
        ):
            feature = build_hosts_feature()
            self.assertTrue(feature.warm_page_data_cache())

        # Прогрев только читает: снимок собирается один раз, файлы не меняются.
        load_snapshot.assert_called_once_with()
        self.assertEqual(metric_stages, ["hosts_warmup.snapshot"])
        self.assertFalse(hasattr(feature, "consume_warmed_services_catalog_plan"))

    def test_telegram_proxy_page_is_prepared_after_interactive_without_opening_it(self) -> None:
        from app.page_names import PageName
        from main import post_startup_telegram_proxy_warmup
        from main.post_startup_telegram_proxy_warmup import install_telegram_proxy_page_warmup

        class Signal:
            def __init__(self) -> None:
                self._callback = None

            def connect(self, callback) -> None:
                self._callback = callback

            def emit(self, value: str = "") -> None:
                self._callback(value)

        signal = Signal()
        startup_host = SimpleNamespace(
            startup_interactive_ready=signal,
            startup_state=SimpleNamespace(interactive_logged=False),
            is_alive=Mock(return_value=True),
            ensure_page=Mock(return_value=object()),
            show_page=Mock(),
        )
        metric = Mock()
        queued: list[tuple[str, object, int]] = []
        speculative_flags: list[bool] = []
        idle_tasks = SimpleNamespace(
            add=lambda name, callback, *, delay_ms=0, needs_shown_window=True, speculative=False: (
                speculative_flags.append(speculative) or queued.append((name, callback, delay_ms))
            )
        )

        install_telegram_proxy_page_warmup(
            startup_host,
            log_startup_metric=metric,
            idle_tasks=idle_tasks,
        )
        signal.emit("interactive")

        # Страница строится в GUI-потоке, поэтому её время выбирает очередь
        # пауз пользователя, а не таймер: 3000 мс — это «не раньше чем».
        # Сборка идёт про запас: только когда заминку окна некому заметить.
        self.assertEqual(speculative_flags, [True])
        self.assertEqual([(name, delay) for name, _cb, delay in queued], [("TelegramProxyPageWarmup", 3000)])
        startup_host.ensure_page.assert_not_called()
        queued[0][1]()
        startup_host.ensure_page.assert_called_once_with(PageName.TELEGRAM_PROXY)
        startup_host.show_page.assert_not_called()
        metric.assert_any_call("StartupTelegramProxyPageWarmupQueued", "3000ms after interactive")
        metric.assert_any_call("StartupTelegramProxyPageWarmupFinished", "ui_page")

    def test_premium_page_uses_warmed_device_snapshot_before_worker(self) -> None:
        from app.feature_facades.premium import PremiumPageData
        from donater.ui.page import PremiumPage

        snapshot = {"device_token": "token", "pair_code": None, "last_check": 123, "device_id": "dev"}
        page = SimpleNamespace(
            _premium_action_runtime=SimpleNamespace(is_running=Mock(return_value=False)),
            _premium=SimpleNamespace(consume_warmed_page_data=Mock(return_value=PremiumPageData(device_info=snapshot))),
            _start_worker_thread=Mock(),
            _on_premium_init_complete=Mock(),
        )

        PremiumPage._start_premium_init_worker(page)

        page._premium.consume_warmed_page_data.assert_called_once_with()
        page._on_premium_init_complete.assert_called_once_with(snapshot)
        page._start_worker_thread.assert_not_called()

    def test_logs_page_applies_warmed_overview_before_runtime_start(self) -> None:
        from log.commands import LogsPageDataState, LogsListState, LogsStatsState
        from log.ui.page import LogsPage

        logs_state = LogsListState(
            entries=[],
            current_log_file="current.log",
            cleanup_deleted=0,
            cleanup_errors=[],
            cleanup_total=0,
        )
        stats_state = LogsStatsState(
            app_logs=1,
            debug_logs=0,
            total_size_mb=0.1,
            max_logs=50,
            max_debug_logs=10,
        )
        cached_state = LogsPageDataState(logs_state=logs_state, stats_state=stats_state)
        calls: list[str] = []
        logs_feature = SimpleNamespace(
            consume_warmed_page_data=Mock(return_value=cached_state),
            run_runtime_init=Mock(return_value=(True, True)),
        )
        page = LogsPage.__new__(LogsPage)
        page._cleanup_in_progress = False
        page.isVisible = Mock(return_value=True)
        page._runtime_initialized = False
        page._runtime_started = False
        page._logs = logs_feature
        page._apply_logs_list_state = Mock(side_effect=lambda *_args, **_kwargs: calls.append("cached_logs"))
        page._apply_logs_stats_state = Mock(side_effect=lambda *_args, **_kwargs: calls.append("cached_stats"))
        page._update_stats = Mock()
        page._start_log_source = Mock()
        page._log_ui_timing = Mock()

        LogsPage._run_runtime_init_once(page)

        self.assertEqual(calls, ["cached_logs", "cached_stats"])
        logs_feature.consume_warmed_page_data.assert_called_once_with()
        logs_feature.run_runtime_init.assert_called_once()
        self.assertTrue(page._runtime_initialized)
        self.assertTrue(page._runtime_started)

    def test_logs_page_applies_warmed_overview_immediately_on_activation(self) -> None:
        from log.commands import LogsPageDataState, LogsListState, LogsStatsState
        from log.ui.page import LogsPage

        cached_state = LogsPageDataState(
            logs_state=LogsListState(
                entries=[],
                current_log_file="current.log",
                cleanup_deleted=0,
                cleanup_errors=[],
                cleanup_total=0,
            ),
            stats_state=LogsStatsState(
                app_logs=1,
                debug_logs=0,
                total_size_mb=0.1,
                max_logs=50,
                max_debug_logs=10,
            ),
        )
        calls: list[str] = []
        page = LogsPage.__new__(LogsPage)
        page._logs = SimpleNamespace(consume_warmed_page_data=Mock(return_value=cached_state))
        page._apply_logs_list_state = Mock(side_effect=lambda *_args, **_kwargs: calls.append("cached_logs"))
        page._apply_logs_stats_state = Mock(side_effect=lambda *_args, **_kwargs: calls.append("cached_stats"))
        page._schedule_runtime_init = Mock(side_effect=lambda: calls.append("runtime_scheduled"))
        page._schedule_logs_secondary_panels = Mock(side_effect=lambda: calls.append("secondary_scheduled"))
        page._first_log_content_timing_done = False

        LogsPage.on_page_activated(page)

        self.assertEqual(
            calls,
            ["runtime_scheduled", "secondary_scheduled", "cached_logs", "cached_stats"],
        )

    def test_logs_page_still_schedules_runtime_when_warmed_overview_is_invalid(self) -> None:
        from log.ui.page import LogsPage

        page = LogsPage.__new__(LogsPage)
        page._logs = SimpleNamespace(
            consume_warmed_page_data=Mock(side_effect=RuntimeError("bad cache")),
        )
        page._schedule_runtime_init = Mock()
        page._schedule_logs_secondary_panels = Mock()
        page._first_log_content_timing_done = False

        LogsPage.on_page_activated(page)

        page._schedule_runtime_init.assert_called_once_with()
        page._schedule_logs_secondary_panels.assert_called_once_with()

    def test_appearance_initial_state_cache_is_consumed_once(self) -> None:
        from settings import appearance

        state = appearance.AppearancePageInitialStatePlan(
            display_mode="dark",
            ui_language="ru",
            background_preset="standard",
            rkn_background=None,
            window_opacity=100,
            accent_color=None,
            follow_windows_accent=False,
            tinted_background=False,
            tinted_intensity=12,
            animations_enabled=True,
            smooth_scroll_enabled=True,
            editor_smooth_scroll_enabled=True,
            sidebar_icon_style="standard",
            garland_enabled=False,
            snowflakes_enabled=False,
        )

        appearance.clear_warmed_page_initial_state_cache()
        appearance.store_warmed_page_initial_state(state)

        self.assertIs(appearance.consume_warmed_page_initial_state(), state)
        self.assertIsNone(appearance.consume_warmed_page_initial_state())


if __name__ == "__main__":
    unittest.main()