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() в тестах раздувал память).
34 lines
1.2 KiB
Python
34 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
import subprocess
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from xray_fluent.engines.singbox.config_check import check_config
|
|
|
|
|
|
class SingboxConfigCheckTests(unittest.TestCase):
|
|
def test_check_uses_core_as_working_directory_for_local_rule_sets(self) -> None:
|
|
exe = Path("C:/ZapretKVN/core/sing-box.exe")
|
|
config = Path("C:/ZapretKVN/data/runtime/singbox_config.json")
|
|
completed = subprocess.CompletedProcess([], 0, stdout=b"", stderr=b"")
|
|
|
|
# check_config выполняется в worker-пуле: без прокачки Qt-событий.
|
|
with patch(
|
|
"xray_fluent.platform.windows.subprocess_utils.run_text",
|
|
return_value=completed,
|
|
) as run_mock, patch(
|
|
"xray_fluent.platform.windows.subprocess_utils.run_text_pumped",
|
|
) as pumped_mock:
|
|
self.assertEqual(check_config(exe, config), (True, ""))
|
|
|
|
self.assertEqual(
|
|
run_mock.call_args.args[0],
|
|
[str(exe), "check", "-D", str(exe.parent), "-c", str(config)],
|
|
)
|
|
pumped_mock.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|