loop-uh
6ac583e90c
Some checks failed
Windows project source guards / test (push) Failing after 39s
Кнопка «Запустить/Остановить» дублировала кнопку питания и убрана. Сфера стала полноценной кнопкой: фокус по Tab, Enter и пробел, рамка фокуса с клавиатуры, подсказка и имя для диктора с действием. Карточка подключения ниже: состояние текстом слева, сцена справа, на узком окне сцена над текстом. Пояснение, что идёт через туннель, ушло из шапки в подпись переключателя «Системный прокси Windows».
703 lines
31 KiB
Python
703 lines
31 KiB
Python
"""Главная панель: бюджет CPU анимаций, плитки режима, приложения, боковое меню.
|
||
|
||
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).
|
||
|
||
Общая страница создаётся один раз на модуль и не уничтожается (см.
|
||
предупреждение про крэш Windows в tests/test_app_scrollable_page.py);
|
||
единственное исключение — LifecycleTest, который проверяет именно
|
||
синхронное разрушение одной страницы.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from types import SimpleNamespace
|
||
import unittest
|
||
from unittest.mock import Mock
|
||
|
||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||
|
||
from PyQt6.QtWidgets import QApplication
|
||
|
||
_existing = QApplication.instance()
|
||
if _existing is not None and not isinstance(_existing, QApplication):
|
||
raise RuntimeError(
|
||
"A bare QCoreApplication was created before test_app_dashboard_motion "
|
||
"was imported; widget tests need a QApplication."
|
||
)
|
||
app = _existing or QApplication([])
|
||
|
||
from qfluentwidgets import NavigationDisplayMode
|
||
|
||
from xray_fluent.profiles.models import AppSettings, Node
|
||
from xray_fluent.ui.connection_orb import CONNECTED, CONNECTING, ERROR, IDLE
|
||
from xray_fluent.ui.dashboard_page import DashboardPage
|
||
from xray_fluent.ui.dashboard_widgets import latency_bars
|
||
from xray_fluent.ui.main_window import _WINDOW_OWNED_SETTINGS, MainWindow
|
||
|
||
_page: DashboardPage | None = None
|
||
_nav_window_ref = None
|
||
|
||
|
||
def _shared_page() -> DashboardPage:
|
||
global _page
|
||
if _page is None:
|
||
_page = DashboardPage()
|
||
_page.resize(1200, 900)
|
||
return _page
|
||
|
||
|
||
def _reset(page: DashboardPage) -> None:
|
||
page._tun_intent.clear()
|
||
page._proxy_intent.clear()
|
||
page.set_active_tun_mode(None)
|
||
page.set_transition_busy(False)
|
||
page.set_runtime_status("idle", "")
|
||
page.set_connection(False)
|
||
page.set_settings_snapshot(AppSettings(tun_mode=True))
|
||
page._do_refresh_dashboard()
|
||
|
||
|
||
class SceneCpuBudgetTest(unittest.TestCase):
|
||
"""Таймер кадров только в подключении/подключено и только пока сцена видна."""
|
||
|
||
def setUp(self) -> None:
|
||
self.page = _shared_page()
|
||
_reset(self.page)
|
||
self.page.show()
|
||
QApplication.processEvents()
|
||
self.scene = self.page.connection_scene
|
||
|
||
def tearDown(self) -> None:
|
||
self.page.hide()
|
||
_reset(self.page)
|
||
QApplication.processEvents()
|
||
|
||
def test_idle_has_no_frame_timers(self) -> None:
|
||
self.assertEqual(self.scene.state(), IDLE)
|
||
self.assertFalse(self.scene.is_animating())
|
||
self.assertFalse(self.page.connection_orb.is_animating())
|
||
|
||
def test_connected_animates_and_stops_on_hide(self) -> None:
|
||
self.page.set_connection(True)
|
||
self.page._do_refresh_dashboard()
|
||
self.assertEqual(self.scene.state(), CONNECTED)
|
||
self.assertTrue(self.scene.is_animating())
|
||
self.page.hide()
|
||
QApplication.processEvents()
|
||
self.assertFalse(self.scene.is_animating())
|
||
self.assertFalse(self.page.connection_orb.is_animating())
|
||
|
||
def test_connected_without_traffic_rests_and_wakes_on_traffic(self) -> None:
|
||
self.page.set_connection(True)
|
||
self.page._do_refresh_dashboard()
|
||
self.assertTrue(self.scene.is_animating())
|
||
self.scene._burst_started = None
|
||
self.scene.set_traffic(0.0, 0.0)
|
||
self.scene._tick() # первый тихий кадр запоминает начало покоя
|
||
self.scene._quiet_since -= 10.0
|
||
self.scene._tick()
|
||
self.assertFalse(self.scene.is_animating())
|
||
self.assertTrue(self.scene.is_resting())
|
||
self.page.set_live_metrics(2e6, 1e5, 40)
|
||
self.assertTrue(self.scene.is_animating())
|
||
|
||
def test_error_is_static(self) -> None:
|
||
self.page.set_runtime_status("error", "Не удалось запустить sing-box")
|
||
self.page._do_refresh_dashboard()
|
||
self.assertEqual(self.scene.state(), ERROR)
|
||
self.assertFalse(self.scene.is_animating())
|
||
self.assertEqual(self.page.connection_state_label.text(), "Ошибка подключения")
|
||
self.assertFalse(self.page.logs_btn.isHidden())
|
||
|
||
def test_starting_runs_probe_animation(self) -> None:
|
||
self.page.set_runtime_status("starting", "Запуск")
|
||
self.page._do_refresh_dashboard()
|
||
self.assertEqual(self.scene.state(), CONNECTING)
|
||
self.assertTrue(self.scene.is_animating())
|
||
|
||
def test_frame_dirty_region_is_the_tunnel_band_through_the_orb(self) -> None:
|
||
# Поток огоньков идёт через центр сферы, поэтому центр перерисовывается;
|
||
# экономия CPU — кадр трогает только полосу туннеля, а не всю сцену.
|
||
self.page.set_connection(True)
|
||
self.page._do_refresh_dashboard()
|
||
self.scene._burst_started = None
|
||
self.scene.grab() # строит кэш и выборку пути
|
||
region = self.scene._dirty_region()
|
||
self.assertTrue(region.contains(self.page.connection_orb.geometry().center()))
|
||
band = region.boundingRect()
|
||
self.assertLessEqual(band.height(), 96)
|
||
self.assertLess(band.height(), self.scene.height() + 1)
|
||
|
||
|
||
class DashboardHonestStatusTest(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
self.page = _shared_page()
|
||
_reset(self.page)
|
||
|
||
def tearDown(self) -> None:
|
||
_reset(self.page)
|
||
|
||
def test_only_tun_is_called_protected(self) -> None:
|
||
cases = (
|
||
(AppSettings(tun_mode=True), "Защищено"),
|
||
(AppSettings(tun_mode=False, enable_system_proxy=True), "Подключено"),
|
||
(AppSettings(tun_mode=False, enable_system_proxy=False), "Прокси запущен"),
|
||
)
|
||
for settings, title in cases:
|
||
with self.subTest(title=title):
|
||
self.page.set_settings_snapshot(settings)
|
||
self.page.set_connection(True)
|
||
self.page._do_refresh_dashboard()
|
||
self.assertEqual(self.page.connection_state_label.text(), title)
|
||
self.page.set_connection(False)
|
||
|
||
def test_orb_is_the_only_connect_control_and_names_the_action(self) -> None:
|
||
self.assertFalse(hasattr(self.page, "toggle_btn"))
|
||
self.page.set_settings_snapshot(AppSettings(tun_mode=False, enable_system_proxy=False))
|
||
self.page.set_connection(True)
|
||
self.page._do_refresh_dashboard()
|
||
self.assertEqual(self.page.connection_orb.toolTip(), "Остановить прокси")
|
||
# Что идёт через туннель, говорит подпись переключателя прокси, не шапка.
|
||
self.assertTrue(self.page.connection_status_label.isHidden())
|
||
self.assertIn("ручной настройкой", self.page.proxy_hint_label.text())
|
||
self.page.set_settings_snapshot(AppSettings(tun_mode=False, enable_system_proxy=True))
|
||
self.page._proxy_intent.request(True)
|
||
self.page._do_refresh_dashboard()
|
||
self.assertIn("подхватывают прокси сами", self.page.proxy_hint_label.text())
|
||
|
||
def test_hero_text_left_of_scene_and_stacked_when_narrow(self) -> None:
|
||
from PyQt6.QtCore import Qt
|
||
from PyQt6.QtWidgets import QBoxLayout
|
||
|
||
page = self.page
|
||
page._apply_hero_direction(stacked=False)
|
||
self.assertEqual(page._hero_layout.direction(), QBoxLayout.Direction.LeftToRight)
|
||
self.assertEqual(page._hero_layout.itemAt(0).widget(), page._hero_text)
|
||
self.assertEqual(page._hero_layout.itemAt(1).widget(), page.connection_scene)
|
||
page._apply_hero_direction(stacked=True)
|
||
self.assertEqual(page._hero_layout.direction(), QBoxLayout.Direction.BottomToTop)
|
||
self.assertTrue(page.connection_state_label.alignment() & Qt.AlignmentFlag.AlignHCenter)
|
||
page._apply_hero_direction(stacked=False)
|
||
self.assertTrue(page.connection_state_label.alignment() & Qt.AlignmentFlag.AlignLeft)
|
||
|
||
def test_idle_invites_to_connect(self) -> None:
|
||
self.page._do_refresh_dashboard()
|
||
self.assertEqual(self.page.connection_state_label.text(), "Не подключено")
|
||
self.assertIn("кнопку питания", self.page.connection_status_label.text())
|
||
|
||
|
||
class ModeTilesTest(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
self.page = _shared_page()
|
||
_reset(self.page)
|
||
self.emitted: list[bool] = []
|
||
self.page.tun_toggled.connect(self.emitted.append)
|
||
|
||
def tearDown(self) -> None:
|
||
self.page.tun_toggled.disconnect(self.emitted.append)
|
||
_reset(self.page)
|
||
|
||
def test_tiles_follow_settings(self) -> None:
|
||
self.assertTrue(self.page.vpn_tile.isChecked())
|
||
self.assertFalse(self.page.proxy_tile.isChecked())
|
||
self.assertTrue(self.page.proxy_options.isHidden())
|
||
self.page.set_settings_snapshot(AppSettings(tun_mode=False))
|
||
self.page._do_refresh_dashboard()
|
||
self.assertFalse(self.page.vpn_tile.isChecked())
|
||
self.assertTrue(self.page.proxy_tile.isChecked())
|
||
self.assertFalse(self.page.proxy_options.isHidden())
|
||
|
||
def test_clicking_other_mode_emits_once_and_current_mode_is_noop(self) -> None:
|
||
self.page.vpn_tile.clicked.emit()
|
||
self.assertEqual(self.emitted, [])
|
||
self.page.proxy_tile.clicked.emit()
|
||
self.assertEqual(self.emitted, [False])
|
||
|
||
def test_tiles_disabled_while_busy(self) -> None:
|
||
self.page.set_transition_busy(True)
|
||
self.assertFalse(self.page.vpn_tile.isEnabled())
|
||
self.assertFalse(self.page.proxy_tile.isEnabled())
|
||
self.assertFalse(self.page.connection_orb.isEnabled())
|
||
|
||
|
||
class ProcessesAndServerTest(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
self.page = _shared_page()
|
||
_reset(self.page)
|
||
|
||
def tearDown(self) -> None:
|
||
_reset(self.page)
|
||
|
||
@staticmethod
|
||
def _stat(exe: str, up: int, down: int, proxy: int):
|
||
return SimpleNamespace(
|
||
exe=exe, upload=up, download=down, proxy_bytes=proxy, direct_bytes=up + down - proxy,
|
||
down_speed=0.0, up_speed=0.0, connections=1, total_connections=1, top_host="example.org",
|
||
)
|
||
|
||
def test_top_apps_ranked_and_session_total_is_cumulative_proxy_bytes(self) -> None:
|
||
self.page.set_connection(True)
|
||
stats = [
|
||
self._stat("small.exe", 10, 10, 20),
|
||
self._stat("chrome.exe", 1000, 3000, 4000),
|
||
self._stat("game.exe", 500, 500, 0),
|
||
]
|
||
self.page.set_process_stats(stats)
|
||
self.page._do_refresh_dashboard()
|
||
names = [name for name, _value, _share in self.page.process_bars.items()]
|
||
self.assertEqual(names, ["chrome.exe", "game.exe", "small.exe"])
|
||
self.assertEqual(self.page.process_bars.items()[0][2], 1.0)
|
||
self.assertEqual(self.page.traffic_session_label.text(), self.page._format_bytes(4020))
|
||
|
||
def test_unavailable_stats_are_shown_honestly(self) -> None:
|
||
self.page.set_connection(True)
|
||
self.page.set_process_stats(None)
|
||
self.page._do_refresh_dashboard()
|
||
self.assertEqual(self.page.traffic_session_label.text(), "--")
|
||
self.assertIn("недоступна", self.page.process_summary_label.text())
|
||
self.assertEqual(self.page.process_bars.items(), [])
|
||
|
||
def test_next_server_needs_more_than_one_node(self) -> None:
|
||
nodes = [Node(name="A", scheme="vless", server="a.example", port=1)]
|
||
self.page.set_nodes(nodes, nodes[0].id)
|
||
self.page._do_refresh_dashboard()
|
||
self.assertFalse(self.page.next_server_btn.isEnabled())
|
||
nodes.append(Node(name="B", scheme="vless", server="b.example", port=1))
|
||
self.page.set_nodes(nodes, nodes[0].id)
|
||
self.page._do_refresh_dashboard()
|
||
self.assertTrue(self.page.next_server_btn.isEnabled())
|
||
self.page.set_nodes([], None)
|
||
|
||
def test_signal_bars_scale(self) -> None:
|
||
self.assertEqual([latency_bars(v) for v in (None, 40, 120, 250, 900)], [0, 4, 3, 2, 1])
|
||
|
||
|
||
def _nav_window(mode, expanding: bool, *, width: int = 1400, saved: bool = False):
|
||
panel = SimpleNamespace(
|
||
displayMode=mode,
|
||
expandAni=Mock(property=Mock(return_value=expanding)),
|
||
minimumExpandWidth=1100,
|
||
expand=Mock(),
|
||
)
|
||
settings = AppSettings(nav_expanded=saved)
|
||
controller = Mock()
|
||
controller.state.settings = settings
|
||
window = SimpleNamespace(
|
||
navigationInterface=SimpleNamespace(panel=panel),
|
||
controller=controller,
|
||
_geometry_persistence_ready=True,
|
||
_nav_expanded_pref=saved,
|
||
width=lambda: width,
|
||
)
|
||
window._sync_nav_to_preference = lambda: MainWindow._sync_nav_to_preference(window)
|
||
return window, panel, settings, controller
|
||
|
||
|
||
class NavPreferenceTest(unittest.TestCase):
|
||
def test_click_expand_is_saved(self) -> None:
|
||
window, _panel, settings, controller = _nav_window(NavigationDisplayMode.EXPAND, True)
|
||
MainWindow._on_nav_menu_clicked(window)
|
||
self.assertTrue(settings.nav_expanded)
|
||
controller.schedule_save.assert_called_once()
|
||
|
||
def test_click_collapse_is_saved(self) -> None:
|
||
window, _panel, settings, _controller = _nav_window(NavigationDisplayMode.EXPAND, False, saved=True)
|
||
MainWindow._on_nav_menu_clicked(window)
|
||
self.assertFalse(settings.nav_expanded)
|
||
self.assertFalse(window._nav_expanded_pref)
|
||
|
||
def test_overlay_menu_is_not_a_preference(self) -> None:
|
||
for expanding in (True, False):
|
||
with self.subTest(expanding=expanding):
|
||
window, _panel, settings, controller = _nav_window(NavigationDisplayMode.MENU, expanding, saved=True)
|
||
MainWindow._on_nav_menu_clicked(window)
|
||
self.assertTrue(settings.nav_expanded)
|
||
controller.schedule_save.assert_not_called()
|
||
|
||
def test_restore_expands_only_on_wide_window(self) -> None:
|
||
window, panel, settings, _controller = _nav_window(NavigationDisplayMode.COMPACT, False, saved=True)
|
||
MainWindow._restore_nav_preference(window, settings)
|
||
panel.expand.assert_called_once_with(useAni=False)
|
||
|
||
window, panel, settings, _controller = _nav_window(
|
||
NavigationDisplayMode.COMPACT, False, width=1000, saved=True
|
||
)
|
||
MainWindow._restore_nav_preference(window, settings)
|
||
panel.expand.assert_not_called()
|
||
|
||
def test_collapsed_preference_never_expands(self) -> None:
|
||
window, panel, settings, _controller = _nav_window(NavigationDisplayMode.COMPACT, False, saved=False)
|
||
MainWindow._restore_nav_preference(window, settings)
|
||
MainWindow._sync_nav_to_preference(window)
|
||
panel.expand.assert_not_called()
|
||
|
||
def test_auto_collapse_keeps_preference_and_widening_restores(self) -> None:
|
||
# Библиотека свернула меню на узком окне: клика не было, предпочтение живо.
|
||
window, panel, settings, _controller = _nav_window(
|
||
NavigationDisplayMode.COMPACT, False, width=1000, saved=True
|
||
)
|
||
MainWindow._sync_nav_to_preference(window)
|
||
panel.expand.assert_not_called()
|
||
window.width = lambda: 1300
|
||
MainWindow._sync_nav_to_preference(window)
|
||
panel.expand.assert_called_once_with(useAni=False)
|
||
self.assertTrue(settings.nav_expanded)
|
||
|
||
def test_real_navigation_panel_click_order(self) -> None:
|
||
"""На настоящей панели qfluentwidgets: toggle() панели срабатывает раньше нашего слота."""
|
||
from PyQt6.QtCore import QEventLoop, QTimer
|
||
from qfluentwidgets import FluentWindow
|
||
|
||
global _nav_window_ref
|
||
window = FluentWindow()
|
||
_nav_window_ref = window # общий на процесс, не уничтожаем (см. докстринг модуля)
|
||
window.resize(1400, 800)
|
||
window.show()
|
||
QApplication.processEvents()
|
||
# Экран может ужать окно (у отключённой сессии Windows ~1036 px):
|
||
# порог развёртывания считаем от фактической ширины, иначе панель
|
||
# уходит во всплывающий режим MENU, который предпочтением не считается.
|
||
window.navigationInterface.setMinimumExpandWidth(max(1, window.width() - 100))
|
||
settings = AppSettings(nav_expanded=False)
|
||
controller = Mock()
|
||
controller.state.settings = settings
|
||
facade = SimpleNamespace(
|
||
navigationInterface=window.navigationInterface,
|
||
controller=controller,
|
||
_geometry_persistence_ready=True,
|
||
_nav_expanded_pref=False,
|
||
)
|
||
button = window.navigationInterface.panel.menuButton
|
||
button.clicked.connect(lambda: MainWindow._on_nav_menu_clicked(facade))
|
||
|
||
def settle() -> None:
|
||
loop = QEventLoop()
|
||
QTimer.singleShot(300, loop.quit)
|
||
loop.exec()
|
||
|
||
button.click()
|
||
settle()
|
||
self.assertTrue(settings.nav_expanded)
|
||
button.click()
|
||
settle()
|
||
self.assertFalse(settings.nav_expanded)
|
||
window.hide()
|
||
|
||
def test_setting_roundtrip(self) -> None:
|
||
self.assertTrue(AppSettings.from_dict(AppSettings(nav_expanded=True).to_dict()).nav_expanded)
|
||
self.assertFalse(AppSettings.from_dict({}).nav_expanded)
|
||
|
||
|
||
class SettingsPageSaveGuardTest(unittest.TestCase):
|
||
def test_window_owned_fields_survive_settings_page_snapshot(self) -> None:
|
||
self.assertIn("nav_expanded", _WINDOW_OWNED_SETTINGS)
|
||
self.assertIn("nodes_column_widths", _WINDOW_OWNED_SETTINGS)
|
||
self.assertIn("window_width", _WINDOW_OWNED_SETTINGS)
|
||
current = AppSettings(nav_expanded=True, window_width=1500, nodes_column_widths={"name": 240})
|
||
controller = Mock()
|
||
controller.state.settings = current
|
||
window = SimpleNamespace(controller=controller)
|
||
stale = AppSettings(nav_expanded=False, window_width=1000, theme="dark")
|
||
|
||
MainWindow._on_settings_page_saved(window, stale)
|
||
|
||
saved = controller.update_settings.call_args.args[0]
|
||
self.assertTrue(saved.nav_expanded)
|
||
self.assertEqual(saved.window_width, 1500)
|
||
self.assertEqual(saved.nodes_column_widths, {"name": 240})
|
||
self.assertEqual(saved.theme, "dark")
|
||
self.assertIsNot(saved.nodes_column_widths, current.nodes_column_widths)
|
||
|
||
|
||
class LifecycleTest(unittest.TestCase):
|
||
"""Нет циклов «C++-владение ↔ __dict__»: окно разрушается сразу, пока жив QApplication."""
|
||
|
||
def test_frame_gate_holds_no_strong_reference_to_window(self) -> None:
|
||
import gc
|
||
import weakref
|
||
|
||
from PyQt6.QtWidgets import QWidget
|
||
|
||
from xray_fluent.ui.motion import FrameGate
|
||
|
||
window = QWidget()
|
||
child = QWidget(window)
|
||
gate = FrameGate(child)
|
||
gate.set_interval(40)
|
||
window.show()
|
||
QApplication.processEvents()
|
||
self.assertTrue(gate.is_running())
|
||
destroyed = []
|
||
window.destroyed.connect(lambda *_: destroyed.append(True))
|
||
ref = weakref.ref(window)
|
||
del window, child, gate
|
||
gc.collect()
|
||
self.assertIsNone(ref())
|
||
self.assertEqual(destroyed, [True])
|
||
|
||
def test_connected_dashboard_is_freed_without_gc(self) -> None:
|
||
import gc
|
||
import weakref
|
||
|
||
page = DashboardPage()
|
||
page.resize(1000, 800)
|
||
page.show()
|
||
page.set_connection(True)
|
||
page._do_refresh_dashboard()
|
||
QApplication.processEvents()
|
||
self.assertTrue(page.connection_scene.is_animating())
|
||
ref = weakref.ref(page)
|
||
gc.disable()
|
||
try:
|
||
del page
|
||
self.assertIsNone(ref(), "dashboard page is kept alive by a reference cycle")
|
||
finally:
|
||
gc.enable()
|
||
|
||
def test_dispose_windows_deletes_idle_windows_and_keeps_running_threads(self) -> None:
|
||
import threading
|
||
|
||
from PyQt6 import sip
|
||
from PyQt6.QtCore import QThread
|
||
from PyQt6.QtWidgets import QWidget
|
||
|
||
from xray_fluent.ui.qt_lifecycle import dispose_windows
|
||
|
||
release = threading.Event()
|
||
|
||
class _Blocker(QThread):
|
||
def run(self) -> None:
|
||
release.wait(5)
|
||
|
||
idle = QWidget()
|
||
busy = QWidget()
|
||
thread = _Blocker(busy)
|
||
thread.start()
|
||
try:
|
||
windows = [idle, busy]
|
||
kept = dispose_windows(windows)
|
||
self.assertTrue(sip.isdeleted(idle))
|
||
self.assertFalse(sip.isdeleted(busy))
|
||
self.assertEqual(kept, [busy])
|
||
self.assertEqual(windows, [busy])
|
||
finally:
|
||
release.set()
|
||
thread.wait(5000)
|
||
dispose_windows(windows)
|
||
self.assertTrue(sip.isdeleted(busy))
|
||
|
||
def test_dispose_all_widgets_sweeps_orphan_top_levels(self) -> None:
|
||
from PyQt6 import sip
|
||
from PyQt6.QtWidgets import QMenu, QWidget
|
||
|
||
from xray_fluent.ui.qt_lifecycle import dispose_all_widgets
|
||
|
||
from unittest.mock import patch
|
||
|
||
from xray_fluent.ui import qt_lifecycle
|
||
|
||
window = QWidget()
|
||
orphan_menu = QMenu() # как меню трея без родителя
|
||
# Подменяем список окон верхнего уровня: настоящий снёс бы общие
|
||
# страницы других тестовых модулей в этом процессе.
|
||
fake_app = SimpleNamespace(topLevelWidgets=lambda: [orphan_menu])
|
||
with patch.object(qt_lifecycle.QApplication, "instance", return_value=fake_app):
|
||
dispose_all_widgets([window])
|
||
self.assertTrue(sip.isdeleted(window))
|
||
self.assertTrue(sip.isdeleted(orphan_menu))
|
||
|
||
|
||
|
||
class TunnelLoadScaleTest(unittest.TestCase):
|
||
"""Огоньки следуют реальной скорости до 1 Гбит/с; полоса меняется только под нагрузкой."""
|
||
|
||
def test_level_scale_distinguishes_high_speeds(self) -> None:
|
||
from xray_fluent.ui.dashboard_widgets import ConnectionScene
|
||
|
||
level = ConnectionScene._traffic_level
|
||
mbit = 125_000.0
|
||
points = [level(v) for v in (1_000.0, 1_000_000.0, 50 * mbit, 100 * mbit, 500 * mbit, 1000 * mbit)]
|
||
self.assertEqual(points, sorted(points))
|
||
self.assertAlmostEqual(level(1000 * mbit), 1.0, places=3)
|
||
self.assertEqual(level(5000 * mbit), 1.0) # потолок — 1 Гбит/с
|
||
self.assertGreater(level(500 * mbit) - level(50 * mbit), 0.2) # 500 и 50 Мбит заметно различаются
|
||
self.assertEqual(level(0), 0.0)
|
||
|
||
def test_background_traffic_does_not_thicken_the_tunnel(self) -> None:
|
||
from xray_fluent.ui.dashboard_widgets import ConnectionScene
|
||
|
||
scene = ConnectionScene()
|
||
try:
|
||
for bps, expect_load in ((20_000.0, False), (1_000_000.0, False), (60_000_000.0, True)):
|
||
scene._level = ConnectionScene._traffic_level(bps)
|
||
self.assertEqual(scene.load_level() > 0.05, expect_load, bps)
|
||
scene._level = 1.0
|
||
self.assertEqual(scene.load_level(), 1.0)
|
||
finally:
|
||
_KEEP_SCENES.append(scene)
|
||
|
||
|
||
_KEEP_SCENES: list = []
|
||
|
||
|
||
|
||
class SystemProxySwitchIntentTest(unittest.TestCase):
|
||
"""Клик по «Системному прокси» не откатывается, пока Windows применяет изменение."""
|
||
|
||
def setUp(self) -> None:
|
||
from xray_fluent.platform.windows.proxy_manager import SystemProxyState
|
||
|
||
self.State = SystemProxyState
|
||
self.page = _shared_page()
|
||
_reset(self.page)
|
||
self.page.set_settings_snapshot(AppSettings(tun_mode=False, enable_system_proxy=True))
|
||
self.page.set_connection(True)
|
||
self.page.set_system_proxy_state(self._registry(enabled=True))
|
||
self.emitted: list[bool] = []
|
||
self.page.proxy_toggled.connect(self.emitted.append)
|
||
|
||
def tearDown(self) -> None:
|
||
self.page.proxy_toggled.disconnect(self.emitted.append)
|
||
self.page._proxy_intent.clear()
|
||
self.page.set_system_proxy_state(None)
|
||
_reset(self.page)
|
||
|
||
def _registry(self, *, enabled: bool):
|
||
fields = getattr(self.State, "__dataclass_fields__", {})
|
||
values = {name: None for name in fields}
|
||
values.update(supported=True, enabled=enabled, is_ours=enabled)
|
||
try:
|
||
return self.State(**{k: v for k, v in values.items() if k in fields})
|
||
except TypeError:
|
||
return SimpleNamespace(supported=True, enabled=enabled, is_ours=enabled)
|
||
|
||
def test_click_off_shows_intent_until_windows_catches_up(self) -> None:
|
||
switch = self.page.proxy_switch
|
||
self.assertTrue(switch.isChecked())
|
||
switch.setChecked(False) # пользователь кликнул «выкл»
|
||
self.assertEqual(self.emitted, [False])
|
||
# Настройки уже «выкл», а реестр Windows ещё «вкл» — раньше тут был откат.
|
||
self.page.set_settings_snapshot(AppSettings(tun_mode=False, enable_system_proxy=False))
|
||
self.page.set_system_proxy_state(self._registry(enabled=True))
|
||
self.assertFalse(switch.isChecked())
|
||
self.assertTrue(switch.text.endswith("…"))
|
||
# Очередь прокси применила изменение — пришёл свежий снимок.
|
||
self.page.set_system_proxy_state(self._registry(enabled=False))
|
||
self.assertFalse(switch.isChecked())
|
||
self.assertEqual(switch.text, "Выкл")
|
||
|
||
def test_intent_times_out_to_the_truth(self) -> None:
|
||
switch = self.page.proxy_switch
|
||
switch.setChecked(False)
|
||
self.page._proxy_intent._deadline = 0.0 # таймаут истёк
|
||
self.page.set_system_proxy_state(self._registry(enabled=True))
|
||
self.assertTrue(switch.isChecked())
|
||
self.assertEqual(switch.text, "Вкл")
|
||
|
||
|
||
class ModeTileIntentTest(unittest.TestCase):
|
||
"""Плитка VPN/Прокси показывает выбор, пока переход применяет его к сессии."""
|
||
|
||
def setUp(self) -> None:
|
||
self.page = _shared_page()
|
||
_reset(self.page)
|
||
# Подключено в режиме прокси: сессия — прокси, настройка — прокси.
|
||
self.page.set_settings_snapshot(AppSettings(tun_mode=False, enable_system_proxy=True))
|
||
self.page.set_connection(True)
|
||
self.page.set_active_tun_mode(False)
|
||
self.emitted: list[bool] = []
|
||
self.page.tun_toggled.connect(self._on_toggled)
|
||
|
||
def tearDown(self) -> None:
|
||
self.page.tun_toggled.disconnect(self._on_toggled)
|
||
_reset(self.page)
|
||
|
||
def _on_toggled(self, tun: bool) -> None:
|
||
# Как MainWindow: настройка меняется синхронно, сессия — после перехода.
|
||
self.emitted.append(tun)
|
||
self.page.set_settings_snapshot(AppSettings(tun_mode=tun, enable_system_proxy=not tun))
|
||
|
||
def _assert_shows(self, tun: bool, *, applying: bool) -> None:
|
||
self.assertEqual(self.page.vpn_tile.isChecked(), tun)
|
||
self.assertEqual(self.page.proxy_tile.isChecked(), not tun)
|
||
chosen = self.page.vpn_tile if tun else self.page.proxy_tile
|
||
other = self.page.proxy_tile if tun else self.page.vpn_tile
|
||
self.assertEqual(chosen.title_label.text().endswith("…"), applying)
|
||
self.assertEqual(chosen.toolTip(), "Применяется…" if applying else "")
|
||
self.assertFalse(other.is_applying())
|
||
|
||
def test_click_shows_intent_until_session_switches(self) -> None:
|
||
self.page.vpn_tile.clicked.emit()
|
||
self.assertEqual(self.emitted, [True])
|
||
self._assert_shows(True, applying=True)
|
||
# Переход начался, старая сессия ещё жива — откатываться нельзя.
|
||
self.page.set_active_tun_mode(False)
|
||
self.page.set_transition_busy(True)
|
||
self._assert_shows(True, applying=True)
|
||
# Старая сессия остановлена, новой ещё нет — факта нет, выбор держится.
|
||
self.page.set_active_tun_mode(None)
|
||
self.page.set_connection(False)
|
||
self._assert_shows(True, applying=True)
|
||
# TUN-сессия поднялась — обычная подпись.
|
||
self.page.set_active_tun_mode(True)
|
||
self.page.set_connection(True)
|
||
self._assert_shows(True, applying=False)
|
||
self.assertEqual(self.page.vpn_tile.title_label.text(), "VPN (TUN)")
|
||
self.page.set_transition_busy(False)
|
||
self._assert_shows(True, applying=False)
|
||
|
||
def test_abandoned_transition_shows_the_running_session(self) -> None:
|
||
self.page.vpn_tile.clicked.emit()
|
||
self.page.set_transition_busy(True)
|
||
self._assert_shows(True, applying=True)
|
||
# Переход отменён до остановки старой сессии (например, DNS сервера
|
||
# для Zapret не определился): «занято» снято, работает прокси.
|
||
self.page.set_active_tun_mode(False)
|
||
self.page.set_transition_busy(False)
|
||
self._assert_shows(False, applying=False)
|
||
|
||
def test_intent_times_out_to_the_running_session(self) -> None:
|
||
self.page.vpn_tile.clicked.emit()
|
||
self.page._tun_intent._deadline = 0.0 # таймаут истёк
|
||
self.page.set_active_tun_mode(False)
|
||
self._assert_shows(False, applying=False)
|
||
# Плитка показывает факт — клик по VPN снова запрашивает переход.
|
||
self.page.vpn_tile.clicked.emit()
|
||
self.assertEqual(self.emitted, [True, True])
|
||
self._assert_shows(True, applying=True)
|
||
|
||
def test_timeout_without_session_falls_back_to_setting(self) -> None:
|
||
self.page.vpn_tile.clicked.emit()
|
||
self.page.set_active_tun_mode(None)
|
||
self.page.set_connection(False)
|
||
self.page._tun_intent._deadline = 0.0
|
||
self.page.set_transition_busy(True)
|
||
# Сессии нет, факта нет — показываем настройку (не «Прокси» из None).
|
||
self._assert_shows(True, applying=False)
|
||
|
||
def test_second_click_retargets_intent(self) -> None:
|
||
self.page.vpn_tile.clicked.emit()
|
||
self._assert_shows(True, applying=True)
|
||
# Второй клик до начала перехода: побеждает последний выбор, и он уже
|
||
# совпадает с работающей сессией — ждать нечего.
|
||
self.page.proxy_tile.clicked.emit()
|
||
self.assertEqual(self.emitted, [True, False])
|
||
self._assert_shows(False, applying=False)
|
||
# Устаревшее «VPN» не возвращается, когда переход отработает.
|
||
self.page.set_transition_busy(True)
|
||
self.page.set_transition_busy(False)
|
||
self._assert_shows(False, applying=False)
|
||
|
||
def test_disconnected_click_applies_at_once(self) -> None:
|
||
self.page.set_active_tun_mode(None)
|
||
self.page.set_connection(False)
|
||
self.page.vpn_tile.clicked.emit()
|
||
# Без подключения фактом служит настройка — она уже сменилась.
|
||
self._assert_shows(True, applying=False)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|