# -*- 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, _, payload = running.json("/api/session")
            assert status == 200
            assert payload["data"]["auth_enabled"] is False
            assert payload["data"]["authenticated"] is True
            status, _, payload = running.json("/api/candidates")
            assert status == 200 and payload["ok"] is True
            status, _, payload = running.json("/api/system")
            assert status == 200 and payload["ok"] is True
            status, _, payload = running.json("/api/about")
            assert status == 200
            assert payload["data"]["schema_version"] == 7
            assert payload["data"]["analytics"]["min_duration_seconds"] == 600

            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

            # 伪静态前端路由：直接访问 /rank、/video/... 返回 index.html
            for route_path in ("/rank", "/videos", "/ups", "/about", "/video/114775988111033", "/up/230020975", "/fans"):
                status, route_headers, route_payload = running.request(route_path)
                assert status == 200, route_path
                assert "text/html" in route_headers["Content-Type"], route_path
                assert b"<html" in route_payload.lower(), route_path
            status, _, _ = running.request("/nope.js")
            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, _, payload = running.json("/api/videos?year=2026")
            assert status == 200
            assert payload["data"]["year"] == 2026
            status, _, payload = running.json("/api/videos?year=1800")
            assert status == 400
            status, _, payload = running.json("/api/ups?year=2026")
            assert status == 200
            assert payload["data"]["year"] == 2026

            status, _, payload = running.request("/api/ups")
            assert status == 200
            assert json.loads(payload)["data"]["sort"] == "efficiency"

            status, _, payload = running.request("/api/fans?days=7&recent_days=14")
            assert status == 200
            fans = json.loads(payload)["data"]
            assert fans["recent_days"] == 14
            assert all(
                point["stat_date"] >= "2026-10-18"
                for point in fans["recent_points"]
            )
            assert all("delta" in point for point in fans["recent_points"])

            status, _, payload = running.request(
                "/api/fans?start=2026-10-29&end=2026-10-31"
            )
            assert status == 200
            data = json.loads(payload)["data"]
            assert data["mode"] == "daily"
            assert data["dates"] == ["2026-10-29", "2026-10-30", "2026-10-31"]

            running.db.write(lambda conn: conn.execute(
                "INSERT OR REPLACE INTO up_stat"
                "(stat_date, mid, follower, following, captured_at, ok) "
                "VALUES ('2025-01-01', 37754047, 500, NULL, 1, 1)"
            ))
            status, _, payload = running.request(
                "/api/fans?start=2025-01-01&end=2026-10-31"
            )
            assert status == 200
            assert json.loads(payload)["data"]["mode"] == "monthly"

            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 == 404

            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="test-token")
        try:
            # 公共页面与公共 API 访客可见
            status, _, body = running.json("/api/overview")
            assert status == 200 and body["ok"] is True
            status, _, body = running.json("/api/videos")
            assert status == 200 and body["ok"] is True
            status, _, body = running.json("/api/fans?days=7")
            assert status == 200 and body["ok"] is True
            status, _, payload = running.request("/")
            assert status == 200 and b"<html" in payload.lower()
            status, _, payload = running.request("/rank")
            assert status == 200 and b"<html" in payload.lower()

            # 候选与系统需要登录
            status, _, body = running.json("/api/candidates")
            assert status == 401 and body["error"]["code"] == "unauthorized"
            status, _, body = running.json("/api/system")
            assert status == 401
            status, headers, _ = running.request("/candidates")
            assert status == 303 and headers["Location"] == "/login?next=/candidates"
            status, headers, _ = running.request("/system")
            assert status == 303 and headers["Location"] == "/login?next=/system"

            status, _, body = running.json("/api/session")
            assert status == 200
            assert body["data"]["auth_enabled"] is True
            assert body["data"]["authenticated"] is False

            status, _, payload = running.request("/login?next=/candidates")
            assert status == 200 and b"web.auth_token" in payload
            assert b"action=\"/login?next=/candidates\"" 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?next=/candidates",
                method="POST",
                headers={"Content-Type": "application/x-www-form-urlencoded"},
                body="token=test-token",
            )
            assert status == 303 and headers["Location"] == "/candidates"
            cookie = headers["Set-Cookie"].split(";", 1)[0]
            assert cookie.startswith("rb_session=")

            status, _, body = running.json("/api/candidates", headers={"Cookie": cookie})
            assert status == 200 and body["ok"] is True
            status, _, body = running.json("/api/system", headers={"Cookie": cookie})
            assert status == 200 and body["ok"] is True
            status, _, body = running.json(
                "/api/candidates", headers={"Authorization": "Bearer test-token"}
            )
            assert status == 200

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

            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()