292 lines
12 KiB
Python
292 lines
12 KiB
Python
"""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
|