"""Web App 站点产物模式测试：签名令牌 / 站点发现 / 能力降级 / 构建脚本。"""

from __future__ import annotations

import json
import subprocess
import sys
import time
from pathlib import Path
from types import SimpleNamespace

import pytest

from deerflow.webapp.engine import (
    SITE_NAME_RE_OK,
    capability_status,
    find_sites,
    guess_site_mime,
)
from deerflow.webapp.security import mint_preview_token, verify_preview_token

BACKEND_DIR = Path(__file__).resolve().parents[1]
SCRIPT_PATH = BACKEND_DIR / "templates" / "web-app-template" / "scripts" / "build_site.py"


# ── 签名令牌 ────────────────────────────────────────────────────────────────


def test_preview_token_roundtrip(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
    import deerflow.webapp.security as security

    security._secret_cache = None
    token, expires_at = mint_preview_token("thread-1", "my-site", ttl_seconds=600)
    assert verify_preview_token(token) == ("thread-1", "my-site")
    assert expires_at > time.time()


def test_preview_token_expires(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
    import deerflow.webapp.security as security

    security._secret_cache = None
    token, expires_at = mint_preview_token("thread-1", "my-site", ttl_seconds=60)
    assert verify_preview_token(token) == ("thread-1", "my-site")
    # 把时钟拨到过期之后：签名仍有效，但应命中 expired 分支
    monkeypatch.setattr(security.time, "time", lambda: expires_at + 1)
    with pytest.raises(ValueError, match="expired"):
        verify_preview_token(token)


def test_preview_token_tamper_rejected(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
    import deerflow.webapp.security as security

    security._secret_cache = None
    token, _ = mint_preview_token("thread-1", "site-a", ttl_seconds=600)
    # 换 secret 后旧签名失效
    security._secret_cache = b"1" * 64
    with pytest.raises(ValueError, match="signature"):
        verify_preview_token(token)
    with pytest.raises(ValueError, match="malformed"):
        verify_preview_token("not-a-token")
    with pytest.raises(ValueError, match="malformed"):
        verify_preview_token("a.b.c")


# ── 站点发现 ────────────────────────────────────────────────────────────────


def _make_site(outputs: Path, name: str, title: str = "示例", with_dist: bool = False) -> None:
    site_dir = outputs / name
    (site_dir / "public").mkdir(parents=True, exist_ok=True)
    (site_dir / "site.json").write_text(
        json.dumps({"title": title, "nav": [{"page": "overview", "label": "总览"}]}, ensure_ascii=False),
        encoding="utf-8",
    )
    if with_dist:
        (site_dir / "dist").mkdir(exist_ok=True)
        (site_dir / "dist" / "index.html").write_text("<html></html>", encoding="utf-8")


def test_find_sites_discovers_manifest_dirs(tmp_path: Path) -> None:
    outputs = tmp_path / "outputs"
    outputs.mkdir()
    _make_site(outputs, "alpha-site", title="甲")
    _make_site(outputs, "beta-site", title="乙", with_dist=True)
    (outputs / "plain.txt").write_text("not a dir", encoding="utf-8")
    (outputs / "not-a-site").mkdir()
    (outputs / ".hidden").mkdir()
    (outputs / ".hidden" / "site.json").write_text("{}", encoding="utf-8")

    sites = find_sites(outputs)
    # .hidden 不满足站点名规则（必须字母数字开头），被排除
    assert sorted(s["name"] for s in sites) == ["alpha-site", "beta-site"]
    beta = next(s for s in sites if s["name"] == "beta-site")
    assert beta["title"] == "乙"
    assert beta["dist"]["exists"] is True
    alpha = next(s for s in sites if s["name"] == "alpha-site")
    assert alpha["dist"]["exists"] is False


def test_find_sites_missing_dir_returns_empty(tmp_path: Path) -> None:
    assert find_sites(tmp_path / "missing") == []


def test_guess_site_mime_overrides(tmp_path: Path) -> None:
    assert "text/javascript" in guess_site_mime(tmp_path / "app.js")
    assert "text/javascript" in guess_site_mime(tmp_path / "app.mjs")
    assert "application/json" in guess_site_mime(tmp_path / "data.geojson")
    assert guess_site_mime(tmp_path / "blob.wasm") == "application/wasm"
    assert "text/html" in guess_site_mime(tmp_path / "index.html")


def test_site_name_regex_blocks_traversal() -> None:
    assert SITE_NAME_RE_OK.match("my-site")
    assert SITE_NAME_RE_OK.match("Site_2.b")
    assert not SITE_NAME_RE_OK.match("../etc")
    assert not SITE_NAME_RE_OK.match(".hidden")
    assert not SITE_NAME_RE_OK.match("a/b")


# ── 能力降级 ────────────────────────────────────────────────────────────────


def _stub_config(*, enabled: bool, template_dir: str) -> SimpleNamespace:
    return SimpleNamespace(
        web_app=SimpleNamespace(
            enabled=enabled,
            template_dir=template_dir,
            build_timeout_seconds=10,
            preview_token_ttl_seconds=600,
            node_path="",
        )
    )


def test_capability_unavailable_when_disabled(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    import deerflow.webapp.engine as engine

    monkeypatch.setattr(engine, "get_app_config", lambda: _stub_config(enabled=False, template_dir=str(tmp_path)))
    cap = capability_status()
    assert cap.available is False
    assert cap.checks["enabled"]["ok"] is False


def test_capability_unavailable_when_template_missing(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    import deerflow.webapp.engine as engine

    monkeypatch.setattr(
        engine, "get_app_config", lambda: _stub_config(enabled=True, template_dir=str(tmp_path / "nope"))
    )
    cap = capability_status()
    assert cap.available is False
    assert cap.checks["template"]["ok"] is False


def test_capability_never_raises_on_bad_config(monkeypatch: pytest.MonkeyPatch) -> None:
    import deerflow.webapp.engine as engine

    def boom() -> None:
        raise RuntimeError("config broken")

    monkeypatch.setattr(engine, "get_app_config", boom)
    cap = capability_status()
    assert cap.available is False
    assert cap.checks["internal"]["ok"] is False


# ── build_site.py 脚本 ──────────────────────────────────────────────────────


def _run_script(args: list[str], cwd: Path) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        [sys.executable, str(SCRIPT_PATH), *args],
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        cwd=str(cwd),
        timeout=120,
    )


def test_script_scaffold_creates_site_without_runtime_artifacts(tmp_path: Path) -> None:
    workdir = tmp_path / "work"
    workdir.mkdir()
    proc = _run_script(["scaffold", "my-site"], cwd=workdir)
    assert proc.returncode == 0, proc.stderr
    payload = json.loads(proc.stdout.strip().splitlines()[-1])
    assert payload["ok"] is True
    site = workdir / "my-site"
    assert (site / "site.json").is_file()
    assert (site / "src" / "pages" / "overview.tsx").is_file()
    assert (site / "scripts").exists() is False  # 构建脚本不入站点
    assert (site / "node_modules").exists() is False


def test_script_scaffold_rejects_traversal(tmp_path: Path) -> None:
    workdir = tmp_path / "work"
    workdir.mkdir()
    proc = _run_script(["scaffold", "../escape"], cwd=workdir)
    payload = json.loads(proc.stdout.strip().splitlines()[-1])
    assert payload["ok"] is False
    assert "非法" in payload["error"]
    assert not (tmp_path / "escape").exists()


def test_script_scaffold_accepts_host_rewritten_path(tmp_path: Path) -> None:
    """LocalSandbox 会把 /mnt/user-data/... 改写为宿主绝对路径，脚本须接受该形态。"""
    user_data = tmp_path / "data" / "user-data"
    (user_data / "outputs").mkdir(parents=True)
    site_dir = user_data / "outputs" / "host-site"
    proc = _run_script(["scaffold", str(site_dir)], cwd=tmp_path)
    payload = json.loads(proc.stdout.strip().splitlines()[-1])
    assert payload["ok"] is True, payload
    assert (site_dir / "site.json").is_file()


def test_script_scaffold_rejects_out_of_root_absolute_path(tmp_path: Path) -> None:
    outside = tmp_path / "elsewhere"
    outside.mkdir()
    proc = _run_script(["scaffold", str(outside / "site")], cwd=tmp_path)
    payload = json.loads(proc.stdout.strip().splitlines()[-1])
    assert payload["ok"] is False
    assert "user-data/outputs" in payload["error"]
    assert not (outside / "site").exists()


def test_script_build_requires_site_json(tmp_path: Path) -> None:
    workdir = tmp_path / "work"
    (workdir / "not-a-site").mkdir(parents=True)
    proc = _run_script(["build", "not-a-site"], cwd=workdir)
    payload = json.loads(proc.stdout.strip().splitlines()[-1])
    assert payload["ok"] is False
    assert "site.json" in payload["error"]


def test_script_check_reports_environment(tmp_path: Path) -> None:
    proc = _run_script(["check"], cwd=tmp_path)
    payload = json.loads(proc.stdout.strip().splitlines()[-1])
    assert payload["ok"] is True  # check 命令本身成功
    assert set(payload["checks"]) == {"enabled", "template", "node_modules", "node"}


def test_script_check_respects_config_flag(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    """check 探测读取 config.yaml 的 web_app.enabled（宿主侧语义）。"""
    config = tmp_path / "config.yaml"

    def check_with(cfg_text: str | None, env_flag: str | None = None) -> bool:
        monkeypatch.delenv("DEERFLOW_WEBAPP_ENABLED", raising=False)
        if env_flag is not None:
            monkeypatch.setenv("DEERFLOW_WEBAPP_ENABLED", env_flag)
        if cfg_text is None:
            monkeypatch.delenv("DEERFLOW_WEBAPP_CONFIG", raising=False)
        else:
            config.write_text(cfg_text, encoding="utf-8")
            monkeypatch.setenv("DEERFLOW_WEBAPP_CONFIG", str(config))
        proc = _run_script(["check"], cwd=tmp_path)
        payload = json.loads(proc.stdout.strip().splitlines()[-1])
        return bool(payload["checks"]["enabled"]["ok"])

    # 显式 true → 开
    assert check_with("web_app:\n  enabled: true\n") is True
    # 显式 false → 关
    assert check_with("web_app:\n  enabled: false\n") is False
    # 有配置文件但无 web_app 段 → 与 pydantic 默认一致：关
    assert check_with("sandbox:\n  use: x\n") is False
    # env 覆盖优先于配置文件
    assert check_with("web_app:\n  enabled: false\n", env_flag="true") is True
    assert check_with("web_app:\n  enabled: true\n", env_flag="false") is False
    # 指定的配置文件不存在（沙箱视图：模板挂载点向上没有 config.yaml）→
    # 视为开启，由「平台是否挂载模板」表达开关
    monkeypatch.delenv("DEERFLOW_WEBAPP_ENABLED", raising=False)
    monkeypatch.setenv("DEERFLOW_WEBAPP_CONFIG", str(tmp_path / "missing-config.yaml"))
    proc = _run_script(["check"], cwd=tmp_path)
    payload = json.loads(proc.stdout.strip().splitlines()[-1])
    assert payload["checks"]["enabled"]["ok"] is True


def test_script_status_reports_dist(tmp_path: Path) -> None:
    workdir = tmp_path / "work"
    workdir.mkdir()
    _run_script(["scaffold", "my-site"], cwd=workdir)
    (workdir / "my-site" / "dist").mkdir()
    (workdir / "my-site" / "dist" / "index.html").write_text("<html></html>", encoding="utf-8")
    proc = _run_script(["status", "my-site"], cwd=workdir)
    payload = json.loads(proc.stdout.strip().splitlines()[-1])
    assert payload["ok"] is True
    assert payload["dist"]["exists"] is True
    assert payload["dist"]["file_count"] == 1