姚章浩
40d84994cb
check_changelog.py：正式版本条目必须四节齐备且非空（改了什么/兼容性与迁移影响/怎么验收/已知问题）、版本号三段无前导零、[Unreleased] 在最上面；新增 --tags 核对正式 tag（必须注释 tag）与它指到那个提交里的版本记录一致。只有 Unreleased 的新工程放行；没有正式 tag 不判红（开发版本不打 tag 是常态）。ci_checks.py：新增必需检查 version（门禁 15 项），策略版本 ci-checks/2。ci_checkout.sh：记录远端 tag 清单并取 tag 对象供核对。release_preflight.py：只读预检（版本建议/完整 SHA/CI 证据/CHANGELOG/远端 tag 与 Release/遗留项），写类 git 命令与 POST 在预检路径上被拒（--self-test 证明）；批准后 --execute 才打注释 tag、推 tag、建 Release 并读回。CHANGELOG 0.1.0 补上原本缺的两节。
677 lines
34 KiB
Python
677 lines
34 KiB
Python
#!/usr/bin/env python3
|
||
"""本仓**唯一权威门禁入口**（正本 §3.1 / 试点草案 §4）。
|
||
|
||
make verify # = python3 scripts/ci_checks.py
|
||
python3 scripts/ci_checks.py --list
|
||
python3 scripts/ci_checks.py --only tests-web # 单跑一项（**不算全绿**）
|
||
python3 scripts/ci_checks.py --self-test # 只跑汇总逻辑的反向自检
|
||
|
||
一条命令跑完全部必需检查，**串行、失败即停**，并产出结构化报告 `ci-report.json`。
|
||
CI 跑的就是这一条（`.gitea/workflows/build.yml` 里 `make verify`）—— 本地与 CI 同一份逻辑，
|
||
不做第二套步骤（正本 §3.1：两边各写一套，迟早只改一边）。
|
||
|
||
三条纪律写进代码里：
|
||
1. **失败即停，但汇总照做**：第一个必需项失败后，剩下的记为 `not_run`，
|
||
报告如实写"未运行"而不是留空 —— 留空会被下游当成"没有这一项"。
|
||
2. **被跳过的必需项不算过**：`aggregate()` 只认 `success`；
|
||
不适用要**在清单里声明**（`required_in`），不能在运行时悄悄 skip。
|
||
3. **工具缺失 / 依赖没按锁装 / 输出解析不了 ⇒ 失败**，不当成"检查通过"。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import datetime as dt
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Callable
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
WEB = ROOT / "web"
|
||
REPORT_DEFAULT = ROOT / "ci-report.json"
|
||
POLICY_REVISION = "ci-checks/2" # 判据变更时改这里，报告里跟着变
|
||
MANIFESTS = ["go.mod", "go.sum", "web/package.json", "web/package-lock.json"]
|
||
|
||
SUCCESS, FAILURE, SKIPPED, NOT_RUN = "success", "failure", "skipped", "not_run"
|
||
|
||
|
||
# ── 工具解析：找不到就报错，不回落成"-- 跳过"（正本 §4.4）──────────────
|
||
def tool(name: str, *extra: str) -> str:
|
||
for cand in (shutil.which(name), *extra):
|
||
if cand and Path(cand).exists():
|
||
return cand
|
||
raise FileNotFoundError(name)
|
||
|
||
|
||
def go_bin() -> str:
|
||
return tool("go", "/opt/citools/go/bin/go")
|
||
|
||
|
||
def gofmt_bin() -> str:
|
||
return tool("gofmt", "/opt/citools/go/bin/gofmt")
|
||
|
||
|
||
def npm_bin() -> str:
|
||
return tool("npm")
|
||
|
||
|
||
def py(*args: str) -> list[str]:
|
||
return [sys.executable or "python3", *args]
|
||
|
||
|
||
def go_env() -> dict:
|
||
env = dict(os.environ)
|
||
env["GOFLAGS"] = "-mod=readonly" # 禁止隐式改写 go.mod / go.sum
|
||
env.setdefault("GOPROXY", "https://goproxy.cn,direct")
|
||
env.setdefault("GOTOOLCHAIN", "local")
|
||
# 缓存位置是**环境相关**的（CI 用主机挂载的 /opt/citools-cache），而缓存只加速、
|
||
# 不作为通过证据，所以这里"可用就用、不可用就用 Go 自己的默认"，不写死。
|
||
cache = Path("/opt/citools-cache")
|
||
if cache.is_dir() and os.access(cache, os.W_OK):
|
||
for key, sub in (("GOCACHE", "gocache"), ("GOMODCACHE", "gomod"), ("GOPATH", "gopath")):
|
||
path = cache / sub
|
||
try:
|
||
path.mkdir(parents=True, exist_ok=True)
|
||
except OSError:
|
||
continue
|
||
env.setdefault(key, str(path))
|
||
return env
|
||
|
||
|
||
# ── 判定函数：拿到 (退出码, stdout, stderr) → (状态, 说明) ──────────────
|
||
def plain(exit_code: int, out: str, err: str) -> tuple[str, str]:
|
||
if exit_code == 0:
|
||
return SUCCESS, "退出码 0"
|
||
tail = (err or out).strip().splitlines()[-12:]
|
||
return FAILURE, f"退出码 {exit_code}｜" + " ／ ".join(tail)
|
||
|
||
|
||
SCOPE_FILE = ROOT / "scripts" / "test-scope.json"
|
||
|
||
|
||
def load_test_scope(path: Path = SCOPE_FILE) -> tuple[list[dict], list[str]]:
|
||
"""读"哪些跳过是允许的"声明。
|
||
|
||
极性是**默认必需**：没声明过的测试一旦被跳过，门禁就红。
|
||
文件缺失 / 解析不了 ⇒ 返回问题（门禁据此判红）—— 声明不出来时不许默认放行。
|
||
"""
|
||
if not path.is_file():
|
||
return [], [f"找不到 {path.relative_to(ROOT)}：测试范围声明不出来 ⇒ 不许默认放行"
|
||
f"（必需与可选必须显式区分）"]
|
||
try:
|
||
raw = json.loads(path.read_text(encoding="utf-8"))
|
||
except json.JSONDecodeError as e:
|
||
return [], [f"{path.name} 解析失败：{e}"]
|
||
entries = raw.get("optional") if isinstance(raw, dict) else None
|
||
if not isinstance(entries, list):
|
||
return [], [f"{path.name} 里没有 optional 列表"]
|
||
return [e for e in entries if isinstance(e, dict)], []
|
||
|
||
|
||
def _optional_hit(entries: list[dict], package: str, test: str) -> dict | None:
|
||
"""这条跳过算不算"已声明的可选未运行"。package / test 都是"给了才比"。"""
|
||
for e in entries:
|
||
if e.get("package") and e["package"] != package:
|
||
continue
|
||
if e.get("test") and e["test"] != test:
|
||
continue
|
||
return e
|
||
return None
|
||
|
||
|
||
def judge_go_tests(exit_code: int, out: str, err: str) -> tuple[str, str]:
|
||
"""`go test -json` 的盘点：**必需用例必须真的跑过**，可选集成用例的跳过要如实记出来。
|
||
|
||
两条为什么必须这样：
|
||
1. 退出码 0 只说明"没有失败" —— 一个用例都没跑的包也是 0（假绿）；
|
||
2. 必需用例被 SKIP 也不等于通过（被跳过的断言等于没有断言）。
|
||
哪些跳过允许，读 `scripts/test-scope.json`；没声明的跳过一律判红。
|
||
"""
|
||
entries, scope_problems = load_test_scope()
|
||
scope_ok = not scope_problems
|
||
|
||
passed = failed = 0
|
||
skipped: list[tuple[str, str, str]] = [] # (package, test, 原因)
|
||
reasons: dict[str, str] = {}
|
||
pkgs: set[str] = set()
|
||
for line in out.splitlines():
|
||
line = line.strip()
|
||
if not line.startswith("{"):
|
||
continue
|
||
try:
|
||
ev = json.loads(line)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
pkg = ev.get("Package") or ""
|
||
test = ev.get("Test") or ""
|
||
if pkg:
|
||
pkgs.add(pkg)
|
||
chunk = ev.get("Output") or ""
|
||
m = re.match(r"\s*\S+_test\.go:\d+:\s*(.+)$", chunk)
|
||
if m and test:
|
||
reasons[test] = m.group(1).strip()
|
||
action = ev.get("Action")
|
||
if test and action == "pass":
|
||
passed += 1
|
||
elif test and action == "fail":
|
||
failed += 1
|
||
elif test and action == "skip":
|
||
skipped.append((pkg, test, reasons.get(test, "未取到原因")))
|
||
|
||
required_skips = [(p, t, r) for p, t, r in skipped if not _optional_hit(entries, p, t)]
|
||
detail = (f"执行包 {len(pkgs)}｜用例 通过 {passed} / 失败 {failed} / 跳过 {len(skipped)}"
|
||
f"（其中已声明的可选集成 {len(skipped) - len(required_skips)}，必需 {len(required_skips)}）")
|
||
|
||
if scope_problems:
|
||
return FAILURE, detail + "｜" + "；".join(scope_problems)
|
||
if exit_code != 0 or failed:
|
||
tail = (err or out).strip().splitlines()
|
||
return FAILURE, detail + (f"｜{tail[-1][:200]}" if tail else "")
|
||
if passed == 0:
|
||
return FAILURE, detail + "｜一个用例都没真的执行 —— 无测试不能算全绿"
|
||
if required_skips:
|
||
head = "；".join(f"{p}::{t}（{r}）" for p, t, r in required_skips[:4])
|
||
return FAILURE, (detail + f"｜**必需用例被跳过**（没在 scripts/test-scope.json 里声明为可选）：{head}"
|
||
+ ("…" if len(required_skips) > 4 else ""))
|
||
if not scope_ok: # 理论上到不了，留作兜底
|
||
return FAILURE, detail + "｜测试范围声明有问题"
|
||
opt = {}
|
||
for p, _t, _r in skipped:
|
||
opt[p] = opt.get(p, 0) + 1
|
||
if opt:
|
||
detail += "｜可选未运行：" + "、".join(f"{p.split('/')[-1]}×{n}" for p, n in sorted(opt.items()))
|
||
return SUCCESS, detail
|
||
|
||
|
||
def judge_vitest_file(path: Path) -> tuple[str, str]:
|
||
try:
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
except (OSError, json.JSONDecodeError) as e:
|
||
return FAILURE, f"前端测试报告读不到/解析不了（{e}）—— 未完成不算过"
|
||
total = int(data.get("numTotalTests") or 0)
|
||
failed = int(data.get("numFailedTests") or 0)
|
||
suites = int(data.get("numTotalTestSuites") or 0)
|
||
passed = int(data.get("numPassedTests") or 0)
|
||
pending = int(data.get("numPendingTests") or 0) + int(data.get("numTodoTests") or 0)
|
||
detail = f"套件 {suites}｜用例 通过 {passed} / 失败 {failed} / 待跑或待办 {pending} / 共 {total}"
|
||
if failed:
|
||
return FAILURE, detail
|
||
if total == 0 or suites == 0:
|
||
return FAILURE, detail + "｜一个用例都没执行（空套件不能算通过）"
|
||
if pending:
|
||
# 前端这套都是必需用例：pending/todo 等于"没验"
|
||
return FAILURE, detail + ("｜有未执行的用例（skip/todo）——必需用例不许跳过；"
|
||
"确实不需要的请删掉，或在 scripts/test-scope.json 里声明理由")
|
||
return SUCCESS, detail
|
||
|
||
|
||
# ── 检查清单（唯一一份定义；Makefile / CI / 文档都指它）─────────────────
|
||
@dataclass
|
||
class Check:
|
||
id: str
|
||
title: str
|
||
argv: list[str]
|
||
cwd: Path = ROOT
|
||
env: dict = field(default_factory=dict)
|
||
judge: Callable[[int, str, str], tuple[str, str]] = plain
|
||
required_in: tuple[str, ...] = ("local", "ci")
|
||
note: str = ""
|
||
|
||
|
||
def checks() -> list[Check]:
|
||
g, gf, npm, py3 = go_bin(), gofmt_bin(), npm_bin(), sys.executable or "python3"
|
||
env = go_env()
|
||
out_dir = Path(os.environ.get("CI_BUILD_OUT") or (ROOT / "bin"))
|
||
table = [
|
||
Check("identity", "检出身份（实际 HEAD == 事件完整 SHA）",
|
||
[], required_in=("ci",),
|
||
note="本地没有事件负载 ⇒ 记录为 skipped 并在这里说明；CI 上它是必需项"),
|
||
Check("commits", "提交信息规范（本次推送的范围）",
|
||
[py3, ".gitea/scripts/ci_gate.py"]),
|
||
Check("deps-install", "依赖按锁文件安装（缺锁即失败，不退化）",
|
||
[py3, "scripts/ci_checks.py", "--internal-deps", str(out_dir)],
|
||
env=env),
|
||
Check("format", "格式（Go gofmt + 前端 prettier）", [py3, "scripts/check_format.py"], env=env),
|
||
Check("static-go", "Go 静态检查（go vet）", [g, "vet", "./..."], env=env),
|
||
Check("static-web", "前端类型检查（tsc）", [npm, "--prefix", "web", "run", "typecheck"], env=env),
|
||
Check("build-go", "Go 构建（带版本注入）",
|
||
[py3, "scripts/ci_checks.py", "--internal-build", str(out_dir)], env=env),
|
||
Check("build-web", "前端构建（vite build）", [npm, "--prefix", "web", "run", "build"], env=env),
|
||
Check("tests-go", "Go 测试（禁用测试缓存，强制真跑）",
|
||
[g, "test", "-count=1", "-json", "./..."], env=env, judge=judge_go_tests),
|
||
Check("tests-web", "前端测试（vitest，必须真跑）",
|
||
[py3, "scripts/ci_checks.py", "--internal-web-test"], env=env),
|
||
Check("contracts", "版本与变更记录契约（不许 SKIP 后算过）",
|
||
[py3, "scripts/ci_checks.py", "--internal-contracts", g], env=env),
|
||
Check("version", "CHANGELOG 与版本记录（四节齐备 + 正式 tag 与提交内版本一致）",
|
||
[py3, "scripts/check_changelog.py", "--tags"]),
|
||
Check("security-secrets", "密钥泄漏扫描（含扫描器自检）",
|
||
[py3, "scripts/check_secrets.py"]),
|
||
Check("security-deps", "依赖漏洞（govulncheck + npm audit，分级 + 例外）",
|
||
[py3, "scripts/check_deps_vuln.py"], env=env),
|
||
Check("manifests-locked", "依赖清单在这次门禁里没有被改写",
|
||
[py3, "scripts/ci_checks.py", "--internal-manifests"]),
|
||
]
|
||
# 声明与实现必须一致：漏声明会让汇总把真项当"未知项"，多出一项又会让它逃过核对
|
||
drift = manifest_mismatch(CHECK_MANIFEST, [(c.id, c.required_in) for c in table])
|
||
if drift:
|
||
raise SystemExit("❌ 必需检查清单（CHECK_MANIFEST）与实现不一致，先对齐再跑：\n · "
|
||
+ "\n · ".join(drift))
|
||
return table
|
||
|
||
|
||
# ── 内部子命令（把"一条复杂命令"收进脚本，YAML/清单里只留一行）─────────
|
||
def cmd_internal_deps(out_dir: str) -> int:
|
||
"""按锁文件安装依赖。缺锁**不**退化成 `npm install`（那会静默改写清单）。"""
|
||
lock = WEB / "package-lock.json"
|
||
if not lock.is_file():
|
||
print("❌ web/package-lock.json 不存在：不接受无约束安装，请先把锁文件入库", file=sys.stderr)
|
||
return 1
|
||
if subprocess.run([npm_bin(), "ci", "--prefix", "web"], cwd=ROOT, env=go_env()).returncode:
|
||
print("❌ npm ci 失败：锁文件与 package.json 不一致（先跑 npm install 并提交锁文件）", file=sys.stderr)
|
||
return 1
|
||
p = subprocess.run([go_bin(), "mod", "download"], cwd=ROOT, env=go_env(),
|
||
capture_output=True, text=True)
|
||
if p.returncode:
|
||
print(f"❌ go mod download 失败：{(p.stderr or p.stdout)[:400]}", file=sys.stderr)
|
||
return 1
|
||
print("✅ 依赖按锁文件安装完成（npm ci + go mod download）")
|
||
return 0
|
||
|
||
|
||
def cmd_internal_build(out_dir: str) -> int:
|
||
out = Path(out_dir)
|
||
out.mkdir(parents=True, exist_ok=True)
|
||
version = subprocess.run(py("scripts/version.py"), cwd=ROOT, capture_output=True,
|
||
text=True).stdout.strip() or "dev"
|
||
sha = subprocess.run(["git", "rev-parse", "--short=12", "HEAD"], cwd=ROOT,
|
||
capture_output=True, text=True).stdout.strip() or "unknown"
|
||
ts = dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
ldflags = " ".join([
|
||
f"-X knowledge-governance/internal/version.Number={version}",
|
||
f"-X knowledge-governance/internal/version.CommitSHA={sha}",
|
||
f"-X knowledge-governance/internal/version.BuildTime={ts}",
|
||
])
|
||
p = subprocess.run([go_bin(), "build", "-trimpath", "-ldflags", ldflags, "-o", f"{out}/", "./cmd/..."],
|
||
cwd=ROOT, env=go_env(), capture_output=True, text=True)
|
||
print(f"-- go build（版本 {version} / 提交 {sha}）--")
|
||
print(p.stdout[-2000:] if p.stdout else "", end="")
|
||
if p.returncode:
|
||
print(f"❌ 构建失败：{(p.stderr or '')[:2000]}", file=sys.stderr)
|
||
return 1
|
||
print(f"✅ 构建完成：{out}")
|
||
return 0
|
||
|
||
|
||
def cmd_internal_web_test() -> int:
|
||
"""前端测试必须真跑（空套件不算通过），所以拿 vitest 的 JSON 报告看数字。"""
|
||
with tempfile.TemporaryDirectory() as d:
|
||
rep = Path(d) / "vitest.json"
|
||
p = subprocess.run([npm_bin(), "--prefix", "web", "run", "test", "--",
|
||
"--reporter=json", f"--outputFile={rep}"],
|
||
cwd=ROOT, env=go_env(), capture_output=True, text=True)
|
||
print(p.stdout[-3000:], end="")
|
||
if p.stderr.strip():
|
||
print(p.stderr[-2000:], end="", file=sys.stderr)
|
||
status, detail = judge_vitest_file(rep)
|
||
print(("✅ " if status == SUCCESS else "❌ ") + detail, file=sys.stdout if status == SUCCESS else sys.stderr)
|
||
return 0 if status == SUCCESS and p.returncode == 0 else 1
|
||
|
||
|
||
def cmd_internal_contracts(go: str) -> int:
|
||
"""版本契约：注入进二进制的号必须等于 CHANGELOG 顶部，而且**必须真的 PASS**。"""
|
||
rc = subprocess.run(py("scripts/check_changelog.py"), cwd=ROOT).returncode
|
||
if rc:
|
||
return 1
|
||
v = subprocess.run(py("scripts/version.py"), cwd=ROOT, capture_output=True, text=True)
|
||
version = v.stdout.strip()
|
||
if v.returncode or not version:
|
||
print("❌ CHANGELOG 顶部解析不到版本号：没有条目就没有版本号，不许回落成假版本", file=sys.stderr)
|
||
return 1
|
||
print(f"-- 版本契约：CHANGELOG 顶部 = {version} --")
|
||
p = subprocess.run([go, "test", f"-ldflags=-X knowledge-governance/internal/version.Number={version}",
|
||
"-count=1", "-run", "TestNumberMatchesChangelog", "-v", "-json",
|
||
"./internal/version/..."],
|
||
cwd=ROOT, env=go_env(), capture_output=True, text=True)
|
||
text = p.stdout + p.stderr
|
||
print(text[-3000:], end="")
|
||
# ⚠️ 被跳过的断言等于没有断言：必须看到 PASS，SKIP 一律判红。
|
||
# 用 in 而不是 `| grep -q`：pipefail 下 grep -q 会因 SIGPIPE 谎报失败（trap-7）。
|
||
if '--- PASS: TestNumberMatchesChangelog' not in text.replace("\\n", "\n"):
|
||
print("❌ 版本契约没有真正 PASS（被 SKIP 或没跑到）—— 缺 -ldflags，或 CHANGELOG 解析失败", file=sys.stderr)
|
||
return 1
|
||
if "--- SKIP: TestNumberMatchesChangelog" in text:
|
||
print("❌ 版本契约被跳过（未注入版本号）—— 跳过不算通过", file=sys.stderr)
|
||
return 1
|
||
print("✅ 版本契约成立：注入 == CHANGELOG 顶部，且断言真的执行了")
|
||
return 0
|
||
|
||
|
||
def cmd_internal_manifests() -> int:
|
||
"""比对门禁开始前后依赖清单的摘要 —— 抓'某一步顺手改了 go.mod/锁文件'（静默漂移）。"""
|
||
baseline_path = Path(os.environ.get("CI_MANIFEST_BASELINE", str(ROOT / ".ci-manifests.json")))
|
||
if not baseline_path.is_file():
|
||
print("❌ 找不到门禁开始时的清单摘要（说明汇总流程被改坏了）—— 判红", file=sys.stderr)
|
||
return 1
|
||
before = json.loads(baseline_path.read_text(encoding="utf-8"))
|
||
after = {f: digest(ROOT / f) for f in MANIFESTS}
|
||
drift = [f for f in MANIFESTS if before.get(f) != after.get(f)]
|
||
if drift:
|
||
print(f"❌ 这些依赖清单在门禁过程中被改写了：{drift}", file=sys.stderr)
|
||
print(" （GOFLAGS=-mod=readonly / npm ci 就是为了防这个；改写必须由人显式提交）", file=sys.stderr)
|
||
return 1
|
||
print(f"✅ 依赖清单未被改写（{len(MANIFESTS)} 个文件摘要一致）")
|
||
return 0
|
||
|
||
|
||
def digest(p: Path) -> str:
|
||
return hashlib.sha256(p.read_bytes()).hexdigest() if p.is_file() else "missing"
|
||
|
||
|
||
# ── 汇总：纯函数，--self-test 直接打它 ────────────────────────────────────
|
||
# 必需检查清单的**声明**（id → 在哪些档案下必需）。它是汇总的唯一权威依据：
|
||
# 汇总不遍历"跑过的项"，而是拿这份声明去**核对报告**。这样"某项没跑到/被删掉"才会露出来。
|
||
CHECK_MANIFEST: dict[str, tuple[str, ...]] = {
|
||
"identity": ("ci",),
|
||
"commits": ("local", "ci"),
|
||
"deps-install": ("local", "ci"),
|
||
"format": ("local", "ci"),
|
||
"static-go": ("local", "ci"),
|
||
"static-web": ("local", "ci"),
|
||
"build-go": ("local", "ci"),
|
||
"build-web": ("local", "ci"),
|
||
"tests-go": ("local", "ci"),
|
||
"tests-web": ("local", "ci"),
|
||
"contracts": ("local", "ci"),
|
||
"version": ("local", "ci"),
|
||
"security-secrets": ("local", "ci"),
|
||
"security-deps": ("local", "ci"),
|
||
"manifests-locked": ("local", "ci"),
|
||
}
|
||
|
||
FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
|
||
|
||
|
||
def manifest_mismatch(declared: dict, actual: list[tuple[str, tuple[str, ...]]]) -> list[str]:
|
||
"""声明与实现必须逐项一致 —— 否则"清单"就是一句空话（漏声明的项会被当不存在）。"""
|
||
problems = []
|
||
seen = {cid: tuple(req) for cid, req in actual}
|
||
for cid in declared:
|
||
if cid not in seen:
|
||
problems.append(f"清单里声明了 {cid}，但实现里没有这一项")
|
||
for cid, req in seen.items():
|
||
if cid not in declared:
|
||
problems.append(f"实现里有 {cid}，但清单没声明它 —— 汇总会把它当成未知项")
|
||
elif tuple(declared[cid]) != req:
|
||
problems.append(f"{cid} 的档案声明不一致：清单 {declared[cid]} vs 实现 {req}")
|
||
return problems
|
||
|
||
|
||
def aggregate(results: list[dict], profile: str,
|
||
declared: dict[str, tuple[str, ...]] | None = None) -> tuple[str, list[str]]:
|
||
"""把"跑出来的结果"与"声明的必需清单"对账。四类问题都要露出来：
|
||
|
||
1. **重复项**：同一个 id 出现两次（报告被拼接/覆盖过）—— 不许当成一项；
|
||
2. **未知项**：报告里有清单之外的 id ⇒ 清单与实现不一致，不能放行；
|
||
3. **缺项**：清单里在本档案下必需的项，报告里没有 ⇒ 报告不完整，不能报全绿；
|
||
4. **非 success**：failure / skipped / not_run 一律不算过。
|
||
"""
|
||
declared = declared if declared is not None else CHECK_MANIFEST
|
||
problems: list[str] = []
|
||
counts: dict[str, int] = {}
|
||
for r in results:
|
||
cid = str(r.get("id") or "")
|
||
counts[cid] = counts.get(cid, 0) + 1
|
||
if cid not in declared:
|
||
problems.append(f"报告里有清单之外的检查项 {cid} —— 清单与实现不一致，不能放行")
|
||
continue
|
||
if profile not in set(declared[cid]):
|
||
if r["status"] in (FAILURE, "unknown"):
|
||
problems.append(f"{cid} 在 {profile} 档案下不要求，但它状态是 {r['status']} —— 不静默放过")
|
||
continue
|
||
if r["status"] != SUCCESS:
|
||
# 摘要只留前 200 字符：完整原因在报告 JSON 的 reason 字段里
|
||
why = f"｜{str(r.get('reason') or '')[:200]}"
|
||
problems.append(f"{cid} 状态 {r['status']}{why}")
|
||
|
||
for cid, n in counts.items():
|
||
if n > 1:
|
||
problems.append(f"报告里 {cid} 重复出现 {n} 次 —— 重复项会让'哪一次为准'变得不可知")
|
||
for cid, req in declared.items():
|
||
if profile in set(req) and cid not in counts:
|
||
problems.append(f"清单里的必需项 {cid} 在报告里**缺失** —— 报告不完整，不能报全绿")
|
||
return (SUCCESS if not problems else FAILURE), problems
|
||
|
||
|
||
def self_test() -> int:
|
||
def R(cid, status, required=("local", "ci")):
|
||
return {"id": cid, "status": status, "required_in": list(required), "reason": ""}
|
||
|
||
D = {"a": ("local", "ci"), "b": ("local", "ci"), "identity": ("ci",)}
|
||
cases = [
|
||
("全绿", [R("a", SUCCESS), R("b", SUCCESS)], "local", 0),
|
||
("一项失败", [R("a", SUCCESS), R("b", FAILURE)], "local", 1),
|
||
("一项未运行（失败即停的后果）", [R("a", FAILURE), R("b", NOT_RUN)], "local", 2),
|
||
("必需项被跳过", [R("a", SKIPPED), R("b", SUCCESS)], "local", 1),
|
||
("某项在本地档案下不要求且被跳过 ⇒ 不算问题", [R("a", SUCCESS), R("b", SUCCESS), R("identity", SKIPPED, ("ci",))], "local", 0),
|
||
("但它在 ci 档案下就是问题", [R("a", SUCCESS), R("b", SUCCESS), R("identity", SKIPPED, ("ci",))], "ci", 1),
|
||
("ci 档案下本地不要求的项 unknown 也要拦", [R("a", SUCCESS), R("b", SUCCESS), R("identity", "unknown", ("ci",))], "ci", 1),
|
||
("必需项在报告里缺失 ⇒ 不能报全绿", [R("a", SUCCESS)], "local", 1),
|
||
("重复项 ⇒ 不能报全绿", [R("a", SUCCESS), R("a", SUCCESS), R("b", SUCCESS)], "local", 1),
|
||
("清单之外的项 ⇒ 不能报全绿", [R("a", SUCCESS), R("b", SUCCESS), R("zz", SUCCESS)], "local", 1),
|
||
]
|
||
problems = []
|
||
for name, results, profile, want in cases:
|
||
_, got = aggregate(results, profile, D)
|
||
if len(got) != want:
|
||
problems.append(f"{name}：期望 {want} 个问题，实际 {len(got)} 个 {got}")
|
||
# 清单与实现必须一致（声明漏了项会被当成"不存在"，那是静默漏检）
|
||
mm = manifest_mismatch({"a": ("local",)}, [("a", ("local",)), ("b", ("ci",))])
|
||
if not mm:
|
||
problems.append("manifest_mismatch 没报出实现比清单多一项")
|
||
mm2 = manifest_mismatch({"a": ("local",), "b": ("ci",)}, [("a", ("local",))])
|
||
if not mm2:
|
||
problems.append("manifest_mismatch 没报出清声明了但实现没有的项")
|
||
if manifest_mismatch({"a": ("local",)}, [("a", ("local",))]):
|
||
problems.append("manifest_mismatch 对一致的两份数据误报")
|
||
# 完整 SHA 的形状
|
||
if not FULL_SHA.match("a" * 40) or FULL_SHA.match("a" * 12):
|
||
problems.append("完整 SHA 的形状判定不对（12 位短 sha 必须判为不完整）")
|
||
if problems:
|
||
print("❌ 汇总逻辑自检失败：", file=sys.stderr)
|
||
for p in problems:
|
||
print(f" · {p}", file=sys.stderr)
|
||
return 1
|
||
print(f"✅ 汇总逻辑自检通过：{len(cases)} 个分支"
|
||
f"（含'必需项未运行/被跳过/缺失/重复/未知项都不得报全绿'）+ 清单一致性 + 完整 SHA 形状")
|
||
return 0
|
||
|
||
|
||
# ── 身份检查（只在 CI 上有意义，所以单独一段）──────────────────────────
|
||
def identity_result() -> tuple[str, str]:
|
||
want = (os.environ.get("GITHUB_SHA") or "").strip()
|
||
if not want:
|
||
return SKIPPED, "本地运行没有事件负载（GITHUB_SHA 未设置）；CI 上这一项是必需项"
|
||
# 必须是**完整 40 位** SHA：拿短 sha 比起不了身份作用（前缀撞车不报错）
|
||
if not FULL_SHA.match(want):
|
||
return FAILURE, f"事件 SHA 不是完整 40 位十六进制：{want[:20]!r} —— 身份核对拿它做不了"
|
||
got = subprocess.run(["git", "rev-parse", "HEAD"], cwd=ROOT,
|
||
capture_output=True, text=True).stdout.strip()
|
||
if not FULL_SHA.match(got or ""):
|
||
return FAILURE, f"实际 HEAD 不是完整 SHA：{got[:20]!r}"
|
||
if got != want:
|
||
return FAILURE, f"实际 HEAD {got[:12]} != 事件 SHA {want[:12]}（分支在推送后又前进了？）"
|
||
return SUCCESS, f"检出 {got} 与事件完整 SHA 逐字符相同"
|
||
|
||
|
||
def main(argv: list[str]) -> int:
|
||
ap = argparse.ArgumentParser(add_help=True)
|
||
ap.add_argument("--list", action="store_true")
|
||
ap.add_argument("--only", default="")
|
||
ap.add_argument("--self-test", action="store_true")
|
||
ap.add_argument("--report", default=str(REPORT_DEFAULT))
|
||
ap.add_argument("--profile", default="ci" if os.environ.get("GITHUB_ACTIONS") else "local",
|
||
choices=["local", "ci"])
|
||
ap.add_argument("--internal-deps", default="")
|
||
ap.add_argument("--internal-build", default="")
|
||
ap.add_argument("--internal-web-test", action="store_true")
|
||
ap.add_argument("--internal-contracts", default="")
|
||
ap.add_argument("--internal-manifests", action="store_true")
|
||
args, _ = ap.parse_known_args(argv)
|
||
|
||
# 内部子命令：把"一条复杂命令"收进本文件，清单与 YAML 里只留一行。
|
||
# 先处理它们，避免被下面的门禁流程当成一次普通运行。
|
||
if args.internal_deps:
|
||
return cmd_internal_deps(args.internal_deps)
|
||
if args.internal_build:
|
||
return cmd_internal_build(args.internal_build)
|
||
if args.internal_contracts:
|
||
return cmd_internal_contracts(args.internal_contracts)
|
||
if args.internal_web_test:
|
||
return cmd_internal_web_test()
|
||
if args.internal_manifests:
|
||
return cmd_internal_manifests()
|
||
if args.self_test:
|
||
return self_test()
|
||
|
||
table = checks()
|
||
if args.list:
|
||
print(f"── 本仓必需检查清单（策略版本 {POLICY_REVISION}）──")
|
||
for c in table:
|
||
print(f" {c.id:<18} {'/'.join(c.required_in):<12} {c.title}")
|
||
if c.note:
|
||
print(f" {'':<18} {'':<12} ⚠️ {c.note}")
|
||
return 0
|
||
|
||
if args.only:
|
||
want = [x for x in args.only.split(",") if x]
|
||
unknown = [x for x in want if x not in {c.id for c in table}]
|
||
if unknown:
|
||
print(f"❌ 没有这些检查 id：{unknown}", file=sys.stderr)
|
||
return 2
|
||
table = [c for c in table if c.id in want]
|
||
print(f"⚠️ --only {args.only}：只跑了子集，**不算全绿**（报告里 overall_status 会写 not_run_only）")
|
||
|
||
# 门禁开始前记录依赖清单摘要：末尾断言"没有任何一步改写了它们"
|
||
baseline = ROOT / ".ci-manifests.json"
|
||
baseline.write_text(json.dumps({f: digest(ROOT / f) for f in MANIFESTS}), encoding="utf-8")
|
||
os.environ["CI_MANIFEST_BASELINE"] = str(baseline)
|
||
|
||
results: list[dict] = []
|
||
failed_early = False
|
||
started = time.time()
|
||
for c in table:
|
||
if failed_early:
|
||
results.append({"id": c.id, "title": c.title, "required_in": list(c.required_in),
|
||
"status": NOT_RUN, "reason": "前一项失败即停，未运行", "command": "",
|
||
"exit_code": None, "duration_s": 0})
|
||
continue
|
||
if c.id == "identity":
|
||
status, reason = identity_result()
|
||
results.append({"id": c.id, "title": c.title, "required_in": list(c.required_in),
|
||
"status": status, "reason": reason, "command": "git rev-parse HEAD",
|
||
"exit_code": 0 if status == SUCCESS else 1, "duration_s": 0})
|
||
failed_early = status == FAILURE
|
||
print(f"\n── [{c.id}] {c.title} ──\n {'✅' if status == SUCCESS else ('⏭️' if status == SKIPPED else '❌')} {reason}")
|
||
continue
|
||
|
||
print(f"\n── [{c.id}] {c.title} ──")
|
||
print(" $ " + " ".join(c.argv))
|
||
t0 = time.time()
|
||
try:
|
||
p = subprocess.run(c.argv, cwd=c.cwd, env={**os.environ, **c.env},
|
||
capture_output=True, text=True)
|
||
exit_code, out, err = p.returncode, p.stdout, p.stderr
|
||
except FileNotFoundError as e:
|
||
exit_code, out, err = 127, "", f"工具缺失：{e}（工具缺失 ⇒ 失败，不当成检查通过）"
|
||
dur = round(time.time() - t0, 2)
|
||
status, reason = c.judge(exit_code, out, err)
|
||
if out.strip():
|
||
print("\n".join(" | " + l for l in out.strip().splitlines()[-40:]))
|
||
if err.strip() and status != SUCCESS:
|
||
print("\n".join(" ! " + l for l in err.strip().splitlines()[-20:]), file=sys.stderr)
|
||
print(f" {'✅' if status == SUCCESS else '❌'} {reason}（{dur}s）")
|
||
results.append({"id": c.id, "title": c.title, "required_in": list(c.required_in),
|
||
"status": status, "reason": reason, "command": " ".join(c.argv),
|
||
"exit_code": exit_code, "duration_s": dur})
|
||
failed_early = status != SUCCESS
|
||
|
||
head_sha = subprocess.run(["git", "rev-parse", "HEAD"], cwd=ROOT, capture_output=True,
|
||
text=True).stdout.strip()
|
||
env_sha = (os.environ.get("GITHUB_SHA") or "").strip()
|
||
result_sha = env_sha or head_sha
|
||
|
||
extra: list[str] = []
|
||
# 档案必须与运行环境相符：在 CI 里报一个 local 档案，identity 会被合法跳过 —— 那是后门
|
||
if os.environ.get("GITHUB_ACTIONS") and args.profile != "ci":
|
||
extra.append(f"运行环境是 CI（GITHUB_ACTIONS 已设置），但档案写的是 {args.profile} —— "
|
||
f"local 档案下 identity 不要求，等于把身份核对关掉")
|
||
if not FULL_SHA.match(result_sha or ""):
|
||
extra.append(f"报告的 commit_sha 不是完整 40 位 SHA：{result_sha[:20]!r} —— 归属无法确认")
|
||
if env_sha and not FULL_SHA.match(env_sha):
|
||
extra.append(f"事件 SHA 不是完整 40 位：{env_sha[:20]!r}")
|
||
if env_sha and head_sha != env_sha:
|
||
extra.append(f"实际 HEAD {head_sha[:12]} != 事件 SHA {env_sha[:12]}")
|
||
|
||
# `--only` 只跑子集：这时"必需清单"就是被选中的那几项 ——
|
||
# 否则汇总会把没选的项全报成"缺失"，单项调试就永远红（那是噪声，不是判据）。
|
||
# 报告里会明确写 run_mode=only / overall_status=not_run_only，不会被当成 CI 证据。
|
||
declared = ({c.id: c.required_in for c in table} if args.only else CHECK_MANIFEST)
|
||
overall, problems = aggregate(results, args.profile, declared)
|
||
problems = problems + extra
|
||
if args.only:
|
||
overall = "not_run_only" if overall == SUCCESS and not problems else FAILURE
|
||
elif problems and overall == SUCCESS:
|
||
overall = FAILURE
|
||
report = {
|
||
"schema": "ci-report/1",
|
||
"policy_revision": POLICY_REVISION,
|
||
"repository": os.environ.get("GITHUB_REPOSITORY", ""),
|
||
"commit_sha": result_sha,
|
||
"event_sha": env_sha,
|
||
"branch": os.environ.get("GITHUB_REF_NAME", ""),
|
||
"run_id": os.environ.get("GITHUB_RUN_ID", ""),
|
||
"event": os.environ.get("GITHUB_EVENT_NAME", "local"),
|
||
"profile": args.profile,
|
||
"started_at": dt.datetime.fromtimestamp(started, dt.timezone.utc).isoformat(timespec="seconds"),
|
||
"duration_s": round(time.time() - started, 2),
|
||
"checks": results,
|
||
"run_mode": "only" if args.only else "full",
|
||
# 把"这一次跑的时候，清单长什么样"一起写进报告：上报方要拿它核对，
|
||
# 而不是只看 overall_status 一个字段（只看它等于信一句没有任何依据的结论）。
|
||
"manifest": {cid: list(req) for cid, req in CHECK_MANIFEST.items()},
|
||
"overall_status": overall,
|
||
"problems": problems,
|
||
"notification_status": "not_attempted",
|
||
"deployed": False,
|
||
}
|
||
Path(args.report).write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
baseline.unlink(missing_ok=True)
|
||
|
||
print("\n════════ 汇总 ════════")
|
||
n = len(results)
|
||
print(f" 检查 {n} 项：" + "｜".join(f"{r['id']}={r['status']}" for r in results))
|
||
print(f" 总体：{overall}（档案 {args.profile}） 报告：{args.report}")
|
||
if args.only:
|
||
ok = all(r["status"] == SUCCESS for r in results)
|
||
print(f" {'（子集全过，但**不算全绿** —— 只跑了 ' + args.only + '）' if ok else '（子集里有失败项）'}")
|
||
return 0 if ok else 1
|
||
if problems:
|
||
for p in problems:
|
||
print(f" ❌ {p}", file=sys.stderr)
|
||
print(" ⚠️ 注意：'未运行'不是'通过' —— 失败即停之后剩下的项一律不参与'全绿'判定。", file=sys.stderr)
|
||
return 1
|
||
print(" ✅ 全部必需检查通过（本结论含全部必需项，不含任何未运行项）")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main(sys.argv[1:])) |