402 lines
20 KiB
Python
402 lines
20 KiB
Python
"""PR base/head 门禁（scripts/task_pr_check.py）：临时真实 Git 仓库（bare origin + clone）+ 进程内 Gitea 桩。
|
||
|
||
全部离线运行：不联网、不读取真实 ~/.secrets（HOME 指向临时目录），token 为合成值。
|
||
"""
|
||
import http.server
|
||
import json
|
||
import os
|
||
import socket
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import unittest
|
||
import urllib.parse
|
||
from pathlib import Path
|
||
|
||
SOURCE = Path(__file__).resolve().parents[1]
|
||
SCRIPT = SOURCE / "scripts" / "task_pr_check.py"
|
||
|
||
TOKEN = "synthetic-gitea-token-not-real"
|
||
|
||
|
||
def card_text(branch, kind, root="main", immediate="main", accepting="false", registry=None):
|
||
slug = branch.replace("/", "-")
|
||
lines = [
|
||
f"# {branch} 任务卡",
|
||
"任务主记录：fixture；状态：进行中；owner：PR Check Test",
|
||
"- 权威仓：fixture；目标分支：main；R0（initial_base_sha）= 0",
|
||
f"- 工作区：fixture（worktree）；分支：{branch}",
|
||
f"- 分支类型：{kind}；initiative_id：{slug}；root_target：{root}；"
|
||
f"immediate_target：{immediate}；accepting_requirements：{accepting}",
|
||
"",
|
||
"## 目标与非目标",
|
||
"",
|
||
"- 交付 PR 门禁",
|
||
"",
|
||
]
|
||
if registry is not None:
|
||
lines += ["## 子任务登记", "", "| 子任务分支 | 必需 | 状态 | PR |", "| --- | --- | --- | --- |"]
|
||
lines += [f"| {branch_name} | {required} | {status} | {pr} |"
|
||
for branch_name, required, status, pr in registry]
|
||
lines.append("")
|
||
return "\n".join(lines) + "\n"
|
||
|
||
|
||
def pr(number, branch, base, sha, state="open", merged=False):
|
||
return {"number": number, "state": state, "merged": merged,
|
||
"base": {"ref": base}, "head": {"ref": branch, "sha": sha}}
|
||
|
||
|
||
class FakeGitea:
|
||
"""Gitea pulls API 桩（127.0.0.1）：校验 token，列表（分页+state）与单 PR 查询；从不联网。"""
|
||
|
||
PAGE = 50
|
||
|
||
def __init__(self, token=TOKEN):
|
||
self.token = token
|
||
self.prs = []
|
||
stub = self
|
||
|
||
class Handler(http.server.BaseHTTPRequestHandler):
|
||
def log_message(self, *args):
|
||
pass
|
||
|
||
def _send(self, status, obj):
|
||
body = json.dumps(obj).encode()
|
||
self.send_response(status)
|
||
self.send_header("Content-Type", "application/json")
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
def do_GET(self):
|
||
if self.headers.get("Authorization") != f"token {stub.token}":
|
||
return self._send(401, {"message": "unauthorized"})
|
||
parsed = urllib.parse.urlsplit(self.path)
|
||
query = urllib.parse.parse_qs(parsed.query)
|
||
parts = parsed.path.split("/")
|
||
if parts[1:4] != ["api", "v1", "repos"] or len(parts) not in (7, 8) or parts[6] != "pulls":
|
||
return self._send(404, {"message": "not found"})
|
||
if len(parts) == 8:
|
||
found = [item for item in stub.prs if str(item["number"]) == parts[7]]
|
||
return self._send(200, found[0]) if found else self._send(404, {"message": "not found"})
|
||
state = query.get("state", [""])[0]
|
||
items = [item for item in stub.prs if not state or item["state"] == state]
|
||
page = int(query.get("page", ["1"])[0])
|
||
start = (page - 1) * stub.PAGE
|
||
return self._send(200, items[start:start + stub.PAGE])
|
||
|
||
self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
||
self.url = f"http://127.0.0.1:{self.server.server_address[1]}/api/v1"
|
||
threading.Thread(target=self.server.serve_forever, daemon=True).start()
|
||
|
||
def close(self):
|
||
self.server.shutdown()
|
||
self.server.server_close()
|
||
|
||
|
||
def closed_port_url():
|
||
with socket.socket() as sock:
|
||
sock.bind(("127.0.0.1", 0))
|
||
port = sock.getsockname()[1]
|
||
return f"http://127.0.0.1:{port}/api/v1"
|
||
|
||
|
||
class PrCheckTests(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory(prefix="pr-check-")
|
||
self.addCleanup(self.tmp.cleanup)
|
||
root = Path(self.tmp.name)
|
||
self.remote = root / "remote.git"
|
||
self.repo = root / "repo"
|
||
self.env = {k: v for k, v in os.environ.items()
|
||
if not k.startswith(("GITHUB_", "GIT_")) and k not in {
|
||
"GITEA_API_URL", "GITEA_REPO", "GITEA_TOKEN", "TASK_TARGET_BRANCH"}}
|
||
self.env.update({"GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": "/dev/null",
|
||
"GIT_ALLOW_PROTOCOL": "file", "GIT_TERMINAL_PROMPT": "0",
|
||
"GIT_AUTHOR_NAME": "PR Check Test", "GIT_AUTHOR_EMAIL": "pr-check@example.invalid",
|
||
"GIT_COMMITTER_NAME": "PR Check Test", "GIT_COMMITTER_EMAIL": "pr-check@example.invalid",
|
||
"GIT_CONFIG_COUNT": "2", "GIT_CONFIG_KEY_0": "gc.auto",
|
||
"GIT_CONFIG_VALUE_0": "0", "GIT_CONFIG_KEY_1": "maintenance.auto",
|
||
"GIT_CONFIG_VALUE_1": "false",
|
||
"HOME": str(root / "no-home"),
|
||
"NO_PROXY": "127.0.0.1,localhost", "no_proxy": "127.0.0.1,localhost",
|
||
"GITEA_REPO": "fixture/repo", "GITEA_TOKEN": TOKEN})
|
||
self.git(root, "init", "--bare", "--initial-branch=main", str(self.remote))
|
||
self.git(root, "init", "--initial-branch=main", str(self.repo))
|
||
(self.repo / "README").write_text("fixture\n")
|
||
self.git(self.repo, "add", "README")
|
||
self.git(self.repo, "commit", "-m", "Create fixture")
|
||
self.git(self.repo, "remote", "add", "origin", str(self.remote))
|
||
self.git(self.repo, "push", "-u", "origin", "main")
|
||
self.gitea = FakeGitea()
|
||
self.addCleanup(self.gitea.close)
|
||
self.env["GITEA_API_URL"] = self.gitea.url
|
||
|
||
# ---- 夹具 ----
|
||
|
||
def git(self, cwd, *args):
|
||
return subprocess.run(["git", *args], cwd=cwd, env=self.env, capture_output=True,
|
||
text=True, check=True).stdout.strip()
|
||
|
||
def head(self, ref="HEAD"):
|
||
return self.git(self.repo, "rev-parse", ref)
|
||
|
||
def make_branch(self, branch, base="main"):
|
||
self.git(self.repo, "switch", "-c", branch, base)
|
||
self.git(self.repo, "push", "-u", "origin", branch)
|
||
return self.head()
|
||
|
||
def commit_card(self, branch, text, base="main"):
|
||
self.make_branch(branch, base)
|
||
path = self.repo / ".task" / (branch.replace("/", "-") + ".md")
|
||
path.parent.mkdir(exist_ok=True)
|
||
path.write_text(text, encoding="utf-8")
|
||
self.git(self.repo, "add", ".task")
|
||
self.git(self.repo, "commit", "-m", f"Record {branch}")
|
||
self.git(self.repo, "push", "origin", branch)
|
||
return self.head()
|
||
|
||
def set_card(self, branch, text):
|
||
self.git(self.repo, "switch", branch)
|
||
path = self.repo / ".task" / (branch.replace("/", "-") + ".md")
|
||
path.write_text(text, encoding="utf-8")
|
||
self.git(self.repo, "add", ".task")
|
||
self.git(self.repo, "commit", "-m", f"Update {branch}")
|
||
self.git(self.repo, "push", "origin", branch)
|
||
return self.head()
|
||
|
||
def commit_work(self, branch, push=True):
|
||
self.git(self.repo, "switch", branch)
|
||
count = len(list(self.repo.glob("work-*.txt")))
|
||
(self.repo / f"work-{count}.txt").write_text("more work\n")
|
||
self.git(self.repo, "add", ".")
|
||
self.git(self.repo, "commit", "-m", "More work")
|
||
if push:
|
||
self.git(self.repo, "push", "origin", "HEAD")
|
||
return self.head()
|
||
|
||
def build_child(self):
|
||
self.commit_card("feature/init", card_text("feature/init", "initiative", accepting="true"))
|
||
self.commit_card("feature/init-child",
|
||
card_text("feature/init-child", "child", root="main", immediate="feature/init"),
|
||
base="feature/init")
|
||
return self.head()
|
||
|
||
def invoke(self, *args, env=None, drop=()):
|
||
merged = {k: v for k, v in (self.env | (env or {})).items() if k not in drop}
|
||
result = subprocess.run([sys.executable, str(SCRIPT), *args], cwd=self.repo, env=merged,
|
||
capture_output=True, text=True, timeout=60)
|
||
result.out = result.stdout + result.stderr
|
||
self.assertNotIn(TOKEN, result.out)
|
||
return result
|
||
|
||
def local(self, *args, env=None, drop=()):
|
||
return self.invoke("local", *args, env=env, drop=drop)
|
||
|
||
@staticmethod
|
||
def fails(result):
|
||
return [line for line in result.stdout.splitlines() if line.startswith("[pr-check] FAIL")]
|
||
|
||
# ---- local：基址与分支类型 ----
|
||
|
||
def test_child_pr_into_main_is_rejected(self):
|
||
self.build_child()
|
||
self.gitea.prs = [pr(1, "feature/init", "main", self.head("feature/init")),
|
||
pr(2, "feature/init-child", "main", self.head())]
|
||
result = self.local()
|
||
self.assertEqual(result.returncode, 1, result.out)
|
||
self.assertEqual(len(self.fails(result)), 1, result.out)
|
||
self.assertIn("expected 'feature/init' for child branch feature/init-child", result.out)
|
||
self.assertIn("[verdict] PR_CHECK_FAILED", result.out)
|
||
|
||
def test_child_pr_into_initiative_passes(self):
|
||
self.build_child()
|
||
head = self.head()
|
||
self.gitea.prs = [pr(1, "feature/init", "main", self.head("feature/init")),
|
||
pr(2, "feature/init-child", "feature/init", head)]
|
||
result = self.local()
|
||
self.assertEqual(result.returncode, 0, result.out)
|
||
self.assertIn(f"[verdict] PR_CHECK_OK base=feature/init head={head} kind=child", result.out)
|
||
|
||
def test_child_is_rejected_when_parent_initiative_pr_is_closed(self):
|
||
self.build_child()
|
||
self.gitea.prs = [pr(1, "feature/init", "main", self.head("feature/init"), state="closed"),
|
||
pr(2, "feature/init-child", "feature/init", self.head())]
|
||
result = self.local()
|
||
self.assertEqual(result.returncode, 1, result.out)
|
||
self.assertEqual(len(self.fails(result)), 1, result.out)
|
||
self.assertIn("parent initiative feature/init has no open PR into main", result.out)
|
||
|
||
def test_standalone_and_initiative_base_gate(self):
|
||
self.make_branch("feature/other")
|
||
self.commit_card("feature/solo", card_text("feature/solo", "standalone"))
|
||
self.commit_card("feature/init-x", card_text("feature/init-x", "initiative", accepting="true"))
|
||
for branch, kind in (("feature/solo", "standalone"), ("feature/init-x", "initiative")):
|
||
self.git(self.repo, "switch", branch)
|
||
head = self.head()
|
||
for base, expected_ok in (("main", True), ("feature/other", False)):
|
||
with self.subTest(kind=kind, base=base):
|
||
self.gitea.prs = [pr(1, branch, base, head)]
|
||
result = self.local()
|
||
if expected_ok:
|
||
self.assertEqual(result.returncode, 0, result.out)
|
||
self.assertIn(f"[verdict] PR_CHECK_OK base=main head={head} kind={kind}", result.out)
|
||
else:
|
||
self.assertEqual(result.returncode, 1, result.out)
|
||
self.assertIn(f"expected 'main' for {kind} branch {branch}", result.out)
|
||
|
||
# ---- local：PR 状态与 head ----
|
||
|
||
def test_stale_pr_head_is_rejected(self):
|
||
self.commit_card("feature/solo", card_text("feature/solo", "standalone"))
|
||
stale = self.head()
|
||
self.gitea.prs = [pr(1, "feature/solo", "main", stale)]
|
||
fresh = self.commit_work("feature/solo")
|
||
result = self.local()
|
||
self.assertEqual(result.returncode, 1, result.out)
|
||
self.assertEqual(len(self.fails(result)), 1, result.out)
|
||
self.assertIn(f"PR #1 head {stale} != local HEAD {fresh}", result.out)
|
||
|
||
def test_remote_branch_head_differing_from_local_is_rejected(self):
|
||
self.commit_card("feature/solo", card_text("feature/solo", "standalone"))
|
||
remote = self.head()
|
||
local = self.commit_work("feature/solo", push=False)
|
||
self.gitea.prs = [pr(1, "feature/solo", "main", local)]
|
||
result = self.local()
|
||
self.assertEqual(result.returncode, 1, result.out)
|
||
self.assertEqual(len(self.fails(result)), 1, result.out)
|
||
self.assertIn(f"origin/feature/solo is {remote}, local HEAD is {local}", result.out)
|
||
|
||
def test_closed_and_merged_prs_are_rejected(self):
|
||
self.commit_card("feature/solo", card_text("feature/solo", "standalone"))
|
||
head = self.head()
|
||
for kwargs, reason in (({"state": "closed"}, "PR #1 is not open (state=closed)"),
|
||
({"state": "closed", "merged": True}, "PR #1 is already merged")):
|
||
with self.subTest(reason=reason):
|
||
self.gitea.prs = [pr(1, "feature/solo", "main", head, **kwargs)]
|
||
result = self.local("--pr", "1")
|
||
self.assertEqual(result.returncode, 1, result.out)
|
||
self.assertIn(reason, result.out)
|
||
|
||
def test_pr_lookup_requires_one_open_pr_or_explicit_number(self):
|
||
self.commit_card("feature/solo", card_text("feature/solo", "standalone"))
|
||
head = self.head()
|
||
self.make_branch("feature/other")
|
||
self.git(self.repo, "switch", "feature/solo")
|
||
self.gitea.prs = [pr(1, "feature/solo", "feature/other", head), pr(2, "feature/solo", "main", head)]
|
||
ambiguous = self.local()
|
||
self.assertEqual(ambiguous.returncode, 1, ambiguous.out)
|
||
self.assertIn("2 open PRs have head feature/solo; pass --pr N", ambiguous.out)
|
||
explicit = self.local("--pr", "2")
|
||
self.assertEqual(explicit.returncode, 0, explicit.out)
|
||
self.gitea.prs = []
|
||
missing = self.local()
|
||
self.assertEqual(missing.returncode, 1, missing.out)
|
||
self.assertIn("no open PR has head feature/solo", missing.out)
|
||
|
||
# ---- 远端不可核验：fail closed ----
|
||
|
||
def test_unreachable_or_unauthorized_api_fails_closed_with_exit_6(self):
|
||
self.build_child()
|
||
self.gitea.prs = [pr(1, "feature/init", "main", self.head("feature/init")),
|
||
pr(2, "feature/init-child", "feature/init", self.head())]
|
||
for label, env in (("unreachable", {"GITEA_API_URL": closed_port_url()}),
|
||
("unauthorized", {"GITEA_TOKEN": "wrong-token"})):
|
||
with self.subTest(label):
|
||
result = self.local(env=env)
|
||
self.assertEqual(result.returncode, 6, result.out)
|
||
self.assertIn("[verdict] REMOTE_UNVERIFIED", result.out)
|
||
|
||
def test_token_is_first_access_token_line_of_secrets_file(self):
|
||
self.commit_card("feature/solo", card_text("feature/solo", "standalone"))
|
||
home = Path(self.tmp.name) / "secrets-home"
|
||
(home / ".secrets").mkdir(parents=True)
|
||
(home / ".secrets" / "gitea").write_text(
|
||
f"URL_PREFIX=https://example.invalid\n- ACCESS_TOKEN={TOKEN}\nACCESS_TOKEN=other-token\n")
|
||
self.gitea.prs = [pr(1, "feature/solo", "main", self.head())]
|
||
result = self.local(env={"HOME": str(home)}, drop=("GITEA_TOKEN",))
|
||
self.assertEqual(result.returncode, 0, result.out)
|
||
|
||
# ---- local --final ----
|
||
|
||
def test_final_rejects_initiative_still_accepting_requirements(self):
|
||
self.commit_card("feature/init", card_text("feature/init", "initiative", accepting="true", registry=[]))
|
||
self.gitea.prs = [pr(1, "feature/init", "main", self.head())]
|
||
result = self.local("--final")
|
||
self.assertEqual(result.returncode, 1, result.out)
|
||
self.assertIn("still accepts requirements", result.out)
|
||
|
||
def test_final_requires_required_children_merged_or_deferred(self):
|
||
self.commit_card("feature/init", card_text("feature/init", "initiative", accepting="true"))
|
||
scenarios = [
|
||
("merged and deferred pass",
|
||
[("feature/init-a", "是", "已合入", "#2"), ("feature/init-b", "否", "进行中", "-"),
|
||
("feature/init-c", "是", "已延期", "-")],
|
||
[pr(2, "feature/init-a", "feature/init", "a" * 40, state="closed", merged=True)],
|
||
None),
|
||
("required child still in progress",
|
||
[("feature/init-d", "是", "进行中", "-")], [], "required child feature/init-d is '进行中'"),
|
||
("open PR is not merged",
|
||
[("feature/init-e", "是", "已合入", "#3")],
|
||
[pr(3, "feature/init-e", "feature/init", "b" * 40, state="open")],
|
||
"child feature/init-e PR #3 is not merged into feature/init"),
|
||
("merged into main instead",
|
||
[("feature/init-f", "是", "已合入", "#4")],
|
||
[pr(4, "feature/init-f", "main", "c" * 40, state="closed", merged=True)],
|
||
"child feature/init-f PR #4 is not merged into feature/init"),
|
||
]
|
||
for label, registry, children, reason in scenarios:
|
||
with self.subTest(label):
|
||
head = self.set_card("feature/init", card_text("feature/init", "initiative",
|
||
accepting="false", registry=registry))
|
||
self.gitea.prs = [pr(1, "feature/init", "main", head)] + children
|
||
result = self.local("--final")
|
||
if reason is None:
|
||
self.assertEqual(result.returncode, 0, result.out)
|
||
self.assertIn("[verdict] PR_CHECK_OK base=main", result.out)
|
||
else:
|
||
self.assertEqual(result.returncode, 1, result.out)
|
||
self.assertIn(reason, result.out)
|
||
|
||
# ---- ci 模式 ----
|
||
|
||
def ci(self, base=None, head=None, event="pull_request"):
|
||
env = {"GITHUB_EVENT_NAME": event}
|
||
if base is not None:
|
||
env["GITHUB_BASE_REF"] = base
|
||
if head is not None:
|
||
env["GITHUB_HEAD_REF"] = head
|
||
# ci 模式不得需要网络或 token：指向关闭端口并移除 token，证明零依赖。
|
||
return self.invoke("ci", env=env | {"GITEA_API_URL": closed_port_url()}, drop=("GITEA_TOKEN",))
|
||
|
||
def test_ci_mode_gates_pull_request_base_against_head_card(self):
|
||
self.build_child()
|
||
result = self.ci(base="feature/init", head="feature/no-card")
|
||
self.assertEqual(result.returncode, 1, result.out) # 无卡 → standalone → 期望 main
|
||
self.assertIn("expected 'main' for standalone branch feature/no-card", result.out)
|
||
cases = [
|
||
("skips non pull_request events", dict(event="push", base="main", head="feature/init-child"), 0,
|
||
"[pr-check] skip: event push is not pull_request"),
|
||
("child into main is rejected", dict(base="main", head="feature/init-child"), 1,
|
||
"expected 'feature/init' for child branch feature/init-child"),
|
||
("child into initiative passes", dict(base="feature/init", head="feature/init-child"), 0,
|
||
"[verdict] PR_CHECK_OK base=feature/init head=feature/init-child kind=child"),
|
||
("card-less head defaults to main", dict(base="main", head="feature/no-card"), 0,
|
||
"[verdict] PR_CHECK_OK base=main head=feature/no-card kind=standalone"),
|
||
("card-less head into other base is rejected", dict(base="feature/init", head="feature/no-card"), 1,
|
||
"expected 'main' for standalone branch feature/no-card"),
|
||
("missing base ref is rejected", dict(base=None, head="feature/init-child"), 1,
|
||
"lacks GITHUB_BASE_REF or GITHUB_HEAD_REF"),
|
||
]
|
||
for label, kwargs, code, text in cases:
|
||
with self.subTest(label):
|
||
result = self.ci(**kwargs)
|
||
self.assertEqual(result.returncode, code, result.out)
|
||
self.assertIn(text, result.out)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|