姚章浩
a3a1319a11
一次性把本地与 CI 的门禁收成一条命令：make verify = python3 scripts/ci_checks.py。 两边跑的是同一份逻辑，不再各写一套步骤（两边各写一套的结局是迟早只改一边）。 14 项必需检查：identity（检出 == 事件完整 SHA，只在 CI 必需）、commits、deps-install、 format、static-go、static-web、build-go、build-web、tests-go、tests-web、contracts、 security-secrets、security-deps、manifests-locked。make list 打印清单。 三条判据写死在代码里： - 失败即停，但剩下的记 not_run 并如实写进报告；aggregate() 只认 success， 未运行/被跳过都不算通过（不适用要在清单里声明 required_in，不许运行时悄悄 skip）。 - 工具缺失、依赖没按锁装、输出解析不了 ⇒ 失败，不当成"检查通过"。 - 测试必须真跑：go test 加 -count=1 并盘点通过/跳过数；前端读 vitest 的 JSON 报告， 空套件不算通过。跳过原因会打进报告（当前 72 个集成用例因缺 TEST_DATABASE_URL 跳过）。 新增四个自带反向自检的脚本（--self-test 故意弄坏必须红）： scripts/check_format.py（Go 判输出而非退出码 + 前端只能减的基线）、 scripts/check_secrets.py（只打印位置与类别，绝不打印值）、 scripts/check_deps_vuln.py（govulncheck 锁 v1.1.4 + npm audit 分级 + 到期例外）、 scripts/ci_checks.py（门禁 + 汇总）。 验证（本机实跑，全部为真实退出码）： make verify → 总体 success（档案 local），identity=skipped 已在清单里声明为 CI 必需； tests-go 执行 31 个包、通过 549 / 失败 0 / 跳过 72；tests-web 106 个套件、191 个用例全过； security-deps：govulncheck v1.1.4 0 命中、npm audit 生产与全量均 0。
498 lines
24 KiB
Python
498 lines
24 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/1" # 判据变更时改这里，报告里跟着变
|
||
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)
|
||
|
||
|
||
def judge_go_tests(exit_code: int, out: str, err: str) -> tuple[str, str]:
|
||
"""`go test -json` 的盘点：必须**真的执行过**用例，且把跳过的原因说清楚。
|
||
|
||
为什么不用退出码了事：退出码 0 只说明"没有失败"，一个用例都没跑的包也是 0。
|
||
被跳过的断言等于没有断言（正本 §5.5）。"""
|
||
passed = failed = skipped = 0
|
||
reasons: list[str] = []
|
||
pending = ""
|
||
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
|
||
if ev.get("Package"):
|
||
pkgs.add(ev["Package"])
|
||
action, test = ev.get("Action"), ev.get("Test")
|
||
if test and action in ("pass", "fail", "skip"):
|
||
passed += action == "pass"
|
||
failed += action == "fail"
|
||
skipped += action == "skip"
|
||
chunk = ev.get("Output") or ""
|
||
m = re.match(r"\s*\S+_test\.go:\d+:\s*(.+)$", chunk)
|
||
if m:
|
||
pending = m.group(1).strip()
|
||
if "--- SKIP:" in chunk and pending and pending not in reasons:
|
||
reasons.append(pending)
|
||
detail = f"执行包 {len(pkgs)}｜用例 通过 {passed} / 失败 {failed} / 跳过 {skipped}"
|
||
if skipped:
|
||
detail += f"｜跳过的原因（前 3 条）：{reasons[:3] or '未取到'}"
|
||
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 + "｜一个用例都没真的执行 —— 无测试不能算全绿"
|
||
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)
|
||
detail = f"套件 {suites}｜用例 通过 {passed} / 失败 {failed} / 共 {total}"
|
||
if failed:
|
||
return FAILURE, detail
|
||
if total == 0 or suites == 0:
|
||
return FAILURE, detail + "｜一个用例都没执行（空套件不能算通过）"
|
||
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"))
|
||
return [
|
||
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("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"]),
|
||
]
|
||
|
||
|
||
# ── 内部子命令（把"一条复杂命令"收进脚本，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 直接打它 ────────────────────────────────────
|
||
def aggregate(results: list[dict], profile: str) -> tuple[str, list[str]]:
|
||
problems = []
|
||
for r in results:
|
||
declared = set(r["required_in"])
|
||
if profile not in declared:
|
||
if r["status"] in (FAILURE, "unknown"):
|
||
problems.append(f"{r['id']} 在 {profile} 档案下不要求，但它状态是 {r['status']} —— 不静默放过")
|
||
continue
|
||
if r["status"] != SUCCESS:
|
||
# 摘要只留前 200 字符：完整原因在报告 JSON 的 reason 字段里
|
||
why = f"｜{r['reason'][:200]}" if r.get("reason") else ""
|
||
problems.append(f"{r['id']} 状态 {r['status']}{why}")
|
||
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": ""}
|
||
|
||
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)], "local", 1),
|
||
("某项在本地档案下不要求且被跳过 ⇒ 不算问题", [R("identity", SKIPPED, ("ci",))], "local", 0),
|
||
("但它在 ci 档案下就是问题", [R("identity", SKIPPED, ("ci",))], "ci", 1),
|
||
("ci 档案下本地不要求的项 unknown 也要拦", [R("identity", "unknown", ("ci",))], "ci", 1),
|
||
]
|
||
problems = []
|
||
for name, results, profile, want in cases:
|
||
_, got = aggregate(results, profile)
|
||
if len(got) != want:
|
||
problems.append(f"{name}：期望 {want} 个问题，实际 {len(got)} 个 {got}")
|
||
if problems:
|
||
print("❌ 汇总逻辑自检失败：", file=sys.stderr)
|
||
for p in problems:
|
||
print(f" · {p}", file=sys.stderr)
|
||
return 1
|
||
print(f"✅ 汇总逻辑自检通过：{len(cases)} 个分支（含'必需项未运行/被跳过不得报全绿'）")
|
||
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 上这一项是必需项"
|
||
got = subprocess.run(["git", "rev-parse", "HEAD"], cwd=ROOT,
|
||
capture_output=True, text=True).stdout.strip()
|
||
if got != want:
|
||
return FAILURE, f"实际 HEAD {got[:12]} != 事件 SHA {want[:12]}（分支在推送后又前进了？）"
|
||
return SUCCESS, f"检出 {got[:12]} == 事件完整 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
|
||
|
||
overall, problems = aggregate(results, args.profile)
|
||
if args.only:
|
||
overall = "not_run_only"
|
||
report = {
|
||
"schema": "ci-report/1",
|
||
"policy_revision": POLICY_REVISION,
|
||
"repository": os.environ.get("GITHUB_REPOSITORY", ""),
|
||
"commit_sha": subprocess.run(["git", "rev-parse", "HEAD"], cwd=ROOT, capture_output=True,
|
||
text=True).stdout.strip(),
|
||
"event_sha": os.environ.get("GITHUB_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,
|
||
"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 problems:
|
||
for p in problems:
|
||
print(f" ❌ {p}", file=sys.stderr)
|
||
print(" ⚠️ 注意：'未运行'不是'通过' —— 失败即停之后剩下的项一律不参与'全绿'判定。", file=sys.stderr)
|
||
return 1
|
||
if args.only:
|
||
return 0 if all(r["status"] == SUCCESS for r in results) else 1
|
||
print(" ✅ 全部必需检查通过（本结论含全部必需项，不含任何未运行项）")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main(sys.argv[1:])) |