knowledge-governance/scripts/ci_preview_negative_test.py

#!/usr/bin/env python3
"""预览发布负例：**真实调用路径**上验证"核对不过就一个请求都不发"。

为什么要有这个文件：只测判定函数（`verify()` 返回几个问题）证明不了"发布顺序"。
真正要紧的是：预览这一步到底有没有发出 HTTP 请求。所以这里起一个**假的 ci-bridge**，
让它记录收到的每一个请求，然后拿各种坏报告去跑**真实的** `.gitea/scripts/ci_preview.py`，
断言坏报告下 publish 请求数 **= 0**、好报告下 **= 1**。

    python3 scripts/ci_preview_negative_test.py

不联网、不碰真 bridge、不写任何远端状态；只在本机 127.0.0.1 的临时端口上起一个记录用的服务。
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
import tempfile
import threading
import shutil
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "scripts"))
import ci_checks  # noqa: E402  用**本仓真实清单**造报告，而不是另编一份

REQUESTS: list[str] = []


class Handler(BaseHTTPRequestHandler):
    def log_message(self, *a):        # 静音（否则每个请求都打一行）
        pass

    def do_GET(self):
        REQUESTS.append(f"GET {self.path}")
        body = b"<html>preview</html>" if self.path.startswith("/preview/") else b"ok"
        self.send_response(200)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self):
        n = int(self.headers.get("Content-Length") or 0)
        self.rfile.read(n)
        REQUESTS.append(f"POST {self.path}")
        payload = json.dumps({"url": f"http://{self.headers.get('Host')}/preview/x/index.html",
                              "url_tailscale": f"http://{self.headers.get('Host')}/preview/x/index.html"})
        body = payload.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)


def good_report(sha: str) -> dict:
    checks = [{"id": cid, "status": "success", "reason": "", "required_in": list(req), "duration_s": 1}
              for cid, req in ci_checks.CHECK_MANIFEST.items()]
    return {"schema": "ci-report/1", "policy_revision": ci_checks.POLICY_REVISION,
            "repository": "Weknora/knowledge-governance", "commit_sha": sha, "event_sha": sha,
            "branch": "main", "event": "push", "profile": "ci", "run_mode": "full", "duration_s": 1.0,
            "checks": checks, "manifest": {c: list(r) for c, r in ci_checks.CHECK_MANIFEST.items()},
            "overall_status": "success", "problems": [],
            "notification_status": "not_attempted", "deployed": False}


def run_case(name: str, rep: dict, sha: str, want_posts: int, work: Path) -> tuple[bool, str]:
    global REQUESTS
    REQUESTS = []
    sandbox = work / "repo"
    shutil.rmtree(sandbox, ignore_errors=True)
    sandbox.mkdir(parents=True)
    shutil.copytree(ROOT / ".gitea", sandbox / ".gitea")
    shutil.copytree(ROOT / "scripts", sandbox / "scripts",
                    ignore=shutil.ignore_patterns("__pycache__"))
    (sandbox / "web" / "dist").mkdir(parents=True)
    (sandbox / "web" / "dist" / "index.html").write_text("<html>ok</html>", encoding="utf-8")
    (sandbox / "ci-report.json").write_text(json.dumps(rep, ensure_ascii=False), encoding="utf-8")

    env = {**os.environ, "GITHUB_ACTIONS": "true", "GITHUB_SHA": sha,
           "GITHUB_REPOSITORY": rep.get("repository", ""), "BRIDGE_TOKEN": "fake-token",
           "CI_REPORT": str(sandbox / "ci-report.json")}
    env.pop("GITHUB_RUN_ID", None)
    p = subprocess.run([sys.executable, str(sandbox / ".gitea" / "scripts" / "ci_preview.py")],
                       cwd=sandbox, env=env, capture_output=True, text=True, timeout=180)
    posts = [r for r in REQUESTS if r.startswith("POST")]
    ok = len(posts) == want_posts and p.returncode == 0
    detail = (f"退出码 {p.returncode}｜publish 请求 {len(posts)} 次（期望 {want_posts}）"
              f"｜其他请求 {len(REQUESTS) - len(posts)}")
    if not ok:
        detail += f"\n      stdout: {p.stdout.strip()[-300:]}\n      stderr: {p.stderr.strip()[-300:]}"
    return ok, detail


def main() -> int:
    sha = "5f3a71cadee1f2b27c0b71d0c9fd1ea1b57a4b0f"
    srv = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    port = srv.server_address[1]
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    os.environ["CI_BRIDGE_RESOLVED"] = f"http://127.0.0.1:{port}"

    good = good_report(sha)
    bad = good_report("b" * 40)
    cases = [
        # 名字, 报告, 期望的 publish 次数
        ("正例：完整合规报告 ⇒ 发 1 次", good, 1),
        ("缺一项必需检查 ⇒ 0 次", {**good, "checks": [c for c in good["checks"] if c["id"] != "tests-go"]}, 0),
        ("多一项清单外的检查 ⇒ 0 次", {**good, "checks": good["checks"] + [{"id": "zz", "status": "success"}]}, 0),
        ("重复项 ⇒ 0 次", {**good, "checks": good["checks"] + [good["checks"][0]]}, 0),
        ("短 SHA ⇒ 0 次", {**good, "commit_sha": sha[:12], "event_sha": ""}, 0),
        ("归错提交 ⇒ 0 次", bad, 0),
        ("CI 里用 local 档案 ⇒ 0 次", {**good, "profile": "local"}, 0),
        ("只跑子集（not_run_only）⇒ 0 次", {**good, "overall_status": "not_run_only", "run_mode": "only"}, 0),
        ("没有清单 ⇒ 0 次", {k: v for k, v in good.items() if k != "manifest"}, 0),
        ("某项 failure 但自称 success ⇒ 0 次",
         {**good, "checks": [{**c, "status": "failure"} if c["id"] == "security-deps" else c
                             for c in good["checks"]]}, 0),
    ]
    problems, lines = [], []
    with tempfile.TemporaryDirectory() as d:
        for name, rep, want in cases:
            ok, detail = run_case(name, rep, sha, want, Path(d))
            lines.append(f"{'✅' if ok else '❌'} {name}：{detail}")
            if not ok:
                problems.append(name)
    srv.shutdown()
    print("── 预览发布负例（真实 ci_preview.py + 假 bridge 计数）──")
    for line in lines:
        print("  " + line)
    if problems:
        print(f"\n❌ {len(problems)} 个分支不符合预期：{problems}", file=sys.stderr)
        return 1
    print(f"\n✅ {len(cases)} 个分支全部符合预期：坏报告一个 publish 请求都不发，好报告正好发 1 次")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())