6de5d52363
build / build (push) Successful in 51s
按使用者要求定的版本规则(自动升的是**开发标识**,发布版本与 tag 仍由人批准): - **开发号自动递增**:display_version(development) = `<正式版本的主.次>.<自 v<正式版本> 以来的提交数>` (正式 0.1.0 之后第 1 个提交 0.1.1、第 12 个 0.1.12;恰在发布提交上 = 正式号)。 它**不写进 CHANGELOG**(发布记录只在发版时动),由 git 算出来 ⇒ 每个提交都有自己的号、数字只增。 - **浅克隆不猜号**:`--depth` 会把条数算少 ⇒ 浅仓库里拒绝算号,退回 `<版本>-dev+g<短SHA>`; 检出步骤在仓库有 tag 时 `--unshallow` 补全历史。 - 接线三处齐全:`version.py --identity` → 门禁 `-X …CommitsSinceRelease=<n>`;Makefile 的 `COMMITS_SINCE_RELEASE`;Dockerfile 同名 build-arg;Go 侧新增 `CommitsSinceRelease` 与 DisplayVersion() 新规则;契约测试**交叉核对**"注入的提交数 == git 算出的"(防注入错号)。 - **首个正式版本 0.1.0**:CHANGELOG 的 `[Unreleased]` 提成 `## [0.1.0] - 2026-10-13`; 头部补"版本怎么走"两行;RELEASE_RUNBOOK 补递增规则与"下一次发布把次版本加一"。 - 发版预检新增:打印**当前开发号**,并挡住"候选正式版本 <= 当前开发号"(会把开发号当正式号;建议次版本加一)。 验证:version.py --self-test(含 tag 后第 2 个提交 = 0.2.2、恰在 tag 上 = 0.2.0、无 tag 不猜号); release_preflight --self-test(21 分支,含开发号两条);go test 本地三态 + 临时仓库 tag+2 提交实测 (Python 与 Go 都得出 0.2.2);反向夹具 85/0;make verify 15/15 success。
722 lines
36 KiB
Python
722 lines
36 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/3" # 判据变更时改这里,报告里跟着变
|
||
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 build_identity() -> tuple[dict | None, str]:
|
||
"""构建身份(`scripts/version.py --identity`)。
|
||
|
||
⚠️ 刻意**不**跑裸的 `scripts/version.py`:那个在"还没有正式版本"时会非 0 退出,
|
||
而开发构建**必须**能正常产出身份(release_version=null / display_version=dev+g<短SHA>)。
|
||
"""
|
||
p = subprocess.run(py("scripts/version.py", "--identity"), cwd=ROOT, capture_output=True, text=True)
|
||
if p.returncode:
|
||
return None, ((p.stderr or p.stdout).strip().splitlines() or [f"退出码 {p.returncode}"])[0][:400]
|
||
try:
|
||
return json.loads(p.stdout), ""
|
||
except ValueError as e:
|
||
return None, f"--identity 的输出不是 JSON:{e}"
|
||
|
||
|
||
def ldflags_for(ident: dict, *, build_kind: str | None = None,
|
||
release_version: str | None = None) -> str:
|
||
"""把构建身份编成 -ldflags(负例可以覆盖 kind / release_version)。"""
|
||
kind = build_kind or str(ident.get("build_kind") or "development")
|
||
rel = ident.get("release_version") if release_version is None else release_version
|
||
return " ".join([
|
||
f"-X knowledge-governance/internal/version.Number={rel or 'dev'}",
|
||
f"-X knowledge-governance/internal/version.ReleaseVersion={rel or ''}",
|
||
f"-X knowledge-governance/internal/version.BuildKind={kind}",
|
||
f"-X knowledge-governance/internal/version.CommitsSinceRelease="
|
||
f"{'' if ident.get('commits_since_release') is None else ident['commits_since_release']}",
|
||
f"-X knowledge-governance/internal/version.CommitSHA={ident.get('commit_sha') or 'unknown'}",
|
||
f"-X knowledge-governance/internal/version.BuildTime={ident.get('built_at') or 'unknown'}",
|
||
])
|
||
|
||
|
||
def cmd_internal_build(out_dir: str) -> int:
|
||
out = Path(out_dir)
|
||
out.mkdir(parents=True, exist_ok=True)
|
||
ident, err = build_identity()
|
||
if ident is None:
|
||
print(f"❌ 拿不到构建身份:{err}", file=sys.stderr)
|
||
return 1
|
||
ldflags = ldflags_for(ident)
|
||
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(显示版本 {ident['display_version']}|{ident['build_kind']}|"
|
||
f"提交 {str(ident['commit_sha'])[:12]})--")
|
||
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:
|
||
"""版本契约:结构 + 注入身份必须**真 PASS**(SKIP 判红),
|
||
并且**开发构建不许自称正式版**(负例:标成 release ⇒ 必须红)。"""
|
||
rc = subprocess.run(py("scripts/check_changelog.py"), cwd=ROOT).returncode
|
||
if rc:
|
||
return 1
|
||
ident, err = build_identity()
|
||
if ident is None:
|
||
print(f"❌ 拿不到构建身份:{err}", file=sys.stderr)
|
||
return 1
|
||
print(f"-- 版本契约:显示版本 {ident['display_version']}({ident['build_kind']}"
|
||
f"|release_version={ident['release_version']!r})--")
|
||
tests = ("TestNumberMatchesChangelog", "TestDevBuildDoesNotClaimRelease")
|
||
|
||
def run(ld: str) -> tuple[int, str]:
|
||
p = subprocess.run([go, "test", f"-ldflags={ld}", "-count=1", "-run", "Test", "-v", "-json",
|
||
"./internal/version/..."], cwd=ROOT, env=go_env(),
|
||
capture_output=True, text=True)
|
||
return p.returncode, (p.stdout + p.stderr).replace("\\n", "\n")
|
||
|
||
rc1, text = run(ldflags_for(ident))
|
||
print(text[-2000:], end="")
|
||
# ⚠️ 被跳过的断言等于没有断言:必须看到 PASS,SKIP 一律判红。
|
||
# 用 in 而不是 `| grep -q`:pipefail 下 grep -q 会因 SIGPIPE 谎报失败(trap-7)。
|
||
for name in tests:
|
||
if f"--- PASS: {name}" not in text:
|
||
print(f"❌ 版本契约 {name} 没有真正 PASS(被 SKIP 或没跑到)—— "
|
||
f"缺 -ldflags,或 CHANGELOG / 构建身份有问题", file=sys.stderr)
|
||
return 1
|
||
if f"--- SKIP: {name}" in text:
|
||
print(f"❌ 版本契约 {name} 被跳过(构建身份没注入)—— 跳过不算通过", file=sys.stderr)
|
||
return 1
|
||
|
||
# 负例:把开发构建标成正式版(ReleaseVersion 指向 CHANGELOG 里没有的号)⇒ **必须红**。
|
||
# 没有这条,"开发版沿用正式版号"这个病会静默回归。
|
||
bogus = "9.9.9" if ident.get("release_version") != "9.9.9" else "8.8.8"
|
||
rc2, _ = run(ldflags_for(ident, build_kind="release", release_version=bogus))
|
||
if rc2 == 0:
|
||
print(f"❌ 把构建标成正式版(BuildKind=release / ReleaseVersion={bogus})竟然还是绿的 —— "
|
||
f"这条断言是死的", file=sys.stderr)
|
||
return 1
|
||
print(f"-- 负例确认:自称正式版({bogus})⇒ 红 ✓ --")
|
||
print("✅ 版本契约成立:注入 == CHANGELOG 顶部(没有正式版本时为空),"
|
||
"开发构建带 dev 标识,且断言真的执行了")
|
||
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:])) |