zapret-kvn/tests/test_auto_switch_dead_link.py

"""Dead-link detection in the auto-switch service.

Auto-switch leaves a server only when it is dead: the metrics worker's
link verdict (link_alive) — requests unanswered, then one confirmation
through the tunnel — switches away while no payload traffic flows; nothing
is sent to the server without that signal. A slow but working server is never
switched, and UDP/QUIC protocols explicitly avoid the TCP verdict.
"""

from __future__ import annotations

import ctypes
import sys
import unittest
from unittest.mock import patch

from xray_fluent.application.selection_source import SelectionSource
from xray_fluent.application import auto_switch_service
from xray_fluent.application.auto_switch_service import (
    AUTO_SWITCH_DEAD_LINK_SEC,
    check_auto_switch,
)
from xray_fluent.profiles.models import AppSettings, Node


class _Recorder:
    def __init__(self):
        self.calls: list = []

    def emit(self, *args) -> None:
        self.calls.append(args)


class _State:
    def __init__(self, nodes):
        self.settings = AppSettings()
        self.nodes = nodes
        self.selected_node_id = nodes[0].id if nodes else ""


class FakeController:
    """Bare attribute bag: the service only touches state and signals."""

    def __init__(self, node_count: int = 3):
        nodes = [
            Node(id=f"n{i}", name=f"node-{i}", server=f"s{i}.example.com", port=443, scheme="vless")
            for i in range(node_count)
        ]
        self.state = _State(nodes)
        self.connected = True
        self._switching = False
        self._reconnecting = False
        self._auto_switch_last_switch = 0.0
        self._auto_switch_cycle_attempts = 0
        self._auto_switch_exhausted = False
        self._auto_switch_transitioning = False
        self._auto_switch_link_down_since = 0.0
        self._auto_switch_manual_hold = False
        self._auto_switch_warmup_until = 0.0
        self._transition_active = False
        self._transition_pending = False
        self._connecting = False
        self._disconnecting = False
        self.status = _Recorder()
        self.auto_switch_triggered = _Recorder()
        self.logs: list[str] = []
        self.selected: list[tuple] = []

    def _log(self, message: str) -> None:
        self.logs.append(message)

    @property
    def selected_node(self):
        return next(
            (node for node in self.state.nodes if node.id == self.state.selected_node_id),
            None,
        )

    def set_selected_node(self, node_id: str, *, source: SelectionSource) -> None:
        self.selected.append((node_id, source))


def _tick(
    controller,
    at: float,
    *,
    down_bps: float = 0.0,
    link_alive=None,
    traffic_valid: bool = True,
) -> None:
    with patch.object(auto_switch_service.time, "monotonic", return_value=at):
        check_auto_switch(
            controller,
            down_bps,
            link_alive,
            traffic_valid=traffic_valid,
        )


def _hysteria_controller() -> FakeController:
    controller = FakeController()
    from xray_fluent.importer.link_parser import parse_single

    node = parse_single(
        "hysteria2://secret@udp.example:443/?obfs=salamander"
        "&obfs-password=cover&sni=udp.example#Hysteria"
    )
    node.id = "hysteria-active"
    controller.state.nodes[0] = node
    controller.state.selected_node_id = node.id
    return controller


class DeadLinkTriggerTests(unittest.TestCase):
    def test_sustained_ping_failures_trigger_switch(self) -> None:
        controller = FakeController()
        _tick(controller, 100.0, link_alive=False)          # arm
        _tick(controller, 100.0 + AUTO_SWITCH_DEAD_LINK_SEC - 1, link_alive=False)
        self.assertEqual(controller.selected, [])            # not yet
        _tick(controller, 100.0 + AUTO_SWITCH_DEAD_LINK_SEC + 1, link_alive=False)
        self.assertEqual(len(controller.selected), 1)
        node_id, source = controller.selected[0]
        self.assertNotEqual(node_id, controller.state.selected_node_id)
        self.assertIs(source, SelectionSource.AUTO_SWITCH)   # автоматика: анти-дребезг сохранён
        self.assertTrue(any("unreachable" in line for line in controller.logs))
        self.assertTrue(controller._auto_switch_transitioning)

    def test_successful_ping_resets_the_window(self) -> None:
        controller = FakeController()
        _tick(controller, 100.0, link_alive=False)
        _tick(controller, 110.0, link_alive=True)            # link recovered
        _tick(controller, 100.0 + AUTO_SWITCH_DEAD_LINK_SEC + 5, link_alive=False)
        # Window restarted at the last failure, so no switch yet.
        self.assertEqual(controller.selected, [])

    def test_traffic_flow_blocks_dead_verdict(self) -> None:
        # Ping fails but payload bytes still arrive: not a dead server.
        controller = FakeController()
        _tick(controller, 100.0, down_bps=500 * 1024.0, link_alive=False)
        _tick(controller, 200.0, down_bps=500 * 1024.0, link_alive=False)
        self.assertEqual(controller.selected, [])
        self.assertEqual(controller._auto_switch_link_down_since, 0.0)

    def test_no_ping_configured_never_triggers(self) -> None:
        controller = FakeController()
        _tick(controller, 100.0, link_alive=None)
        _tick(controller, 100.0 + AUTO_SWITCH_DEAD_LINK_SEC * 10, link_alive=None)
        self.assertEqual(controller.selected, [])

    def test_cooldown_defers_the_switch(self) -> None:
        controller = FakeController()
        controller._auto_switch_last_switch = 95.0           # just switched
        _tick(controller, 100.0, link_alive=False)
        at = 100.0 + AUTO_SWITCH_DEAD_LINK_SEC + 1
        _tick(controller, at, link_alive=False)
        self.assertEqual(controller.selected, [])            # cooldown holds
        cooled = 95.0 + controller.state.settings.auto_switch_cooldown_sec + 1
        _tick(controller, max(at, cooled), link_alive=False)
        self.assertEqual(len(controller.selected), 1)

    def test_exhaustion_applies_to_dead_link_path(self) -> None:
        controller = FakeController(node_count=2)
        controller._auto_switch_cycle_attempts = 1           # max_attempts == 1
        _tick(controller, 100.0, link_alive=False)
        _tick(controller, 100.0 + AUTO_SWITCH_DEAD_LINK_SEC + 1, link_alive=False)
        self.assertEqual(controller.selected, [])
        self.assertTrue(controller._auto_switch_exhausted)
        self.assertEqual(len(controller.status.calls), 1)

    def test_disabled_setting_wins(self) -> None:
        controller = FakeController()
        controller.state.settings.auto_switch_enabled = False
        _tick(controller, 100.0, link_alive=False)
        _tick(controller, 100.0 + AUTO_SWITCH_DEAD_LINK_SEC + 1, link_alive=False)
        self.assertEqual(controller.selected, [])

    def test_speed_drop_on_a_live_server_never_switches(self) -> None:
        # A finished download followed by a trickle of traffic is not a dead
        # server: this used to read as degradation and drop a healthy node.
        controller = FakeController()
        for i in range(10):
            _tick(controller, 100.0 + i, down_bps=200 * 1024.0, link_alive=True)
        for i in range(0, 600, 5):
            _tick(controller, 111.0 + i, down_bps=10 * 1024.0, link_alive=True)
        self.assertEqual(controller.selected, [])
        self.assertEqual(controller.auto_switch_triggered.calls, [])

    def test_hysteria_tcp_failure_never_uses_dead_link_fallback(self) -> None:
        controller = _hysteria_controller()

        _tick(controller, 100.0, link_alive=False)
        _tick(controller, 100.0 + AUTO_SWITCH_DEAD_LINK_SEC * 2, link_alive=False)

        self.assertEqual(controller.selected, [])
        self.assertEqual(controller._auto_switch_link_down_since, 0.0)

    def test_invalid_traffic_sample_is_not_treated_as_zero_speed(self) -> None:
        controller = FakeController()

        _tick(controller, 100.0, down_bps=0.0, link_alive=False, traffic_valid=False)
        _tick(
            controller,
            100.0 + AUTO_SWITCH_DEAD_LINK_SEC * 2,
            down_bps=0.0,
            link_alive=False,
            traffic_valid=False,
        )

        self.assertEqual(controller.selected, [])
        self.assertEqual(controller._auto_switch_link_down_since, 0.0)

    def test_hysteria_slow_traffic_never_switches(self) -> None:
        controller = _hysteria_controller()
        controller.state.settings.auto_switch_cooldown_sec = 1
        base = 1000.0

        for offset in range(10):
            _tick(controller, base + offset, down_bps=200 * 1024.0, link_alive=False)
        for offset in range(0, 600, 5):
            _tick(controller, base + 10 + offset, down_bps=10 * 1024.0, link_alive=False)

        self.assertEqual(controller.selected, [])

    def test_manual_hold_and_transition_guard_block_switch(self) -> None:
        controller = FakeController()
        controller._auto_switch_manual_hold = True
        _tick(controller, 100.0, link_alive=False)
        _tick(controller, 100.0 + AUTO_SWITCH_DEAD_LINK_SEC + 1, link_alive=False)
        self.assertEqual(controller.selected, [])

        controller._auto_switch_manual_hold = False
        controller._transition_active = True
        _tick(controller, 200.0, link_alive=False)
        _tick(controller, 200.0 + AUTO_SWITCH_DEAD_LINK_SEC + 1, link_alive=False)
        self.assertEqual(controller.selected, [])

    def test_startup_warmup_blocks_dead_link_until_runtime_is_observed(self) -> None:
        controller = FakeController()
        controller._auto_switch_warmup_until = 200.0

        _tick(controller, 100.0, link_alive=False)
        _tick(controller, 199.0, link_alive=False)
        self.assertEqual(controller.selected, [])

        _tick(controller, 200.0, link_alive=False)  # warmup has just ended
        _tick(controller, 200.0 + AUTO_SWITCH_DEAD_LINK_SEC + 1, link_alive=False)
        self.assertEqual(len(controller.selected), 1)


class MetricsWorkerPingTargetTests(unittest.TestCase):
    """set_ping_target re-points the surviving worker after a hot-switch."""

    def _worker_class(self):
        if sys.platform == "win32":
            from xray_fluent.diagnostics.live_metrics_worker import LiveMetricsWorker
            return LiveMetricsWorker
        original_windll = getattr(ctypes, "windll", None)

        class _AnyLib:
            def __getattr__(self, name):
                lib = _AnyLib()
                setattr(self, name, lib)
                return lib

            def __call__(self, *args, **kwargs):
                return 0

        ctypes.windll = _AnyLib()  # type: ignore[attr-defined]
        try:
            from xray_fluent.diagnostics.live_metrics_worker import LiveMetricsWorker
        finally:
            if original_windll is None:
                del ctypes.windll
            else:
                ctypes.windll = original_windll  # type: ignore[attr-defined]
        return LiveMetricsWorker

    def test_set_ping_target_repoints_and_forces_probe(self) -> None:
        worker_class = self._worker_class()
        worker = worker_class("xray.exe", 0, ping_host="old.example.com", ping_port=443)
        worker._last_ping_ms = 42
        worker._last_ping_ts = 1234.5
        worker._link_verdict = False
        worker.set_ping_target("new.example.com", 8443)
        self.assertEqual(worker._ping_host, "new.example.com")
        self.assertEqual(worker._ping_port, 8443)
        self.assertIsNone(worker._last_ping_ms)     # stale value cleared
        self.assertIsNone(worker._link_verdict)     # verdict of the old server dropped
        self.assertEqual(worker._last_ping_ts, 0.0)  # next tick reads now
        self.assertTrue(worker.pings_active_node())

    def _watching(self, **kwargs):
        worker_class = self._worker_class()
        module = sys.modules[worker_class.__module__]
        reads: list[tuple] = []
        confirms: list[int] = []
        verdict = kwargs.pop("verdict", False)

        def reader(addresses, port):
            reads.append((addresses, port))
            return 42

        def confirmer():
            confirms.append(1)
            return verdict

        worker = worker_class(
            "xray.exe", 0, ping_host="203.0.113.10", ping_port=443,
            rtt_reader=reader, link_confirmer=confirmer,
        )
        return module, worker, reads, confirms

    @staticmethod
    def _settle(worker) -> None:
        import time

        deadline = time.monotonic() + 2.0
        while worker._confirming and time.monotonic() < deadline:
            time.sleep(0.01)

    def test_latency_is_read_passively_from_the_core_connection(self) -> None:
        # Незаметность: задержка — RTT уже открытого соединения ядра, ни одного
        # своего соединения к серверу.
        module, worker, reads, confirms = self._watching()
        worker._watch_link(100.0, 0.0, 0.0)
        worker._watch_link(100.5, 0.0, 0.0)  # чаще RTT_READ_INTERVAL_SEC не читаем
        worker._watch_link(100.0 + module.RTT_READ_INTERVAL_SEC, 0.0, 0.0)
        self.assertEqual(reads, [(("203.0.113.10",), 443)] * 2)
        self.assertEqual(worker._last_ping_ms, 42)
        self.assertEqual(confirms, [])
        self.assertIsNone(worker._link_verdict)

    def test_last_rtt_is_kept_while_the_core_has_no_connection(self) -> None:
        module, worker, _reads, _confirms = self._watching()
        values = iter([42, None, None])
        worker._rtt_reader = lambda _a, _p: next(values)
        worker._watch_link(100.0, 0.0, 0.0)
        worker._watch_link(100.0 + module.RTT_READ_INTERVAL_SEC, 0.0, 0.0)
        self.assertEqual(worker._last_ping_ms, 42)
        worker._watch_link(100.0 + module.RTT_KEEP_SEC + 5, 0.0, 0.0)
        self.assertIsNone(worker._last_ping_ms)

    def test_idle_link_is_never_probed(self) -> None:
        module, worker, _reads, confirms = self._watching()
        for second in range(600):
            worker._watch_link(100.0 + second, 0.0, 0.0)
        self.assertEqual(confirms, [])
        self.assertIsNone(worker._link_verdict)

    def test_unanswered_requests_trigger_one_confirmation(self) -> None:
        from xray_fluent.application.auto_switch_service import AUTO_SWITCH_DEAD_LINK_SEC

        module, worker, _reads, confirms = self._watching(verdict=False)
        up, down = module.PING_UNANSWERED_UP_BPS * 2, 0.0
        now = 100.0
        while now < 100.0 + module.LINK_SUSPECT_SEC + 1:
            worker._watch_link(now, down, up)
            now += 1.0
        self._settle(worker)
        self.assertEqual(confirms, [1])
        self.assertIs(worker._link_verdict, False)
        # Пока подозрение держится, повтор — не чаще кулдауна.
        for _ in range(int(module.LINK_CONFIRM_COOLDOWN_SEC) - 2):
            worker._watch_link(now, down, up)
            now += 1.0
        self._settle(worker)
        self.assertEqual(confirms, [1])
        # Подозрение + подтверждение укладываются в разумное окно до переключения.
        self.assertLessEqual(
            module.LINK_SUSPECT_SEC + module.LINK_CONFIRM_TIMEOUT_SEC + AUTO_SWITCH_DEAD_LINK_SEC, 35.0
        )

    def test_answered_traffic_clears_the_verdict(self) -> None:
        module, worker, _reads, confirms = self._watching()
        worker._link_verdict = False
        worker._watch_link(100.0, module.PING_TRAFFIC_BPS * 4, 1000.0)
        self.assertIs(worker._link_verdict, True)
        self.assertEqual(confirms, [])

    def test_confirmation_is_journaled_as_current_server_check(self) -> None:
        from xray_fluent.network import probe_gate

        module, worker, _reads, _confirms = self._watching()
        gate = probe_gate.ProbeGate()
        with patch.object(probe_gate, "GATE", gate):
            up = module.PING_UNANSWERED_UP_BPS * 2
            for second in range(int(module.LINK_SUSPECT_SEC) + 2):
                worker._watch_link(100.0 + second, 0.0, up)
            self._settle(worker)
        self.assertEqual([(r.kind, r.target) for r in gate.journal()], [(probe_gate.CURRENT, "203.0.113.10")])

    def test_udp_transport_does_not_start_tcp_probe(self) -> None:
        worker_class = self._worker_class()
        worker = worker_class(
            "xray.exe",
            0,
            ping_host="udp.example.com",
            ping_port=443,
            transport_kind="udp",
        )

        self.assertFalse(worker.pings_active_node())
        worker.set_ping_target("tcp.example.com", 443, "tcp")
        self.assertTrue(worker.pings_active_node())

    def test_pings_active_node_requires_target(self) -> None:
        worker_class = self._worker_class()
        worker = worker_class("xray.exe", 0)
        self.assertFalse(worker.pings_active_node())


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