diff --git a/.github/workflows/gate.yml b/.github/workflows/gate.yml index d7ff436..cd8a7dd 100644 --- a/.github/workflows/gate.yml +++ b/.github/workflows/gate.yml @@ -89,7 +89,10 @@ jobs: bash -n governance/apply.sh && bash -n governance/drift-check.sh && bash -n scripts/new-repo-init.sh && bash -n scripts/gh-app-token.sh && bash -n scripts/ghcb \ && bash -n governance/auto-fix-limit.sh && bash -n governance/cost-check.sh \ && bash -n governance/butler-reconcile.sh \ - && bash -n governance/deadman-trip.sh && bash -n governance/butler-audit.sh + && bash -n governance/deadman-trip.sh && bash -n governance/butler-audit.sh \ + && bash -n governance/tests/test-drill.sh \ + && bash -n governance/drill/tests/test-samples.sh && bash -n governance/drill/tests/test-select.sh \ + && bash -n governance/drill/tests/test-verify.sh echo "OK scripts" - name: 治理脚本自测(governance/tests) run: | diff --git a/governance/drill/drill.py b/governance/drill/drill.py index 1624df9..b50c1d1 100644 --- a/governance/drill/drill.py +++ b/governance/drill/drill.py @@ -1,16 +1,19 @@ #!/usr/bin/env python3 -"""drill.py —— 周种子缺陷演习引擎·读面(宪法 §4B/§6 / ADR-0069 / .github#223 W4-C4) +"""drill.py —— 周种子缺陷演习引擎(宪法 §4B/§6 / ADR-0069 / .github#223 W4-C4) 子命令(全部 fail-closed:任何失败非零退出,绝不静默降级为"通过"): select 随机选缺陷样本 + 随机目标仓(--seed 注入可复现——测试与事后复盘) + inject 向目标仓开演习分支(只开分支不开 PR)注入缺陷,输出 head_sha decode owner 审阅用:解码样本缺陷内容(ADR-0069 决策 1 样本库 owner 直管) + record 演习记录追加 governance/drill/history.jsonl(append-only 台账=dashboard 数据源) + redrate 聚合红率(目标 ≈100%,AC-4)+ 样本难度按周分布(防"为演习写代码"Goodhart) -注入/独立验证/台账(inject/record/redrate)随后续 PR 落地——红绿判定与样本库 -分离(verify_gate.py,ADR-0069 决策 2"注入者与判定者分离")。注入物只落在 -演习分支(隔离执行,不进 agent 工作区,ADR-0069 风险缓解);分支验后即删。 +红绿判定不在本文件——verify_gate.py 与样本库分离(独立验证步骤,ADR-0069 +决策 2"注入者与判定者分离")。注入物只落在演习分支(隔离执行,不进 agent +工作区,ADR-0069 风险缓解);分支验后即删。 """ import argparse -import base64 # decode 用(样本 defect_b64) +import base64 import json import os import random @@ -139,6 +142,60 @@ def cmd_select(a): ensure_ascii=False)) +def _git(env_extra, *args, cwd=None, check=True): + env = dict(os.environ, GIT_TERMINAL_PROMPT="0", **env_extra) + p = subprocess.run(["git", *args], cwd=cwd, env=env, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) + if check and p.returncode != 0: + die(f"git {' '.join(args[:3])} 失败(rc={p.returncode}): {p.stdout.strip()[:400]}") + return p + + +def cmd_inject(a): + samples = {s["id"]: s for s in load_samples(a.samples)} + if a.sample_id not in samples: + die(f"样本不存在: {a.sample_id}") + s = samples[a.sample_id] + date = a.date or datetime.now(timezone.utc).strftime("%Y%m%d") + prefix = os.environ.get("DRILL_GIT_PREFIX", "https://github.com") + push_url = f"https://github.com/{ORG}/{a.repo}.git" # push 一律直连(镜像仅 fetch) + auth = {} + tok = os.environ.get("DRILL_TOKEN") + if tok: # CI: GOVERNANCE_TOKEN 经 git env-config 下发 extraheader,不落 remote URL/日志 + hdr = "Authorization: Basic " + base64.b64encode(f"x-access-token:{tok}".encode()).decode() + auth = {"GIT_CONFIG_COUNT": "1", + "GIT_CONFIG_KEY_0": "http.https://github.com/.extraheader", + "GIT_CONFIG_VALUE_0": hdr} + branch = a.branch or f"drill/seed-{date}" + with tempfile.TemporaryDirectory(prefix="drill-") as tmp: + _git({}, "clone", "--depth", "1", f"{prefix}/{ORG}/{a.repo}.git", tmp) + _git({}, "checkout", "-B", branch, cwd=tmp) + path = os.path.join(tmp, *s["payload_path"].format(DATE=date).split("/")) + os.makedirs(os.path.dirname(path), exist_ok=True) + if s["payload_kind"] == "file": + data = base64.b64decode(s["defect_b64"]) + else: # generated: 零填充大文件(hygiene >5MB 禁入规则的靶) + data = b"\0" * s["size_bytes"] + with open(path, "wb") as f: + f.write(data) + _git({}, "add", path, cwd=tmp) + _git({}, "-c", "user.name=drill-seed-bot", "-c", "user.email=drill-bot@users.noreply.github.com", + "commit", "-m", f"drill(seed): 注入演习样本 {s['id']}(ADR-0069 周演习,验后即删)", + cwd=tmp) + sha = _git({}, "rev-parse", "HEAD", cwd=tmp).stdout.strip() + last_rc = 1 + for _ in range(3): # push 间歇失败重试(org 网络现实) + p = _git(auth, "push", push_url, f"HEAD:refs/heads/{branch}", cwd=tmp, check=False) + last_rc = p.returncode + if last_rc == 0: + break + if last_rc != 0: + die(f"演习分支 push 失败(3 次重试后): {p.stdout.strip()[:400]}") + print(json.dumps({"branch": branch, "head_sha": sha, + "path": s["payload_path"].format(DATE=date), "sample_id": s["id"]}, + ensure_ascii=False)) + + def cmd_decode(a): samples = {s["id"]: s for s in load_samples(a.samples)} if a.id not in samples: @@ -150,6 +207,59 @@ def cmd_decode(a): sys.stdout.write(base64.b64decode(s["defect_b64"]).decode("utf-8")) +def cmd_record(a): + """append-only 台账:时间戳严格递增 + 同 run 不重复(AC-4 聚合的数据完整性前提)。""" + try: + rec = json.loads(a.json) + except Exception as e: + die(f"--json 解析失败: {e}") + if not rec.get("ts") or not rec.get("kind"): + die("记录缺 ts/kind 字段") + lines = [] + if os.path.exists(a.history): + lines = [l for l in open(a.history, encoding="utf-8").read().splitlines() if l.strip()] + for l in lines: + prev = json.loads(l) # 既有行畸形=台账已损坏,拒绝追加(fail-closed) + if prev["ts"] >= rec["ts"]: + die(f"append-only 破坏: 新 ts {rec['ts']} 不晚于末行 {prev['ts']}") + if (prev.get("kind"), prev.get("run_id")) == (rec["kind"], rec.get("run_id")) \ + and rec.get("run_id") is not None: + die(f"同一 run 的 {rec['kind']} 记录已存在(run_id={rec['run_id']})") + with open(a.history, "a", encoding="utf-8", newline="\n") as f: + f.write(json.dumps(rec, ensure_ascii=False, sort_keys=True) + "\n") + print(f"OK append 1 行(现有 {len(lines) + 1} 行)") + + +def cmd_redrate(a): + """红率(目标 ≈100%)+ 难度按周分布(AC-4;Goodhart 防护=趋势可见而非打分)。""" + if not os.path.exists(a.history): + die(f"台账不存在: {a.history}") + rows = [json.loads(l) for l in open(a.history, encoding="utf-8").read().splitlines() if l.strip()] + seeds = [r for r in rows if r.get("kind") == "seed-drill"] + verdicts = {} + for r in seeds: + verdicts[r.get("verdict", "?")] = verdicts.get(r.get("verdict", "?"), 0) + 1 + red, green = verdicts.get("red", 0), verdicts.get("green", 0) + denom = red + green + # 零分母 → null(诚实口径:不除零不出假 100%,同 dashboard-update SLI 原则) + rate = round(red / denom, 4) if denom else None + weeks = {} + for r in seeds: + wk = r["ts"][:10] + d = r.get("difficulty", "?") + weeks.setdefault(wk, {}).setdefault(d, 0) + weeks[wk][d] += 1 + fails = [r for r in rows if r.get("kind") == "failclose-drill"] + out = {"total_records": len(rows), "seed_drills": len(seeds), "verdicts": verdicts, + "red_rate": rate, "red_rate_note": None if denom else "N/A(尚无可判定演习)", + "difficulty_trend_by_day": {k: dict(sorted(v.items())) for k, v in sorted(weeks.items())}, + "failclose_drills": len(fails), + "failclose_last": fails[-1].get("outcome") if fails else None} + print(json.dumps(out, ensure_ascii=False, indent=2)) + if a.fail_unhealthy and denom and rate < 0.999: + die(f"红率 {rate} < 100%——存在关卡失灵(green={green}),按 AC-4 告警") + + def main(): ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) sub = ap.add_subparsers(dest="cmd", required=True) @@ -163,11 +273,29 @@ def main(): p.add_argument("--target-repo", help="跳过随机,指定目标仓(复盘/首演用)") p.set_defaults(func=cmd_select) + p = sub.add_parser("inject", help="开演习分支注入缺陷") + p.add_argument("--samples", default=os.path.join(here, "samples", "registry.yaml")) + p.add_argument("--sample-id", required=True) + p.add_argument("--repo", required=True) + p.add_argument("--branch", default=None, help="缺省 drill/seed-YYYYMMDD") + p.add_argument("--date", default=None, help="payload {DATE} 占位值(缺省今天)") + p.set_defaults(func=cmd_inject) + p = sub.add_parser("decode", help="owner 审阅:解码样本内容") p.add_argument("--samples", default=os.path.join(here, "samples", "registry.yaml")) p.add_argument("--id", required=True) p.set_defaults(func=cmd_decode) + p = sub.add_parser("record", help="追加 history.jsonl(append-only)") + p.add_argument("--history", default=os.path.join(here, "history.jsonl")) + p.add_argument("--json", required=True) + p.set_defaults(func=cmd_record) + + p = sub.add_parser("redrate", help="红率+难度趋势聚合(AC-4)") + p.add_argument("--history", default=os.path.join(here, "history.jsonl")) + p.add_argument("--fail-unhealthy", action="store_true", + help="红率<100% 时非零退出(workflow 告警用)") + p.set_defaults(func=cmd_redrate) a = ap.parse_args() if a.cmd == "select": diff --git a/governance/drill/tests/test-select.sh b/governance/drill/tests/test-select.sh new file mode 100644 index 0000000..cb7aade --- /dev/null +++ b/governance/drill/tests/test-select.sh @@ -0,0 +1,49 @@ +#!/usr/bin/env bash +# test-select.sh —— 选样随机性自测(W4-C4):seed 可注入(确定性复盘)+ 目标池约束 +set -uo pipefail +HERE="$(cd "$(dirname "$0")" && pwd)" +ROOT="$(cd "$HERE/.." && pwd)" +source "$(cd "$(dirname "$0")" && pwd)/lib.sh" +PYTHON="$(pick_py)" || { echo "::error::无可用 python(含 pyyaml)"; exit 2; } +PASS=0; FAIL=0 +TMP="$(mktemp -d)"; trap 'rm -rf "$TMP"' EXIT + +sel() { "$PYTHON" "$ROOT/drill.py" select --samples "$ROOT/samples/registry.yaml" \ + --repos "$ROOT/../REPOS.yaml" "$@"; } + +echo "== 1) 同 seed 确定性(随机可注入=可复盘可审计)" +sel --seed 20260822 > "$TMP/a.json"; sel --seed 20260822 > "$TMP/b.json" +if cmp -s "$TMP/a.json" "$TMP/b.json"; then PASS=$((PASS+1)); echo "ok 同 seed 输出一致" +else FAIL=$((FAIL+1)); echo "FAIL 同 seed 输出漂移"; cat "$TMP/a.json" "$TMP/b.json"; fi + +echo "== 2) 目标池约束(holdout 永不入池;github-scope 样本只打治理总仓)" +BAD=0 +for s in 1 2 3 5 8 13 21 34 55; do + R=$(sel --seed "$s" | "$PYTHON" -c 'import json,sys;print(json.load(sys.stdin)["target_repo"])') + [[ "$R" == "holdout" ]] && { echo "FAIL seed=$s 选中 holdout(隔离面污染)"; BAD=1; } +done +[[ $BAD -eq 0 ]] && { PASS=$((PASS+1)); echo "ok 多 seed 采样 holdout 零命中(owner 直管封存面隔离)"; } || FAIL=$((FAIL+1)) + +G=$(sel --seed 7 --sample-id gate-yaml-parse-corrupt | "$PYTHON" -c 'import json,sys;print(json.load(sys.stdin)["target_repo"])') +[[ "$G" == ".github" ]] && { PASS=$((PASS+1)); echo "ok github-scope 样本固定目标=.github"; } \ + || { FAIL=$((FAIL+1)); echo "FAIL github-scope 样本目标=$G"; } + +echo "== 3) 固定样本/目标(首演与复盘通道)" +O=$(sel --seed 99 --sample-id hygiene-gitleaks-aws-key --target-repo .github) +echo "$O" | grep -q '"sample_id": "hygiene-gitleaks-aws-key"' && echo "$O" | grep -q '"target_repo": ".github"' \ + && { PASS=$((PASS+1)); echo "ok --sample-id/--target-repo 钉选生效"; } \ + || { FAIL=$((FAIL+1)); echo "FAIL 钉选失效: $O"; } + +echo "== 4) 选样输出字段完整(AC-2 关卡 ID + AC-4 难度随记录可溯)" +sel --seed 3 | "$PYTHON" -c ' +import json, sys +d = json.load(sys.stdin) +assert d["sample_id"] and d["gate"] and d["difficulty"] in ("easy", "medium", "hard") and d["target_repo"], d +' && { PASS=$((PASS+1)); echo "ok 字段齐全"; } || { FAIL=$((FAIL+1)); echo "FAIL 字段缺失"; } + +echo "== 5) 非法钉选被拒(fail-closed,不静默回退随机)" +sel --seed 1 --sample-id no-such-sample >/dev/null 2>&1; [[ $? -ne 0 ]] \ + && { PASS=$((PASS+1)); echo "ok 未知样本 id 非零退出"; } || { FAIL=$((FAIL+1)); echo "FAIL 未知样本被放行"; } + +echo "选样自测: pass=$PASS fail=$FAIL" +[[ $FAIL -eq 0 ]] diff --git a/governance/drill/tests/test-verify.sh b/governance/drill/tests/test-verify.sh new file mode 100644 index 0000000..f2a6a7f --- /dev/null +++ b/governance/drill/tests/test-verify.sh @@ -0,0 +1,72 @@ +#!/usr/bin/env bash +# test-verify.sh —— 红绿判定断言自测(W4-C4 AC-1 判定核):离线注入 check-runs, +# 红=演习成功(0) / 绿=演习失败(1) / 中间态(3)——"没看到"绝不装绿。 +set -uo pipefail +HERE="$(cd "$(dirname "$0")" && pwd)" +ROOT="$(cd "$HERE/.." && pwd)" +source "$(cd "$(dirname "$0")" && pwd)/lib.sh" +PYTHON="$(pick_py)" || { echo "::error::无可用 python(含 pyyaml)"; exit 2; } +export DRILL_VERIFY_GRACE_S=0 # 静默期压零——离线模式立即判定(默认 120s 仅在线用) +PASS=0; FAIL=0 +TMP="$(mktemp -d)"; trap 'rm -rf "$TMP"' EXIT + +vf() { "$PYTHON" "$ROOT/verify_gate.py" --repo dot --sha deadbeef "$@"; } +verdict_of() { "$PYTHON" -c 'import json,sys;print(json.load(sys.stdin)["verdict"])'; } + +mkcr() { printf '{"check_runs": [%s]}' "$1" > "$TMP/$2"; } + +echo "== 1) 红=演习成功(关卡 conclusion=failure → 退出码 0)" +mkcr '{"name":"org-hygiene","status":"completed","conclusion":"failure"}' red.json +OUT=$(vf --gate org-hygiene --checkruns-file "$TMP/red.json"); RC=$? +V=$(verdict_of <<<"$OUT") +if [[ "$V" == "RED" && $RC -eq 0 ]]; then PASS=$((PASS+1)); echo "ok failure→RED/exit0" +else FAIL=$((FAIL+1)); echo "FAIL 期望 RED/0 实得 $V/$RC"; fi + +echo "== 2) 绿=演习失败(关卡 success → 退出码 1,绝不装绿)" +mkcr '{"name":"org-hygiene","status":"completed","conclusion":"success"}' green.json +OUT=$(vf --gate org-hygiene --checkruns-file "$TMP/green.json"); RC=$? +V=$(verdict_of <<<"$OUT") +if [[ "$V" == "GREEN" && $RC -eq 1 ]]; then PASS=$((PASS+1)); echo "ok success→GREEN/exit1" +else FAIL=$((FAIL+1)); echo "FAIL 期望 GREEN/1 实得 $V/$RC"; fi + +echo "== 3) NO-SURFACE(空 check 面=push 分支无 CI 局限,如实记录不装绿)" +mkcr '' empty.json +OUT=$(vf --gate org-hygiene --checkruns-file "$TMP/empty.json"); RC=$? +V=$(verdict_of <<<"$OUT") +if [[ "$V" == "NO-SURFACE" && $RC -eq 3 ]]; then PASS=$((PASS+1)); echo "ok 空→NO-SURFACE/exit3" +else FAIL=$((FAIL+1)); echo "FAIL 期望 NO-SURFACE/3 实得 $V/$RC"; fi + +echo "== 4) MISSING-GATE(有别的 check、无目标关卡=名字漂移,不猜)" +mkcr '{"name":"lint","status":"completed","conclusion":"success"}' other.json +OUT=$(vf --gate org-hygiene --checkruns-file "$TMP/other.json"); RC=$? +V=$(verdict_of <<<"$OUT") +if [[ "$V" == "MISSING-GATE" && $RC -eq 3 ]]; then PASS=$((PASS+1)); echo "ok 缺关卡→MISSING-GATE/exit3" +else FAIL=$((FAIL+1)); echo "FAIL 期望 MISSING-GATE/3 实得 $V/$RC"; fi + +echo "== 5) 子串唯一兜底 + 进行中不算结论(等不到=TIMEOUT)" +mkcr '{"name":"org-hygiene / gitleaks","status":"completed","conclusion":"failure"}' substr.json +OUT=$(vf --gate gitleaks --checkruns-file "$TMP/substr.json"); RC=$? +V=$(verdict_of <<<"$OUT") +[[ "$V" == "RED" && $RC -eq 0 ]] && { PASS=$((PASS+1)); echo "ok 精确名缺席时子串唯一命中"; } \ + || { FAIL=$((FAIL+1)); echo "FAIL 子串匹配: $V/$RC"; } +mkcr '{"name":"org-hygiene","status":"in_progress","conclusion":null}' wip.json +OUT=$(vf --gate org-hygiene --checkruns-file "$TMP/wip.json" --timeout 0); RC=$? +V=$(verdict_of <<<"$OUT") +[[ "$V" == "TIMEOUT" && $RC -eq 3 ]] && { PASS=$((PASS+1)); echo "ok in_progress 不判结论→TIMEOUT"; } \ + || { FAIL=$((FAIL+1)); echo "FAIL 进行中语义: $V/$RC"; } + +echo "== 6) 歧义子串不猜(两个候选=MISSING-GATE)" +mkcr '{"name":"org-hygiene-a","status":"completed","conclusion":"failure"},{"name":"org-hygiene-b","status":"completed","conclusion":"failure"}' amb.json +OUT=$(vf --gate org-hygiene --checkruns-file "$TMP/amb.json"); RC=$? +V=$(verdict_of <<<"$OUT") +[[ "$V" == "MISSING-GATE" ]] && { PASS=$((PASS+1)); echo "ok 歧义拒绝猜测"; } \ + || { FAIL=$((FAIL+1)); echo "FAIL 歧义应 MISSING-GATE 实得 $V"; } + +echo "== 7) API 模式无凭据 fail-closed(不静默跳过)" +unset GH_TOKEN || true +vf --gate org-hygiene >/dev/null 2>&1; RC=$? +[[ $RC -eq 2 ]] && { PASS=$((PASS+1)); echo "ok 无 GH_TOKEN 拒跑(exit2)"; } \ + || { FAIL=$((FAIL+1)); echo "FAIL 无凭据 rc=$RC"; } + +echo "验证核自测: pass=$PASS fail=$FAIL" +[[ $FAIL -eq 0 ]] diff --git a/governance/drill/verify_gate.py b/governance/drill/verify_gate.py new file mode 100644 index 0000000..6addc4f --- /dev/null +++ b/governance/drill/verify_gate.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python3 +"""verify_gate.py —— 周演习的独立验证步骤(与样本库分离;ADR-0069 / .github#223) + +职责单一:断言"该缺陷应触发关卡 X 失败"。不读样本库(关卡名由调用方传参), +不做注入——注入者与判定者分离,被审者不能组织对自己的审计的同款精神。 + +判定(红=演习成功——关卡活着;绿=演习失败——关卡死了,宪法 §4B): + RED 目标关卡 check conclusion == failure + GREEN 目标关卡 success(→ 调用方开 P0 + 该关卡标 suspect) + NO-SURFACE 该 SHA 上没有任何 check-run(目标仓 workflow 不跑 push 分支——已知 + 局限,如实记录;调用方决定是否走 draft PR 面补救) + MISSING-GATE / TIMEOUT 中间态,不计入红率分母(诚实口径,防"没看到=绿") + +check-runs 数据源: GitHub API(GH_TOKEN 必须提供)或 --checkruns-file +(测试注入用,不碰网络——选样随机性/判定逻辑可离线自测)。 +""" +import argparse +import json +import os +import sys +import time +import urllib.request + +API = "https://api.github.com" + + +def fetch_check_runs(org, repo, sha, token): + req = urllib.request.Request( + f"{API}/repos/{org}/{repo}/commits/{sha}/check-runs?per_page=100", + headers={"Authorization": f"Bearer {token}", "Accept": "application/vnd.github+json", + "User-Agent": "seed-drill-verify"}) + with urllib.request.urlopen(req) as r: + return json.loads(r.read() or b"{}").get("check_runs", []) + + +def match_gate(checks, gate): + """关卡 check 匹配: 精确名优先;子串兜底须唯一(歧义=MISSING-GATE,不猜)。""" + exact = [c for c in checks if c.get("name") == gate] + if exact: + return exact + sub = [c for c in checks if gate in c.get("name", "")] + return sub if len(sub) == 1 else [] + + +def verdict(org, repo, sha, gate, deadline_s, token, offline_file): + checks, waited = [], 0.0 + # 静默期(可注入缩短——自测试用):空 check 需过静默期才判 NO-SURFACE(push 后 + # workflow 排队中也会呈现为空,立即判=假阴性);目标关卡缺席判定取其半 + grace = int(os.environ.get("DRILL_VERIFY_GRACE_S", "120")) + while True: + if offline_file: + checks = json.load(open(offline_file, encoding="utf-8")).get("check_runs", []) + else: + checks = fetch_check_runs(org, repo, sha, token) + hits = match_gate(checks, gate) + if hits and hits[0].get("status") == "completed": + c = hits[0] + v = "RED" if c.get("conclusion") == "failure" else "GREEN" + return {"verdict": v, "check_name": c.get("name"), + "conclusion": c.get("conclusion"), "waited_s": int(waited), + "checks_seen": {x.get("name"): x.get("conclusion") + for x in checks if x.get("status") == "completed"}} + if not checks: + # 一个 check 都没有:须先过静默期再判 NO-SURFACE + if waited >= grace: + return {"verdict": "NO-SURFACE", "check_name": None, "conclusion": None, + "waited_s": int(waited), "checks_seen": {}} + elif not hits and waited >= max(grace // 2, 1): + # 有其他 check 而目标关卡迟迟未现身 → 名字漂移,不硬等 + return {"verdict": "MISSING-GATE", "check_name": None, "conclusion": None, + "waited_s": int(waited), + "checks_seen": {x.get("name"): x.get("status") for x in checks}} + if waited >= deadline_s: + return {"verdict": "TIMEOUT", "check_name": gate, "conclusion": None, + "waited_s": int(waited), + "checks_seen": {x.get("name"): x.get("conclusion") + for x in checks if x.get("status") == "completed"}} + time.sleep(0.05 if offline_file else 20) # 离线注入模式快进(自测试) + waited += 0.05 if offline_file else 20 + + +def main(): + ap = argparse.ArgumentParser(description="演习独立验证:断言关卡 X 应红") + ap.add_argument("--repo", required=True) + ap.add_argument("--sha", required=True) + ap.add_argument("--gate", required=True, help="预期触发失败的关卡 check 名") + ap.add_argument("--org", default=os.environ.get("DRILL_ORG", "Cloudbird-Software")) + ap.add_argument("--timeout", type=int, default=600, help="轮询上限秒(默认 600)") + ap.add_argument("--checkruns-file", default=None, + help="离线注入 check-runs JSON(自测试用;设置时不访问网络)") + a = ap.parse_args() + token = os.environ.get("GH_TOKEN", "") + if not a.checkruns_file and not token: + print("::error::缺 GH_TOKEN(API 模式必需;离线测试用 --checkruns-file)", file=sys.stderr) + raise SystemExit(2) + out = verdict(a.org, a.repo, a.sha, a.gate, a.timeout, token, a.checkruns_file) + print(json.dumps(out, ensure_ascii=False)) + # RED=演习成功(0);GREEN=演习失败(1,调用方须开 P0);中间态(3)不装绿 + raise SystemExit({"RED": 0, "GREEN": 1}.get(out["verdict"], 3)) + + +if __name__ == "__main__": + main() diff --git a/governance/tests/test-drill.sh b/governance/tests/test-drill.sh new file mode 100644 index 0000000..29217a8 --- /dev/null +++ b/governance/tests/test-drill.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env bash +# test-drill.sh —— W4-C4 活体演习链路自测入口(.github#223 / ADR-0069) +# gate"治理脚本自测"自动纳入 governance/tests/test-*.sh;本入口聚合 +# governance/drill/tests/ 下全部 test-*.sh(样本库 schema/选样随机性/红绿判定/ +# 台账 append-only/缺席 fail-closed dry-run)。目录为空=测试面丢失,fail-closed。 +set -uo pipefail +HERE="$(cd "$(dirname "$0")/../drill/tests" && pwd)" +shopt -s nullglob; tests=("$HERE"/test-*.sh); shopt -u nullglob +if [[ ${#tests[@]} -eq 0 ]]; then + echo "::error::governance/drill/tests 无 test-*.sh——演习测试面丢失(fail-closed)" + exit 1 +fi +for t in "${tests[@]}"; do echo "-- $t"; bash "$t" || exit 1; done +echo "drill 全部自测通过(${#tests[@]} 套)"