loop-uh
ab0384c1a5
sing-box, Hysteria, Amnezia и Zapret получили start_steps/stop_steps по образцу XrayManager: ожидания процессов — сигналы QProcess (wait_process_*), пробы портов/релея, sing-box check, поиск осиротевших процессов (PowerShell), WinAPI-резолв аплинка и PUT Clash API — в run_in_worker, паузы — таймеры. Синхронные start()/stop() остались драйверами тех же шагов для выхода из приложения и тестов. - sing-box: запись конфига + check в воркере, готовность прокси/TUN и снятие TUN-адаптера — шагами; should_continue прерывает устаревший старт без ложной ошибки; request_stop() — kill без ожидания для обработчиков сигналов. - Clash API: select_outbound_when_ready_steps (попытка в воркере, пауза таймером). - Hysteria: повтор без Chrome-parrot выполняется внутри идущего start_steps, а вне старта — собственным TransitionRunner, без вложенного start(). - Amnezia: physical_network_steps, ожидание relay_ready/рукопожатия шагами. - Zapret: start() асинхронный (ожидание выхода прежнего winws2, поиск осиротевших, пауза драйвера — шаги), stop() без waitForFinished; блокирующее ожидание оставлено только для выхода (wait=True). - async_steps: run_in_worker(executor=...), жёсткий дедлайн шага (60 с) в обоих драйверах; run_steps_blocking падает TypeError на не-генераторе и на не-шаге вместо бесконечного цикла (Mock.send() в тестах раздувал память).
88 lines
5.1 KiB
Python
88 lines
5.1 KiB
Python
"""Regressions from 0.6.1 diagnostics: DNS bootstrap and misleading startup errors."""
|
||
import ipaddress
|
||
import unittest
|
||
from unittest.mock import Mock, patch
|
||
from concurrent.futures import Future
|
||
from PyQt6.QtCore import QProcess
|
||
from xray_fluent.engines.hysteria.manager import HysteriaManager, _FUNCTIONAL_HTTPS_ENDPOINTS
|
||
from xray_fluent.engines.socks_probe import probe_https
|
||
from xray_fluent.engines.hysteria.runtime_contract import HysteriaFailureCode
|
||
from xray_fluent.diagnostics.connection_message import connection_message
|
||
|
||
|
||
class HysteriaStartupRegressions(unittest.TestCase):
|
||
def test_probes_send_domain_connect_resolved_remotely_with_matching_tls_name(self):
|
||
# Remote resolution needs no local/front DNS and proves node egress.
|
||
for host, name, path in _FUNCTIONAL_HTTPS_ENDPOINTS:
|
||
with self.assertRaises(ValueError):
|
||
ipaddress.ip_address(host)
|
||
raw = Mock()
|
||
raw.recv.side_effect = [b'\x05\x02', b'\x01\x00', b'\x05\x00\x00\x01', b'\0' * 6]
|
||
tls = Mock()
|
||
tls.recv.return_value = b'HTTP/'
|
||
context = Mock()
|
||
context.wrap_socket.return_value.__enter__ = Mock(return_value=tls)
|
||
context.wrap_socket.return_value.__exit__ = Mock(return_value=False)
|
||
with patch('xray_fluent.engines.socks_probe.socket.create_connection', return_value=raw), \
|
||
patch('xray_fluent.engines.socks_probe.ssl.create_default_context', return_value=context):
|
||
probe_https(11809, username='user', password='secret', endpoint=(host, name, path), timeout=4)
|
||
request = raw.sendall.call_args_list[-1].args[0]
|
||
encoded = host.encode('idna')
|
||
self.assertEqual(request[:5], b'\x05\x01\x00\x03' + bytes([len(encoded)]))
|
||
self.assertEqual(request[5:5 + len(encoded)], encoded)
|
||
self.assertEqual(request[-2:], (443).to_bytes(2, 'big'))
|
||
context.wrap_socket.assert_called_once_with(raw, server_hostname=name)
|
||
self.assertIn(f'HEAD {path} HTTP/1.1'.encode(), tls.sendall.call_args.args[0])
|
||
self.assertIn(('Host: '+name).encode(), tls.sendall.call_args.args[0])
|
||
raw.close.assert_called_once()
|
||
|
||
def test_initial_timeout_is_logged_without_premature_error_and_success_clears_it(self):
|
||
manager = HysteriaManager()
|
||
manager._starting = True
|
||
errors = []
|
||
manager.error.connect(errors.append)
|
||
manager._emit_process_line('WARN connect error: timeout: no recent network activity')
|
||
self.assertEqual(errors, [])
|
||
self.assertEqual(manager.last_failure_code, HysteriaFailureCode.TARGET_NETWORK_TIMEOUT)
|
||
self.assertTrue(manager._last_output_lines)
|
||
manager._mark_running()
|
||
self.assertIsNone(manager.last_failure_code)
|
||
manager.deleteLater()
|
||
|
||
def test_security_failure_is_not_deferred_or_retried_with_identical_quic(self):
|
||
manager = HysteriaManager()
|
||
manager._starting = True
|
||
manager._compatibility_config = {'quic': {'disableChromeParrot': True}}
|
||
errors = []
|
||
manager.error.connect(errors.append)
|
||
with patch.object(manager, '_schedule_chrome_parrot_fallback') as retry:
|
||
manager._emit_process_line('connect error: CRYPTO_ERROR 0x150 (remote): tls: internal error')
|
||
retry.assert_not_called()
|
||
self.assertEqual(len(errors), 1)
|
||
manager.deleteLater()
|
||
|
||
def test_cancelled_readiness_stops_without_another_probe_wave(self):
|
||
manager = HysteriaManager()
|
||
clock = [0.0]
|
||
executor = Mock()
|
||
def submit(*args, **kwargs):
|
||
future = Future(); future.set_exception(TimeoutError('timed out')); return future
|
||
executor.submit.side_effect = submit
|
||
def cancel(seconds):
|
||
clock[0] += seconds
|
||
manager._compatibility_generation += 1
|
||
with patch.object(manager._process, 'state', return_value=QProcess.ProcessState.Running), \
|
||
patch('xray_fluent.engines.hysteria.manager.ThreadPoolExecutor', return_value=executor), \
|
||
patch('xray_fluent.engines.hysteria.manager.time.monotonic', side_effect=lambda: clock[0]), \
|
||
patch('xray_fluent.application.async_steps.sleep_with_events', side_effect=cancel):
|
||
self.assertFalse(manager._wait_until_remote_ready(11809, username='u', password='p'))
|
||
self.assertEqual(executor.submit.call_count, 3)
|
||
manager.deleteLater()
|
||
|
||
def test_ui_distinguishes_transport_timeout_from_https_probe_timeout(self):
|
||
remote = '[hysteria][attempt=3 node="Server"] SOCKS5 TCP error {"error":"connect error: timeout: no recent network activity"}'
|
||
https = 'Hysteria relay локально открыт, но удалённый handshake не готов: functional HTTPS probes failed: _ssl.c:993: timed out'
|
||
self.assertIn('соединение с сервером', connection_message(remote))
|
||
self.assertIn('Проверка HTTPS', connection_message(https))
|
||
self.assertNotIn('attempt=', connection_message(remote))
|
||
self.assertEqual(connection_message('Неверный пароль профиля'), 'Неверный пароль профиля')
|