loop-uh
f9971d5a1f
Some checks failed
Windows project source guards / test (push) Failing after 31s
Движок разложен на части с одной задачей каждая: - endpoint: вид сервера (TCP / QUIC / WireGuard), хосты, порты, DNS; - strategies: каталоги winws2 и правило для выбранного сервера; - presets: файлы пресетов; - command: сборка командной строки winws2 (чистая функция); - manager: процесс winws2 с одним входом apply(план) и ревизиями — готовность только по started процесса нужной ревизии, таймаут 5 с. Склейка с подключением (~450 строк в контроллере, два наслоенных механизма «UDP-защиты» и «профиля сервера», кэш и прогрев DNS) заменена шлюзом application/server_bypass.py. Сохранены: DNS только в воркере, остановка туннеля перед сменой правила (кроме восстановления Hysteria), забор перед стартом ядра, fail-closed при потере winws2 (теперь и по таймауту перезапуска), маскировка IP сервера в логах до старта ядра; адреса серверов пула изучаются в фоне после подключения. Поведение: - у QUIC и WireGuard теперь свои стратегии; ключ zapret_target совместим, старые сборки после отката читают udp_* как раньше; - в каталогах вернулись 82 составные стратегии с ветками --payload (валидатор принимал только --lua-desync и молча их выбрасывал); - смена пресета под защищённой сессией идёт через переход: туннель останавливается, winws2 перезапускается, VPN подключается снова; - пропавший файл выбранного пресета — откат к Default, а не отказ; - правка работающего пресета перезапускает его; переименование больше не оставляет копию, занятое имя отклоняется; - настройки Zapret не откатываются при сохранении страницы «Настройки». Интерфейс: карточка состояния (работает/нет, пресет, одна кнопка, автозапуск), выбранный VPN-сервер с живым правилом, три строки видов серверов с подстраницей стратегии, пресеты — отдельной подстраницей. Объяснение — в окне «Как это б#&^ь работает?» с иллюстрацией.
82 lines
2.8 KiB
Python
82 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from unittest.mock import Mock
|
|
|
|
from PyQt6.QtCore import Qt
|
|
|
|
from xray_fluent.network.background_workers import EndpointResolver, StateWriter
|
|
|
|
|
|
class BackgroundWorkerTests(unittest.TestCase):
|
|
def test_endpoint_resolution_reports_generation_and_result(self) -> None:
|
|
events: list[tuple[int, str, set[str], Exception | None]] = []
|
|
worker = EndpointResolver(
|
|
7,
|
|
"proxy.example.com",
|
|
lambda _server: {"203.0.113.7"},
|
|
)
|
|
worker.resolved.connect(lambda *args: events.append(args))
|
|
|
|
worker.run()
|
|
|
|
self.assertEqual(events, [(7, "proxy.example.com", {"203.0.113.7"}, None)])
|
|
|
|
def test_endpoint_resolution_returns_errors_instead_of_raising(self) -> None:
|
|
error = OSError("DNS unavailable")
|
|
events = []
|
|
|
|
def fail(_server: str) -> set[str]:
|
|
raise error
|
|
|
|
worker = EndpointResolver(8, "proxy.example.com", fail)
|
|
worker.resolved.connect(lambda *args: events.append(args))
|
|
|
|
worker.run()
|
|
|
|
self.assertEqual(events, [(8, "proxy.example.com", None, error)])
|
|
|
|
def test_state_writer_writes_off_the_caller_thread(self) -> None:
|
|
caller = threading.get_ident()
|
|
writes = []
|
|
storage = Mock()
|
|
storage.write_serialized.side_effect = lambda payload, pw: writes.append((payload, pw, threading.get_ident()))
|
|
writer = StateWriter(storage)
|
|
writer.submit("{}", "pw")
|
|
self.assertTrue(writer.close())
|
|
self.assertEqual([(p, pw) for p, pw, _ in writes], [("{}", "pw")])
|
|
self.assertNotEqual(writes[0][2], caller)
|
|
|
|
def test_state_writer_coalesces_to_latest_snapshot(self) -> None:
|
|
gate = threading.Event()
|
|
writes = []
|
|
|
|
def slow_write(payload, pw):
|
|
gate.wait(2)
|
|
writes.append(payload)
|
|
|
|
storage = Mock()
|
|
storage.write_serialized.side_effect = slow_write
|
|
writer = StateWriter(storage)
|
|
writer.submit("first", "")
|
|
time.sleep(0.05) # писатель занят первым снимком
|
|
for payload in ("second", "third", "latest"):
|
|
writer.submit(payload, "")
|
|
gate.set()
|
|
self.assertTrue(writer.close())
|
|
self.assertEqual(writes, ["first", "latest"])
|
|
|
|
def test_state_writer_reports_failures(self) -> None:
|
|
storage = Mock()
|
|
storage.write_serialized.side_effect = OSError("disk full")
|
|
writer = StateWriter(storage)
|
|
errors = []
|
|
writer.failed.connect(errors.append, Qt.ConnectionType.DirectConnection)
|
|
writer.submit("{}", "")
|
|
self.assertTrue(writer.close())
|
|
self.assertEqual(errors, ["disk full"])
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|