from __future__ import annotations

import subprocess
import socket
import os
import shutil
import sys
from pathlib import Path

from pyxray.libs.xray_assets import xray_executable_name
from pyxray.libs.xray_config import XrayConfigSettings, write_transparent_rule_files
from pyxray.libs.xray_runtime import XrayServiceManager
from pyxray.libs.xray_transparent_runtime import TransparentRuntime


def test_transparent_runtime_auto_backend_prefers_iptables_and_starts_local_ip_watcher(tmp_path: Path) -> None:
    settings = XrayConfigSettings()
    settings.transparent.mode = "proxy"
    settings.transparent.type = "redirect"
    write_transparent_rule_files(settings, tmp_path)
    commands: list[list[str]] = []

    runtime = TransparentRuntime(
        transparent_dir=tmp_path,
        log_path=tmp_path / "xray.log",
        backend="auto",
        executor=_executor(commands),
        local_cidrs_provider=lambda: ["198.51.100.10/32"],
        watcher_interval=60,
    )

    runtime.setup(settings)
    runtime.cleanup()

    assert runtime.backend == "iptables"
    assert ["/bin/sh", str(tmp_path / "transparent-iptables-setup.sh")] in commands
    assert ["iptables", "-w", "2", "-t", "nat", "-I", "TP_RULE", "1", "-d", "198.51.100.10/32", "-j", "RETURN"] in commands
    assert ["iptables", "-w", "2", "-t", "nat", "-D", "TP_RULE", "-d", "198.51.100.10/32", "-j", "RETURN"] in commands


def test_transparent_runtime_auto_backend_falls_back_to_nft_when_iptables_setup_fails(tmp_path: Path) -> None:
    settings = XrayConfigSettings()
    settings.transparent.mode = "proxy"
    settings.transparent.type = "redirect"
    write_transparent_rule_files(settings, tmp_path)
    commands: list[list[str]] = []

    runtime = TransparentRuntime(
        transparent_dir=tmp_path,
        log_path=tmp_path / "xray.log",
        backend="auto",
        executor=_executor(commands, failures={"transparent-iptables-setup.sh"}),
        local_cidrs_provider=lambda: [],
    )

    runtime.setup(settings)
    runtime.cleanup()

    assert runtime.backend == "nft"
    assert ["/bin/sh", str(tmp_path / "transparent-iptables-setup.sh")] in commands
    assert ["/bin/sh", str(tmp_path / "transparent-nft-setup.sh")] in commands


def test_xray_service_manager_reports_inbound_port_conflict(tmp_path: Path) -> None:
    _write_fake_xray(tmp_path)
    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    listener.bind(("127.0.0.1", 0))
    listener.listen(1)
    port = listener.getsockname()[1]
    config = tmp_path / "config.json"
    config.write_text(
        f'{{"inbounds":[{{"tag":"conflict-http","protocol":"http","listen":"127.0.0.1","port":{port}}}]}}',
        encoding="utf-8",
    )
    manager = XrayServiceManager(xray_dir=tmp_path, config_path=config, log_path=tmp_path / "xray.log")

    try:
        try:
            manager.start()
        except RuntimeError as exc:
            message = str(exc)
        else:
            raise AssertionError("expected port conflict")
    finally:
        listener.close()

    assert "inbound port conflict" in message
    assert f"conflict-http 127.0.0.1:{port}/tcp" in message
    assert "inbound port conflict" in (tmp_path / "xray.log").read_text(encoding="utf-8")


def _executor(commands: list[list[str]], failures: set[str] | None = None):
    failures = failures or set()

    def execute(command: list[str]) -> subprocess.CompletedProcess[str]:
        commands.append(command)
        script = Path(command[-1]).name
        return subprocess.CompletedProcess(
            args=command,
            returncode=1 if script in failures else 0,
            stdout="",
            stderr=f"{script} failed" if script in failures else "",
        )

    return execute


def _write_fake_xray(directory: Path) -> Path:
    xray = directory / xray_executable_name()
    if os.name == "nt":
        try:
            os.link(sys.executable, xray)
        except OSError:
            shutil.copy2(sys.executable, xray)
        (directory / "run").write_text("import time\ntime.sleep(30)\n", encoding="utf-8")
        return xray
    xray.write_text("#!/bin/sh\nsleep 30\n", encoding="utf-8")
    xray.chmod(0o755)
    return xray