"""用真实的 Nginx 验证网关配置：反向代理、SPA 回退、SSE 不缓冲、限流、Cookie 灰度。

运行：python -m pytest -q test_gateway.py（需要本机安装 nginx；没有安装时自动跳过）
测试会启动两个 Python 后端（稳定版 / 灰度版）和一个非 root 的 Nginx，全部使用临时端口和临时目录。
"""
import http.client
import shutil
import socket
import subprocess
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

import pytest

NGINX = shutil.which("nginx") or ("/usr/sbin/nginx" if Path("/usr/sbin/nginx").exists() else None)
pytestmark = pytest.mark.skipif(NGINX is None, reason="没有安装 nginx")
HERE = Path(__file__).resolve().parent


def free_port() -> int:
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


def backend(version: str) -> ThreadingHTTPServer:
    class H(BaseHTTPRequestHandler):
        protocol_version = "HTTP/1.1"

        def log_message(self, *a):  # 安静
            pass

        def do_GET(self):
            if self.path.startswith("/ai/stream"):
                self.send_response(200)
                self.send_header("Content-Type", "text/event-stream")
                self.send_header("Cache-Control", "no-cache")
                self.send_header("Connection", "close")
                self.end_headers()
                for i in range(3):
                    self.wfile.write(f"data: token{i}\n\n".encode())
                    self.wfile.flush()
                    time.sleep(0.3)
                self.close_connection = True
                return
            body = f'{{"version":"{version}","xff":"{self.headers.get("X-Forwarded-For", "")}"}}'.encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

    srv = ThreadingHTTPServer(("127.0.0.1", 0), H)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    return srv


@pytest.fixture(scope="module")
def gateway(tmp_path_factory):
    prefix = tmp_path_factory.mktemp("nginx")
    (prefix / "html").mkdir()
    (prefix / "html" / "index.html").write_text("<div id=app>SPA</div>")
    stable, canary = backend("stable"), backend("canary")
    port = free_port()
    conf = (HERE / "nginx.conf.template").read_text()
    for k, v in {"PREFIX": str(prefix), "STABLE_PORT": stable.server_address[1],
                 "CANARY_PORT": canary.server_address[1], "LISTEN_PORT": port}.items():
        conf = conf.replace("{{" + k + "}}", str(v))
    (prefix / "nginx.conf").write_text(conf)
    check = subprocess.run([NGINX, "-t", "-p", str(prefix), "-c", str(prefix / "nginx.conf")],
                           capture_output=True, text=True)
    assert check.returncode == 0, check.stderr
    proc = subprocess.Popen([NGINX, "-p", str(prefix), "-c", str(prefix / "nginx.conf"), "-g", "daemon off;"],
                            stderr=subprocess.PIPE)
    for _ in range(50):
        try:
            socket.create_connection(("127.0.0.1", port), timeout=0.1).close()
            break
        except OSError:
            time.sleep(0.1)
    yield port
    proc.terminate()
    proc.wait(5)
    stable.shutdown()
    canary.shutdown()


def get(port: int, path: str, headers: dict | None = None) -> tuple[int, str]:
    c = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
    c.request("GET", path, headers=headers or {})
    r = c.getresponse()
    body = r.read().decode()
    c.close()
    return r.status, body


def test_spa_fallback(gateway):
    assert get(gateway, "/orders/42") == (200, "<div id=app>SPA</div>")


def test_proxy_and_canary(gateway):
    status, body = get(gateway, "/api/hello")
    assert status == 200 and '"version":"stable"' in body and '"xff":"127.0.0.1"' in body
    time.sleep(1.2)  # 等限流桶恢复
    status, body = get(gateway, "/api/hello", {"Cookie": "canary=1"})
    assert '"version":"canary"' in body


def test_sse_not_buffered(gateway):
    c = http.client.HTTPConnection("127.0.0.1", gateway, timeout=5)
    start = time.monotonic()
    c.request("GET", "/ai/stream")
    r = c.getresponse()
    first = r.readline()  # 读到第一条事件就返回
    first_at = time.monotonic() - start
    rest = r.read()
    total = time.monotonic() - start
    c.close()
    assert first.startswith(b"data: token0")
    assert b"token2" in rest
    # 不缓冲时，第一条事件远早于整个响应结束（后端共耗时约 0.9 秒）
    assert first_at < 0.5 < total, (first_at, total)


def test_rate_limit(gateway):
    time.sleep(1.5)
    codes = [get(gateway, "/api/hello")[0] for _ in range(20)]
    assert codes.count(200) >= 5 and 429 in codes, codes