import io import json import os import shutil import subprocess import sys import tempfile import time import unittest from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager, redirect_stdout from pathlib import Path from threading import Event from unittest import mock from scripts import agent_worktree, check_public_text, release_tag from scripts.agent_worktree import ( WorkspaceError, _git, _lines, create, integrate, processes_under, prune, ready, ) from scripts.version_bump import read_version from support.gitrepo import seed_repository #: 仓库里版本化的 git hook，测试原样装进临时仓库。 HOOKS = Path(__file__).resolve().parents[1] / agent_worktree.HOOKS_PATH AUTHOR = "Co-Authored-By: Codex (GPT-6) " def install_commit_hooks(repo: Path) -> None: hooks = repo / agent_worktree.HOOKS_PATH hooks.mkdir(parents=True) sources = list(HOOKS.iterdir()) sources += [HOOKS.parent / name for name in ("check_commit_message.py", "check_readme_impact.py", "co_author.py") if (HOOKS.parent / name).is_file()] for source in sources: relative = source.relative_to(HOOKS.parent.parent) target = repo / relative shutil.copyfile(source, target) target.chmod(0o755) _git(repo, "add", relative.as_posix()) _git(repo, "commit", "-m", "hooks") #: 版本号的唯一来源，`src/peach/__init__.py` 在测试仓库里的最小复刻。 VERSION_SEED = b'"""Peach application package."""\n\n__version__ = "0.7.14"\n' class StuckRemoval: """复刻 2026-09-01 Windows 上的实测：目录句柄被别的进程占着， `git worktree remove` 报 Permission denied。 `residue`：git 已经把文件删光、注册也摘掉了，只剩一个空目录删不掉。 `stuck`：注册还在，工作树整个没摘掉。 """ def __init__(self, real, modes: dict[Path, str]): self.real = real self.modes = {path.resolve(): mode for path, mode in modes.items()} def __call__(self, repo, *args, **kwargs): mode = None if args[:2] == ("worktree", "remove"): mode = self.modes.get(Path(args[-1]).resolve()) if mode is None: return self.real(repo, *args, **kwargs) path = Path(args[-1]) if mode == "residue": self.real(repo, *args, check=False) path.mkdir(parents=True, exist_ok=True) return subprocess.CompletedProcess( list(args), 1, "", f"fatal: failed to delete '{path}': Permission denied") def commit(repo: Path, message: str) -> None: _git(repo, "add", "tracked.txt", "worker.txt") _git(repo, "commit", "-m", message) class _WorktreeCase(unittest.TestCase): mock_pull_request = True def setUp(self): self.verification = mock.patch.object(agent_worktree, "require_verified") self.verification.start() self.addCleanup(self.verification.stop) if self.mock_pull_request: pull_request = mock.patch.object(agent_worktree, "require_pull_request", side_effect= lambda worker, branch, target, head: {"url": "https://github.com/example/peach/pull/1", "head": head, "base": target, "refs": []}) pull_request.start() self.addCleanup(pull_request.stop) self.tmp = tempfile.TemporaryDirectory() # 先 resolve：prune 报告的是 git worktree list 给出的真实路径，而 CI runner 的 # 临时目录是别名（macOS /var 软链到 /private/var，Windows 的 RUNNER~1 短名展开 # 成 runneradmin）。拿未 resolve 的路径去比对，本机全绿、CI 全红。 self.root = Path(self.tmp.name).resolve() self.repo = seed_repository(self.root / "repo", { ".gitignore": "build/\n", "tracked.txt": "base\n", "worker.txt": "base\n", "src/peach/__init__.py": VERSION_SEED, }, "base") def tearDown(self): self.tmp.cleanup() class AgentWorktreeTests(_WorktreeCase): def test_document_integration_keeps_its_checks_without_waiting_for_full_suite(self): result = create(self.repo, "Codex", "readme", self.root / "worktrees") worker = Path(result["path"]) (worker / "README.md").write_text("# Example\n", encoding="utf-8") _git(worker, "add", "README.md") _git(worker, "commit", "-m", "docs: example") folder = agent_worktree.test_evidence.evidence_dir(self.repo) with agent_worktree.test_evidence.held(folder / "full-suite.lock", scope="full"): report = integrate(self.repo, str(result["branch"]), lock_timeout=0) self.assertEqual(report["files"], ["README.md"]) agent_worktree.require_verified.assert_called_once_with(worker, "master", {"README.md"}) self.assertEqual((self.repo / "README.md").read_text(encoding="utf-8"), "# Example\n") def test_other_document_paths_use_the_same_checks_boundary(self): for index, name in enumerate(("CONTRIBUTING.md", ".github/PULL_REQUEST_TEMPLATE.md")): with self.subTest(path=name): result = create(self.repo, "Codex", f"document-{index}", self.root / "worktrees") worker = Path(result["path"]) (worker / name).parent.mkdir(parents=True, exist_ok=True) (worker / name).write_text("# Example\n", encoding="utf-8") _git(worker, "add", name) _git(worker, "commit", "-m", "docs: example") folder = agent_worktree.test_evidence.evidence_dir(self.repo) with agent_worktree.test_evidence.held(folder / "full-suite.lock", scope="full"): report = integrate(self.repo, str(result["branch"]), lock_timeout=0) self.assertEqual(report["files"], [name]) def test_runtime_markdown_still_waits_for_full_suite(self): result = create(self.repo, "Codex", "runtime-document", self.root / "worktrees") worker = Path(result["path"]) name = "frontend/src/react/boardui/ORIGIN.md" (worker / name).parent.mkdir(parents=True) (worker / name).write_text("# Origin\n", encoding="utf-8") _git(worker, "add", name) _git(worker, "commit", "-m", "docs: origin") folder = agent_worktree.test_evidence.evidence_dir(self.repo) with agent_worktree.test_evidence.held(folder / "full-suite.lock", scope="full"), \ self.assertRaisesRegex(WorkspaceError, "全量验证"): integrate(self.repo, str(result["branch"]), lock_timeout=0) self.assertFalse((self.repo / name).exists()) def test_waiting_code_integration_does_not_block_documents(self): """文档先合进去，排队的代码集成拿到锁后 rebase 到新的 master 上再合，不必回工作树重来。""" code = create(self.repo, "Codex", "queued-code", self.root / "worktrees") worker = Path(code["path"]) (worker / "worker.txt").write_text("code\n", encoding="utf-8") commit(worker, "worker change") origin = self.root / "origin.git" subprocess.run(["git", "init", "--bare", "-q", str(origin)], check=True) _git(self.repo, "remote", "add", "origin", str(origin)) _git(worker, "push", "-q", "origin", str(code["branch"])) for name in ("_verify_worker", "_await_pull_request_head"): patcher = mock.patch.object(agent_worktree, name) patcher.start() self.addCleanup(patcher.stop) docs = create(self.repo, "Codex", "parallel-docs", self.root / "worktrees") (Path(docs["path"]) / "README.md").write_text("# Example\n", encoding="utf-8") _git(Path(docs["path"]), "add", "README.md") _git(Path(docs["path"]), "commit", "-m", "docs: example") folder = agent_worktree.test_evidence.evidence_dir(self.repo) waiting = Event() held = agent_worktree.test_evidence.held @contextmanager def observed(lock, **kwargs): if kwargs.get("branch") == code["branch"] and lock.name == "full-suite.lock": waiting.set() with held(lock, **kwargs): yield with ThreadPoolExecutor() as executor, mock.patch.object( agent_worktree.test_evidence, "held", observed): with held(folder / "full-suite.lock", scope="full"): future = executor.submit(integrate, self.repo, str(code["branch"]), lock_timeout=120) self.assertTrue(waiting.wait(5), "代码集成尚未进入全量锁等待") report = integrate(self.repo, str(docs["branch"]), lock_timeout=0) self.assertEqual(report["files"], ["README.md"]) self.assertFalse(future.done()) queued = future.result(timeout=60) self.assertEqual(queued["files"], ["worker.txt"]) self.assertIsNotNone(queued["refreshed"], "排队期间 master 前进，拿到锁后自动 rebase") self.assertEqual((self.repo / "worker.txt").read_text(encoding="utf-8"), "code\n") self.assertEqual((self.repo / "README.md").read_text(encoding="utf-8"), "# Example\n") def test_code_integration_waits_for_full_suite_without_changing_master(self): result = create(self.repo, "Codex", "code", self.root / "worktrees") worker = Path(result["path"]) (worker / "worker.txt").write_text("worker\n", encoding="utf-8") (worker / "README.md").write_text("# Example\n", encoding="utf-8") _git(worker, "add", "README.md") commit(worker, "worker change") before = _git(self.repo, "rev-parse", "HEAD").stdout.strip() folder = agent_worktree.test_evidence.evidence_dir(self.repo) with agent_worktree.test_evidence.held(folder / "full-suite.lock", scope="full"), \ self.assertRaisesRegex(WorkspaceError, "全量验证"): integrate(self.repo, str(result["branch"]), lock_timeout=0) self.assertEqual(_git(self.repo, "rev-parse", "HEAD").stdout.strip(), before) self.assertEqual((self.repo / "worker.txt").read_text(encoding="utf-8"), "base\n") self.assertFalse((self.repo / "README.md").exists()) def test_create_ready_and_integrate_non_overlapping_branch(self): result = create(self.repo, "Claude", "metadata batch", self.root / "worktrees") worker = Path(result["path"]) (worker / "worker.txt").write_text("worker\n", encoding="utf-8") commit(worker, "worker change") report = ready(worker) self.assertEqual(report["same_file_changes"], []) merged = integrate(self.repo, str(result["branch"])) self.assertIn("worker.txt", merged["files"]) self.assertEqual((self.repo / "worker.txt").read_text(encoding="utf-8"), "worker\n") def test_ready_rejects_dirty_worker(self): result = create(self.repo, "Codex", "dirty task", self.root / "worktrees") worker = Path(result["path"]) (worker / "worker.txt").write_text("dirty\n", encoding="utf-8") with self.assertRaisesRegex(WorkspaceError, "dirty"): ready(worker) def test_integrate_refuses_same_file_changed_on_both_branches(self): result = create(self.repo, "Claude", "overlap", self.root / "worktrees") worker = Path(result["path"]) (worker / "tracked.txt").write_text("worker\n", encoding="utf-8") commit(worker, "worker overlap") (self.repo / "tracked.txt").write_text("main\n", encoding="utf-8") commit(self.repo, "main overlap") with self.assertRaisesRegex(WorkspaceError, "same-file"): integrate(self.repo, str(result["branch"])) self.assertEqual(_git(worker, "log", "-1", "--format=%s").stdout.strip(), "worker overlap", "有重叠时不自动 rebase") class QueuedRebaseTests(_WorktreeCase): """排队等锁期间目标分支前进：拿到锁后把工作者分支 rebase 上去、补验、推送，再合入。""" def setUp(self): super().setUp() self.verifier = mock.patch.object(agent_worktree, "_verify_worker") self.verify = self.verifier.start() self.addCleanup(self.verifier.stop) head = mock.patch.object(agent_worktree, "_await_pull_request_head") self.head = head.start() self.addCleanup(head.stop) self.origin = self.root / "origin.git" subprocess.run(["git", "init", "--bare", "-q", str(self.origin)], check=True) _git(self.repo, "remote", "add", "origin", str(self.origin)) def worker(self, name: str, path: str, text: str) -> tuple[Path, str]: result = create(self.repo, "Claude", name, self.root / "worktrees") worker = Path(result["path"]) (worker / path).parent.mkdir(parents=True, exist_ok=True) (worker / path).write_text(text, encoding="utf-8") _git(worker, "add", path) _git(worker, "commit", "-m", f"fix(tooling): {name}") _git(worker, "push", "-q", "origin", str(result["branch"])) return worker, str(result["branch"]) def test_a_target_that_moved_while_queued_is_rebased_verified_and_pushed(self): worker, branch = self.worker("queued", "worker.txt", "worker\n") before = _git(worker, "rev-parse", "HEAD").stdout.strip() (self.repo / "tracked.txt").write_text("main moved\n", encoding="utf-8") commit(self.repo, "main moved") report = integrate(self.repo, branch) after = _git(worker, "rev-parse", "HEAD").stdout.strip() self.assertEqual(report["refreshed"], {"from": before, "to": after}) self.assertNotEqual(before, after) self.verify.assert_called_once_with(self.repo, worker, frontend=False) self.head.assert_called_once_with(worker, branch, after) self.assertEqual(_git(self.origin, "rev-parse", branch).stdout.strip(), after, "PR 分支推到 rebase 后的提交") self.assertEqual((self.repo / "worker.txt").read_text(encoding="utf-8"), "worker\n") self.assertEqual((self.repo / "tracked.txt").read_text(encoding="utf-8"), "main moved\n") def test_a_failed_reverification_merges_nothing(self): worker, branch = self.worker("red-after-rebase", "worker.txt", "worker\n") (self.repo / "tracked.txt").write_text("main moved\n", encoding="utf-8") commit(self.repo, "main moved") master = _git(self.repo, "rev-parse", "HEAD").stdout.strip() self.verify.side_effect = WorkspaceError("rebase 后补验未通过") with self.assertRaisesRegex(WorkspaceError, "补验未通过"): integrate(self.repo, branch) self.assertEqual(_git(self.repo, "rev-parse", "HEAD").stdout.strip(), master) self.assertEqual(_git(worker, "status", "--porcelain").stdout, "") self.head.assert_not_called() def test_an_up_to_date_branch_and_a_documents_branch_are_not_rebased(self): worker, branch = self.worker("current", "worker.txt", "worker\n") self.assertIsNone(integrate(self.repo, branch)["refreshed"]) _, docs = self.worker("docs", "README.md", "# Example\n") (self.repo / "tracked.txt").write_text("main moved\n", encoding="utf-8") commit(self.repo, "main moved") self.assertIsNone(integrate(self.repo, docs)["refreshed"], "文档集成不持全量锁，不在锁内补验") self.verify.assert_not_called() def test_reverification_follows_the_test_entry_and_keeps_the_full_lock(self): self.verifier.stop() worker = self.root / "worker" calls = [] with mock.patch.object(agent_worktree.subprocess, "run", side_effect=lambda command, **kwargs: calls.append((command, kwargs)) or subprocess.CompletedProcess(command, 0)): agent_worktree._verify_worker(self.repo, worker, frontend=True) commands = [command for command, _ in calls] self.assertIn("scripts/test_environment.py", commands[0]) self.assertEqual(commands[1][-3:], ["--prefix", "frontend", "ci"]) self.assertEqual(commands[2][1:], ["scripts/test_runner.py", "--scope", "auto", "--jobs", "auto", "--full-lock-held"]) self.assertTrue(all(kwargs["cwd"] == worker for _, kwargs in calls)) self.assertEqual(calls[2][1]["env"]["PYTHONPATH"], str(worker / "src")) with mock.patch.object(agent_worktree.subprocess, "run", return_value=subprocess.CompletedProcess([], 1)), \ self.assertRaisesRegex(WorkspaceError, "integrate-verify.log"): agent_worktree._verify_worker(self.repo, worker, frontend=False) self.verifier.start() class PullRequestIntegrationTests(_WorktreeCase): mock_pull_request = False def setUp(self): super().setUp() result = create(self.repo, "Codex", "pr-boundary", self.root / "worktrees") self.worker = Path(result["path"]) self.branch = result["branch"] (self.worker / "worker.txt").write_text("worker\n", encoding="utf-8") commit(self.worker, "fix(tooling): 示例变更") self.head = _git(self.worker, "rev-parse", "HEAD").stdout.strip() self.pr = {"url": "https://github.com/example/peach/pull/12", "state": "OPEN", "headRefOid": self.head, "baseRefName": "master"} @contextmanager def github(self, stdout=None, returncode=0, stderr=""): run = subprocess.run query = mock.Mock(return_value=subprocess.CompletedProcess( ["gh"], returncode, json.dumps(self.pr) if stdout is None else stdout, stderr)) def call(command, *args, **kwargs): return query(command, *args, **kwargs) if command[0] == "gh" else run(command, *args, **kwargs) with mock.patch.object(agent_worktree.subprocess, "run", side_effect=call): yield query def assert_refused(self, message): before = _git(self.repo, "rev-parse", "HEAD").stdout.strip() with self.assertRaisesRegex(WorkspaceError, message): integrate(self.repo, self.branch) self.assertEqual(_git(self.repo, "rev-parse", "HEAD").stdout.strip(), before) self.assertEqual((self.repo / "worker.txt").read_text(encoding="utf-8"), "base\n") entry = next(row for row in agent_worktree._worktree_entries(self.repo) if row.get("branch") == self.branch) self.assertIn("locked", entry) def test_missing_pull_request_preserves_master(self): with self.github(returncode=1, stderr="no pull requests found for branch"): self.assert_refused("未取得 PR") def test_open_matching_pull_request_is_bound_to_the_merge(self): with self.github() as query: result = integrate(self.repo, self.branch) self.assertEqual(result["pull_request"], {"url": self.pr["url"], "head": self.head, "base": "master", "refs": [], "body": ""}) self.assertEqual((self.repo / "worker.txt").read_text(encoding="utf-8"), "worker\n") command = query.call_args.args[0] self.assertEqual(command[:4], ["gh", "pr", "view", self.branch]) self.assertEqual(query.call_args.kwargs["cwd"], self.worker) self.assertEqual(_git(self.repo, "log", "-1", "--format=%B").stdout.strip(), f"Merge branch '{self.branch}' (#12)") def test_issue_link_lines_in_the_body_follow_into_the_merge_commit(self): body = ("排查记录里顺带提到 #3，不算关联。\n\nRefs #45\r\n" " Refs peach-mitao/peach-roadmap#7\n" "Fixes https://github.com/example/peach/issues/8\n" "Refs Telegram 用户反馈（非 GitHub issue）\n") with self.github(json.dumps({**self.pr, "body": body})): result = integrate(self.repo, self.branch) refs = ["Refs #45", "Refs peach-mitao/peach-roadmap#7", "Fixes https://github.com/example/peach/issues/8"] self.assertEqual(result["pull_request"]["refs"], refs) self.assertEqual(_git(self.repo, "log", "-1", "--format=%B").stdout.strip(), f"Merge branch '{self.branch}' (#12)\n\n" + "\n".join(refs)) def test_library_lists_in_the_body_or_a_commit_message_are_refused(self): """PR 正文和提交说明公开在 GitHub 上，成串列出本机馆藏的不合入；问题需要的一两个例子照常合入。""" library = check_public_text.Library(codes={"ZZQ-101", "ZZQ-102", "ZZQ-103"}, names={"测试女优甲", "测试女优乙"}) listed = "剩下要人判的：ZZQ-101、ZZQ-102、ZZQ-103。" with mock.patch.object(check_public_text, "load_library", return_value=library): with self.github(json.dumps({**self.pr, "body": listed})): self.assert_refused("PR 正文（番号 3 个") (self.worker / "worker.txt").write_text("worker again\n", encoding="utf-8") commit(self.worker, "fix(tooling): 测试女优甲与测试女优乙的别名") self.pr["headRefOid"] = _git(self.worker, "rev-parse", "HEAD").stdout.strip() with self.github(json.dumps({**self.pr, "body": "只修 ZZQ-101 的前缀。"})): self.assertRaisesRegex(WorkspaceError, "提交 [0-9a-f]+（人物 2 个", integrate, self.repo, self.branch) def test_insider_terms_in_the_pr_body_are_refused_but_commit_messages_may_name_the_tooling(self): with mock.patch.object(check_public_text, "load_library", return_value=None): self.assertRaisesRegex(WorkspaceError, "内部用语.*合批", agent_worktree.require_public_text, self.repo, {"PR 正文": "合批验证：两处反馈修复。"}) agent_worktree.require_public_text(self.repo, {"PR 正文": "两处反馈修复一起合入。", "提交 abc1234": "fix(worktree): 工作树残留检查跳过空目录"}) def test_a_single_example_or_a_machine_without_a_ledger_still_integrates(self): library = check_public_text.Library(codes={"ZZQ-101", "ZZQ-102", "ZZQ-103"}) for loaded, body in ((library, "只修 ZZQ-101 的前缀。"), (None, "ZZQ-101、ZZQ-102、ZZQ-103")): with self.subTest(body=body), mock.patch.object(check_public_text, "load_library", return_value=loaded): agent_worktree.require_public_text(self.repo, {"PR 正文": body}) def test_an_older_remote_head_and_another_base_are_refused(self): for field, value, message in (("headRefOid", "0" * 40, "提交.*不一致"), ("baseRefName", "other", "目标分支"), ("state", "CLOSED", "必须开放"), ("state", "MERGED", "必须开放")): with self.subTest(field=field, value=value), self.github(json.dumps({**self.pr, field: value})): self.assert_refused(message) def test_unreadable_or_invalid_receipts_never_count_as_a_pull_request(self): cases = (("", "有效 JSON"), ("{", "有效 JSON"), ("[]", "字段无效"), (json.dumps({**self.pr, "headRefOid": None}), "字段无效"), (json.dumps({**self.pr, "url": "file:///private"}), "链接无效")) for stdout, message in cases: with self.subTest(stdout=stdout), self.github(stdout): self.assert_refused(message) def test_an_unavailable_client_and_a_timeout_preserve_master(self): for error in (FileNotFoundError("gh"), subprocess.TimeoutExpired(["gh"], 60)): with self.subTest(error=type(error).__name__), self.github() as query: query.side_effect = error self.assert_refused("未取得 PR") def test_a_worker_change_during_the_remote_read_prevents_merge(self): before = _git(self.repo, "rev-parse", "HEAD").stdout.strip() with self.github() as query: reply = query.return_value def advance(*_args, **_kwargs): (self.worker / "worker.txt").write_text("parallel\n", encoding="utf-8") commit(self.worker, "fix(tooling): 并行变更") return reply query.side_effect = advance with self.assertRaisesRegex(WorkspaceError, "工作者分支改变"): integrate(self.repo, self.branch) self.assertEqual(_git(self.repo, "rev-parse", "HEAD").stdout.strip(), before) self.assertNotEqual(_git(self.worker, "rev-parse", "HEAD").stdout.strip(), self.head) def test_the_requested_target_branch_is_preserved(self): _git(self.repo, "branch", "-m", "master", "develop") with self.github(json.dumps({**self.pr, "baseRefName": "develop"})): result = integrate(self.repo, self.branch, "develop") self.assertEqual(result["pull_request"]["base"], "develop") class PreflightTests(_WorktreeCase): def worker(self): item = create(self.repo, "Codex", "preflight", self.root / "worktrees") worker = Path(item["path"]) path = worker / "src/peach/module.py" path.write_text("VALUE = 1\n", encoding="utf-8") _git(worker, "add", "src/peach/module.py") _git(worker, "commit", "-m", "fix(tooling): 示例规则", "-m", AUTHOR) return worker def test_the_cli_checks_metadata_without_reading_or_signing_test_proof(self): worker = self.worker() output = io.StringIO() with mock.patch.object(sys, "argv", ["agent_worktree.py", "--repo", str(worker), "preflight"]), \ mock.patch.object(agent_worktree.test_evidence, "key") as key, redirect_stdout(output): self.assertEqual(agent_worktree.main(), 0) result = json.loads(output.getvalue()) self.assertEqual((result["action"], result["metadata"], result["tests"]), ("preflight", "passed", "not_checked")) self.assertEqual(result["head"], _git(worker, "rev-parse", "HEAD").stdout.strip()) key.assert_not_called() agent_worktree.require_verified.assert_not_called() agent_worktree.require_pull_request.assert_not_called() self.assertFalse((self.repo / "build/agent-verification").exists()) def test_single_readme_change_passes_preflight_with_authorship_only(self): worker = self.worker() (worker / 'README.md').write_text('# Example\n', encoding='utf-8') _git(worker, 'add', 'README.md') _git(worker, 'commit', '-m', 'docs: 更新说明', '-m', AUTHOR) with mock.patch.object(agent_worktree.test_evidence, "key") as key: result = agent_worktree.preflight(worker) self.assertEqual(result['metadata'], 'passed') key.assert_not_called() def test_dirty_content_and_a_newer_target_require_preparation(self): worker = self.worker() path = worker / "src/peach/module.py" path.write_text("VALUE = 2\n", encoding="utf-8") with self.assertRaisesRegex(WorkspaceError, "dirty"): agent_worktree.preflight(worker) path.write_text("VALUE = 1\n", encoding="utf-8") (self.repo / "tracked.txt").write_text("parallel\n", encoding="utf-8") commit(self.repo, "docs(tooling): 并行变更") with self.assertRaisesRegex(WorkspaceError, "已前进"): agent_worktree.preflight(worker) class MasterCommitGuardTests(_WorktreeCase): """主检出的 master 只收 `integrate` 与 `release_tag.py` 的提交。 2026-09 进 master 的五个红提交，四个是在主检出上直接提交，一个是手工 merge， 都没经过 ready / integrate 的检查。这里用仓库里那份 hook 原样装进测试仓库。 """ def setUp(self): super().setUp() install_commit_hooks(self.repo) environment = mock.patch.dict(os.environ, {"PATH": str(Path(sys.executable).parent) + os.pathsep + os.environ.get("PATH", "")}) environment.start() self.addCleanup(environment.stop) result = create(self.repo, "Claude", "guarded", self.root / "worktrees") self.worker, self.branch = Path(result["path"]), str(result["branch"]) (self.worker / "worker.txt").write_text("worker\n", encoding="utf-8") commit(self.worker, "test: worker change\n\n" + AUTHOR) def test_create_points_the_repository_at_the_versioned_hooks(self): self.assertEqual(_git(self.repo, "config", "core.hooksPath").stdout.strip(), agent_worktree.HOOKS_PATH) def test_a_direct_commit_on_master_in_the_main_checkout_is_refused(self): before = _git(self.repo, "rev-parse", "HEAD").stdout (self.repo / "tracked.txt").write_text("main\n", encoding="utf-8") with self.assertRaisesRegex(WorkspaceError, "主检出只做集成"): commit(self.repo, "direct") self.assertEqual(_git(self.repo, "rev-parse", "HEAD").stdout, before) def test_a_hand_made_merge_into_master_is_refused(self): before = _git(self.repo, "rev-parse", "HEAD").stdout with self.assertRaisesRegex(WorkspaceError, "主检出只做集成"): _git(self.repo, "merge", "--no-ff", "-m", "merge by hand", self.branch) self.assertEqual(_git(self.repo, "rev-parse", "HEAD").stdout, before) def test_integrate_still_merges_the_worker_branch(self): self.assertEqual(_git(self.worker, "status", "--porcelain").stdout, "") integrate(self.repo, self.branch) self.assertEqual((self.repo / "worker.txt").read_text(encoding="utf-8"), "worker\n") def test_the_release_commit_carries_the_pass(self): """`release_tag.py` 发出的实际参数由 `test_scripts.ShipTests` 钉住。""" (self.repo / "tracked.txt").write_text("release\n", encoding="utf-8") _git(self.repo, "add", "tracked.txt") _git(self.repo, "-c", f"{release_tag.MASTER_WRITER}=release", "commit", "-m", "release") self.assertEqual(_git(self.repo, "log", "-1", "--format=%s").stdout.strip(), "release") class CommitMessageGuardTests(_WorktreeCase): def setUp(self): super().setUp() install_commit_hooks(self.repo) environment = mock.patch.dict(os.environ, {"PATH": str(Path(sys.executable).parent) + os.pathsep + os.environ.get("PATH", "")}) environment.start() self.addCleanup(environment.stop) result = create(self.repo, "codex", "message", self.root / "worktrees with spaces") self.worker = Path(result["path"]) (self.worker / "worker.txt").write_text("worker\n", encoding="utf-8") _git(self.worker, "add", "worker.txt") def attempt(self, message): path = self.worker / "message.txt" path.write_text(message, encoding="utf-8") return _git(self.worker, "commit", "-F", str(path), check=False) def test_a_missing_or_invalid_author_refuses_the_commit_and_keeps_the_index(self): before = _git(self.worker, "rev-parse", "HEAD").stdout for trailer in ("", "Co-Authored-By: Codex ", "Co-Authored-By: Codex (GPT-6) "): with self.subTest(trailer=trailer): result = self.attempt("test: check\n\n" + trailer) self.assertNotEqual(result.returncode, 0) self.assertIn("署名" if trailer else "Co-Authored-By", result.stderr) self.assertEqual(_git(self.worker, "rev-parse", "HEAD").stdout, before) self.assertEqual(_git(self.worker, "diff", "--cached", "--name-only").stdout.strip(), "worker.txt") def test_optional_readme_trailers_are_accepted_as_commit_text(self): for declaration in ("README-Impact: none; 内部测试\n\n", "README-Impact: none\n", "README-Impact: none; 内部测试\nREADME-Impact: updated; 文档\n", "README-Reviewed: README.zh-CN.md; 已核对\n\n", "README-Reviewed: README.zh-CN.md;\n", "README-Reviewed: unknown.md; 已核对\n"): with self.subTest(declaration=declaration): self.assertEqual(check_commit_message.check(self.worker, "test: check\n\n" + declaration + AUTHOR), []) def test_valid_trailers_and_amend_preserve_the_message(self): message = ("test: 中文提交\n\nREADME-Impact: updated; 两种语言\n" + AUTHOR + "\nCo-Authored-By: Claude Code (Opus 5) \n") result = self.attempt(message) self.assertEqual(result.returncode, 0, result.stderr) amended = _git(self.worker, "commit", "--amend", "--no-edit", check=False) self.assertEqual(amended.returncode, 0, amended.stderr) self.assertEqual(_git(self.worker, "show", "-s", "--format=%B").stdout.strip(), message.strip()) def test_human_and_agent_trailers_are_preserved_in_the_commit(self): message = ("test: 共同作者署名\n\n" + AUTHOR + "\nCo-Authored-By: Contributor \n") result = self.attempt(message) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(_git(self.worker, "show", "-s", "--format=%B").stdout.strip(), message.strip()) def test_commit_accepts_authorship_without_readme_trailers(self): result = self.attempt("test: intermediate\n\n" + AUTHOR) self.assertEqual(result.returncode, 0, result.stderr) def test_author_lines_separated_into_different_blocks_are_rejected(self): result = self.attempt("test: authors\n\n" + AUTHOR + "\n\n" "Co-Authored-By: Claude Code (Opus 5) ") self.assertNotEqual(result.returncode, 0) self.assertIn("Co-Authored-By", result.stderr) def test_git_comment_lines_do_not_count_as_declarations(self): result = self.attempt("test: comments\n\n" + AUTHOR + "\n\n# Co-Authored-By: example\n") self.assertEqual(result.returncode, 0, result.stderr) def test_merge_messages_are_exempt_from_authorship_trailers(self): result = self.attempt("test: worker\n\n" + AUTHOR) self.assertEqual(result.returncode, 0, result.stderr) _git(self.repo, "switch", "-c", "parallel") (self.repo / "tracked.txt").write_text("parallel\n", encoding="utf-8") _git(self.repo, "add", "tracked.txt") _git(self.repo, "commit", "-m", "test: parallel") merged = _git(self.worker, "merge", "--no-ff", "-m", "Merge parallel", "parallel", check=False) self.assertEqual(merged.returncode, 0, merged.stderr) if __name__ == "__main__": unittest.main() class PruneTests(_WorktreeCase): """回收已经并入 master 的隔离工作树。 `create` 一直有人用，回收却从来没有入口——2026-08-29 清到 3 个，两天后长回 74 个、 占 868 MB。一次性清理解决不了这个问题，所以把回收做成命令，并在这里锁住它的边界。 """ def _integrated(self, name: str) -> Path: result = create(self.repo, "Codex", name, self.root / "worktrees") worker = Path(result["path"]) (worker / "worker.txt").write_text(name + " done\n", encoding="utf-8") commit(worker, f"{name} change") integrate(self.repo, str(result["branch"])) return worker def test_reporting_does_not_remove_anything(self): """默认只报告。回收不可逆，不该是运行一下就顺手做了的事。""" worker = self._integrated("reportable") report = prune(self.repo) self.assertIn(str(worker), report["reclaimed"]) self.assertFalse(report["applied"]) self.assertTrue(worker.is_dir(), "没给 --apply 就不该动手") def test_apply_reclaims_the_worktree_and_its_branch(self): worker = self._integrated("reclaimable") branch = _git(self.repo, "branch", "--list", "agent/codex/reclaimable").stdout self.assertIn("reclaimable", branch) prune(self.repo, apply=True) self.assertFalse(worker.exists(), "工作树应当被回收") self.assertNotIn("reclaimable", _git(self.repo, "branch", "--list", "agent/codex/reclaimable").stdout, "已并入的分支也应一并删掉") def test_a_worktree_with_files_under_build_is_kept_even_when_clean_and_merged(self): """`build/` 被 Git 忽略，`status` 看不见它；里面却是会话的日志、报告和派单，删了没有备份。""" (self.repo / ".git" / "info").mkdir(exist_ok=True) (self.repo / ".git" / "info" / "exclude").write_text("build/\n", encoding="utf-8") worker = self._integrated("has-build") (worker / "build").mkdir() (worker / "build" / "report.md").write_text("复盘报告\n", encoding="utf-8") report = prune(self.repo, apply=True) self.assertTrue((worker / "build" / "report.md").is_file()) self.assertEqual(report["reclaimed"], []) self.assertIn("build/", next(row["why"] for row in report["kept"] if row["path"] == str(worker))) def test_a_worktree_a_process_is_working_in_is_kept(self): """会话、终端或预览服务还开在里面的工作树不回收：分支合入了，人还在用。""" worker = self._integrated("in-use") process = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"], cwd=worker) try: report = prune(self.repo, apply=True) finally: process.kill() process.wait(timeout=15) self.assertTrue(worker.is_dir()) why = next(row["why"] for row in report["kept"] if row["path"] == str(worker)) self.assertIn(str(process.pid), why) def test_a_dirty_worktree_is_refused_even_when_its_branch_is_merged(self): """分支已合入不等于工作区里没东西。 实测就有这样的工作树：分支早已并入 master，里面却躺着一份成形的未提交改动 （按钮等待态改用 aria-busy，还配了对应的测试）。所以脏的一律拒收并单独列出， 交给人看，不给 --force 这个口子。 """ worker = self._integrated("has-wip") (worker / "worker.txt").write_text("未提交的后续改动\n", encoding="utf-8") report = prune(self.repo, apply=True) self.assertTrue(worker.is_dir(), "脏工作树不能被回收") self.assertEqual([row["path"] for row in report["dirty"]], [str(worker)]) self.assertEqual(report["reclaimed"], []) def test_an_unmerged_branch_is_kept_with_a_reason(self): result = create(self.repo, "Codex", "still-open", self.root / "worktrees") worker = Path(result["path"]) (worker / "worker.txt").write_text("still open wip\n", encoding="utf-8") commit(worker, "still open") report = prune(self.repo, apply=True) self.assertTrue(worker.is_dir()) self.assertEqual([row["branch"] for row in report["kept"]], ["agent/codex/still-open"]) def test_the_main_worktree_is_never_a_candidate(self): self._integrated("bystander") report = prune(self.repo) listed = report["reclaimed"] + [row["path"] for row in report["dirty"] + report["kept"]] self.assertNotIn(str(self.repo), listed, "主检出不能出现在回收清单里") def test_a_directory_git_could_not_delete_is_deleted_by_the_script(self): """git 摘了注册却删不掉目录时，脚本自己把目录删掉，不留给人。 2026-09-01 实测：连跑 5 次 prune --apply 才把 5 个工作树摘完，5 条分支一条没删， 最后是手工 `git branch -d` 收的尾——因为 remove 一失败就抛错中止了整轮。分支的 那半边当时补上了，目录这半边还写着「待人工清理」，于是残留目录照样留在原地： 长得和真工作树一样，在里面跑 git 全作用于主检出的 master。 """ worker = self._integrated("stuck-handle") with mock.patch.object(agent_worktree, "_git", StuckRemoval(agent_worktree._git, {worker: "residue"})): report = prune(self.repo, apply=True) self.assertTrue(report["ok"], "一个目录删不掉不该把整轮判成失败") self.assertFalse(worker.exists(), "git 删不掉的空目录由脚本删掉") self.assertEqual(report["residue"], [], "删掉了就不是残留") self.assertEqual(report["reclaimed"], [str(worker)]) self.assertNotIn("stuck-handle", _git(self.repo, "branch", "--list", "agent/codex/stuck-handle").stdout, "注册已经摘掉了，分支就该跟着删") def test_a_directory_nobody_can_delete_is_reported_with_the_reason(self): """脚本也删不掉才算残留：报出原因交给人，分支照删。""" worker = self._integrated("locked-handle") with mock.patch.object(agent_worktree, "_git", StuckRemoval(agent_worktree._git, {worker: "residue"})), \ mock.patch.object(agent_worktree, "_delete_tree", return_value="PermissionError: 目录被别的进程占着"): report = prune(self.repo, apply=True) self.assertTrue(worker.is_dir(), "删不掉就还在原地") self.assertEqual([row["path"] for row in report["residue"]], [str(worker)]) self.assertIn("目录被别的进程占着", report["residue"][0]["why"]) self.assertEqual(report["reclaimed"], []) self.assertNotIn("locked-handle", _git(self.repo, "branch", "--list", "agent/codex/locked-handle").stdout) def test_one_stuck_worktree_does_not_block_the_others(self): """回收失败只波及它自己，剩下的工作树照常处理，不用人反复重跑。""" stuck = self._integrated("stuck-first") other = self._integrated("second-in-line") with mock.patch.object(agent_worktree, "_git", StuckRemoval(agent_worktree._git, {stuck: "stuck"})): report = prune(self.repo, apply=True) self.assertEqual(report["reclaimed"], [str(other)]) self.assertFalse(other.exists()) self.assertNotIn("second-in-line", _git(self.repo, "branch", "--list", "agent/codex/second-in-line").stdout) self.assertEqual([row["branch"] for row in report["failed"]], ["agent/codex/stuck-first"]) self.assertTrue(stuck.is_dir()) self.assertIn("stuck-first", _git(self.repo, "branch", "--list", "agent/codex/stuck-first").stdout, "注册还在就是没回收成，分支不能删") class UnregisteredLeftoverTests(_WorktreeCase): """清掉两个工作树落点里已经不登记的残留目录。 `create` 的落点和 Claude Code 内置的落点都会留下这种目录：它不在 `git worktree list` 里，所以按登记项遍历的那轮回收永远碰不到它；人却会走进去当 工作树用，而在里面跑 git 全部作用于主检出的 master。两处同一条规则，只盖其中 一处的话，另一处的残留就只能靠人想起来。 """ def roots(self) -> tuple[Path, ...]: """`create` 建在主检出旁边，Claude Code 建在主检出里。""" return (self.repo.parent / agent_worktree.WORKTREE_ROOT, self.repo / agent_worktree.BUILTIN_WORKTREES) def leftover(self, root: Path, name: str) -> Path: path = root / name path.mkdir(parents=True) return path def test_an_unregistered_empty_leftover_is_swept(self): for root in self.roots(): with self.subTest(root=root.name): empty = self.leftover(root, "distracted-lamarr") report = prune(self.repo, apply=True) self.assertEqual(report["swept"], [str(empty)]) self.assertFalse(empty.exists()) def test_reporting_lists_the_leftover_without_deleting_it(self): for root in self.roots(): with self.subTest(root=root.name): empty = self.leftover(root, "reportable-leftover") report = prune(self.repo) self.assertEqual(report["swept"], [str(empty)]) self.assertTrue(empty.is_dir(), "没给 --apply 就不该动手") shutil.rmtree(empty) def test_a_leftover_that_still_holds_files_is_left_alone(self): """空目录才扫。里面还有文件就可能是别人正开着的检出或没提交的东西。""" for root in self.roots(): with self.subTest(root=root.name): used = self.leftover(root, "sweet-newton") (used / "worker.txt").write_text("别人正在用\n", encoding="utf-8") report = prune(self.repo, apply=True) self.assertEqual(report["swept"], []) self.assertTrue(used.is_dir()) self.assertEqual([row["path"] for row in report["kept"]], [str(used)]) shutil.rmtree(used) def test_a_registered_worktree_under_either_root_is_not_swept(self): """登记着的工作树归上面那轮按分支状态处理，扫残留这步不许碰。""" for index, root in enumerate(self.roots()): with self.subTest(root=root.name): task = f"in-place-{index}" result = create(self.repo, "Claude", task, root) worker = Path(result["path"]) (worker / "worker.txt").write_text("还在做\n", encoding="utf-8") commit(worker, f"{task} wip") report = prune(self.repo, apply=True) self.assertEqual(report["swept"], []) self.assertTrue(worker.is_dir()) self.assertIn(f"agent/claude/{task}", [row["branch"] for row in report["kept"]]) class LeftoverProcessTests(_WorktreeCase): """找出工作树落点下还活着的进程：谁占着删不掉的目录、机器变卡时谁是自己的残留。""" def sleeper(self, cwd: Path) -> subprocess.Popen: cwd.mkdir(parents=True, exist_ok=True) child = subprocess.Popen([sys.executable, "-c", "import sys; sys.stdin.read()"], cwd=cwd, stdin=subprocess.PIPE) deadline = time.monotonic() + 10 while time.monotonic() < deadline: rows = processes_under([cwd]) if any(row["pid"] == child.pid for row in rows): break time.sleep(0.1) return child def stop(self, child: subprocess.Popen) -> None: # EOF 让解释器自己退出；Windows venv 启动器也会等其子进程结束。 child.communicate(timeout=10) def test_a_process_working_in_a_worktree_is_listed_with_its_details(self): child = self.sleeper(self.repo.parent / agent_worktree.WORKTREE_ROOT / "busy") try: report = agent_worktree.processes(self.repo) finally: self.stop(child) row = next(row for row in report["processes"] if row["pid"] == child.pid) self.assertFalse(row["orphan"], "父进程（测试本身）还在") self.assertFalse(row["caller"]) self.assertGreaterEqual(row["age_minutes"], 0) self.assertIn("cpu_percent", row) self.assertIn("memory_percent", report["machine"]) def test_the_caller_and_processes_elsewhere_are_told_apart(self): child = self.sleeper(self.root / "elsewhere") try: rows = processes_under([self.root / "elsewhere", Path(os.getcwd())]) finally: self.stop(child) self.assertIn(child.pid, [row["pid"] for row in rows]) self.assertNotIn(child.pid, [row["pid"] for row in processes_under([self.repo])]) mine = [row for row in rows if row["pid"] == os.getpid()] self.assertEqual([row["caller"] for row in mine], [True], "调用者自己这条链要标出来") def test_a_directory_nobody_can_delete_names_the_process_holding_it(self): held = self.repo / agent_worktree.BUILTIN_WORKTREES / "held-open" child = self.sleeper(held) try: with mock.patch.object(agent_worktree, "_delete_tree", return_value="PermissionError: 目录被别的进程占着"): report = prune(self.repo, apply=True) finally: self.stop(child) self.assertEqual([row["path"] for row in report["residue"]], [str(held)]) self.assertIn(child.pid, [row["pid"] for row in report["residue"][0]["holders"]]) def test_without_psutil_the_scan_says_it_was_not_obtained(self): with mock.patch.dict(sys.modules, {"psutil": None}): self.assertTrue(processes_under([self.repo]).startswith("未取得")) report = create(self.repo, "Codex", "no-psutil", self.root / "worktrees") self.assertTrue(report["orphan_processes"].startswith("未取得")) class IntegrationVersionTests(_WorktreeCase): """集成报告版本号，但把推进它的权力留给发布入口。 一天几十次集成共用一个版本号：跑着的是哪一份由 commit 认定，`build-info.json` 随包走，托盘按它算自己落后多少个提交。版本号一次发布推一格，于是每个 `X.Y.Z` 都对应一份能下载到的制品（ADR-0012）。 """ def worker_touching(self, name: str, path: str, text: str = "改了\n", *, subject: str | None = None) -> str: result = create(self.repo, "Claude", name, self.root / "worktrees") worker = Path(result["path"]) target = worker / path target.parent.mkdir(parents=True, exist_ok=True) target.write_text(text, encoding="utf-8") _git(worker, "add", path) _git(worker, "commit", "-m", subject or f"fix: {name}") return str(result["branch"]) def test_a_runtime_fix_merges_without_touching_the_version(self): branch = self.worker_touching("runtime-change", "src/peach/module.py") report = integrate(self.repo, branch) self.assertEqual(report["version"], "0.7.14") self.assertEqual(read_version(self.repo), "0.7.14") self.assertEqual(_git(self.repo, "log", "-1", "--format=%s").stdout.strip(), f"Merge branch '{branch}' (#1)") touched = _lines(_git(self.repo, "show", "--name-only", "--format=", "HEAD")) self.assertNotIn("src/peach/__init__.py", touched) def test_a_feature_commit_stays_on_the_same_version_too(self): """`feat` 决定的是发布时推哪一位，不是集成时推不推。""" branch = self.worker_touching("feature-work", "src/peach/module.py", subject="feat: 新功能") report = integrate(self.repo, branch) self.assertEqual(report["version"], "0.7.14") self.assertEqual(read_version(self.repo), "0.7.14") def test_a_new_migration_stays_on_the_same_version_too(self): branch = self.worker_touching("schema", "migrations/0099_new.sql", subject="fix: 补一列") self.assertEqual(integrate(self.repo, branch)["version"], "0.7.14") self.assertEqual(read_version(self.repo), "0.7.14") def test_integration_never_creates_a_version_tag_itself(self): """打标签只有 `scripts/release_tag.py` 一个入口，它遇到同名本地标签就拒绝。 集成这一步再造一个本地标签，等于把唯一的发布入口挡在门外。 """ branch = self.worker_touching("taggable", "src/peach/module.py") report = integrate(self.repo, branch) self.assertEqual(report["release_tag_entry"], "scripts/release_tag.py") self.assertEqual(_lines(_git(self.repo, "tag", "--list")), []) def test_every_integration_reports_whether_a_release_is_due(self): """判断挂在集成的输出里，不靠谁记着去查——「记着」正是它要替掉的东西。""" branch = self.worker_touching("user-visible", "src/peach/module.py", subject="fix(web): 修一处使用者看得见的") report = integrate(self.repo, branch) self.assertEqual(report["release"]["entries"], 1) self.assertEqual(report["release"]["groups"], {"修复": 1}) self.assertIn("due", report["release"]) def test_development_only_work_never_asks_for_a_release(self): branch = self.worker_touching("inside-only", "src/peach/module.py", subject="refactor(web): 只动开发过程") report = integrate(self.repo, branch) self.assertEqual((report["release"]["entries"], report["release"]["due"]), (0, False)) from scripts.check_readme_impact import git from scripts import check_commit_message, co_author class CoAuthorTests(unittest.TestCase): """交付提交的署名。追责要落到具体的模型，所以工具名单独一个词不够。""" def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.repo = Path(self.temp.name).resolve() git(self.repo, "init", "-b", "master") git(self.repo, "config", "user.name", "Test") git(self.repo, "config", "user.email", "test@example.invalid") git(self.repo, "commit", "--allow-empty", "-m", "base") git(self.repo, "checkout", "-b", "worker") def signed(self, *trailers): git(self.repo, "commit", "--allow-empty", "-m", "测试", "-m", "\n".join(trailers)) return co_author.check(self.repo, "master") def test_an_unsigned_commit_cannot_be_traced_back_to_a_model(self): self.assertTrue(self.signed()) def test_the_tool_alone_does_not_say_which_model_wrote_it(self): self.assertTrue(self.signed("Co-Authored-By: Claude Code ")) def test_each_registered_tool_passes_with_its_own_vendor(self): self.assertEqual( self.signed("Co-Authored-By: Claude Code (Opus 5) "), []) self.assertEqual( self.signed("Co-Authored-By: Codex (GPT-5.5) "), []) self.assertEqual( self.signed("Co-Authored-By: opencode (GLM 5.3 Flash) "), []) def test_an_address_that_does_not_belong_to_the_tool_is_rejected(self): self.assertTrue(self.signed("Co-Authored-By: Codex (GPT-5.5) ")) def test_an_unregistered_tool_waits_until_someone_registers_it(self): """名单在 `co_author.VENDORS`：换用别的智能体是要动一行代码的决定，不是随手写。""" self.assertTrue(self.signed("Co-Authored-By: Gemini CLI (3.5) ")) def test_two_agents_writing_one_commit_may_both_sign(self): self.assertEqual( self.signed("Co-Authored-By: Claude Code (Opus 5) ", "Co-Authored-By: Codex (GPT-5.5) "), []) def test_human_and_agent_coauthors_share_one_commit(self): human = "Co-Authored-By: Contributor " agent = "Co-Authored-By: Codex (GPT-6) " for signatures in ((human, agent), (agent, human)): with self.subTest(signatures=signatures): self.assertEqual(self.signed(*signatures), []) def test_human_coauthor_requires_an_agent_signature(self): self.assertTrue(self.signed("Co-Authored-By: Contributor ")) def contributed(self, *trailers): """采纳的社区补丁：作者是贡献者本人，不是本机提交者。""" git(self.repo, "commit", "--allow-empty", "-m", "测试", "-m", "\n".join(trailers) or "说明", "--author", "Contributor ") return co_author.check(self.repo, "master") def test_a_contributor_commit_is_not_given_an_agent_signature_it_never_had(self): """外部贡献者是否用了 AI 由本人说明；门槛不能逼维护者替他补一行 AI 署名。""" self.assertEqual(self.contributed(), []) def test_a_contributor_may_name_a_tool_the_maintainer_has_not_registered(self): """工具名单约束的是本机智能体，贡献者用什么工具是他自己的事。""" self.assertEqual(self.contributed("Co-Authored-By: Qoder (Qwen 3) "), []) def test_a_contributor_signature_still_has_to_be_well_formed(self): self.assertTrue(self.contributed("Co-Authored-By: Codex (GPT-6) ")) self.assertTrue(co_author.problems("测试", ["Codex "], own=False)) def test_the_local_committer_signs_even_beside_an_adopted_commit(self): self.contributed() self.assertEqual(len(self.signed()), 1) def test_an_unknown_local_identity_is_treated_as_the_local_committer(self): """取不到本机邮箱时无法判断作者是谁，按本机提交从严检查。""" self.assertTrue(co_author.is_own("contributor@example.invalid", "")) self.assertFalse(co_author.is_own("contributor@example.invalid", "test@example.invalid")) self.assertTrue(co_author.is_own("Test@Example.invalid", "test@example.invalid")) def test_the_commit_hook_reads_the_author_actually_used(self): self.assertEqual(check_commit_message.author_mail(self.repo), "test@example.invalid") self.assertEqual(check_commit_message.check(self.repo, "测试\n", own=False), []) self.assertTrue(check_commit_message.check(self.repo, "测试\n", own=True)) def test_human_coauthor_requires_a_name_and_email(self): agent = "Co-Authored-By: Codex (GPT-6) " for human in ("Contributor", "Contributor ", ""): with self.subTest(human=human): self.assertTrue(co_author.problems("测试", [agent, human])) def test_registered_agent_and_vendor_addresses_require_model_metadata(self): agent = "Codex (GPT-6) " for signature in ("Codex ", "Contributor "): with self.subTest(signature=signature): self.assertTrue(co_author.problems("测试", [agent, signature])) def test_every_commit_on_the_branch_is_checked_not_only_head(self): """一次交付常常是好几个提交。只判尖端等于只判收尾那一次签得对不对， 而要追的恰恰是写出某一行的那个提交，所以问题里要报出它的短 hash 和主题。 """ self.signed("Co-Authored-By: Gemini CLI (3.5) ") problems = self.signed( "Co-Authored-By: Claude Code (Opus 5) ") self.assertEqual(len(problems), 1) self.assertIn(git(self.repo, "log", "-1", "--format=%h", "HEAD~1").strip(), problems[0]) self.assertIn("Gemini CLI", problems[0]) def test_merges_from_the_target_branch_are_not_the_worker_s_commits(self): """目标分支上的提交由它自己那次交付负责，合进来不等于这个分支重写了一遍。""" git(self.repo, "checkout", "master") git(self.repo, "commit", "--allow-empty", "-m", "上游的提交") git(self.repo, "checkout", "worker") self.signed("Co-Authored-By: Claude Code (Opus 5) ") git(self.repo, "merge", "--no-ff", "-m", "Merge branch 'master' into worker", "master") self.assertEqual(co_author.check(self.repo, "master"), []) def test_a_commit_already_on_the_remote_master_cannot_be_signed_any_more(self): """网页上直接改的那条已经在 `origin/master` 上，分支并回它时改写 hash 只能强推； 没有远端引用时什么都不豁免。""" git(self.repo, "commit", "--allow-empty", "-m", "Remove section from README") published = git(self.repo, "rev-parse", "HEAD").strip() self.signed("Co-Authored-By: Claude Code (Opus 5) ") self.assertEqual(len(co_author.check(self.repo, "master")), 1) git(self.repo, "update-ref", "refs/remotes/origin/master", published) self.assertEqual(co_author.check(self.repo, "master"), []) def test_the_signature_gate_runs_before_the_test_evidence_is_read(self): """署名在读取测试记录前检查。""" (self.repo / "tests").mkdir() (self.repo / "tests" / "example.py").write_text("x = 1\n", encoding="utf-8") git(self.repo, "add", "--", "tests/example.py") git(self.repo, "commit", "-m", "测试") with mock.patch.object(agent_worktree.test_evidence, "key") as key: with self.assertRaisesRegex(agent_worktree.WorkspaceError, "Co-Authored-By"): agent_worktree.require_verified(self.repo, "master", {"tests/example.py"}) key.assert_not_called() from scripts import commit_subject class CommitSubjectTests(unittest.TestCase): """提交主题的形状。变更日志按它分组，写歪的提交不报错，只是静默漏掉。""" def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.repo = Path(self.temp.name).resolve() git(self.repo, "init", "-b", "master") git(self.repo, "config", "user.name", "Test") git(self.repo, "config", "user.email", "test@example.invalid") git(self.repo, "commit", "--allow-empty", "-m", "base") git(self.repo, "checkout", "-b", "worker") def titled(self, *subjects): for subject in subjects: git(self.repo, "commit", "--allow-empty", "-m", subject) return commit_subject.check(self.repo, "master") def test_a_conventional_subject_passes_with_or_without_scope(self): self.assertEqual(self.titled("fix(web): 补齐图标声明", "docs: 补记四个版本", "refactor(media)!: 删掉旧快照根"), []) def test_a_bare_sentence_is_not_a_subject(self): self.assertTrue(self.titled("修复关注页按手柄查不到")) def test_the_space_after_the_colon_is_part_of_the_shape(self): self.assertTrue(self.titled("fix(web):保留必需凭据")) def test_an_unlisted_type_waits_until_someone_lists_it(self): """清单在 `commit_subject.TYPES`：加类型要先想清它在变更日志里归哪一组。""" problems = self.titled("release(app): 版本 0.30.0") self.assertEqual(len(problems), 1) self.assertIn("release", problems[0]) def test_every_commit_on_the_branch_is_checked_not_only_head(self): problems = self.titled("补齐 JAV 元数据", "fix(web): 后面这条是对的") self.assertEqual(len(problems), 1) self.assertIn("补齐 JAV 元数据", problems[0]) def test_merges_from_the_target_branch_are_not_the_worker_s_subjects(self): git(self.repo, "checkout", "master") git(self.repo, "commit", "--allow-empty", "-m", "chore(deps-dev): bump vitest") git(self.repo, "checkout", "worker") git(self.repo, "commit", "--allow-empty", "-m", "fix(web): 自己的提交") git(self.repo, "merge", "--no-ff", "-m", "Merge branch 'master' into worker", "master") self.assertEqual(commit_subject.check(self.repo, "master"), []) def test_an_adopted_community_commit_keeps_its_own_subject(self): """`merge -s ours` 记入的社区提交原样保留，主题门槛只管本机提交者写的。""" git(self.repo, "checkout", "-b", "community", "master") git(self.repo, "-c", "user.email=someone@users.noreply.github.com", "commit", "--allow-empty", "-m", "Translate frontend UI to English") git(self.repo, "checkout", "worker") git(self.repo, "merge", "-s", "ours", "--no-ff", "-m", "chore(i18n): 记入社区译法", "community") self.assertEqual(commit_subject.check(self.repo, "master"), []) self.assertTrue(self.titled("Translate it myself")) def test_a_subject_already_on_the_remote_master_is_not_the_worker_s_to_fix(self): problems = self.titled("Remove section from README", "docs(readme): 并回网页上的改动") self.assertEqual(len(problems), 1) git(self.repo, "update-ref", "refs/remotes/origin/master", git(self.repo, "rev-parse", "HEAD~1").strip()) self.assertEqual(commit_subject.check(self.repo, "master"), []) def test_the_subject_gate_runs_before_the_test_evidence_is_read(self): (self.repo / "tests").mkdir() (self.repo / "tests" / "example.py").write_text("x = 1\n", encoding="utf-8") git(self.repo, "add", "--", "tests/example.py") git(self.repo, "commit", "-m", "加个测试", "-m", "Co-Authored-By: Claude Code (Opus 5) ") with mock.patch.object(agent_worktree.test_evidence, "key") as key: with self.assertRaisesRegex(agent_worktree.WorkspaceError, "提交主题"): agent_worktree.require_verified(self.repo, "master", {"tests/example.py"}) key.assert_not_called()