zapret-kvn/tests/test_engine_process_stop.py

from __future__ import annotations

import shutil
import time
import unittest
from pathlib import Path
from unittest.mock import Mock, patch

from PyQt6.QtCore import QCoreApplication, QProcess

from xray_fluent.application.async_steps import TransitionRunner
from tests.step_fakes import drive, steps_via
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 setUp(self) -> None:
        # Таблица TCP на Windows настоящая: без подмены «порт закрыт» и пробы
        # (подменённые ниже) не вызываются. Здесь проверяется логика проб.
        patcher = patch("xray_fluent.engines.xray.manager.ports_listening", return_value=None)
        patcher.start()
        self.addCleanup(patcher.stop)

    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):
    # stop() — синхронный драйвер stop_steps() (только shutdown/тесты); ожидания
    # выхода процесса — шаги WaitProcessFinishedStep, не waitForFinished в GUI.
    def test_stop_kills_immediately_without_terminate_grace(self) -> None:
        # terminate() (WM_CLOSE) консольный sing-box не обрабатывает: ожидание
        # «мягкой» остановки только задерживало каждое переключение.
        manager = SingBoxManager()
        fake = Mock()
        fake.state.side_effect = [_RUNNING, _NOT_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.terminate.assert_not_called()
        fake.kill.assert_called_once_with()
        timeouts = [call.args[1] for call in wait_mock.call_args_list]
        self.assertEqual(timeouts, [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
        manager._tun_interface_name = "xftunabc123"
        released_mock = Mock()

        with patch(
            "xray_fluent.application.async_steps.wait_for_qprocess_finished",
            return_value=True,
        ), patch.object(SingBoxManager, "_wait_tun_released_steps", staticmethod(steps_via(released_mock))):
            self.assertTrue(manager.stop())

        released_mock.assert_called_once_with("xftunabc123")

    def test_request_stop_kills_without_waiting(self) -> None:
        manager = SingBoxManager()
        fake = Mock()
        fake.state.return_value = _RUNNING
        manager._process = fake

        with patch("xray_fluent.application.async_steps.wait_for_qprocess_finished") as wait_mock:
            manager.request_stop(expected=True)

        fake.kill.assert_called_once_with()
        wait_mock.assert_not_called()
        self.assertTrue(manager._stop_requested)


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
        sleeps: list[int] = []

        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],
        ):
            self.assertFalse(drive(manager._wait_until_tun_ready_steps("xftun0", max_wait=1.0), sleeps=sleeps))

        probe_mock.assert_called_once_with("xftun0")
        self.assertEqual(sleeps, [250])


class SingBoxProxyReadinessTests(unittest.TestCase):
    def setUp(self) -> None:
        # Таблица TCP на Windows настоящая: без подмены «порт закрыт» и пробы
        # (подменённые ниже) не вызываются. Здесь проверяется логика проб.
        patcher = patch("xray_fluent.engines.singbox.manager.ports_listening", return_value=None)
        patcher.start()
        self.addCleanup(patcher.stop)

    @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
        sleeps: list[int] = []

        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:
            self.assertTrue(
                drive(manager._wait_until_proxy_ready_steps(self._config(), max_wait=1.0), sleeps=sleeps)
            )

        # Пауза между пробами — таймерный шаг, сами пробы — в воркере.
        self.assertEqual(sleeps, [100, 100])
        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
        stop_mock = Mock(return_value=True)
        manager.stop_steps = steps_via(stop_mock)
        errors: list[str] = []
        manager.error.connect(errors.append)

        with patch(
            "xray_fluent.engines.singbox.manager.probe_listener_role",
            return_value=False,
        ):
            self.assertFalse(drive(manager._wait_until_proxy_ready_steps(self._config(), max_wait=0.0)))

        stop_mock.assert_called_once_with(expected=True)
        self.assertTrue(manager.last_start_failure_retryable)
        self.assertTrue(any("SOCKS 1390" in message for message in errors))

    def test_superseded_start_stops_without_reporting_failure(self) -> None:
        manager = SingBoxManager()
        fake = Mock()
        fake.state.return_value = _RUNNING
        manager._process = fake
        stop_mock = Mock(return_value=True)
        manager.stop_steps = steps_via(stop_mock)
        errors: list[str] = []
        manager.error.connect(errors.append)

        with patch("xray_fluent.engines.singbox.manager.probe_listener_role") as probe_mock:
            self.assertFalse(
                drive(manager._wait_until_proxy_ready_steps(self._config(), should_continue=lambda: False))
            )

        probe_mock.assert_not_called()
        stop_mock.assert_called_once_with(expected=True)
        self.assertEqual(errors, [])


if __name__ == "__main__":
    unittest.main()


class SingBoxTunRestartGapTests(unittest.TestCase):
    """Новый TUN — не раньше TUN_RESTART_GAP_S после остановки прежнего."""

    def _sleeps_before_spawn(self, stopped_ago: float | None) -> list[int]:
        from xray_fluent.engines.singbox import manager as singbox_manager

        manager = SingBoxManager()
        fake = Mock()
        fake.state.return_value = _NOT_RUNNING
        manager._process = fake
        now = 100.0
        manager._tun_stopped_at = None if stopped_ago is None else now - stopped_ago
        sleeps: list[int] = []

        class Spawned(Exception):
            pass

        with patch.object(singbox_manager, "_kill_orphaned_blocking", return_value=False), patch.object(
            singbox_manager.time, "monotonic", return_value=now
        ), patch.object(manager, "_spawn", side_effect=Spawned):
            with self.assertRaises(Spawned):
                drive(
                    manager._launch_steps(Path("sing-box.exe"), {}, Path("c.json"), "xftun0", None),
                    sleeps=sleeps,
                )
        return sleeps

    def test_waits_only_the_rest_of_the_gap_after_a_tun_stop(self) -> None:
        from xray_fluent.engines.singbox.manager import TUN_RESTART_GAP_S

        self.assertEqual(self._sleeps_before_spawn(0.5), [int((TUN_RESTART_GAP_S - 0.5) * 1000)])

    def test_no_wait_long_after_or_without_a_previous_tun(self) -> None:
        self.assertEqual(self._sleeps_before_spawn(10.0), [])
        self.assertEqual(self._sleeps_before_spawn(None), [])