姚章浩
05c8395bc0
必需 context 只能来自 --require-context 或仓库 .gitea/required-contexts.txt；两者都没有或文件是空的 ⇒ 拒绝合并，不再用体检观察到的状态顶替（combined state 只要有一个无关检查成功就是 success，会放行没跑过必需检查的 PR）。同时核对 PR 目标分支 == 默认分支、state=open、head 就是本仓。
682 lines
36 KiB
Python
682 lines
36 KiB
Python
#!/usr/bin/env python3
|
||
"""把一个**既有** Gitea 工程接入工程规范（正本 §8 的可执行版）。
|
||
|
||
python3 adopt-existing.py <owner/repo> # DRY-RUN：只体检 + 打印计划
|
||
python3 adopt-existing.py <owner/repo> --apply # 执行（加指针 = 分支 + PR，**不合并**）
|
||
python3 adopt-existing.py <owner/repo> --apply --with-protection
|
||
python3 adopt-existing.py <owner/repo> --merge-pr <n> --approve-sha <完整40位SHA> \
|
||
[--require-context "build / build (push)"] # 显式批准后合并
|
||
|
||
**合并是独立动作**：本工具建完 PR 就**停下**，把状态如实打给你，等你批准。
|
||
批准时必须给出**完整的 40 位 SHA**（短 sha 起不了身份作用），它会去核对：
|
||
① PR 头提交就是那个 SHA（期间没人再推）；② 每个必需 status context 都是 success；
|
||
③ 汇总状态是 success。任一条不成立就不请求合并。
|
||
不自动合并、不自动审批的理由是它自己的教训：
|
||
① 检查失败 / 取消 / 从未运行 / 状态拿不到（文件缺失、超时、未知）时合并，等于把没验过的东西合进 main；
|
||
② "等一会儿就合"把**审批人**变成了这个脚本 —— 而脚本没有资格替人拍板。
|
||
判据见 `merge_decision()`（纯函数，`--self-test` 直接打它）。
|
||
|
||
**分支保护默认不开**（原稿口径：所有改动直接进 main，正本 §1.3）。
|
||
只有加 `--with-protection` 才动它。
|
||
|
||
为什么开保护要单独一个开关、而且还得分两步入（体检 / 执行）：
|
||
"强制状态检查"开错了会把 PR **永久锁死** —— 检查永远没人报，PR 永远合不了。
|
||
所以必须先体检判定该仓的工作流**能不能覆盖 PR 头提交**，判定通过才允许开。
|
||
|
||
只依赖 Python 标准库 + git。凭据运行时从 ~/.git-credentials 读，**永不打印**。
|
||
"""
|
||
import base64
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
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):
|
||
"""JSON 对象，非 dict 一律当空。"""
|
||
return v if isinstance(v, dict) else {}
|
||
|
||
|
||
def L(v):
|
||
"""JSON 数组，非 list 一律当空。"""
|
||
return v if isinstance(v, list) else []
|
||
|
||
|
||
def load_token(netloc: str) -> str:
|
||
"""~/.git-credentials 里主机可能是 URL 编码的（`192.168.50.223%3a3000`），
|
||
所以先解码 netloc 再比 —— 按字面比会找不到凭据，而症状是"没有凭据"。"""
|
||
fp = Path.home() / ".git-credentials"
|
||
want = netloc.lower()
|
||
for line in fp.read_text(encoding="utf-8", errors="replace").splitlines():
|
||
m = re.match(r"^https?://([^:]+):([^@]+)@(.+)$", line)
|
||
if not m:
|
||
continue
|
||
got = urllib.parse.unquote(m.group(3)).strip().lower()
|
||
if got == want:
|
||
return urllib.parse.unquote(m.group(2))
|
||
return ""
|
||
|
||
|
||
def pick_host():
|
||
for base in HOSTS:
|
||
try:
|
||
with urllib.request.urlopen(f"{base}/api/v1/version", timeout=6) as x:
|
||
if x.status == 200:
|
||
return base
|
||
except Exception:
|
||
continue
|
||
return None
|
||
|
||
|
||
class Gitea:
|
||
def __init__(self):
|
||
self.base = pick_host() or ""
|
||
if not self.base:
|
||
sys.exit("❌ 连不上任何 Gitea 候选地址，先确认网络/代理")
|
||
self.host = urllib.parse.urlparse(self.base).netloc # 用于 git clone / 显示
|
||
self.tok = load_token(self.host)
|
||
if not self.tok:
|
||
sys.exit(f"❌ ~/.git-credentials 里没有 {self.host} 的凭据")
|
||
# 提交令牌走 URL 里的 basic auth 会进 shell 历史/进程表，所以一律用 credential helper
|
||
|
||
def api(self, path, method="GET", body=None):
|
||
"""返回 (status, value)。value 可能是 dict/list；空 body 归一化成 {}。
|
||
⚠️ 空 body 是 Gitea 合并**成功**的返回，不能当成失败（正本里写过的坑）。"""
|
||
data = json.dumps(body).encode() if body is not None else None
|
||
req = urllib.request.Request(f"{self.base}/api/v1{path}", data=data, method=method)
|
||
req.add_header("Authorization", "token " + self.tok)
|
||
req.add_header("Content-Type", "application/json")
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=30) as x:
|
||
raw = x.read().decode()
|
||
if not raw.strip():
|
||
return x.status, {}
|
||
try:
|
||
return x.status, json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
return x.status, {"_raw": raw[:200]}
|
||
except urllib.error.HTTPError as e:
|
||
return e.code, {"_error": e.read().decode()[:300]}
|
||
|
||
def file(self, repo, path, ref=None):
|
||
p = f"/repos/{repo}/contents/{urllib.parse.quote(path)}"
|
||
if ref:
|
||
p += f"?ref={urllib.parse.quote(ref)}"
|
||
st, b = self.api(p)
|
||
if st != 200:
|
||
return None
|
||
try:
|
||
return base64.b64decode(D(b).get("content", "")).decode("utf-8", "replace")
|
||
except Exception:
|
||
return ""
|
||
|
||
def branch(self, repo, name):
|
||
st, b = self.api(f"/repos/{repo}/branches/{urllib.parse.quote(name, safe='')}")
|
||
return D(D(b).get("commit")).get("id") if st == 200 else None
|
||
|
||
|
||
def workflow_triggers(text: str):
|
||
"""启发式解析 `on:` 块（不引 YAML：本仓不许引依赖）。"""
|
||
lines = text.splitlines()
|
||
start = next((i for i, l in enumerate(lines) if re.match(r"^on\s*:", l)), None)
|
||
if start is None:
|
||
return {"push_all": False, "push_any_branch": False, "pull_request": False}
|
||
block = []
|
||
for l in lines[start + 1:]:
|
||
if l.strip() and not l.startswith((" ", "\t", "#")):
|
||
break
|
||
block.append(l)
|
||
raw = "\n".join(block)
|
||
has_push = bool(re.search(r"^\s*push\s*:", raw, re.M))
|
||
# "推任意分支"：push 块里没有 branches 过滤，或 branches 里写了 **
|
||
push_all = has_push and (not re.search(r"^\s{2,}branches\s*:", raw, re.M) or "**" in raw)
|
||
return {"push_all": push_all, "push_any_branch": push_all,
|
||
"pull_request": bool(re.search(r"^\s*pull_request\s*:", raw, re.M))}
|
||
|
||
|
||
def preflight(g: Gitea, repo: str):
|
||
rep = {}
|
||
st, r = g.api(f"/repos/{repo}")
|
||
if st != 200:
|
||
sys.exit(f"❌ 取不到仓库 {repo}：HTTP {st} {str(r)[:120]}")
|
||
rep["default_branch"] = D(r).get("default_branch") or "main"
|
||
rep["empty"] = bool(D(r).get("empty"))
|
||
|
||
st, lst = g.api(f"/repos/{repo}/contents/.gitea/workflows")
|
||
rep["workflows"] = [D(x).get("name", "") for x in L(lst)] if st == 200 else []
|
||
rep["triggers"] = {}
|
||
for wf in rep["workflows"]:
|
||
t = g.file(repo, f".gitea/workflows/{wf}")
|
||
if t is not None:
|
||
rep["triggers"][wf] = workflow_triggers(t)
|
||
|
||
st, rr = g.api(f"/repos/{repo}/actions/runners")
|
||
rep["runners"] = [D(x).get("name", "") for x in L(D(rr).get("runners"))] if st == 200 else []
|
||
rep["runners_scope"] = "仓库级" if rep["runners"] else ""
|
||
if not rep["runners"]: # aiball 的 runner 挂在组织上，只查仓库级会误报"无 runner"
|
||
st2, orr = g.api(f"/orgs/{repo.split('/')[0]}/actions/runners")
|
||
if st2 == 200:
|
||
rep["runners"] = [D(x).get("name", "") for x in L(D(orr).get("runners"))]
|
||
rep["runners_scope"] = "组织级" if rep["runners"] else ""
|
||
|
||
st, cs = g.api(f"/repos/{repo}/commits?limit=1")
|
||
sha = D(L(cs)[0]).get("sha") if st == 200 and L(cs) else None
|
||
rep["head"] = sha
|
||
rep["contexts"], rep["combined"] = [], None
|
||
if sha:
|
||
st, s = g.api(f"/repos/{repo}/commits/{sha}/status")
|
||
if st == 200:
|
||
rep["contexts"] = sorted({c for c in (D(x).get("context") for x in L(D(s).get("statuses"))) if c})
|
||
rep["combined"] = D(s).get("state")
|
||
|
||
rep["has_agents"] = g.file(repo, "AGENTS.md") is not None
|
||
rep["has_hermes_md"] = g.file(repo, ".hermes.md") is not None
|
||
st, bp = g.api(f"/repos/{repo}/branch_protections")
|
||
rep["protections"] = [(D(x).get("branch_name"), D(x).get("enable_status_check"),
|
||
D(x).get("status_check_contexts")) for x in L(bp)] if st == 200 else []
|
||
return rep
|
||
|
||
|
||
def verdict(rep):
|
||
"""能不能开"强制状态检查"？只有"推任意分支都会跑"的工作流，PR 头提交才拿得到状态。"""
|
||
why = []
|
||
for wf, t in rep["triggers"].items():
|
||
if t["push_all"]:
|
||
why.append(f"{wf}: on.push 覆盖所有分支 → PR 头提交带着 (push) 状态 ✅ 可用")
|
||
elif t["pull_request"]:
|
||
why.append(f"{wf}: push 只覆盖 main、但有 pull_request → PR 走 (pull_request) 上下文，"
|
||
f"与 (push) 不匹配，需人工确认后按 (pull_request) 开 ⛔ 本工具不自动开")
|
||
else:
|
||
why.append(f"{wf}: 只在 push[main] 触发且无 pull_request → PR 不产生任何状态 ⛔ 不可开")
|
||
allowed = any(t["push_all"] for t in rep["triggers"].values())
|
||
ctxs = [c for c in rep["contexts"] if c and c.endswith("(push)")] if allowed else []
|
||
return allowed and bool(ctxs), ctxs, why
|
||
|
||
|
||
def agents_md(repo: str, rep: dict):
|
||
tpl = Path(__file__).resolve().parent.parent / "templates" / "README.md"
|
||
m = re.search(r"^````markdown\n(.*?)^````", tpl.read_text(encoding="utf-8"), re.S | re.M)
|
||
if not m:
|
||
sys.exit("❌ templates/README.md 里找不到 ```markdown 骨架块")
|
||
name = repo.split("/")[1]
|
||
body = m.group(1).replace("<PROJECT>", name).replace("<ns>/<repo>", repo)
|
||
# 如实写：默认**不开**分支保护（正本 §1.3）。别把模板里的假设写进别人的仓库 ——
|
||
# 一行"已启用分支保护"在没开保护时就是假的，而下一个人会信它。
|
||
branch_line = (f"| 默认分支 | `{rep['default_branch']}` —— 已启用分支保护（禁直推 + 必须 PR） |"
|
||
if rep.get("protections")
|
||
else f"| 默认分支 | `{rep['default_branch']}` —— **直推**（默认不开分支保护，正本 §1.3） |")
|
||
# 用 .get() 而不是下标：缺 key 时抛 KeyError 会**掩盖**下面那条真正要报的错
|
||
# （实测：夹具少一个 key，于是"锚点坏了"的自检报的是 KeyError，等于没验）。
|
||
_wf = "、".join(rep.get("workflows") or []) or "（无）"
|
||
_ctx = "、".join(rep.get("contexts") or []) or "（无）"
|
||
_rn = "、".join(rep.get("runners") or []) or "（无本仓 runner —— 有工作流也会永远 queued）"
|
||
if rep.get("runners_scope"):
|
||
_rn += f"（{rep['runners_scope']}）"
|
||
facts = "\n".join([
|
||
"## 0b. 接入时核实的现状（事实变了就更新；别让它过期）", "",
|
||
"| 项 | 值 |", "|---|---|",
|
||
branch_line,
|
||
f"| CI | 工作流 {_wf}；status context {_ctx} |",
|
||
f"| Runner | {_rn} |", "",
|
||
])
|
||
# ⚠️ 锚点必须真的存在。`str.replace` 找不到时**静默返回原文**，不会报错 ——
|
||
# 实测踩过：锚点写成 `## 2. 硬约束`（骨架里其实是 `## 1. 硬约束`），
|
||
# 于是这张现状表从来没被插进去，而工具一切正常地"成功"了。
|
||
# 所以这里显式断言锚点存在（正本 §3.2-3：断言校验器真的作用在目标上）。
|
||
anchor = "## 1. 硬约束"
|
||
if anchor not in body:
|
||
sys.exit(f"❌ AGENTS 骨架里找不到锚点 {anchor!r}；现状表会**静默**插不进去，先修锚点再跑")
|
||
return body.replace(anchor, facts + "\n" + anchor, 1)
|
||
|
||
|
||
def run_git(*args, cwd=None, stdin=None):
|
||
return subprocess.run(["git", *args], cwd=cwd, input=stdin, capture_output=True, text=True)
|
||
|
||
|
||
def head_status(g: Gitea, repo: str, head: str, tries: int = 12, interval: int = 5):
|
||
"""**有界**轮询 PR 头提交的检查状态。
|
||
|
||
返回 (state, contexts, waited)。state ∈ success / failure / pending / none / unknown。
|
||
没有 CI 的仓会一直 pending ⇒ 必须设上限，否则整批预算被一个仓耗光（正本 §2.6）。
|
||
拿不到状态时返回 none/unknown —— 那两种都**不允许合并**。
|
||
"""
|
||
state, ctxs = None, []
|
||
for i in range(tries):
|
||
st, s = g.api(f"/repos/{repo}/commits/{head}/status")
|
||
if st == 200:
|
||
state = D(s).get("state")
|
||
ctxs = sorted({c for c in (D(x).get("context") for x in L(D(s).get("statuses"))) if c})
|
||
if state and state != "pending":
|
||
return state, ctxs, i * interval
|
||
time.sleep(interval)
|
||
return (state or "unknown"), ctxs, tries * interval
|
||
|
||
|
||
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` 直接打它）。
|
||
|
||
默认拒绝。只有下面**全部**成立才放行：
|
||
· 显式批准，且批准的是一个**完整 40 位 SHA**（短 sha 起不了身份作用）；
|
||
· PR 头提交 == 那个被批准的 SHA；
|
||
· 要求的每一个 status context 都在头提交上 **success**（缺一个都不行）；
|
||
· 汇总状态是 success。
|
||
检查失败 / 取消 / 从未运行 / 拿不到状态 / 超时，一律不请求合并。
|
||
"""
|
||
if not explicitly_approved:
|
||
return False, ("没有显式批准：合并是独立动作，要用 --merge-pr <n> --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 头提交 ≠ 被批准的 SHA（期间有人推了新提交）—— 不合并"
|
||
if missing_contexts:
|
||
return False, f"这些必需 status context 没有在头提交上 success：{missing_contexts} —— 不合并"
|
||
if state == "success":
|
||
return True, "检查 success + 显式批准（完整 SHA 一致、必需 context 齐）"
|
||
if state == "pending":
|
||
return False, f"检查仍是 {state}（可能在等一个不会出现的检查，或还在跑）—— 不合并"
|
||
if state in (None, "", "unknown", "none"):
|
||
return False, f"拿不到检查状态（{state}）—— 缺失/未知都不合并"
|
||
return False, f"检查状态是 {state}（失败或被取消）—— 不合并"
|
||
|
||
|
||
def self_test() -> int:
|
||
"""反向自检：每一条"不许合并"的分支都要真的拦下来。"""
|
||
good = "a" * 40
|
||
cases = [
|
||
("全绿 + 已批准（完整 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, 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:
|
||
print("❌ 合并判定自检失败：", file=sys.stderr)
|
||
for p in problems:
|
||
print(f" · {p}", file=sys.stderr)
|
||
return 1
|
||
print(f"✅ 合并判定自检通过：{len(cases)} 个分支"
|
||
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
|
||
|
||
|
||
def add_pointer(g: Gitea, repo: str, rep: dict, apply: bool):
|
||
if rep["has_agents"] or rep["has_hermes_md"]:
|
||
return "已有上下文文件，跳过"
|
||
if not apply:
|
||
return "计划：走分支 + PR 新增 AGENTS.md（不推 main，**建完 PR 即停，等你批准合并**）"
|
||
tmp = tempfile.mkdtemp(prefix="adopt-")
|
||
d = os.path.join(tmp, repo.split("/")[1])
|
||
br = "docs/agents-pointer"
|
||
try:
|
||
if run_git("clone", "-q", "--depth", "1", f"{g.base}/{repo}.git", d).returncode:
|
||
return "❌ clone 失败"
|
||
run_git("switch", "-q", "-c", br, cwd=d)
|
||
with open(os.path.join(d, "AGENTS.md"), "w", encoding="utf-8") as fh:
|
||
fh.write(agents_md(repo, rep))
|
||
run_git("add", "AGENTS.md", cwd=d)
|
||
msg = ("docs(agents): 加 AGENTS.md 操作手册与规范指针\n\n"
|
||
f"指向系统级规范（{STANDARDS_REPO}），并写明本仓接入时核实的现状。\n"
|
||
"为什么用 AGENTS.md 而不是 .hermes.md：Codex / Claude Code / OpenCode 读的也是 AGENTS.md。\n")
|
||
if run_git("commit", "-q", "-F", "-", cwd=d, stdin=msg).returncode:
|
||
return "❌ commit 被提交闸门拦下（提交信息不合规）"
|
||
if run_git("push", "-q", "-u", "origin", br, cwd=d).returncode:
|
||
return "❌ push 失败"
|
||
head = run_git("rev-parse", "HEAD", cwd=d).stdout.strip()
|
||
st, pr = g.api(f"/repos/{repo}/pulls", "POST",
|
||
{"title": "docs(agents): 加 AGENTS.md 操作手册与规范指针",
|
||
"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 <n> --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} --approve-sha <完整40位SHA>")
|
||
finally:
|
||
shutil.rmtree(tmp, ignore_errors=True)
|
||
|
||
|
||
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 ""
|
||
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",
|
||
{"Do": "merge", "head_commit_id": head, "delete_branch_after_merge": True})
|
||
_, back = g.api(f"/repos/{repo}/pulls/{prn}") # 读回；不信 POST 的返回（成功时可能是空 body）
|
||
back = D(back)
|
||
merged = bool(back.get("merged"))
|
||
return (f"合并请求已发出 → 读回 merged={merged} "
|
||
f"state={back.get('state')} merge_commit_sha={str(back.get('merge_commit_sha') or '')[:12]}"
|
||
+ ("" if merged else " ⚠️ 读回显示没合并：**不要重试**，先看 PR 页面"))
|
||
|
||
|
||
def protection_payload(rep, allow, ctxs):
|
||
return {"branch_name": rep["default_branch"], "enable_push": False, "enable_force_push": False,
|
||
"required_approvals": 0, "block_on_outdated_branch": True, "block_on_rejected_reviews": True,
|
||
"dismiss_stale_approvals": True, "require_signed_commits": False,
|
||
"enable_status_check": bool(allow), "status_check_contexts": ctxs if allow else []}
|
||
|
||
|
||
def set_protection(g: Gitea, repo: str, rep: dict, apply: bool, allow: bool, ctxs, with_protection: bool):
|
||
if not with_protection:
|
||
return "默认不开（原稿口径：直推 main，正本 §1.3）；要开请加 --with-protection"
|
||
plan = f"保护 {rep['default_branch']}（禁直推 + 必须 PR；强制状态检查 {'开 ' + str(ctxs) if allow else '不开'}）"
|
||
if not apply:
|
||
return "计划：" + plan
|
||
payload = protection_payload(rep, allow, ctxs)
|
||
st, resp = g.api(f"/repos/{repo}/branch_protections", "POST", payload)
|
||
if st in (403, 409): # 已存在 → 改
|
||
st, resp = g.api(f"/repos/{repo}/branch_protections/{urllib.parse.quote(rep['default_branch'], safe='')}",
|
||
"PATCH", payload)
|
||
if st not in (200, 201):
|
||
return f"❌ 设置失败 HTTP {st} {str(resp)[:100]}"
|
||
return "已设置 —— " + plan
|
||
|
||
|
||
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]
|
||
apply = "--apply" in sys.argv
|
||
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} 模式: "
|
||
f"{'MERGE-PR（只合并，显式批准）' if merge_n else ('APPLY（会改）' if apply else 'DRY-RUN（只体检，不动任何东西）')}\n")
|
||
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(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
|
||
|
||
print("── 体检 ──")
|
||
print(f" 默认分支 : {rep['default_branch']}")
|
||
print(f" 工作流 : {rep['workflows'] or '（无）'}")
|
||
for wf, t in rep["triggers"].items():
|
||
print(f" {wf}: push_覆盖所有分支={t['push_all']} pull_request={t['pull_request']}")
|
||
print(f" 本仓 runner : "
|
||
f"{(str(rep['runners']) + '（' + rep['runners_scope'] + '）') if rep['runners'] else '（无 —— 有工作流也会永远 queued）'}")
|
||
print(f" 最新提交 : {str(rep['head'] or '')[:12]} combined={rep['combined']} "
|
||
f"contexts={rep['contexts']}")
|
||
print(f" AGENTS.md/.hermes.md : {rep['has_agents']}/{rep['has_hermes_md']}")
|
||
print(f" 现有分支保护 : {rep['protections'] or '（无）'}")
|
||
|
||
allow, ctxs, why = verdict(rep)
|
||
print("\n── 状态检查判定（决定 PR 会不会被锁死）──")
|
||
for line in (why or ["（本仓没有工作流 → 不能开状态检查）"]):
|
||
print(" · " + line)
|
||
print(f" ⇒ 强制状态检查: {'✅ 可以开 —— ' + str(ctxs) if allow else '⛔ 不开（开了 PR 永远合不了）'}")
|
||
|
||
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 <n> --approve-sha <完整SHA>；"
|
||
"它会核对头提交 SHA 与每一个必需 context，任一不成立就拒绝合并")
|
||
|
||
if not apply:
|
||
print("\n（DRY-RUN：什么都没改。确认计划无误后加 --apply 重跑）")
|
||
return
|
||
|
||
print("\n── 执行 ──")
|
||
print(" AGENTS.md : " + add_pointer(g, repo, rep, True))
|
||
print(" 分支保护 : " + set_protection(g, repo, rep, True, allow, ctxs, with_protection))
|
||
_, bp = g.api(f"/repos/{repo}/branch_protections")
|
||
for b in L(bp):
|
||
d = D(b)
|
||
print(f" 读回: {d.get('branch_name')} enable_push={d.get('enable_push')} "
|
||
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 <n> --approve-sha <完整SHA>")
|
||
print(" ② 反向自检：推一个故意违规的提交 → 本机 hook 必须红、CI 的 gate 必须红。")
|
||
print(" ⚠️ CI 的 gate 是**推完再跑**的：它拦不住那次推送本身（要拦推送只有两条路：")
|
||
print(" 本机 hook，或开分支保护禁直推）。别把「CI 会拦住直推」当成前提。")
|
||
if with_protection:
|
||
print(f" t=$(mktemp -d); git clone {g.base}/{repo}.git $t; cd $t \\")
|
||
print(" && git commit --allow-empty -m 'chore(probe): 不该进 main' \\")
|
||
print(f" && git push origin {rep['default_branch']}")
|
||
print(" 判据：必须被拒 —— Not allowed to push to protected branch")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |