# -*- coding: utf-8 -*-
"""只读 Web 服务测试（离线）: uv run python tests/test_web_server.py"""

import http.client
import json
import os
import sys
import tempfile
import threading
import urllib.parse
from pathlib import Path

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.stdout.reconfigure(encoding="utf-8")

from reaction_board.config import ConfigStore  # noqa: E402
from reaction_board.database import Database  # noqa: E402
from reaction_board.web.server import create_server  # noqa: E402
from tests.web_fixtures import build_db, write_config  # noqa: E402


class RunningServer:
    def __init__(self, tmp: str, *, token: str = "", preview: bool = False):
        self.config = write_config(tmp, token=token)
        build_db(tmp)
        self.db = Database(Path(tmp) / "app.db")
        self.db.initialize()
        if preview:
            self.db.write(lambda conn: conn.execute(
                "INSERT OR REPLACE INTO kv(k, v) VALUES ('preview:seeded', '1')"
            ))
        self.store = ConfigStore(self.config)
        self.server = create_server(
            self.store, self.db, port=0, preview=preview,
            clock=lambda: 1_793_459_000,
        )
        self.port = self.server.server_address[1]
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()

    def request(self, path: str, *, method: str = "GET", headers=None, body=None):
        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5)
        conn.request(method, path, body=body, headers=headers or {})
        response = conn.getresponse()
        payload = response.read()
        result = response.status, dict(response.getheaders()), payload
        conn.close()
        return result

    def json(self, path: str, **kwargs):
        status, headers, payload = self.request(path, **kwargs)
        return status, headers, json.loads(payload.decode("utf-8"))

    def close(self):
        self.server.shutdown()
        self.server.server_close()
        self.db.close()


def test_readonly_routes_and_static():
    with tempfile.TemporaryDirectory() as tmp:
        running = RunningServer(tmp)
        try:
            status, _, health = running.json("/api/health")
            assert status == 200 and health["ok"] is True
            assert health["data"]["preview"] is False

            status, _, overview = running.json("/api/overview")
            assert status == 200 and overview["ok"] is True
            assert overview["data"]["data_date"] == "2026-10-31"

            status, headers, payload = running.request("/")
            assert status == 200
            assert "text/html" in headers["Content-Type"]
            assert b"<html" in payload.lower()

            for asset in ("/static/style.css", "/static/app.js", "/static/vendor/echarts.min.js"):
                status, asset_headers, asset_payload = running.request(asset)
                assert status == 200, asset
                assert asset_payload, asset
                content_type = asset_headers["Content-Type"]
                assert "text/css" in content_type or "javascript" in content_type, asset

            status, _, payload = running.request("/static/../config.json")
            assert status == 404

            status, _, payload = running.request("/api/videos")
            assert status == 200
            items = json.loads(payload)["data"]["items"]
            assert items and items[0]["aid"] == 102

            status, headers, payload = running.request("/api/videos", method="HEAD")
            assert status == 200 and payload == b""
            assert "Content-Length" in headers

            # 只读：所有业务写方法都被拒绝，且数据库不变
            before = running.db.read(lambda conn: conn.execute(
                "SELECT COUNT(*) FROM candidate"
            ).fetchone()[0])
            for method in ("POST", "PUT", "PATCH", "DELETE"):
                status, _, _ = running.request("/api/videos/101", method=method)
                assert status == 405, method
            after = running.db.read(lambda conn: conn.execute(
                "SELECT COUNT(*) FROM candidate"
            ).fetchone()[0])
            assert after == before

            status, _, error = running.json("/api/videos?sort=bad")
            assert status == 400 and error["ok"] is False
            status, _, error = running.json("/api/videos?page=0")
            assert status == 400
            status, _, error = running.json("/api/videos/abc")
            assert status == 400
            status, _, error = running.json("/api/nope")
            assert status == 404

            status, _, payload = running.request("/api/export.csv?sort=view")
            assert status == 200
            assert payload.startswith(b"\xef\xbb\xbf")
            assert "咻咻满".encode("utf-8") in payload

            status, headers, _ = running.request("/api/overview")
            assert headers["X-Content-Type-Options"] == "nosniff"
            assert "default-src 'self'" in headers["Content-Security-Policy"]
            assert headers["X-Frame-Options"] == "DENY"
        finally:
            running.close()


def test_token_auth_and_login():
    with tempfile.TemporaryDirectory() as tmp:
        running = RunningServer(tmp, token="secret-token")
        try:
            status, _, body = running.json("/api/overview")
            assert status == 401 and body["error"]["code"] == "unauthorized"

            status, headers, _ = running.request("/")
            assert status == 303 and headers["Location"] == "/login"

            status, _, payload = running.request("/login")
            assert status == 200 and b"web.auth_token" in payload

            status, _, _ = running.request(
                "/login",
                method="POST",
                headers={"Content-Type": "application/x-www-form-urlencoded"},
                body="token=wrong",
            )
            assert status == 401

            status, headers, _ = running.request(
                "/login",
                method="POST",
                headers={"Content-Type": "application/x-www-form-urlencoded"},
                body="token=secret-token",
            )
            assert status == 303
            cookie = headers["Set-Cookie"].split(";", 1)[0]
            assert cookie.startswith("rb_session=")

            status, _, body = running.json("/api/overview", headers={"Cookie": cookie})
            assert status == 200 and body["ok"] is True

            status, _, body = running.json(
                "/api/overview", headers={"Authorization": "Bearer secret-token"}
            )
            assert status == 200

            tampered = cookie[:-1] + ("0" if cookie[-1] != "0" else "1")
            status, _, _ = running.json("/api/overview", headers={"Cookie": tampered})
            assert status == 401

            # 健康检查始终匿名
            status, _, _ = running.json("/api/health")
            assert status == 200
        finally:
            running.close()


def test_preview_and_loopback_guards():
    with tempfile.TemporaryDirectory() as tmp:
        config = write_config(tmp)
        build_db(tmp)
        db = Database(Path(tmp) / "app.db")
        db.initialize()
        store = ConfigStore(config)
        try:
            try:
                create_server(store, db, host="0.0.0.0", port=0)
            except ValueError as exc:
                assert "web.auth_token" in str(exc)
            else:
                raise AssertionError("非回环地址且无 token 应拒绝启动")

            try:
                create_server(store, db, port=0, preview=True)
            except ValueError as exc:
                assert "preview:seeded" in str(exc)
            else:
                raise AssertionError("非预览库不应允许 --preview")

            db.write(lambda conn: conn.execute(
                "INSERT OR REPLACE INTO kv(k, v) VALUES ('preview:seeded', '1')"
            ))
            server = create_server(store, db, port=0, preview=True)
            try:
                assert server.context.preview is True
            finally:
                server.server_close()
        finally:
            db.close()


def main_test():
    test_readonly_routes_and_static()
    test_token_auth_and_login()
    test_preview_and_loopback_guards()
    print("OK: Web 只读路由/鉴权/预览/安全边界测试全部通过")


if __name__ == "__main__":
    main_test()