From 05c8395bc0cd92e929232e2d4fc69165a59bb59d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A7=9A=E7=AB=A0=E6=B5=A9?= Date: Mon, 12 Oct 2026 11:15:32 +0800 Subject: [PATCH] =?UTF-8?q?fix(adopt):=20=E5=90=88=E5=B9=B6=E5=BF=85?= =?UTF-8?q?=E9=9C=80=20context=20=E5=8F=AA=E8=AE=A4=E6=98=8E=E7=A1=AE?= =?UTF-8?q?=E7=AD=96=E7=95=A5=EF=BC=8C=E5=B9=B6=E6=A0=B8=E5=AF=B9=20base/s?= =?UTF-8?q?tate/=E6=9D=A5=E6=BA=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 必需 context 只能来自 --require-context 或仓库 .gitea/required-contexts.txt；两者都没有或文件是空的 ⇒ 拒绝合并，不再用体检观察到的状态顶替（combined state 只要有一个无关检查成功就是 success，会放行没跑过必需检查的 PR）。同时核对 PR 目标分支 == 默认分支、state=open、head 就是本仓。 --- scripts/adopt-existing.py | 270 ++++++++++++++++++++++++++++++++++---- 1 file changed, 243 insertions(+), 27 deletions(-) diff --git a/scripts/adopt-existing.py b/scripts/adopt-existing.py index a18475d..beb8397 100644 --- a/scripts/adopt-existing.py +++ b/scripts/adopt-existing.py @@ -4,9 +4,13 @@ python3 adopt-existing.py # DRY-RUN：只体检 + 打印计划 python3 adopt-existing.py --apply # 执行（加指针 = 分支 + PR，**不合并**） python3 adopt-existing.py --apply --with-protection - python3 adopt-existing.py --merge-pr # 显式批准后合并某个 PR + python3 adopt-existing.py --merge-pr --approve-sha <完整40位SHA> \ + [--require-context "build / build (push)"] # 显式批准后合并 **合并是独立动作**：本工具建完 PR 就**停下**，把状态如实打给你，等你批准。 +批准时必须给出**完整的 40 位 SHA**（短 sha 起不了身份作用），它会去核对： + ① PR 头提交就是那个 SHA（期间没人再推）；② 每个必需 status context 都是 success； + ③ 汇总状态是 success。任一条不成立就不请求合并。 不自动合并、不自动审批的理由是它自己的教训： ① 检查失败 / 取消 / 从未运行 / 状态拿不到（文件缺失、超时、未知）时合并，等于把没验过的东西合进 main； ② "等一会儿就合"把**审批人**变成了这个脚本 —— 而脚本没有资格替人拍板。 @@ -37,6 +41,8 @@ from pathlib import Path HOSTS = ["http://192.168.50.223:3000", "https://gitea.zheromiku.top"] STANDARDS_REPO = "zheromiku/eng-standards-main" +# 完整 SHA：批准与身份核对都只认 40 位十六进制 —— 短 sha 起不了身份作用（前缀撞车不报错） +FULL_SHA = re.compile(r"^[0-9a-f]{40}$") def D(v): @@ -259,19 +265,46 @@ def head_status(g: Gitea, repo: str, head: str, tries: int = 12, interval: int = return (state or "unknown"), ctxs, tries * interval -def merge_decision(state, head_matches: bool, explicitly_approved: bool): +def head_contexts(g: Gitea, repo: str, head: str) -> dict[str, str]: + """头提交上**每个** status context 的状态（不是只看总的 combined state）。 + + 为什么不能只看 combined：合并要核对的是"我要求的那几个检查都成功了"， + 而 combined 只给一个汇总值 —— 某个必需 context 根本没报时，combined 仍可能是 success。 + """ + st, s = g.api(f"/repos/{repo}/commits/{head}/status") + if st != 200: + return {} + out = {} + for x in L(D(s).get("statuses")): + ctx, stt = D(x).get("context"), D(x).get("status") + if ctx: + out[ctx] = stt + return out + + +def merge_decision(state, head_matches: bool, explicitly_approved: bool, + approved_sha: str = "", missing_contexts: list[str] | None = None): """能不能**请求**合并？纯函数（`--self-test` 直接打它）。 - 默认拒绝，只有"显式批准 + 头提交没变 + 检查 success"三条同时成立才放行。 + 默认拒绝。只有下面**全部**成立才放行： + · 显式批准，且批准的是一个**完整 40 位 SHA**（短 sha 起不了身份作用）； + · PR 头提交 == 那个被批准的 SHA； + · 要求的每一个 status context 都在头提交上 **success**（缺一个都不行）； + · 汇总状态是 success。 检查失败 / 取消 / 从未运行 / 拿不到状态 / 超时，一律不请求合并。 """ if not explicitly_approved: - return False, ("没有显式批准：合并是独立动作，要用 --merge-pr 明确指定" + return False, ("没有显式批准：合并是独立动作，要用 --merge-pr --approve-sha <完整SHA> 明确指定" "（本工具不替人拍板，也不自动审批）") + if not FULL_SHA.match(approved_sha or ""): + return False, (f"没有给出获批的**完整 40 位** SHA（收到 {str(approved_sha)[:20]!r}）—— " + f"短 sha / 空值都不能作为批准依据") if not head_matches: - return False, "PR 头提交与会话记录的那个提交不一致（期间有人推了新提交）—— 不合并" + return False, "PR 头提交 ≠ 被批准的 SHA（期间有人推了新提交）—— 不合并" + if missing_contexts: + return False, f"这些必需 status context 没有在头提交上 success：{missing_contexts} —— 不合并" if state == "success": - return True, "检查 success + 显式批准" + return True, "检查 success + 显式批准（完整 SHA 一致、必需 context 齐）" if state == "pending": return False, f"检查仍是 {state}（可能在等一个不会出现的检查，或还在跑）—— 不合并" if state in (None, "", "unknown", "none"): @@ -281,19 +314,24 @@ def merge_decision(state, head_matches: bool, explicitly_approved: bool): def self_test() -> int: """反向自检：每一条"不许合并"的分支都要真的拦下来。""" + good = "a" * 40 cases = [ - ("全绿 + 已批准 + 头一致", ("success", True, True), True), - ("全绿但**没有**批准", ("success", True, False), False), - ("失败 + 已批准", ("failure", True, True), False), - ("被取消 + 已批准", ("cancelled", True, True), False), - ("仍在跑 + 已批准", ("pending", True, True), False), - ("拿不到状态 + 已批准", ("none", True, True), False), - ("状态未知 + 已批准", (None, True, True), False), - ("头提交变了 + 已批准", ("success", False, True), False), + ("全绿 + 已批准（完整 SHA）+ 头一致 + context 齐", + ("success", True, True, good, []), True), + ("全绿但**没有**批准", ("success", True, False, good, []), False), + ("已批准但 SHA 是空的（没给批准依据）", ("success", True, True, "", []), False), + ("已批准但 SHA 只有 12 位（短 sha 不算批准依据）", ("success", True, True, good[:12], []), False), + ("失败 + 已批准", ("failure", True, True, good, []), False), + ("被取消 + 已批准", ("cancelled", True, True, good, []), False), + ("仍在跑 + 已批准", ("pending", True, True, good, []), False), + ("拿不到状态 + 已批准", ("none", True, True, good, []), False), + ("状态未知 + 已批准", (None, True, True, good, []), False), + ("头提交变了 + 已批准", ("success", False, True, good, []), False), + ("必需 context 没 success + 已批准", ("success", True, True, good, ["build / build (push)"]), False), ] problems = [] - for name, (state, head_ok, approved), want in cases: - got, why = merge_decision(state, head_ok, approved) + for name, (state, head_ok, approved, sha, missing), want in cases: + got, why = merge_decision(state, head_ok, approved, sha, missing) if got != want: problems.append(f"{name}：期望 {'合并' if want else '拒绝'}，实际 {'合并' if got else '拒绝'}（{why}）") if problems: @@ -302,7 +340,105 @@ def self_test() -> int: print(f" · {p}", file=sys.stderr) return 1 print(f"✅ 合并判定自检通过：{len(cases)} 个分支" - f"（失败/取消/未运行/未知/超时/未批准/头提交变了 —— 全部拒绝合并）") + f"（失败/取消/未运行/未知/超时/未批准/缺完整 SHA/头提交变了/必需 context 不齐 —— 全部拒绝合并）") + return 0 + + +class _StubGitea: + """假 Gitea：只实现 merge_pr / resolve_required_contexts 用到的那几个 API。 + + 为什么要有它：`merge_decision()` 是纯函数，测它只能证明"判定对不对"；真正要证明的是 + **调用链上有没有发出那个 POST /merge**。所以这里把真实的 `merge_pr()` 跑一遍， + 断言看的是 `self.calls` 里有没有 merge 请求 —— 一个真实请求都不发出去。 + """ + + def __init__(self, pr: dict, contexts: dict, policy=None): + self.calls: list[tuple[str, str]] = [] + self.pr, self.contexts, self.policy = pr, contexts, policy + + def api(self, path, method="GET", body=None): + self.calls.append((method, path)) + if method == "POST": + # 真实的 Gitea 合并成功后 body 可能是空的；这里模拟"合并完成" + self.pr["merged"] = True + self.pr["state"] = "closed" + self.pr["merge_commit_sha"] = "e" * 40 + return 200, {} + if path.endswith("/pulls/7"): + return 200, self.pr + if path.endswith("/status"): + # Gitea 的 combined state：**只要报过状态的都成功就是 success** —— + # 这正是"无关检查成功也算 success"的由来，所以不能只看 combined。 + st = "success" if self.contexts and all(v == "success" for v in self.contexts.values()) else "failure" + return 200, {"state": st, + "statuses": [{"context": k, "status": v} for k, v in self.contexts.items()]} + return 404, {} + + def file(self, repo, path, ref=None): + return self.policy + + +def self_test_chain() -> int: + """真实调用链的负例：不同报告/状态组合下，**POST /merge 到底发没发**。""" + sha = "5f3a71cadee1f2b27c0b71d0c9fd1ea1b57a4b0f" + req = ["build / build (push)", "ci / gate (push)"] + base_pr = {"merged": False, "state": "open", "mergeable": True, + "head": {"sha": sha, "repo": {"full_name": "Weknora/knowledge-governance"}}, + "base": {"ref": "main"}} + rep = {"default_branch": "main"} + allok = {"build / build (push)": "success", "ci / gate (push)": "success"} + cases = [ + # 名字, PR, contexts, require_ctx, 仓库策略文件内容, 期望的 merge POST 次数 + ("正例：策略齐 + 必需 context 全 success + 完整 SHA + base/state 对 ⇒ 发 1 次", + base_pr, allok, req, None, 1), + ("只有无关检查 success（必需 context 缺失）⇒ 0 次", + base_pr, {"unrelated / x (push)": "success"}, req, None, 0), + ("必需 context 里有一个 failure ⇒ 0 次", + base_pr, {"build / build (push)": "success", "ci / gate (push)": "failure"}, req, None, 0), + ("必需 context 是 pending ⇒ 0 次", + base_pr, {"build / build (push)": "success", "ci / gate (push)": "pending"}, req, None, 0), + ("没有策略、也没有策略文件（require_ctx 为空）⇒ 0 次", + base_pr, allok, [], None, 0), + ("策略文件是空的（全是注释）⇒ 0 次", + base_pr, allok, [], "# 还没有写\n\n", 0), + ("策略文件只有无关 context ⇒ 0 次", + base_pr, allok, [], "unrelated / x (push)\n", 0), + ("策略文件齐全（不传 --require-context）⇒ 发 1 次", + base_pr, allok, [], "build / build (push)\nci / gate (push)\n", 1), + ("PR 目标分支不是默认分支 ⇒ 0 次", + {**base_pr, "base": {"ref": "release"}}, allok, req, None, 0), + ("PR 不是 open（closed）⇒ 0 次", + {**base_pr, "state": "closed"}, allok, req, None, 0), + ("PR 来自另一个仓库 ⇒ 0 次", + {**base_pr, "head": {"sha": sha, "repo": {"full_name": "someone/fork"}}}, allok, req, None, 0), + ("头提交已经不是获批的 SHA ⇒ 0 次", + {**base_pr, "head": {"sha": "b" * 40, "repo": {"full_name": "Weknora/knowledge-governance"}}}, + allok, req, None, 0), + ("获批 SHA 只给了短 sha ⇒ 0 次", base_pr, allok, req, None, 0), + ] + problems, lines = [], [] + for i, (name, pr, contexts, explicit, policy, want) in enumerate(cases): + approve = sha if "短 sha" not in name else sha[:12] + g = _StubGitea(dict(pr), dict(contexts), policy) + ctxs, source, err = resolve_required_contexts(g, "Weknora/knowledge-governance", list(explicit)) + if err: + result = f"⛔ 不请求合并（{err[:40]}…）" + posts = 0 + else: + result = merge_pr(g, "Weknora/knowledge-governance", rep, "7", approve, ctxs, source) + posts = len([c for c in g.calls if c[0] == "POST" and c[1].endswith("/merge")]) + ok = posts == want and ("不请求合并" in result) == (want == 0) + lines.append(f"{'✅' if ok else '❌'} {name}：merge 请求 {posts} 次（期望 {want}）｜{result[:60]}") + if not ok: + problems.append(name) + print("── 合并调用链负例（假 Gitea，真实 merge_pr()；断言 merge POST 发没发）──") + for line in lines: + print(" " + line) + if problems: + print(f"\n❌ {len(problems)} 个分支不符合预期：{problems}", file=sys.stderr) + return 1 + print(f"\n✅ {len(cases)} 个分支全部符合预期：只有'策略齐 + 必需 context 全 success + " + f"完整获批 SHA + base/state/来源都对'才发出 merge 请求") return 0 @@ -334,28 +470,92 @@ def add_pointer(g: Gitea, repo: str, rep: dict, apply: bool): "head": br, "base": rep["default_branch"], "body": f"指向系统级规范 `{STANDARDS_REPO}`，记录本仓接入时核实的现状。未改动任何代码。\n\n" f"⚠️ 本 PR **未合并**：合并由人批准后单独执行。\n" - f"命令：`python3 ~/eng-standards/scripts/adopt-existing.py {repo} --merge-pr `"}) + f"命令：`python3 ~/eng-standards/scripts/adopt-existing.py {repo} --merge-pr --approve-sha <完整SHA>`"}) if st not in (200, 201): return f"❌ 建 PR 失败 HTTP {st}" prn = D(pr).get("number") state, ctxs, waited = head_status(g, repo, head) return (f"已建 **PR #{prn}**（头提交 {head[:12]}）—— 检查状态 {state}" f"{'，context ' + str(ctxs) if ctxs else ''}（等了 {waited}s 上限）；" - f"**未合并**，等你在 Gitea 上看过并批准后跑 --merge-pr {prn}") + f"**未合并**，等你在 Gitea 上看过并批准后跑 --merge-pr {prn} --approve-sha <完整40位SHA>") finally: shutil.rmtree(tmp, ignore_errors=True) -def merge_pr(g: Gitea, repo: str, rep: dict, prn: str, watch_head: str = ""): - """显式批准的合并：先核对状态，再合并，再读回。""" +CONTEXT_POLICY_PATH = ".gitea/required-contexts.txt" + + +def parse_context_policy(text: str) -> list[str]: + """解析策略文件：一行一个 context，`#` 后是注释，空行忽略。""" + out = [] + for line in (text or "").splitlines(): + line = line.split("#", 1)[0].strip() + if line: + out.append(line) + return out + + +def resolve_required_contexts(g, repo: str, explicit: list[str]): + """**"要求哪些检查"只能来自明确策略**。返回 (列表, 来源, 错误)。 + + 不许用"体检时在该仓观察到的 context"来顶替：那等于把"要求什么"交给当时碰巧报过的状态， + 只要有一个无关检查报过 success，就会把没有跑过必需检查的 PR 放过去（实测会漏）。 + 策略按优先级取：① --require-context 显式给定；② 仓库里的 .gitea/required-contexts.txt。 + 两个都没有 / 文件是空的 ⇒ 报错，不合并（空策略不等于"没有要求"）。 + """ + if explicit: + return list(explicit), "显式 --require-context", "" + text = g.file(repo, CONTEXT_POLICY_PATH) + if text is None: + return [], "", (f"没有任何明确策略：既没有 --require-context，仓库里也没有 " + f"{CONTEXT_POLICY_PATH} —— 没有'要求哪些检查'就不许合并" + f"（不能拿'观察到哪些状态'代替）") + ctxs = parse_context_policy(text) + if not ctxs: + return [], "", f"{CONTEXT_POLICY_PATH} 里没有任何 context（空文件/只有注释）—— 空策略不算'没有要求'" + return ctxs, f"仓库策略文件 {CONTEXT_POLICY_PATH}", "" + + +def merge_pr(g: Gitea, repo: str, rep: dict, prn: str, approved_sha: str, require_ctx: list[str], + ctx_source: str = "显式 --require-context") -> str: + """显式批准的合并：核对"获批的完整 SHA + **明确策略要求的** status context"，再合并，再读回。""" + if not FULL_SHA.match(approved_sha or ""): + return (f"⛔ **不请求合并**：--approve-sha 必须是**完整 40 位** SHA" + f"（收到 {str(approved_sha)[:20]!r}）—— 短 sha 不能作为批准依据") + if not require_ctx: + return ("⛔ **不请求合并**：必需 context 列表为空 —— 没有'要求哪些检查'就不许合并。" + "（用 --require-context，或在仓库里放 .gitea/required-contexts.txt）") st, pr = g.api(f"/repos/{repo}/pulls/{prn}") if st != 200: return f"❌ 取不到 PR #{prn}（HTTP {st}）" pr = D(pr) head = D(pr.get("head")).get("sha") or "" - state, ctxs, waited = head_status(g, repo, head) - ok, why = merge_decision(state, head_matches=True, explicitly_approved=True) - print(f" PR #{prn} 头提交 {head[:12]}｜状态 {state}｜context {ctxs or '（无）'}｜等了 {waited}s") + if bool(pr.get("merged")): + return f"PR #{prn} 已经是 merged 状态（merged=true）—— 不重复合并" + + # 合到哪儿、PR 是什么状态、改动从哪来 —— 三个都要核对： + # 合错分支/合一个已经关掉的 PR/合一个跨仓 PR，都不是"合并获批的那个改动"。 + base_ref = D(pr.get("base")).get("ref") or "" + default_branch = rep.get("default_branch") or "" + if base_ref != default_branch: + return (f"⛔ **不请求合并**：PR #{prn} 的目标分支是 {base_ref!r}，" + f"不是默认分支 {default_branch!r}") + if str(pr.get("state") or "") != "open": + return f"⛔ **不请求合并**：PR #{prn} 的状态是 {pr.get('state')!r}，不是 open" + head_repo = D(D(pr.get("head")).get("repo")).get("full_name") or "" + if head_repo and head_repo != repo: + return (f"⛔ **不请求合并**：PR #{prn} 来自另一个仓库（{head_repo}）—— " + f"跨仓合并不属于本流程，不自动做") + + head_matches = head == approved_sha + state, _ctxs, waited = head_status(g, repo, head) + have = head_contexts(g, repo, head) + missing = [c for c in require_ctx if have.get(c) != "success"] + print(f" PR #{prn} 头提交 {head[:12]}｜获批 SHA {approved_sha[:12]}｜base={base_ref}｜" + f"state={pr.get('state')}｜status {state}｜等了 {waited}s") + print(f" 必需 context（来源：{ctx_source}）：" + + "、".join(f"{c}={have.get(c, '缺失')}" for c in require_ctx)) + ok, why = merge_decision(state, head_matches, True, approved_sha, missing) if not ok: return f"⛔ **不请求合并**：{why}" g.api(f"/repos/{repo}/pulls/{prn}/merge", "POST", @@ -395,6 +595,8 @@ def main(): args = [a for a in sys.argv[1:] if not a.startswith("--")] if "--self-test" in sys.argv: sys.exit(self_test()) + if "--self-test-chain" in sys.argv: + sys.exit(self_test_chain()) if not args: sys.exit(__doc__) repo = args[0] @@ -402,6 +604,11 @@ def main(): with_protection = "--with-protection" in sys.argv merge_index = sys.argv.index("--merge-pr") if "--merge-pr" in sys.argv else -1 merge_n = sys.argv[merge_index + 1] if merge_index >= 0 and merge_index + 1 < len(sys.argv) else "" + sha_index = sys.argv.index("--approve-sha") if "--approve-sha" in sys.argv else -1 + approve_sha = sys.argv[sha_index + 1] if sha_index >= 0 and sha_index + 1 < len(sys.argv) else "" + ctx_index = sys.argv.index("--require-context") if "--require-context" in sys.argv else -1 + require_ctx = ([x for x in sys.argv[ctx_index + 1].split(",") if x] + if ctx_index >= 0 and ctx_index + 1 < len(sys.argv) else []) g = Gitea() print(f"Gitea: {g.base} 仓库: {repo} 模式: " @@ -409,8 +616,16 @@ def main(): rep = preflight(g, repo) if merge_n: + req_ctx, ctx_source, ctx_err = resolve_required_contexts(g, repo, require_ctx) + if ctx_err: + print("── 合并（显式批准模式）──") + print(f" ⛔ **不请求合并**：{ctx_err}") + print(f" （体检在该仓观察到的 context（仅参考，**不用于合并判定**）：" + f"{rep.get('contexts') or '（该仓没有产生过状态）'}）") + return print("── 合并（显式批准模式）──") - print(" " + merge_pr(g, repo, rep, merge_n)) + print(f" 必需 context 来源：{ctx_source}") + print(" " + merge_pr(g, repo, rep, merge_n, approve_sha, req_ctx, ctx_source)) print("\n ⚠️ 合并后仍要读回：state=closed + merged=true + merge_commit_sha 非空（上面已打印）。") return @@ -435,7 +650,8 @@ def main(): print("\n── 计划 ──") print(" 1. AGENTS.md : " + add_pointer(g, repo, rep, False)) print(" 2. 分支保护 : " + set_protection(g, repo, rep, False, allow, ctxs, with_protection)) - print(" 3. 合并 : **不自动做**。批准后跑 --merge-pr （状态不 success 时它会拒绝合并）") + print(" 3. 合并 : **不自动做**。批准后跑 --merge-pr --approve-sha <完整SHA>；" + "它会核对头提交 SHA 与每一个必需 context，任一不成立就拒绝合并") if not apply: print("\n（DRY-RUN：什么都没改。确认计划无误后加 --apply 重跑）") @@ -451,7 +667,7 @@ def main(): f"status_check={d.get('enable_status_check')} ctx={d.get('status_check_contexts')}") print("\n ⚠️ 还差两步（必须做，别跳）：") print(" ① 你去看那个 PR 并批准，然后：") - print(f" python3 ~/eng-standards/scripts/adopt-existing.py {repo} --merge-pr ") + print(f" python3 ~/eng-standards/scripts/adopt-existing.py {repo} --merge-pr --approve-sha <完整SHA>") print(" ② 反向自检：推一个故意违规的提交 → 本机 hook 必须红、CI 的 gate 必须红。") print(" ⚠️ CI 的 gate 是**推完再跑**的：它拦不住那次推送本身（要拦推送只有两条路：") print(" 本机 hook，或开分支保护禁直推）。别把「CI 会拦住直推」当成前提。")