diff --git a/.gitea/workflows/ci.yaml b/.gitea/workflows/ci.yaml index 8970e45b..9a35d103 100644 --- a/.gitea/workflows/ci.yaml +++ b/.gitea/workflows/ci.yaml @@ -1,7 +1,6 @@ name: ci on: pull_request: - branches: [main] push: branches: [main] @@ -156,6 +155,8 @@ jobs: print("Checkout complete; credentials were not persisted in Git configuration") PY_CHECKOUT unset TOKEN SUBMODULE_TOKEN + - name: PR base gate + run: python3 scripts/task_pr_check.py ci - name: Install Go 1.27.1 run: | set -euo pipefail diff --git a/Makefile b/Makefile index 011ad5b4..2f911918 100644 --- a/Makefile +++ b/Makefile @@ -13,7 +13,7 @@ PLUGINS := plugins COMBO := core=$(shell git -C $(CORE) rev-parse --short HEAD 2>/dev/null)+backend=$(shell git -C $(BACKEND) rev-parse --short HEAD 2>/dev/null)+frontend=$(shell git -C $(FRONTEND) rev-parse --short HEAD 2>/dev/null)+ui=$(shell git -C $(UI) rev-parse --short HEAD 2>/dev/null)+agentcmd=$(shell git -C deps/agent-command rev-parse --short HEAD 2>/dev/null)+documents=$(shell git -C modules/documents rev-parse --short HEAD 2>/dev/null)+native=$(shell git -C deps/agent-native rev-parse --short HEAD 2>/dev/null)+squad=$(shell git -C deps/agent-squad rev-parse --short HEAD 2>/dev/null)+intent=$(shell git -C deps/agent-intent rev-parse --short HEAD 2>/dev/null) VERSION := $(shell git describe --tags --always --dirty 2>/dev/null)+$(COMBO) -.PHONY: help sync status build frontend svc svc-list svc-info svc-commits svc-history svc-update svc-recover svc-watchdog test-backend test-backend-pg test-multi-instance test-lifecycle test-frontend test-ui test-core test-agentcmd test-pins test-libs dev pin clean setup precheck task-triage task-triage-decide task-start task-resume task-sync task-close task-finish facts-verify evidence version agent-exec-validate agent-exec-apply agent-exec-admit +.PHONY: help sync status build frontend svc svc-list svc-info svc-commits svc-history svc-update svc-recover svc-watchdog test-backend test-backend-pg test-multi-instance test-lifecycle test-frontend test-ui test-core test-agentcmd test-pins test-libs dev pin clean setup precheck task-triage task-triage-decide task-start task-resume task-sync task-close task-finish task-pr-check facts-verify evidence version agent-exec-validate agent-exec-apply agent-exec-admit version: @echo "agent-system $(VERSION)" @@ -60,6 +60,10 @@ task-close: task-finish: @bash $(TASK) finish $(ABANDON) +# PR base/head 门禁(两级分支):本地分支卡片期望目标 vs Gitea PR 与 origin;FINAL=1 用于 initiative 收口。 +task-pr-check: + python3 scripts/task_pr_check.py local $(if $(PR),--pr $(PR)) $(if $(FINAL),--final) + # 证据写入:执行命令并把结果行追加进当前任务卡(AGENTS.md「证据与验证门禁」)。 # 本地执行只记 SELF;RECEIPT 捕获启动收据,IMPORT_RECEIPT 导入原始收据而不重跑。 # 用法:CMD="<单行命令>" make evidence [TAIL=15];退出码随被执行命令。 @@ -93,6 +97,7 @@ help: @echo "make task-triage REQ='<需求>' TRIAGE_RECEIPT= 以远端为准发现接纳中的 initiative 并写分流收据(候选存在时 exit 5,待决策)" @echo "make task-triage-decide TRIAGE_RECEIPT= VERDICT=JOIN|NEW [INITIATIVE=] REASON='<理由>' 记录显式分流决策(不按数量自动 JOIN)" @echo "make task-start NAME=feature/ TRIAGE_RECEIPT= [KIND=...] [INITIATIVE=...] 建任务前核验分流收据(失败不建分支)" + @echo "make task-pr-check [PR=N] [FINAL=1] 核验当前任务分支 PR 的 base/head 门禁(远端为准;FINAL=1 用于 initiative 收口)" agent-exec-validate: PYTHONDONTWRITEBYTECODE=1 python3 scripts/agent_exec.py validate --repo . diff --git a/scripts/task_pr_check.py b/scripts/task_pr_check.py new file mode 100644 index 00000000..d928476c --- /dev/null +++ b/scripts/task_pr_check.py @@ -0,0 +1,291 @@ +#!/usr/bin/env python3 +"""PR base/head 门禁(两级分支模型:child → initiative → main)。 + +子命令: + local [--pr N] [--final] + 核验当前检出的任务分支及其 PR(事实来源是 Gitea 与 origin):PR 处于开放状态、base 为卡片期望目标、 + PR head 与本地 HEAD 一致、origin 分支头与本地 HEAD 一致、期望目标分支存在;child 的父 initiative + 须仍有开放 PR 指向其 root_target。--final(仅 initiative):accepting_requirements=false,且 + `## 子任务登记` 中「必需=是」的子任务为「已合入」(PR 已合并入本 initiative)或「已延期」。 + ci + 仅 pull_request 事件:按 GITHUB_HEAD_REF 在检出的 .task/ 顶层取卡片,核对 GITHUB_BASE_REF。 + 不联网、不需要 token;非 pull_request 事件跳过(退出 0)。 + +期望目标:child → immediate_target;其余(initiative / standalone / 无卡)→ root_target(无卡即 main)。 + +退出码:0 通过;1 门禁拒绝(PR_CHECK_FAILED);2 用法;6 远端无法核验(REMOTE_UNVERIFIED,fail closed)。 +仅使用标准库,复用 task_triage 的 Gitea 客户端与 token 读取。token 不打印、不写入任何产物。 +""" +import argparse +import os +import re +import subprocess +import sys +import urllib.parse +from pathlib import Path + +from task_triage import (EXIT_REMOTE, Stop, api_base, as_dict, card_branch, http_get_json, + open_pulls, read_token, repo_slug, target_branch) + +EXIT_FAILED = 1 +EXIT_USAGE = 2 +CHILD_COLUMNS = ("子任务分支", "必需", "状态", "PR") +DONE_STATES = ("已合入", "已延期") + + +def sh(*args): + return subprocess.run(["git", *args], capture_output=True, text=True, + encoding="utf-8", errors="replace") + + +def standalone(target): + return {"kind": "standalone", "initiative_id": "", "root_target": target, + "immediate_target": target, "accepting": "false"} + + +def card_meta(text, target): + """两级分支元数据行;无该行即旧卡,按 standalone 处理(与 task.sh 的 card_branch_kind 一致)。""" + match = re.search(r"^- 分支类型:([^\n]*)$", text, re.M) + if not match: + return standalone(target) + line = match.group(1) + + def field(name, default): + found = re.search(";" + name + r":([^;]*)", line) + return found.group(1).strip() if found else default + + return {"kind": line.split(";", 1)[0].strip(), "initiative_id": field("initiative_id", ""), + "root_target": field("root_target", target), + "immediate_target": field("immediate_target", target), + "accepting": field("accepting_requirements", "false")} + + +def expected_base(meta): + return meta["immediate_target"] if meta["kind"] == "child" else meta["root_target"] + + +def find_card(branch): + """只看检出树的 .task/ 顶层(archive 不参与);返回 (路径, 文本),无卡返回 (None, None)。""" + found = [] + for path in sorted(Path(".task").glob("*.md")): + text = path.read_text(encoding="utf-8", errors="replace") + if card_branch(text) == branch: + found.append((path, text)) + if len(found) > 1: + raise Stop(EXIT_FAILED, f"ambiguous task cards for {branch}: {', '.join(str(p) for p, _ in found)}") + return found[0] if found else (None, None) + + +def branch_meta(branch, target): + _, text = find_card(branch) + return (standalone(target) if text is None else card_meta(text, target)), text + + +def current_branch(): + proc = sh("symbolic-ref", "--quiet", "--short", "HEAD") + if proc.returncode != 0 or not proc.stdout.strip(): + raise Stop(EXIT_USAGE, "detached HEAD: check out the task branch first") + return proc.stdout.strip() + + +def local_head(): + proc = sh("rev-parse", "--verify", "-q", "HEAD^{commit}") + if proc.returncode != 0: + raise Stop(EXIT_USAGE, "cannot resolve local HEAD") + return proc.stdout.strip() + + +def remote_head(branch): + proc = sh("ls-remote", "origin", f"refs/heads/{branch}") + if proc.returncode != 0: + raise Stop(EXIT_REMOTE, "git ls-remote origin failed") + for line in proc.stdout.splitlines(): + sha, _, ref = line.partition("\t") + if ref == f"refs/heads/{branch}": + return sha + return None + + +def remote_context(): + origin = sh("remote", "get-url", "origin") + if origin.returncode != 0: + raise Stop(EXIT_REMOTE, "origin remote is not configured") + slug = repo_slug(origin.stdout.strip()) + token = read_token() + if not token: + raise Stop(EXIT_REMOTE, "no Gitea token: set GITEA_TOKEN or ~/.secrets/gitea ACCESS_TOKEN") + return api_base(), slug, token + + +def pull_request(api, slug, token, number): + owner, name = (urllib.parse.quote(part, safe="") for part in slug.split("/")) + pr = http_get_json(f"{api}/repos/{owner}/{name}/pulls/{number}", token) + if not isinstance(pr, dict): + raise Stop(EXIT_REMOTE, f"Gitea PR #{number} response is malformed") + return pr + + +def base_ref(pr): + return as_dict(pr.get("base")).get("ref") + + +def head_ref(pr): + return as_dict(pr.get("head")).get("ref") + + +def head_sha(pr): + return as_dict(pr.get("head")).get("sha") + + +def parent_pr_open(api, slug, token, parent, root): + return any(isinstance(pr, dict) and pr.get("state") == "open" + and head_ref(pr) == parent and base_ref(pr) == root + for pr in open_pulls(api, slug, token)) + + +def pr_number(cell): + for pattern in (r"pulls/(\d+)", r"#(\d+)", r"^(\d+)$"): + found = re.search(pattern, cell.strip()) + if found: + return int(found.group(1)) + return None + + +def child_registry(text): + """解析 `## 子任务登记` 表(列:子任务分支 | 必需 | 状态 | PR);缺失即拒绝(fail closed)。""" + lines = text.splitlines() + starts = [i for i, line in enumerate(lines) if line.strip() == "## 子任务登记"] + if not starts: + raise Stop(EXIT_FAILED, "initiative card has no ## 子任务登记 section") + rows, columns = [], None + for line in lines[starts[0] + 1:]: + if line.startswith("## "): + break + if not line.strip().startswith("|"): + continue + cells = [cell.strip() for cell in line.strip().strip("|").split("|")] + if columns is None: + if not all(name in cells for name in CHILD_COLUMNS): + raise Stop(EXIT_FAILED, "子任务登记 table must have columns 子任务分支 | 必需 | 状态 | PR") + columns = {name: cells.index(name) for name in CHILD_COLUMNS} + continue + if all(re.fullmatch(r":?-{3,}:?", cell) for cell in cells): + continue + if len(cells) <= max(columns.values()): + raise Stop(EXIT_FAILED, f"malformed 子任务登记 row: {line.strip()}") + rows.append({name: cells[index] for name, index in columns.items()}) + if columns is None: + raise Stop(EXIT_FAILED, "子任务登记 section has no table") + return rows + + +def final_problems(api, slug, token, branch, meta, text): + problems = [] + if meta["accepting"] != "false": + problems.append(f"initiative {branch} still accepts requirements " + f"(accepting_requirements={meta['accepting']}); set false before --final") + for row in child_registry(text): + name, required, status, pr_cell = row["子任务分支"], row["必需"], row["状态"], row["PR"] + if required == "否": + continue + if required != "是": + problems.append(f"child {name} has invalid 必需 value {required!r}") + elif status == "已合入": + number = pr_number(pr_cell) + if number is None: + problems.append(f"child {name} is 已合入 but PR cell {pr_cell!r} names no PR") + continue + pr = pull_request(api, slug, token, number) + if pr.get("merged") is not True or base_ref(pr) != branch: + problems.append(f"child {name} PR #{number} is not merged into {branch}") + elif status not in DONE_STATES: + problems.append(f"required child {name} is {status!r}; must be 已合入 or 已延期") + return problems + + +def verdict(problems, kind, base, head): + if problems: + for problem in problems: + print(f"[pr-check] FAIL {problem}") + print("[verdict] PR_CHECK_FAILED") + return EXIT_FAILED + print(f"[verdict] PR_CHECK_OK base={base} head={head} kind={kind}") + return 0 + + +def cmd_local(args): + if args.pr is not None and args.pr < 1: + raise Stop(EXIT_USAGE, "--pr must be a positive PR number") + branch = current_branch() + head = local_head() + meta, text = branch_meta(branch, target_branch()) + kind, expected = meta["kind"], expected_base(meta) + api, slug, token = remote_context() + if args.pr is not None: + pr = pull_request(api, slug, token, args.pr) + else: + pulls = [pr for pr in open_pulls(api, slug, token) if isinstance(pr, dict) and head_ref(pr) == branch] + if not pulls: + raise Stop(EXIT_FAILED, f"no open PR has head {branch}") + if len(pulls) > 1: + raise Stop(EXIT_FAILED, f"{len(pulls)} open PRs have head {branch}; pass --pr N") + pr = pulls[0] + number = pr.get("number") + problems = [] + if pr.get("merged") is True: + problems.append(f"PR #{number} is already merged") + elif pr.get("state") != "open": + problems.append(f"PR #{number} is not open (state={pr.get('state')})") + if base_ref(pr) != expected: + problems.append(f"PR #{number} base is {base_ref(pr)!r}; expected {expected!r} for {kind} branch {branch}") + if head_sha(pr) != head: + problems.append(f"PR #{number} head {head_sha(pr) or '(none)'} != local HEAD {head}") + remote = remote_head(branch) + if remote is None: + problems.append(f"origin has no branch {branch}") + elif remote != head: + problems.append(f"origin/{branch} is {remote}, local HEAD is {head}") + if remote_head(expected) is None: + problems.append(f"base branch {expected} is absent on origin") + if kind == "child" and not parent_pr_open(api, slug, token, expected, meta["root_target"]): + problems.append(f"parent initiative {expected} has no open PR into {meta['root_target']}") + if args.final and kind == "initiative": + problems.extend(final_problems(api, slug, token, branch, meta, text)) + return verdict(problems, kind, expected, head) + + +def cmd_ci(_args): + event = os.environ.get("GITHUB_EVENT_NAME", "") + if event != "pull_request": + print(f"[pr-check] skip: event {event or '(none)'} is not pull_request") + return 0 + base = os.environ.get("GITHUB_BASE_REF", "").strip() + head = os.environ.get("GITHUB_HEAD_REF", "").strip() + if not base or not head: + raise Stop(EXIT_FAILED, "pull_request event lacks GITHUB_BASE_REF or GITHUB_HEAD_REF") + meta, _ = branch_meta(head, target_branch()) + expected = expected_base(meta) + if base != expected: + raise Stop(EXIT_FAILED, f"PR base {base!r} differs from expected {expected!r} " + f"for {meta['kind']} branch {head}") + return verdict([], meta["kind"], base, head) + + +def main(argv=None): + parser = argparse.ArgumentParser(prog="task_pr_check.py", description="PR base/head 门禁(两级分支模型)") + sub = parser.add_subparsers(dest="cmd", required=True) + local = sub.add_parser("local") + local.add_argument("--pr", type=int) + local.add_argument("--final", action="store_true") + sub.add_parser("ci") + args = parser.parse_args(argv) + try: + return cmd_local(args) if args.cmd == "local" else cmd_ci(args) + except Stop as stop: + print(f"[pr-check] FAIL {stop.message}") + print("[verdict] REMOTE_UNVERIFIED" if stop.code == EXIT_REMOTE else "[verdict] PR_CHECK_FAILED") + return stop.code + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_task_pr_check.py b/tests/test_task_pr_check.py new file mode 100644 index 00000000..4454fd26 --- /dev/null +++ b/tests/test_task_pr_check.py @@ -0,0 +1,401 @@ +"""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()