zapret-kvn/tests/test_core_hot_switch.py

from __future__ import annotations

from tests.step_fakes import steps_via

import unittest
from types import SimpleNamespace
from unittest.mock import Mock, patch

from xray_fluent.application.selection_source import SelectionSource
from xray_fluent.application.controller import AppController
from tests.step_fakes import bridge_controller, drive
from xray_fluent.engines.hysteria.runtime_contract import HysteriaFailureCode
from xray_fluent.application.node_service import set_selected_node
from xray_fluent.application.signature_service import transition_signature
from xray_fluent.application.outbound_pool_service import (
    XRAY_BALANCER_TAG,
    XRAY_OUTBOUND_PREFIX,
    build_xray_outbound_pool,
)
from xray_fluent.engines.singbox.runtime_planner import (
    parse_singbox_document,
    plan_singbox_proxy_runtime,
    plan_singbox_runtime,
)
from xray_fluent.engines.singbox.selector_api import (
    build_selector_url,
    select_outbound,
    select_outbound_when_ready,
)
from xray_fluent.engines.hysteria.manager import HysteriaManager
from xray_fluent.engines.xray.config_builder import build_xray_config
from xray_fluent.importer.link_parser import parse_single
from xray_fluent.profiles.models import AppSettings, RoutingSettings

from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
SINGBOX_TEMPLATE = ROOT / "data" / "templates" / "sing-box" / "default.json"


def xray_nodes():
    return [
        parse_single(
            "vless://11111111-1111-1111-1111-111111111111@one.example:443"
            "?type=tcp&security=tls&sni=one.example#one"
        ),
        parse_single("trojan://secret@two.example:443?security=tls&sni=two.example#two"),
    ]


class XrayPoolTests(unittest.TestCase):
    def test_pool_tags_are_stable_and_not_display_name_based(self) -> None:
        nodes = xray_nodes()
        first = build_xray_outbound_pool(nodes)
        nodes[0].name = "renamed locally"
        second = build_xray_outbound_pool(reversed(nodes))
        self.assertEqual(first.tags, second.tags)
        self.assertTrue(all(tag.startswith(XRAY_OUTBOUND_PREFIX) for tag in first.tags.values()))

    def test_generated_config_keeps_routing_but_uses_core_balancer(self) -> None:
        nodes = xray_nodes()
        routing = RoutingSettings()
        routing.bypass_lan = False
        pool = build_xray_outbound_pool(nodes)
        config = build_xray_config(nodes[0], routing, AppSettings(), outbound_pool=pool)

        tags = [outbound["tag"] for outbound in config["outbounds"]]
        self.assertEqual(tags[:len(pool.nodes)], [pool.tag_for(node.id) for node in pool.nodes])
        self.assertEqual(config["routing"]["balancers"][0]["tag"], XRAY_BALANCER_TAG)
        self.assertTrue(
            any(rule.get("balancerTag") == XRAY_BALANCER_TAG for rule in config["routing"]["rules"])
        )
        self.assertEqual(
            config["api"]["services"],
            ["StatsService", "RoutingService", "HandlerService"],
        )


class SingboxPoolTests(unittest.TestCase):
    def test_native_provider_excludes_all_uri_hysteria2_sidecar_nodes(self) -> None:
        salamander = parse_single(
            "hy2://secret@one.example:443/?obfs=salamander&obfs-password=cover#one"
        )
        gecko = parse_single(
            "hy2://secret@gecko.example:443/?obfs=gecko&obfs-password=cover#gecko"
        )
        vless = parse_single(
            "vless://11111111-1111-1111-1111-111111111111@two.example:443"
            "?type=tcp&security=tls&sni=two.example#two"
        )
        native = parse_single("trojan://secret@native.example:443?security=tls#native")
        nodes = [salamander, gecko, vless, native]
        document = parse_singbox_document(
            SINGBOX_TEMPLATE,
            SINGBOX_TEMPLATE.read_text(encoding="utf-8"),
        )

        plans = (
            plan_singbox_proxy_runtime(document, native, pool_nodes=nodes),
            plan_singbox_runtime(document, native, pool_nodes=nodes),
        )

        for plan in plans:
            with self.subTest(mode="proxy" if plan.socks_port else "tun"):
                self.assertEqual(plan.outcome, "native_singbox")
                self.assertIn(native.id, plan.selector_tags)
                self.assertNotIn(vless.id, plan.selector_tags)
                self.assertNotIn(salamander.id, plan.selector_tags)
                self.assertNotIn(gecko.id, plan.selector_tags)
                provider = (plan.provider_payload or {})["outbounds"]
                self.assertFalse(any(item.get("type") in {"hysteria2", "vless"} for item in provider))

    def test_native_nodes_are_exposed_through_provider_selector(self) -> None:
        nodes = [
            parse_single(
                "trojan://secret@two.example:443?security=tls&sni=two.example#two"
            ),
            parse_single("trojan://secret@three.example:443?security=tls&sni=three.example#three"),
        ]
        document = parse_singbox_document(SINGBOX_TEMPLATE, SINGBOX_TEMPLATE.read_text(encoding="utf-8"))
        plan = plan_singbox_proxy_runtime(document, nodes[0], pool_nodes=nodes)

        selector = next(item for item in plan.singbox_config["outbounds"] if item.get("tag") == "proxy")
        self.assertEqual(selector["type"], "selector")
        self.assertEqual(selector["default"], "direct")
        self.assertTrue(selector["interrupt_exist_connections"])
        self.assertEqual(set(plan.selector_tags or {}), {node.id for node in nodes})
        self.assertEqual(len((plan.provider_payload or {})["outbounds"]), 2)
        self.assertEqual(plan.selected_outbound_tag, plan.selector_tags[nodes[0].id])

    def test_hybrid_sidecar_has_its_own_persistent_xray_pool(self) -> None:
        nodes = [
            parse_single(
                "vless://11111111-1111-1111-1111-111111111111@one.example:443"
                "?type=xhttp&security=tls&sni=one.example&path=%2Fapi#xhttp"
            ),
            parse_single("trojan://secret@two.example:443?security=tls&sni=two.example#trojan"),
        ]
        document = parse_singbox_document(SINGBOX_TEMPLATE, SINGBOX_TEMPLATE.read_text(encoding="utf-8"))
        plan = plan_singbox_proxy_runtime(document, nodes[0], pool_nodes=nodes)

        self.assertTrue(plan.is_hybrid)
        self.assertGreater(plan.xray_sidecar.api_port, 0)
        self.assertEqual(set(plan.selector_tags or {}), {nodes[0].id})
        sidecar = plan.xray_sidecar.config
        self.assertEqual(sidecar["routing"]["balancers"][0]["tag"], XRAY_BALANCER_TAG)
        self.assertIn("RoutingService", sidecar["api"]["services"])
        self.assertIn("HandlerService", sidecar["api"]["services"])
        selector = next(
            item for item in plan.singbox_config["outbounds"] if item.get("tag") == "proxy"
        )
        self.assertEqual(selector["type"], "selector")
        self.assertEqual(tuple(selector["outbounds"]), plan.hybrid_relay_selector_tags)
        self.assertTrue(selector["interrupt_exist_connections"])

    def test_hybrid_signature_is_stable_before_and_after_session_capture(self) -> None:
        nodes = [
            parse_single(
                "vless://11111111-1111-1111-1111-111111111111@one.example:443"
                "?type=xhttp&security=tls&sni=one.example&path=%2Fapi#xhttp"
            ),
            parse_single("trojan://secret@two.example:443?security=tls&sni=two.example#trojan"),
        ]

        class Controller:
            def __init__(self):
                settings = AppSettings()
                settings.proxy_engine = "singbox"
                self.state = SimpleNamespace(
                    settings=settings,
                    routing=RoutingSettings(),
                    nodes=nodes,
                    selected_node_id=nodes[0].id,
                )
                self.selected_node = nodes[0]
                self._active_session = None

            @staticmethod
            def is_singbox_editor_mode(_settings=None):
                return True

            @staticmethod
            def is_singbox_tun_mode(_settings=None):
                return False

            @staticmethod
            def is_singbox_proxy_mode(_settings=None):
                return True

            @staticmethod
            def _inspect_active_singbox_config():
                return SINGBOX_TEMPLATE, "config-hash", True

            def xray_outbound_pool(self):
                return build_xray_outbound_pool(self.state.nodes)

        controller = Controller()
        before = transition_signature(controller)
        controller._active_session = SimpleNamespace(
            hybrid=True,
            outbound_pool_tags=controller.xray_outbound_pool().tags,
        )
        after = transition_signature(controller)

        self.assertEqual(before, after)


class SelectorApiTests(unittest.TestCase):
    def test_url_is_loopback_and_tag_is_escaped(self) -> None:
        self.assertEqual(build_selector_url(19090, "proxy group"), "http://127.0.0.1:19090/proxies/proxy%20group")

    @patch("xray_fluent.engines.singbox.selector_api.build_opener")
    def test_puts_exact_target_without_system_proxy(self, build_opener_mock: Mock) -> None:
        response = Mock(status=204)
        response.__enter__ = Mock(return_value=response)
        response.__exit__ = Mock(return_value=False)
        opener = build_opener_mock.return_value
        opener.open.return_value = response

        ok, message = select_outbound(19090, "proxy", "provider/node")

        self.assertTrue(ok, message)
        request = opener.open.call_args.args[0]
        self.assertEqual(request.method, "PUT")
        self.assertEqual(request.data, b'{"name": "provider/node"}')

    @patch("xray_fluent.engines.singbox.selector_api.build_opener")
    def test_cold_start_waits_for_control_plane_listener(self, build_opener_mock: Mock) -> None:
        from urllib.error import URLError

        response = Mock(status=204)
        response.__enter__ = Mock(return_value=response)
        response.__exit__ = Mock(return_value=False)
        opener = build_opener_mock.return_value
        opener.open.side_effect = [
            URLError("connection refused"),
            URLError("connection refused"),
            response,
        ]
        waits: list[float] = []

        ok, message = select_outbound_when_ready(
            19090,
            "proxy",
            "provider/node",
            timeout_sec=1.0,
            wait=waits.append,
        )

        self.assertTrue(ok, message)
        self.assertEqual(opener.open.call_count, 3)
        self.assertEqual(len(waits), 2)


class ManualSelectionTests(unittest.TestCase):
    def _controller(self, hot_result: bool):
        nodes = xray_nodes()
        controller = Mock()
        controller.state.selected_node_id = nodes[0].id
        controller.state.nodes = nodes
        controller.selected_node = nodes[1]
        controller.connected = True
        controller._desired_connected = True
        controller._try_hot_switch_selected_node.return_value = hot_result
        return controller, nodes

    def test_successful_core_switch_skips_transition_queue(self) -> None:
        controller, nodes = self._controller(True)
        set_selected_node(controller, nodes[1].id, source=SelectionSource.NODES_MENU)
        controller._try_hot_switch_selected_node.assert_called_once_with()
        controller._request_transition.assert_not_called()

    def test_rejected_core_switch_falls_back_to_transition(self) -> None:
        controller, nodes = self._controller(False)
        set_selected_node(controller, nodes[1].id, source=SelectionSource.NODES_MENU)
        controller._request_transition.assert_called_once_with("node switched")


def run_hot_switch_generator(controller):
    """Deliberate async update (hot-switch-hardening, AC7/C1).

    ``_try_hot_switch_selected_node`` больше не выполняет control-plane I/O
    синхронно: проверки — в ``_hot_switch_precheck``, вызовы ядра — воркер-шаги
    генератора ``_hot_switch_selected_node_steps``. Семантические контракты
    (порядок xray→singbox, откат, коммит сессии только после подтверждения)
    проверяются прогоном генератора с замоканными control-plane шагами; сами
    шаги замоканы, поэтому генератор обязан завершиться без единого yield.
    """
    plan = AppController._hot_switch_precheck(controller)
    if plan is None:
        return False
    generator = AppController._hot_switch_selected_node_steps(controller, plan)
    try:
        step = next(generator)
    except StopIteration as stop:
        return stop.value
    raise AssertionError(f"unexpected async step from mocked hot switch: {step!r}")


class LiveConnectionCutoverTests(unittest.TestCase):
    def _controller(self, *, hybrid: bool, apply_results: list[bool] | None = None):
        nodes = xray_nodes()
        tags = {nodes[0].id: "old-tag", nodes[1].id: "new-tag"}
        session = SimpleNamespace(
            node_id=nodes[0].id,
            active_core="singbox" if hybrid else "xray",
            hybrid=hybrid,
            tun_mode=False,
            outbound_pool_tags=tags,
            hybrid_relay_selector_tags=("relay-a", "relay-b") if hybrid else (),
            hybrid_relay_selected_tag="relay-a" if hybrid else "",
        )
        controller = Mock()
        controller.selected_node = nodes[1]
        controller._active_session = session
        controller.connected = True
        controller.bypass.ready_for.return_value = True
        controller.bypass.verify_switch_target_steps = steps_via(lambda _node: True)

        apply_calls: list[unittest.mock._Call] = []
        results = list(apply_results) if apply_results is not None else None

        def apply_core_outbound_tag_steps(core: str, outbound_tag: str):
            apply_calls.append(unittest.mock.call(core, outbound_tag))
            return True if results is None else results.pop(0)
            yield  # unreachable — сохраняет генераторную форму продуктового метода

        controller._apply_core_outbound_tag_steps = apply_core_outbound_tag_steps
        controller._apply_core_calls = apply_calls
        controller._capture_hot_switched_session = Mock()
        return controller, nodes, tags, session

    def test_hybrid_switch_changes_xray_then_interrupts_old_singbox_generation(self) -> None:
        controller, nodes, tags, session = self._controller(hybrid=True)

        self.assertTrue(run_hot_switch_generator(controller))

        self.assertEqual(
            controller._apply_core_calls,
            [
                unittest.mock.call("xray", "new-tag"),
                unittest.mock.call("singbox", "relay-b"),
            ],
        )
        controller._capture_hot_switched_session.assert_called_once_with(
            nodes[1],
            session,
            tags,
            "new-tag",
            hybrid_relay_selected_tag="relay-b",
        )

    def test_hybrid_selector_failure_restores_previous_xray_outbound(self) -> None:
        controller, _nodes, _tags, _session = self._controller(
            hybrid=True,
            apply_results=[True, False, True],
        )

        self.assertFalse(run_hot_switch_generator(controller))

        self.assertEqual(
            controller._apply_core_calls,
            [
                unittest.mock.call("xray", "new-tag"),
                unittest.mock.call("singbox", "relay-b"),
                unittest.mock.call("xray", "old-tag"),
            ],
        )
        controller._capture_hot_switched_session.assert_not_called()

    def test_xray_only_does_not_report_cutover_without_connection_interrupt_api(self) -> None:
        controller, _nodes, _tags, _session = self._controller(hybrid=False)

        self.assertFalse(run_hot_switch_generator(controller))

        self.assertEqual(controller._apply_core_calls, [])
        controller._capture_hot_switched_session.assert_not_called()

    def test_native_singbox_tun_session_switches_by_full_transition(self) -> None:
        controller, _nodes, _tags, _session = self._controller(hybrid=False)
        controller._active_session.active_core = "singbox"
        controller._active_session.tun_mode = True
        self.assertIsNone(AppController._hot_switch_precheck(controller))

    def test_hybrid_tun_session_keeps_the_hot_switch(self) -> None:
        controller, _nodes, _tags, _session = self._controller(hybrid=True)
        controller._active_session.tun_mode = True
        self.assertIsNotNone(AppController._hot_switch_precheck(controller))

    def test_hot_switch_needs_the_zapret_rule_of_the_new_node(self) -> None:
        controller, nodes, _tags, _session = self._controller(hybrid=True)
        controller.bypass.ready_for.return_value = False

        self.assertFalse(run_hot_switch_generator(controller))

        controller.bypass.ready_for.assert_called_with(nodes[1])
        self.assertEqual(controller._apply_core_calls, [])
        controller._capture_hot_switched_session.assert_not_called()


class ProxyProtectionTransitionTests(unittest.TestCase):
    """The transition queue stays shut while the Zapret gate waits."""

    def test_drain_is_held_while_the_gate_waits_for_this_generation(self) -> None:
        controller = Mock()
        controller._transition_active = False
        controller._transition_generation = 11
        controller.bypass.waiting_for.return_value = True

        AppController._schedule_transition_drain(controller, 0)

        controller.bypass.waiting_for.assert_called_once_with(11)
        controller._transition_timer.start.assert_not_called()

    def test_new_request_releases_a_stale_wait(self) -> None:
        controller = Mock()
        controller._transition_active = True
        controller._transition_generation = 3

        AppController._request_transition(controller, "node switched")

        controller.bypass.cancel_wait.assert_called_once_with()
        self.assertEqual(controller._transition_generation, 4)


class HybridRuntimeStartupTests(unittest.TestCase):
    def test_stale_hysteria_state_callback_cannot_clear_new_generation(self) -> None:
        controller = Mock()
        current = Mock()
        current.process_generation = 8
        controller.hysteria = current
        controller._hysteria_active_generation = 9

        AppController._on_hysteria_state_changed(controller, current, False)

        controller._on_core_state_changed.assert_not_called()
        self.assertIn("ignored stale state callback", controller._log.call_args.args[0])

    def test_security_failure_stops_admission_without_tls_downgrade_or_switch(self) -> None:
        controller = Mock()
        manager = Mock()
        manager.process_generation = 4
        controller.hysteria = manager
        controller._hysteria_active_generation = 4
        controller._hysteria_contract.session.session_generation = 12
        controller.connected = True
        controller._disconnecting = False
        controller._desired_connected = True
        controller.singbox.is_running = True

        AppController._on_hysteria_failure(
            controller,
            manager,
            HysteriaFailureCode.TARGET_PIN_MISMATCH.value,
            "certificate pin mismatch",
            4,
        )

        # Admission closes without a GUI-thread wait: kill now, exit arrives as a signal.
        controller.singbox.request_stop.assert_called_once_with(expected=True)
        controller.singbox.stop.assert_not_called()
        controller._request_transition.assert_not_called()
        controller._try_hot_switch_selected_node.assert_not_called()
        self.assertFalse(controller._desired_connected)

    def test_running_manager_security_log_reaches_controller_and_closes_admission(self) -> None:
        controller = Mock()
        manager = HysteriaManager()
        manager._running = True
        manager._failure_reported = False
        manager._process_generation = 4
        controller.hysteria = manager
        controller._hysteria_active_generation = 4
        controller._hysteria_contract.session.session_generation = 12
        controller.connected = True
        controller._disconnecting = False
        controller._desired_connected = True
        controller.singbox.is_running = True
        manager.failure.connect(
            lambda code, message, generation: AppController._on_hysteria_failure(
                controller,
                manager,
                code,
                message,
                generation,
            )
        )

        manager._emit_process_line(
            "outbound connection failed: x509: certificate signed by unknown authority"
        )

        # Admission closes without a GUI-thread wait: kill now, exit arrives as a signal.
        controller.singbox.request_stop.assert_called_once_with(expected=True)
        controller.singbox.stop.assert_not_called()
        controller._request_transition.assert_not_called()
        self.assertFalse(controller._desired_connected)

    def test_official_core_pin_mismatch_line_closes_admission_without_switch(self) -> None:
        # Регресс-щит инцидента 31.08–01.09: строка официального ядра про pin
        # не распознавалась, деградировала до TARGET_NETWORK_TIMEOUT и вызывала
        # тихое авто-переключение вместо fail-closed остановки.
        controller = Mock()
        manager = HysteriaManager()
        manager._running = True
        manager._failure_reported = False
        manager._process_generation = 7
        controller.hysteria = manager
        controller._hysteria_active_generation = 7
        controller._hysteria_contract.session.session_generation = 12
        controller.connected = True
        controller._disconnecting = False
        controller._desired_connected = True
        controller.singbox.is_running = True
        manager.failure.connect(
            lambda code, message, generation: AppController._on_hysteria_failure(
                controller,
                manager,
                code,
                message,
                generation,
            )
        )

        manager._emit_process_line(
            "client connect error: CRYPTO_ERROR 0x12a (local): "
            "no certificate matches the pinned hash"
        )

        self.assertEqual(HysteriaFailureCode.TARGET_PIN_MISMATCH, manager._last_failure_code)
        # Admission closes without a GUI-thread wait: kill now, exit arrives as a signal.
        controller.singbox.request_stop.assert_called_once_with(expected=True)
        controller.singbox.stop.assert_not_called()
        controller._request_transition.assert_not_called()
        controller._try_hot_switch_selected_node.assert_not_called()
        self.assertFalse(controller._desired_connected)

    def test_network_failure_selects_one_other_target_once(self) -> None:
        current = parse_single("hy2://current@one.example:443/#current")
        first = parse_single("hy2://first@two.example:443/#first")
        second = parse_single("hy2://second@three.example:443/#second")
        controller = Mock()
        manager = Mock()
        manager.process_generation = 6
        controller.hysteria = manager
        controller._hysteria_active_generation = 6
        controller._hysteria_contract.session.session_generation = 20
        controller.connected = True
        controller._disconnecting = False
        controller._hysteria_recovery_active = False
        controller._hysteria_cooldown_until = {}
        controller._hysteria_failure_episode_id = 0
        controller.selected_node = current
        controller.state = SimpleNamespace(
            nodes=[current, first, second],
            selected_node_id=current.id,
        )
        controller._pending_transport_node_id = None
        controller.singbox.is_running = True

        for _ in range(2):
            AppController._on_hysteria_failure(
                controller,
                manager,
                HysteriaFailureCode.TARGET_NETWORK_TIMEOUT.value,
                "no recent network activity",
                6,
            )

        self.assertEqual(controller.state.selected_node_id, current.id)
        self.assertEqual(controller._pending_transport_node_id, first.id)
        controller._request_transition.assert_called_once_with("node switched")
        controller._try_hot_switch_selected_node.assert_not_called()
        controller.auto_switch_triggered.emit.assert_called_once_with(first.name)
        self.assertEqual(controller._hysteria_failure_episode_id, 1)

    def test_start_pins_xray_target_and_known_relay_generation(self) -> None:
        sidecar = SimpleNamespace(
            protect_port=19084,
            protect_password="protect",
            relay_port=19080,
            api_port=19085,
            config={"sidecar": True},
        )
        plan = SimpleNamespace(
            provider_payload=None,
            xray_sidecar=sidecar,
            hysteria_sidecar=None,
            selected_outbound_tag="node-tag",
            clash_api_port=19090,
            singbox_config={"main": True},
            is_hybrid=True,
            hybrid_relay_selected_tag="relay-a",
        )
        controller = bridge_controller(Mock())
        controller.state.settings.xray_path = "xray.exe"
        controller.state.settings.singbox_path = "sing-box.exe"
        controller.xray.start.return_value = True
        controller.xray.is_running = True
        controller.singbox.start.return_value = True
        controller._apply_core_outbound_tag.return_value = True
        controller._rejected_xray_tags.return_value = []

        self.assertTrue(drive(AppController._start_singbox_runtime_plan_steps(controller, plan)))

        self.assertEqual(
            controller._apply_core_outbound_tag.call_args_list,
            [
                unittest.mock.call("xray", "node-tag"),
                unittest.mock.call("singbox", "relay-a", startup=True),
            ],
        )
        self.assertEqual(controller._xray_api_port, 19085)
        self.assertEqual(controller._singbox_clash_api_port, 19090)

    def test_start_restarts_singbox_once_when_cold_selector_pin_fails(self) -> None:
        plan = SimpleNamespace(
            provider_payload=None,
            xray_sidecar=None,
            hysteria_sidecar=None,
            selected_outbound_tag="__app_nodes/node-one",
            clash_api_port=19090,
            singbox_config={"main": True},
            is_hybrid=False,
            hybrid_relay_selected_tag="",
        )
        controller = bridge_controller(Mock())
        controller.state.settings.singbox_path = "sing-box.exe"
        controller.singbox.start.return_value = True
        controller._apply_core_outbound_tag.side_effect = [False, True]
        sleeps: list[int] = []

        self.assertTrue(drive(AppController._start_singbox_runtime_plan_steps(controller, plan), sleeps=sleeps))

        self.assertEqual(controller.singbox.start.call_count, 2)
        controller.singbox.stop.assert_called_once()
        # Пауза перед повтором — таймерный шаг, не сон GUI-потока.
        self.assertEqual(sleeps, [500])
        self.assertEqual(
            controller._apply_core_outbound_tag.call_args_list,
            [
                unittest.mock.call("singbox", "__app_nodes/node-one", startup=True),
                unittest.mock.call("singbox", "__app_nodes/node-one", startup=True),
            ],
        )

    def test_start_retries_manager_readiness_timeout_once(self) -> None:
        plan = SimpleNamespace(
            provider_payload=None,
            xray_sidecar=None,
            hysteria_sidecar=None,
            selected_outbound_tag="__app_nodes/node-one",
            clash_api_port=19090,
            singbox_config={"main": True},
            is_hybrid=False,
            hybrid_relay_selected_tag="",
        )
        controller = bridge_controller(Mock())
        controller.state.settings.singbox_path = "sing-box.exe"
        controller.singbox.start.side_effect = [False, True]
        controller.singbox.last_start_failure_retryable = True
        controller._apply_core_outbound_tag.return_value = True
        sleeps: list[int] = []

        self.assertTrue(drive(AppController._start_singbox_runtime_plan_steps(controller, plan), sleeps=sleeps))

        self.assertEqual(controller.singbox.start.call_count, 2)
        controller.singbox.stop.assert_not_called()
        self.assertEqual(sleeps, [500])

    def test_sync_wrapper_rejects_a_non_generator_instead_of_spinning(self) -> None:
        # Регресс: Mock-контроллер отдавал в run_steps_blocking Mock вместо
        # генератора, Mock.send() бесконечно порождал новые «шаги», и память
        # росла без предела. Драйвер обязан падать сразу.
        with self.assertRaises(TypeError):
            AppController._start_singbox_runtime_plan(Mock(), SimpleNamespace())


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