zapret-kvn/tests/test_engine_process_stop.py

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()