loop-uh
2bcafcf599
All checks were successful
Windows project source guards / test (push) Successful in 2m17s
365 lines
13 KiB
Python
365 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import shutil
|
|
import time
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
from PyQt6.QtCore import QCoreApplication, QProcess
|
|
|
|
from xray_fluent.application.async_steps import TransitionRunner
|
|
from xray_fluent.engines.singbox.manager import SingBoxManager
|
|
from xray_fluent.engines.xray.manager import XrayManager
|
|
|
|
_APP = QCoreApplication.instance() or QCoreApplication([])
|
|
|
|
_RUNNING = QProcess.ProcessState.Running
|
|
_NOT_RUNNING = QProcess.ProcessState.NotRunning
|
|
|
|
|
|
class XrayStopTests(unittest.TestCase):
|
|
# После миграции на шаги (AC22) stop() гоняет stop_steps() через синхронный
|
|
# драйвер; блокирующее ожидание живёт в async_steps.WaitProcessFinishedStep.
|
|
def test_stop_kills_immediately_without_terminate(self) -> None:
|
|
manager = XrayManager()
|
|
fake = Mock()
|
|
fake.state.return_value = _RUNNING
|
|
manager._process = fake
|
|
|
|
with patch(
|
|
"xray_fluent.application.async_steps.wait_for_qprocess_finished",
|
|
return_value=True,
|
|
) as wait_mock:
|
|
self.assertTrue(manager.stop())
|
|
|
|
fake.kill.assert_called_once_with()
|
|
fake.terminate.assert_not_called()
|
|
wait_mock.assert_called_once_with(fake, 2000)
|
|
|
|
def test_stop_emits_perf_log_line(self) -> None:
|
|
manager = XrayManager()
|
|
fake = Mock()
|
|
fake.state.return_value = _RUNNING
|
|
manager._process = fake
|
|
lines: list[str] = []
|
|
manager.log_received.connect(lines.append)
|
|
|
|
with patch(
|
|
"xray_fluent.application.async_steps.wait_for_qprocess_finished",
|
|
return_value=True,
|
|
):
|
|
manager.stop()
|
|
|
|
self.assertTrue(any(line.startswith("[xray-perf] stop:") for line in lines))
|
|
|
|
def test_process_start_does_not_report_running_before_readiness(self) -> None:
|
|
manager = XrayManager()
|
|
states: list[bool] = []
|
|
manager.state_changed.connect(states.append)
|
|
|
|
manager._on_started()
|
|
|
|
self.assertFalse(manager.is_running)
|
|
self.assertEqual(states, [])
|
|
|
|
manager._mark_running()
|
|
|
|
self.assertTrue(manager.is_running)
|
|
self.assertEqual(states, [True])
|
|
|
|
|
|
class XrayEnsurePortsTests(unittest.TestCase):
|
|
def test_bindable_port_skips_netstat_diagnostics(self) -> None:
|
|
manager = XrayManager()
|
|
owner_lookup = Mock()
|
|
manager._find_listening_port_owner = owner_lookup
|
|
|
|
with patch(
|
|
"xray_fluent.engines.xray.manager.is_tcp_port_bindable",
|
|
return_value=True,
|
|
) as bind_mock:
|
|
self.assertIsNone(manager._ensure_ports_available({10808: "SOCKS"}))
|
|
|
|
bind_mock.assert_called_once_with("127.0.0.1", 10808)
|
|
owner_lookup.assert_not_called()
|
|
|
|
def test_busy_port_reports_owner_from_diagnostics(self) -> None:
|
|
manager = XrayManager()
|
|
manager._find_listening_port_owner = Mock(return_value=(4242, "other.exe"))
|
|
|
|
with patch(
|
|
"xray_fluent.engines.xray.manager.is_tcp_port_bindable",
|
|
return_value=False,
|
|
):
|
|
message = manager._ensure_ports_available({10808: "SOCKS"})
|
|
|
|
self.assertIsNotNone(message)
|
|
self.assertIn("10808", message)
|
|
self.assertIn("other.exe", message)
|
|
self.assertIn("4242", message)
|
|
|
|
def test_busy_port_without_owner_still_reports_conflict(self) -> None:
|
|
manager = XrayManager()
|
|
manager._find_listening_port_owner = Mock(return_value=None)
|
|
|
|
with patch(
|
|
"xray_fluent.engines.xray.manager.is_tcp_port_bindable",
|
|
return_value=False,
|
|
):
|
|
message = manager._ensure_ports_available({10808: "SOCKS"})
|
|
|
|
self.assertIsNotNone(message)
|
|
self.assertIn("10808", message)
|
|
|
|
def test_stale_xray_owner_is_killed_and_port_reprobed(self) -> None:
|
|
manager = XrayManager()
|
|
manager._find_listening_port_owner = Mock(return_value=(777, "xray.exe"))
|
|
manager._kill_pid = Mock(return_value=True)
|
|
|
|
with patch(
|
|
"xray_fluent.engines.xray.manager.is_tcp_port_bindable",
|
|
side_effect=[False, True],
|
|
), patch("xray_fluent.application.async_steps.sleep_with_events"):
|
|
self.assertIsNone(manager._ensure_ports_available({10808: "SOCKS"}))
|
|
|
|
manager._kill_pid.assert_called_once_with(777)
|
|
|
|
def test_readiness_probes_declared_proxy_protocols(self) -> None:
|
|
manager = XrayManager()
|
|
fake = Mock()
|
|
fake.state.return_value = _RUNNING
|
|
manager._process = fake
|
|
|
|
with patch(
|
|
"xray_fluent.engines.xray.manager.probe_listener_role",
|
|
return_value=True,
|
|
) as probe_mock:
|
|
self.assertTrue(manager._wait_until_ready({1390: "SOCKS", 1391: "HTTP"}))
|
|
|
|
self.assertCountEqual(
|
|
probe_mock.call_args_list,
|
|
[unittest.mock.call(1390, "SOCKS"), unittest.mock.call(1391, "HTTP")],
|
|
)
|
|
|
|
def test_readiness_forwards_socks_credentials_from_own_config(self) -> None:
|
|
manager = XrayManager()
|
|
fake = Mock()
|
|
fake.state.return_value = _RUNNING
|
|
manager._process = fake
|
|
|
|
with patch(
|
|
"xray_fluent.engines.xray.manager.probe_listener_role",
|
|
return_value=True,
|
|
) as probe_mock:
|
|
self.assertTrue(
|
|
manager._wait_until_ready(
|
|
{11808: "SOCKS"},
|
|
credentials={11808: {"username": "sidecar-a1", "password": "s3cret"}},
|
|
)
|
|
)
|
|
|
|
probe_mock.assert_called_once_with(
|
|
11808, "SOCKS", username="sidecar-a1", password="s3cret"
|
|
)
|
|
|
|
def test_extract_socks_credentials_reads_password_auth_inbounds(self) -> None:
|
|
config = {
|
|
"inbounds": [
|
|
{
|
|
"protocol": "socks",
|
|
"port": 11808,
|
|
"settings": {
|
|
"auth": "password",
|
|
"accounts": [{"user": "sidecar-a1", "pass": "s3cret"}],
|
|
"udp": True,
|
|
},
|
|
},
|
|
{"protocol": "socks", "port": 10808, "settings": {"auth": "noauth"}},
|
|
{"protocol": "dokodemo-door", "port": 19085, "tag": "api"},
|
|
]
|
|
}
|
|
self.assertEqual(
|
|
{11808: {"username": "sidecar-a1", "password": "s3cret"}},
|
|
XrayManager._extract_socks_credentials(config),
|
|
)
|
|
self.assertEqual({}, XrayManager._extract_socks_credentials({"inbounds": []}))
|
|
|
|
def test_singbox_extract_socks_credentials_reads_inbound_users(self) -> None:
|
|
from xray_fluent.engines.singbox.manager import SingBoxManager
|
|
|
|
config = {
|
|
"inbounds": [
|
|
{
|
|
"type": "mixed",
|
|
"listen_port": 2080,
|
|
"users": [{"username": "front-user", "password": "front-pass"}],
|
|
},
|
|
{"type": "socks", "listen_port": 10808},
|
|
]
|
|
}
|
|
self.assertEqual(
|
|
{2080: {"username": "front-user", "password": "front-pass"}},
|
|
SingBoxManager._extract_socks_credentials(config),
|
|
)
|
|
|
|
|
|
class XrayStopStepsAsyncTests(unittest.TestCase):
|
|
"""AC22: горячий путь stop_steps() работает через TransitionRunner без пампинга."""
|
|
|
|
@unittest.skipIf(shutil.which("sleep") is None, "требуется бинарь sleep")
|
|
def test_stop_steps_via_transition_runner_kills_real_process(self) -> None:
|
|
manager = XrayManager()
|
|
manager._process.setProgram("sleep")
|
|
manager._process.setArguments(["5"])
|
|
manager._process.start()
|
|
self.assertTrue(manager._process.waitForStarted(3000))
|
|
|
|
runner = TransitionRunner(manager.stop_steps())
|
|
runner.start()
|
|
|
|
deadline = time.monotonic() + 5.0
|
|
while not runner.done and time.monotonic() < deadline:
|
|
_APP.processEvents()
|
|
time.sleep(0.002)
|
|
|
|
self.assertTrue(runner.done)
|
|
self.assertIs(runner.result, True)
|
|
self.assertIsNone(runner.error)
|
|
self.assertEqual(manager._process.state(), _NOT_RUNNING)
|
|
|
|
|
|
class SingBoxStopTests(unittest.TestCase):
|
|
def test_stop_grace_is_at_most_500ms_before_kill(self) -> None:
|
|
manager = SingBoxManager()
|
|
fake = Mock()
|
|
fake.state.side_effect = [_RUNNING, _NOT_RUNNING]
|
|
manager._process = fake
|
|
|
|
with patch(
|
|
"xray_fluent.engines.singbox.manager.wait_for_qprocess_finished",
|
|
side_effect=[False, True],
|
|
) as wait_mock:
|
|
self.assertTrue(manager.stop())
|
|
|
|
fake.terminate.assert_called_once_with()
|
|
fake.kill.assert_called_once_with()
|
|
timeouts = [call.args[1] for call in wait_mock.call_args_list]
|
|
self.assertEqual(timeouts, [500, 2000])
|
|
|
|
def test_stop_waits_for_tun_release_after_tun_session(self) -> None:
|
|
manager = SingBoxManager()
|
|
fake = Mock()
|
|
fake.state.side_effect = [_RUNNING, _NOT_RUNNING]
|
|
manager._process = fake
|
|
manager._uses_tun = True
|
|
|
|
with patch(
|
|
"xray_fluent.engines.singbox.manager.wait_for_qprocess_finished",
|
|
return_value=True,
|
|
), patch.object(SingBoxManager, "_wait_tun_released") as released_mock:
|
|
self.assertTrue(manager.stop())
|
|
|
|
released_mock.assert_called_once_with()
|
|
|
|
|
|
class SingBoxTunReadinessTests(unittest.TestCase):
|
|
def test_slow_fallback_probe_cannot_extend_the_startup_deadline(self) -> None:
|
|
manager = SingBoxManager()
|
|
fake = Mock()
|
|
fake.state.return_value = _RUNNING
|
|
manager._process = fake
|
|
|
|
with patch("xray_fluent.engines.singbox.manager.os.name", "nt"), patch.object(
|
|
manager, "_probe_tun_interface_has_ipv4", return_value=(False, False)
|
|
) as probe_mock, patch(
|
|
"xray_fluent.engines.singbox.manager.time.monotonic",
|
|
side_effect=[10.0, 10.0, 11.0],
|
|
), patch(
|
|
"xray_fluent.engines.singbox.manager.sleep_with_events"
|
|
) as sleep_mock:
|
|
self.assertFalse(manager._wait_until_tun_ready("xftun0", max_wait=1.0))
|
|
|
|
probe_mock.assert_called_once_with("xftun0")
|
|
sleep_mock.assert_called_once_with(0.25)
|
|
|
|
|
|
class SingBoxProxyReadinessTests(unittest.TestCase):
|
|
@staticmethod
|
|
def _config() -> dict:
|
|
return {
|
|
"inbounds": [
|
|
{"type": "mixed", "listen_port": 1390},
|
|
{"type": "http", "listen_port": 1391},
|
|
],
|
|
"experimental": {
|
|
"clash_api": {"external_controller": "127.0.0.1:19090"}
|
|
},
|
|
}
|
|
|
|
def test_proxy_readiness_contract_contains_data_and_control_planes(self) -> None:
|
|
config = self._config()
|
|
self.assertEqual(SingBoxManager._extract_clash_api_port(config), 19090)
|
|
self.assertEqual(
|
|
SingBoxManager._extract_proxy_port_roles(config),
|
|
{1390: "SOCKS", 1391: "HTTP", 19090: "Clash API"},
|
|
)
|
|
|
|
def test_proxy_waits_for_every_runtime_listener(self) -> None:
|
|
manager = SingBoxManager()
|
|
fake = Mock()
|
|
fake.state.return_value = _RUNNING
|
|
manager._process = fake
|
|
http_attempts = 0
|
|
|
|
def probe(_port: int, role: str) -> bool:
|
|
nonlocal http_attempts
|
|
if role == "HTTP":
|
|
http_attempts += 1
|
|
return http_attempts >= 3
|
|
return True
|
|
|
|
with patch(
|
|
"xray_fluent.engines.singbox.manager.probe_listener_role",
|
|
side_effect=probe,
|
|
) as probe_mock, patch(
|
|
"xray_fluent.engines.singbox.manager.sleep_with_events"
|
|
) as sleep_mock:
|
|
self.assertTrue(
|
|
manager._wait_until_proxy_ready(
|
|
self._config(),
|
|
max_wait=1.0,
|
|
)
|
|
)
|
|
|
|
self.assertEqual(sleep_mock.call_count, 2)
|
|
self.assertIn(unittest.mock.call(1390, "SOCKS"), probe_mock.call_args_list)
|
|
self.assertIn(unittest.mock.call(1391, "HTTP"), probe_mock.call_args_list)
|
|
self.assertIn(unittest.mock.call(19090, "Clash API"), probe_mock.call_args_list)
|
|
|
|
def test_proxy_readiness_timeout_stops_the_process(self) -> None:
|
|
manager = SingBoxManager()
|
|
fake = Mock()
|
|
fake.state.return_value = _RUNNING
|
|
manager._process = fake
|
|
manager.stop = Mock(return_value=True)
|
|
errors: list[str] = []
|
|
manager.error.connect(errors.append)
|
|
|
|
with patch(
|
|
"xray_fluent.engines.singbox.manager.probe_listener_role",
|
|
return_value=False,
|
|
):
|
|
self.assertFalse(
|
|
manager._wait_until_proxy_ready(
|
|
self._config(),
|
|
max_wait=0.0,
|
|
)
|
|
)
|
|
|
|
manager.stop.assert_called_once_with(expected=True)
|
|
self.assertTrue(manager.last_start_failure_retryable)
|
|
self.assertTrue(any("SOCKS 1390" in message for message in errors))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|