"""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()