loop-uh
91e8e3fc0d
Три Python-слоя (модель → QSortFilterProxyModel → прокси группировки с кэшами) и TableItemDelegate из qfluentwidgets заменены одной моделью: фильтр, сортировка, группы и свёртка считаются одним проходом, тексты ячеек лениво кэшируются в строке. Вид перерисовывает при наведении и нажатии только затронутые строки, делегат рисует ячейку сам по палитре, собранной один раз на тему/акцент. Замер на 1000 серверах (offscreen): наведение 9,2 → 0,4 мс, прокрутка 14,8 → 3,6 мс, обновление после пинга 40 → 0,9 мс, после «в избранное» 39 → 5 мс. Поведение: - выбор ≠ подключение: клик, стрелки и правый клик только выделяют; подключение — двойной клик, Enter или пункт «Подключить к этому серверу»; детали — пункт «Подробности»; - выделение серверов переживает свёртку/развёртку группы; - у недоступного сервера красный только пинг, остальное приглушено; протокол рисуется бейджем.
225 lines
11 KiB
Python
225 lines
11 KiB
Python
"""Behavioral regressions for grouped servers, offline countries and startup."""
|
||
import os
|
||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||
import time
|
||
import threading
|
||
import unittest
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from unittest.mock import patch, Mock, MagicMock
|
||
|
||
from PyQt6.QtCore import QRect, QPersistentModelIndex, Qt, QTimer, QEventLoop
|
||
from PyQt6.QtTest import QTest
|
||
from PyQt6.QtWidgets import QApplication, QWidget
|
||
_APP = QApplication.instance() or QApplication([])
|
||
|
||
from xray_fluent.profiles.models import Node, AppSettings, Subscription
|
||
from xray_fluent.profiles.geoip import CountryDatabase, endpoint_hosts
|
||
from xray_fluent.network.country_resolver import CountryResolver
|
||
from xray_fluent.ui.nodes_page import NodesPage
|
||
from xray_fluent.ui.nodes_table_model import GROUP_KEY_ROLE, NODE_ID_ROLE, COL_PING
|
||
from xray_fluent.ui.window_geometry import fitted_geometry
|
||
from xray_fluent.ui.deferred_page import DeferredPage
|
||
from xray_fluent.application.startup_service import StartupWorker
|
||
|
||
|
||
# Одна страница на модуль: массовое создание/удаление полных страниц в одном
|
||
# процессе роняет Windows-прогон (см. tests/test_app_detail_pages.py).
|
||
_shared_page: NodesPage | None = None
|
||
|
||
# Окно троттлинга пересортировки по метрикам (300 мс) плюс запас.
|
||
_AFTER_RELAYOUT_WINDOW_MS = 400
|
||
|
||
|
||
def _nodes_page() -> NodesPage:
|
||
global _shared_page
|
||
if _shared_page is None:
|
||
_shared_page = NodesPage()
|
||
page = _shared_page
|
||
page._table_model._relayout_timer.stop()
|
||
page.set_subscriptions([])
|
||
page.apply_view_settings(AppSettings())
|
||
page._clear_filters()
|
||
page.table.clearSelection()
|
||
page.set_active_node(None)
|
||
page.set_nodes([])
|
||
for timer in (page._column_layout_timer, page._viewport_layout_timer, page._search_timer,
|
||
page._ping_batch_timer, page._speed_progress_timer):
|
||
timer.stop()
|
||
return page
|
||
|
||
|
||
class GroupedServersTests(unittest.TestCase):
|
||
def setUp(self):
|
||
self.page = _nodes_page()
|
||
|
||
def _connect(self, signal, slot):
|
||
signal.connect(slot)
|
||
self.addCleanup(signal.disconnect, slot)
|
||
|
||
def test_duplicate_subscription_names_still_have_distinct_groups(self):
|
||
self.page.set_subscriptions([Subscription(id="one", name="Provider", url=""), Subscription(id="two", name="Provider", url="")])
|
||
self.page.set_nodes([Node(id="a", subscription_id="one"), Node(id="b", subscription_id="two"), Node(id="c")])
|
||
model = self.page._table_model
|
||
self.assertEqual(len(model.group_keys()), 3)
|
||
keys = {model.index(row, 0).data(GROUP_KEY_ROLE) for row in range(model.rowCount())} - {None}
|
||
self.assertEqual(keys, {"source:one", "source:two", "source:local"})
|
||
|
||
def test_persistent_selection_and_expansion_survive_metric_sort(self):
|
||
nodes = [Node(id="a", name="A", ping_ms=10), Node(id="b", name="B", ping_ms=20)]
|
||
self.page.set_nodes(nodes, "a")
|
||
model = self.page._table_model
|
||
self.page._table_model.set_sort("ping", False)
|
||
persistent = QPersistentModelIndex(model.index(model.row_of_node("a"), 0))
|
||
resets = []
|
||
self._connect(model.modelReset, lambda: resets.append(True))
|
||
|
||
# Пересортировка по пингу — перенос строк без reset: персистентный
|
||
# индекс и выделение едут вместе с сервером.
|
||
nodes[0].ping_ms = 90
|
||
self.page._table_model.finish_ping_batch({"a"})
|
||
QTest.qWait(_AFTER_RELAYOUT_WINDOW_MS)
|
||
self.assertEqual(persistent.data(NODE_ID_ROLE), "a")
|
||
self.assertEqual(persistent.row(), 2)
|
||
self.assertEqual(self.page._selected_ids(), {"a"})
|
||
self.assertEqual(resets, [])
|
||
|
||
# Свёрнутая группа не разворачивается от пересортировки, а выделение
|
||
# внутри неё не теряется.
|
||
self.page.table.set_group_expanded("source:local", False)
|
||
resets.clear()
|
||
nodes[0].ping_ms = 5
|
||
self.page._table_model.finish_ping_batch({"a"})
|
||
QTest.qWait(_AFTER_RELAYOUT_WINDOW_MS)
|
||
self.assertEqual(resets, [])
|
||
self.assertFalse(self.page.table.is_group_expanded("source:local"))
|
||
self.assertEqual(self.page._selected_ids(), {"a"})
|
||
self.page.table.set_group_expanded("source:local", True)
|
||
self.assertEqual(model.row_of_node("a"), 1)
|
||
self.assertEqual(self.page._selected_ids(), {"a"})
|
||
|
||
def test_favorites_filter_and_settings_roundtrip(self):
|
||
nodes = [Node(id="a", is_favorite=True), Node(id="b")]
|
||
self.page.set_nodes(nodes)
|
||
self.page.favorites_filter.setChecked(True)
|
||
self.assertEqual(self.page._table_model.visible_node_count(), 1)
|
||
self.assertEqual(self.page._table_model.visible_node_ids(), ["a"])
|
||
prefs=[]
|
||
self._connect(self.page.view_prefs_changed, prefs.append)
|
||
self.page.table.horizontalHeader().resizeSection(0, 480)
|
||
self.page._emit_view_prefs()
|
||
restored = AppSettings.from_dict(prefs[-1])
|
||
self.assertTrue(restored.nodes_favorites_only)
|
||
self.assertEqual(restored.nodes_column_widths['name'], 480)
|
||
self.assertTrue(Node.from_dict(nodes[0].to_dict()).is_favorite)
|
||
|
||
def test_large_list_point_update_does_not_reset_or_resize(self):
|
||
nodes=[Node(id=str(i), name=f"Server {i}", subscription_id=str(i%20)) for i in range(10000)]
|
||
self.page.set_nodes(nodes)
|
||
model=self.page._table_model
|
||
self.assertEqual(len(model.group_keys()),20)
|
||
resets, layouts, changes = [], [], []
|
||
self._connect(model.modelReset, lambda: resets.append(True))
|
||
self._connect(model.layoutChanged, lambda *args: layouts.append(True))
|
||
self._connect(model.dataChanged, lambda top, bottom, roles=None: changes.append((top.row(), bottom.row())))
|
||
nodes[5000].ping_ms = 42
|
||
self.page._table_model.finish_ping_batch({'5000'})
|
||
QTest.qWait(_AFTER_RELAYOUT_WINDOW_MS)
|
||
self.assertEqual(resets,[])
|
||
self.assertEqual(layouts,[])
|
||
row = model.row_of_node('5000')
|
||
self.assertEqual(changes, [(row, row)])
|
||
self.assertEqual(self.page.table.horizontalHeader().sectionSize(0),360)
|
||
|
||
|
||
class OfflineCountryTests(unittest.TestCase):
|
||
def test_no_network_calls_even_for_domains_and_ipv6(self):
|
||
factory=MagicMock()
|
||
reader=CountryDatabase()
|
||
reader._reader=Mock()
|
||
reader._reader.get.return_value={'country':{'iso_code':'US'}}
|
||
factory.return_value.__enter__.return_value=reader
|
||
results=[]
|
||
worker=CountryResolver([('a',('vpn.example',),('vpn.example',)),('b',('8.8.8.8',),('8.8.8.8',)),('c',('v6',),('2606:4700:4700::1111',))],database_factory=factory,cache_provider=lambda: {})
|
||
worker.resolved.connect(results.append)
|
||
with patch('socket.getaddrinfo',side_effect=AssertionError('DNS forbidden')), patch('socket.socket',side_effect=AssertionError('socket forbidden')), patch('urllib.request.urlopen',side_effect=AssertionError('HTTP forbidden')):
|
||
worker.run()
|
||
self.assertEqual(results[0]['a'][1], '')
|
||
self.assertEqual(results[0]['b'][1], 'US')
|
||
self.assertEqual(results[0]['c'][1], 'US')
|
||
self.assertEqual(reader._reader.get.call_count,2)
|
||
reader.country.cache_clear()
|
||
|
||
def test_awg_uses_peers_not_interface_or_routes(self):
|
||
node=Node(scheme='awg',server='8.8.8.8',outbound={'address':['10.0.0.2/32'],'peers':[{'address':'8.8.8.8','allowed_ips':['0.0.0.0/0']}]})
|
||
self.assertEqual(endpoint_hosts(node),('8.8.8.8',))
|
||
|
||
def test_old_country_is_discarded_but_manual_choice_is_persisted(self):
|
||
node=Node.from_dict({'country_code':'NL','country_override':'US','is_favorite':True})
|
||
self.assertEqual(node.country_code,'')
|
||
self.assertEqual(node.country_override,'US')
|
||
self.assertNotIn('country_code',node.to_dict())
|
||
|
||
def test_late_result_cannot_override_changed_endpoint_or_manual_country(self):
|
||
from xray_fluent.application.node_runtime_service import on_countries_resolved
|
||
node=Node(id='a',server='new.example')
|
||
ctrl=SimpleNamespace(state=SimpleNamespace(nodes=[node]),nodes_changed=Mock())
|
||
on_countries_resolved(ctrl,{'a':(('old.example',),'NL')})
|
||
self.assertEqual(node.country_code,'')
|
||
node.country_override='US'
|
||
on_countries_resolved(ctrl,{'a':(('new.example',),'NL')})
|
||
self.assertEqual(node.country_override,'US')
|
||
ctrl.nodes_changed.emit.assert_not_called()
|
||
|
||
|
||
class GeometryTests(unittest.TestCase):
|
||
def test_secondary_negative_coordinates(self):
|
||
settings=AppSettings(window_x=-1500,window_y=50,window_width=1000,window_height=720)
|
||
rect, minimum=fitted_geometry(settings,[QRect(0,0,1920,1080),QRect(-1920,0,1920,1080)])
|
||
self.assertEqual(rect.x(),-1500)
|
||
self.assertEqual(minimum.width(),860)
|
||
|
||
def test_disconnected_screen_and_small_desktop(self):
|
||
rect,minimum=fitted_geometry(AppSettings(window_x=3000,window_y=3000,window_width=3000,window_height=2000),[QRect(0,0,800,500)])
|
||
self.assertEqual(rect,QRect(0,0,800,500))
|
||
self.assertEqual(minimum.width(),800)
|
||
|
||
|
||
class StartupTests(unittest.TestCase):
|
||
def test_slow_preparation_leaves_qt_event_loop_running(self):
|
||
storage=SimpleNamespace(_startup_raw='',load_payload=lambda raw:{},passphrase='',_normalize_state_paths=lambda s:s)
|
||
release = threading.Event()
|
||
observed = []
|
||
worker=StartupWorker(storage,prepare=lambda:observed.append(release.wait(2)))
|
||
loop=QEventLoop();worker.finished.connect(loop.quit)
|
||
with patch('xray_fluent.application.startup_service.TrafficHistoryStorage',return_value=object()):
|
||
QTimer.singleShot(20, release.set)
|
||
worker.start();loop.exec();worker.wait()
|
||
self.assertEqual(observed, [True], "GUI must service the timer while startup is waiting")
|
||
|
||
def test_deferred_page_constructs_once_and_replays_latest_snapshot(self):
|
||
class Page(QWidget):
|
||
count=0
|
||
def __init__(self,parent=None):
|
||
super().__init__(parent);Page.count+=1;self.value=None
|
||
def set_values(self,value):self.value=value
|
||
host=DeferredPage(Page,'test');host.set_values(1);host.set_values(2)
|
||
self.assertEqual(Page.count,0)
|
||
page=host.ensure_page();self.assertEqual(page.value,2)
|
||
self.assertIs(host.ensure_page(),page);self.assertEqual(Page.count,1)
|
||
host.deleteLater()
|
||
|
||
|
||
class StartupCancellationTests(unittest.TestCase):
|
||
def test_cancel_before_load_never_saves_default_state(self):
|
||
from xray_fluent.ui.main_window import MainWindow
|
||
controller=Mock()
|
||
window=SimpleNamespace(_startup_loader=Mock(),_state_loaded=False,controller=controller)
|
||
MainWindow._shutdown_controller(window)
|
||
controller.shutdown.assert_not_called()
|
||
controller.save.assert_not_called()
|
||
window._startup_loader.cancel.assert_called_once()
|
||
|
||
def test_initialize_is_idempotent(self):
|
||
from xray_fluent.ui.main_window import MainWindow
|
||
MainWindow.initialize(SimpleNamespace(_initialized=True))
|