1916 lines
107 KiB
Python
1916 lines
107 KiB
Python
"""Lifecycle regressions against real disposable Git repositories and remotes."""
|
||
import os
|
||
import json
|
||
import re
|
||
import hashlib
|
||
import http.server
|
||
import threading
|
||
from pathlib import Path
|
||
import shutil
|
||
import shlex
|
||
import subprocess
|
||
import tempfile
|
||
import unittest
|
||
import urllib.parse
|
||
|
||
|
||
SOURCE = Path(__file__).resolve().parents[1]
|
||
TRIAGE_COMMANDS = ("start", "triage", "triage-decide")
|
||
ROUTING_LINE = ("路由与 fail-stop 规则见根目录 AGENT-EXECUTION.md 与 policy/delivery.md；"
|
||
"每次尝试一行，原始日志留在源码外收据路径。")
|
||
LITE_SECTIONS = ["## 目标与非目标", "## Agent 实施收据", "## 同步记录", "## 证据（Evidence）",
|
||
"## 结果与验证（完成后填写）", "## 合入与清理"]
|
||
|
||
|
||
class FakeGitea:
|
||
"""Gitea pulls API 桩（进程内 127.0.0.1）：校验 token 请求头，按页返回 self.prs；从不联网。"""
|
||
|
||
PAGE = 50
|
||
|
||
def __init__(self, token="dummy"):
|
||
self.token = token
|
||
self.prs = []
|
||
self.posted = []
|
||
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)
|
||
if not parsed.path.startswith("/api/v1/repos/") or not parsed.path.endswith("/pulls"):
|
||
return self._send(404, {"message": "not found"})
|
||
page = int(urllib.parse.parse_qs(parsed.query).get("page", ["1"])[0])
|
||
start = (page - 1) * stub.PAGE
|
||
return self._send(200, stub.prs[start:start + stub.PAGE])
|
||
|
||
def do_POST(self):
|
||
if self.headers.get("Authorization") != f"token {stub.token}":
|
||
return self._send(401, {"message": "unauthorized"})
|
||
body = json.loads(self.rfile.read(int(self.headers.get("Content-Length", "0"))) or b"{}")
|
||
stub.posted.append(body)
|
||
return self._send(201, {"html_url": f"{stub.web}/fixture/repo/pulls/{len(stub.posted)}"})
|
||
|
||
self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
||
self.web = f"http://127.0.0.1:{self.server.server_address[1]}"
|
||
self.url = f"{self.web}/api/v1"
|
||
threading.Thread(target=self.server.serve_forever, daemon=True).start()
|
||
|
||
def close(self):
|
||
self.server.shutdown()
|
||
self.server.server_close()
|
||
|
||
@staticmethod
|
||
def pr(number, branch, sha, base="main", state="open"):
|
||
return {"number": number, "state": state, "base": {"ref": base},
|
||
"head": {"ref": branch, "sha": sha}}
|
||
|
||
|
||
class TaskLifecycleTests(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory(prefix="task-lifecycle-")
|
||
self.addCleanup(self.tmp.cleanup)
|
||
self.root = Path(self.tmp.name)
|
||
self.repo = self.root / "repo"
|
||
self.remote = self.root / "remote.git"
|
||
self.gitea = FakeGitea()
|
||
self.addCleanup(self.gitea.close)
|
||
self.env = os.environ.copy()
|
||
self.env.update({"GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": "/dev/null",
|
||
"GIT_ALLOW_PROTOCOL": "file", "GIT_TERMINAL_PROMPT": "0",
|
||
"GIT_AUTHOR_NAME": "Lifecycle Test", "GIT_AUTHOR_EMAIL": "test@example.invalid",
|
||
"GIT_COMMITTER_NAME": "Lifecycle Test", "GIT_COMMITTER_EMAIL": "test@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",
|
||
# task-start 自动 precheck 收据落在 TMPDIR：限定在夹具目录内随测试清理。
|
||
"TMPDIR": str(self.root),
|
||
"AGENT_EVIDENCE_DIR": str(self.root / "evidence"),
|
||
# 桩位于 127.0.0.1：禁止经代理转发，保证测试不触达外部网络。
|
||
"NO_PROXY": "127.0.0.1,localhost", "no_proxy": "127.0.0.1,localhost"})
|
||
for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GITEA_TOKEN", "GITEA_API_URL",
|
||
"GITEA_REPO", "TRIAGE_RECEIPT", "PRECHECK_RECEIPT",
|
||
"TASK_PARENT_BRANCH", "TASK_PARENT_PR", "FORCE_ABANDON", "TASK_TARGET_BRANCH",
|
||
"TASK_WORK_MODE", "VALIDATE_RECEIPT_PURPOSE", "TASK_BRANCH_KIND", "TASK_INITIATIVE"):
|
||
self.env.pop(key, None)
|
||
self.git(self.root, "init", "--bare", "--initial-branch=main", str(self.remote))
|
||
# receive-pack 默认在后台跑 gc --auto，与 TemporaryDirectory 清理竞态
|
||
# （CI 实测 OSError 39 remote.git not empty）；夹具远端无留存价值，直接关。
|
||
self.git(self.remote, "config", "gc.auto", "0")
|
||
self.git(self.remote, "config", "receive.autogc", "false")
|
||
self.git(self.root, "init", "--initial-branch=main", str(self.repo))
|
||
shutil.copytree(SOURCE / "scripts", self.repo / "scripts")
|
||
(self.repo / "Makefile").write_text("precheck:\n\t@bash scripts/task.sh precheck\n")
|
||
(self.repo / "tracked").write_text("saved\n")
|
||
(self.repo / ".gitignore").write_text("cache/\n")
|
||
self.git(self.repo, "add", ".")
|
||
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")
|
||
|
||
def run_cmd(self, cwd, *args, check=True, env=None):
|
||
result = subprocess.run(args, cwd=cwd, env=env or self.env, text=True,
|
||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=45)
|
||
if check:
|
||
self.assertEqual(result.returncode, 0, result.stdout)
|
||
return result
|
||
|
||
def git(self, cwd, *args, **kwargs):
|
||
return self.run_cmd(cwd, "git", *args, **kwargs)
|
||
|
||
def task(self, cwd, command, *args, **kwargs):
|
||
if command == "finish" and "env" not in kwargs:
|
||
kwargs["env"] = self.env | {"TASK_WRITERS_STOPPED": "1"}
|
||
if command in TRIAGE_COMMANDS:
|
||
# 分流只读远端：注入桩地址与测试专用 token，绝不读取真实 ~/.secrets。
|
||
kwargs["env"] = self.gitea_env() | kwargs.get("env", self.env)
|
||
return self.run_cmd(cwd, "bash", str(self.repo / "scripts/task.sh"), command, *args, **kwargs)
|
||
|
||
def gitea_env(self):
|
||
return {"GITEA_API_URL": self.gitea.url, "GITEA_REPO": "fixture/repo", "GITEA_TOKEN": self.gitea.token}
|
||
|
||
def worktree(self, name="feature/test"):
|
||
wt = self.root / "worktree"
|
||
self.git(self.repo, "worktree", "add", "-b", name, str(wt), "origin/main")
|
||
return wt
|
||
|
||
def advance_remote(self):
|
||
(self.repo / "new-main").write_text("new main\n")
|
||
self.git(self.repo, "add", "new-main")
|
||
self.git(self.repo, "commit", "-m", "Advance main")
|
||
self.git(self.repo, "push", "origin", "main")
|
||
|
||
def record_task_card(self, wt):
|
||
branch = self.git(wt, "branch", "--show-current").stdout.strip()
|
||
r0 = self.git(wt, "rev-parse", "origin/main").stdout.strip()
|
||
card = wt / ".task" / (branch.replace("/", "-") + ".md")
|
||
card.parent.mkdir(exist_ok=True)
|
||
card.write_text(
|
||
f"# {branch}\n任务主记录：fixture；状态：进行中；owner：Lifecycle Test\n"
|
||
f"- 权威仓：fixture；目标分支：main；R0（initial_base_sha）= {r0}\n"
|
||
f"- 工作区：{wt}（worktree）；分支：{branch}\n"
|
||
)
|
||
self.git(wt, "add", str(card.relative_to(wt)))
|
||
self.git(wt, "commit", "-m", "Record fixture task")
|
||
return card
|
||
|
||
def add_submodule(self):
|
||
child = self.root / "component"
|
||
self.git(self.root, "init", "--initial-branch=main", str(child))
|
||
(child / "content").write_text("v1\n")
|
||
self.git(child, "add", ".")
|
||
self.git(child, "commit", "-m", "Component v1")
|
||
self.git(self.repo, "submodule", "add", str(child), "component")
|
||
self.git(self.repo, "commit", "-m", "Pin component")
|
||
self.git(self.repo, "push", "origin", "main")
|
||
return child
|
||
|
||
def test_precheck_rejects_unreachable_authority(self):
|
||
self.git(self.repo, "remote", "set-url", "origin", str(self.root / "missing.git"))
|
||
result = self.task(self.repo, "precheck", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("REMOTE_UNVERIFIED", result.stdout)
|
||
|
||
def test_precheck_feature_behind_requires_sync(self):
|
||
wt = self.worktree()
|
||
self.advance_remote()
|
||
result = self.task(wt, "precheck", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("SYNC_REQUIRED", result.stdout)
|
||
|
||
def test_precheck_routes_existing_project_delivery_policy(self):
|
||
result = self.task(self.repo, "precheck")
|
||
self.assertNotIn("policy/delivery.md", result.stdout)
|
||
policy = self.repo / "policy/delivery.md"
|
||
policy.parent.mkdir()
|
||
policy.write_text("# Project delivery policy\n")
|
||
self.git(self.repo, "add", "policy")
|
||
self.git(self.repo, "commit", "-m", "Adopt project delivery policy")
|
||
result = self.task(self.repo, "precheck")
|
||
reading = next(line for line in result.stdout.splitlines() if line.startswith("[required-reading]"))
|
||
self.assertIn("policy/delivery.md", reading)
|
||
|
||
def test_precheck_facts_row_with_escaped_pipe_stays_fresh(self):
|
||
facts = self.repo / "policy/facts.md"
|
||
facts.parent.mkdir(exist_ok=True)
|
||
facts.write_text(
|
||
"| ID | 事实 | 验证命令 | last_verified | TTL(天) |\n"
|
||
"|---|---|---|---|---|\n"
|
||
"| F1 | 含转义竖线的验证命令 | `echo a\\|b` | 2026-10-13 | 9999 |\n"
|
||
)
|
||
self.git(self.repo, "add", "policy")
|
||
self.git(self.repo, "commit", "-m", "Add facts registry with escaped pipe")
|
||
result = self.task(self.repo, "precheck")
|
||
line = next(l for l in result.stdout.splitlines() if l.startswith("[facts]"))
|
||
self.assertIn("均在有效期内", line)
|
||
self.assertNotIn("日期格式", line)
|
||
|
||
def test_sync_same_base_preserves_head_and_card(self):
|
||
wt = self.worktree()
|
||
card = self.record_task_card(wt)
|
||
head = self.git(wt, "rev-parse", "HEAD").stdout.strip()
|
||
content = card.read_bytes()
|
||
index = self.git(wt, "ls-files", "--stage").stdout
|
||
for _ in range(2):
|
||
result = self.task(wt, "sync")
|
||
self.assertEqual(self.git(wt, "rev-parse", "HEAD").stdout.strip(), head)
|
||
self.assertEqual(card.read_bytes(), content)
|
||
self.assertEqual(self.git(wt, "ls-files", "--stage").stdout, index)
|
||
self.assertEqual(self.git(wt, "status", "--porcelain").stdout, "")
|
||
self.assertIn("changed=0", result.stdout)
|
||
self.assertIn(f"tested_head={head}", result.stdout)
|
||
|
||
def test_sync_new_base_reports_actual_final_head(self):
|
||
wt = self.worktree()
|
||
card = self.record_task_card(wt)
|
||
self.advance_remote()
|
||
base = self.git(self.repo, "rev-parse", "HEAD").stdout.strip()
|
||
result = self.task(wt, "sync")
|
||
head = self.git(wt, "rev-parse", "HEAD").stdout.strip()
|
||
self.git(wt, "merge-base", "--is-ancestor", base, head)
|
||
self.assertIn(f"sync_base={base}", result.stdout)
|
||
self.assertIn(f"tested_head={head}", result.stdout)
|
||
self.assertIn("changed=1", result.stdout)
|
||
content = card.read_bytes()
|
||
self.task(wt, "sync")
|
||
self.assertEqual(self.git(wt, "rev-parse", "HEAD").stdout.strip(), head)
|
||
self.assertEqual(card.read_bytes(), content)
|
||
|
||
def test_sync_preserves_dirty_parent_and_staged_changes(self):
|
||
for staged in (False, True):
|
||
with self.subTest(staged=staged):
|
||
wt = self.worktree(name="feature/staged" if staged else "feature/dirty")
|
||
head = self.git(wt, "rev-parse", "HEAD").stdout.strip()
|
||
(wt / "tracked").write_text("keep owner work\n")
|
||
if staged:
|
||
self.git(wt, "add", "tracked")
|
||
index = self.git(wt, "ls-files", "--stage").stdout
|
||
result = self.task(wt, "sync", check=False)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertEqual(self.git(wt, "rev-parse", "HEAD").stdout.strip(), head)
|
||
self.assertEqual(self.git(wt, "ls-files", "--stage").stdout, index)
|
||
self.assertEqual((wt / "tracked").read_text(), "keep owner work\n")
|
||
self.git(self.repo, "worktree", "remove", "--force", str(wt))
|
||
|
||
def test_sync_rejects_unreachable_target_without_changing_head(self):
|
||
wt = self.worktree()
|
||
self.record_task_card(wt)
|
||
head = self.git(wt, "rev-parse", "HEAD").stdout.strip()
|
||
self.git(wt, "remote", "set-url", "origin", str(self.root / "missing.git"))
|
||
result = self.task(wt, "sync", check=False)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("REMOTE_UNVERIFIED", result.stdout)
|
||
self.assertEqual(self.git(wt, "rev-parse", "HEAD").stdout.strip(), head)
|
||
|
||
def test_sync_conflict_is_not_reported_as_synced(self):
|
||
wt = self.worktree()
|
||
self.record_task_card(wt)
|
||
(wt / "tracked").write_text("task choice\n")
|
||
self.git(wt, "commit", "-am", "Task edit")
|
||
(self.repo / "tracked").write_text("target choice\n")
|
||
self.git(self.repo, "commit", "-am", "Target edit")
|
||
self.git(self.repo, "push", "origin", "main")
|
||
result = self.task(wt, "sync", check=False)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertNotIn("[synced]", result.stdout)
|
||
self.assertIn("UU tracked", self.git(wt, "status", "--porcelain").stdout)
|
||
|
||
def test_sync_materializes_updated_component(self):
|
||
child = self.add_submodule()
|
||
wt = self.worktree()
|
||
self.git(wt, "submodule", "update", "--init", "--recursive")
|
||
(child / "content").write_text("v2\n")
|
||
self.git(child, "commit", "-am", "Component v2")
|
||
latest = self.git(child, "rev-parse", "HEAD").stdout.strip()
|
||
self.git(self.repo / "component", "fetch", "origin")
|
||
self.git(self.repo / "component", "checkout", "--detach", latest)
|
||
self.git(self.repo, "commit", "-am", "Update component")
|
||
self.git(self.repo, "push", "origin", "main")
|
||
result = self.task(wt, "sync")
|
||
self.assertIn("[synced]", result.stdout)
|
||
self.assertEqual((wt / "component/content").read_text(), "v2\n")
|
||
self.assertEqual(self.git(wt / "component", "rev-parse", "HEAD").stdout.strip(), latest)
|
||
|
||
def test_sync_preserves_component_feature_branch(self):
|
||
child = self.add_submodule()
|
||
wt = self.worktree()
|
||
self.git(wt, "submodule", "update", "--init", "--recursive")
|
||
self.git(wt / "component", "switch", "-c", "feature/owned")
|
||
before = self.git(wt / "component", "rev-parse", "HEAD").stdout
|
||
(child / "content").write_text("v2\n")
|
||
self.git(child, "commit", "-am", "Component v2")
|
||
latest = self.git(child, "rev-parse", "HEAD").stdout.strip()
|
||
self.git(self.repo / "component", "fetch", "origin")
|
||
self.git(self.repo / "component", "checkout", "--detach", latest)
|
||
self.git(self.repo, "commit", "-am", "Update component")
|
||
self.git(self.repo, "push", "origin", "main")
|
||
result = self.task(wt, "sync", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertNotIn("[synced]", result.stdout)
|
||
self.assertEqual(self.git(wt / "component", "rev-parse", "HEAD").stdout, before)
|
||
self.assertEqual(self.git(wt / "component", "branch", "--show-current").stdout.strip(), "feature/owned")
|
||
|
||
def test_sync_keeps_unchanged_component_branch(self):
|
||
self.add_submodule()
|
||
wt = self.worktree()
|
||
self.git(wt, "submodule", "update", "--init", "--recursive")
|
||
self.git(wt / "component", "switch", "-c", "feature/owned")
|
||
self.task(wt, "sync")
|
||
self.assertEqual(self.git(wt / "component", "branch", "--show-current").stdout.strip(), "feature/owned")
|
||
|
||
def test_sync_preserves_dirty_child(self):
|
||
self.add_submodule()
|
||
wt = self.worktree()
|
||
self.git(wt, "submodule", "update", "--init", "--recursive")
|
||
(wt / "component/content").write_text("unsaved\n")
|
||
self.advance_remote()
|
||
before = self.git(wt, "rev-parse", "HEAD").stdout
|
||
result = self.task(wt, "sync", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertEqual(self.git(wt, "rev-parse", "HEAD").stdout, before)
|
||
self.assertEqual((wt / "component/content").read_text(), "unsaved\n")
|
||
|
||
def test_finish_preserves_modified_untracked_and_ignored_files(self):
|
||
for kind, rel in (("tracked", "tracked"), ("untracked", "unique"), ("ignored", "cache/unique")):
|
||
with self.subTest(kind=kind):
|
||
wt = self.root / f"worktree-{kind}"
|
||
self.git(self.repo, "worktree", "add", "-b", f"feature/{kind}", str(wt), "origin/main")
|
||
path = wt / rel
|
||
path.parent.mkdir(exist_ok=True)
|
||
path.write_text("unique user content\n")
|
||
result = self.task(wt, "finish", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertTrue(path.exists(), result.stdout)
|
||
self.assertEqual(path.read_text(), "unique user content\n")
|
||
|
||
def test_finish_rejects_subdirectory_without_deleting_it(self):
|
||
wt = self.worktree()
|
||
result = self.task(wt / "scripts", "finish", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertTrue((wt / "scripts/task.sh").is_file(), result.stdout)
|
||
|
||
def test_finish_preserves_locked_worktree(self):
|
||
wt = self.worktree()
|
||
self.git(self.repo, "worktree", "lock", "--reason", "active owner", str(wt))
|
||
result = self.task(wt, "finish", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertTrue(wt.exists(), result.stdout)
|
||
|
||
def test_finish_preserves_component_untracked_file(self):
|
||
self.add_submodule()
|
||
wt = self.worktree()
|
||
self.git(wt, "submodule", "update", "--init", "--recursive")
|
||
unique = wt / "component/unique"
|
||
unique.write_text("cannot regenerate\n")
|
||
result = self.task(wt, "finish", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertTrue(unique.is_file(), result.stdout)
|
||
|
||
def test_finish_preserves_component_commits_off_the_pinned_head(self):
|
||
self.add_submodule()
|
||
wt = self.worktree()
|
||
self.git(wt, "submodule", "update", "--init", "--recursive")
|
||
child = wt / "component"
|
||
pinned = self.git(child, "rev-parse", "HEAD").stdout.strip()
|
||
self.git(child, "switch", "-c", "feature/unique-history")
|
||
(child / "content").write_text("unique committed work\n")
|
||
self.git(child, "commit", "-am", "Preserve this commit")
|
||
unique = self.git(child, "rev-parse", "HEAD").stdout.strip()
|
||
self.git(child, "checkout", "--detach", pinned)
|
||
result = self.task(wt, "finish", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertTrue(wt.exists(), result.stdout)
|
||
self.assertEqual(self.git(child, "rev-parse", "feature/unique-history").stdout.strip(), unique)
|
||
|
||
def test_finish_preserves_linked_component_worktrees_at_the_pinned_head(self):
|
||
self.add_submodule()
|
||
for state in ("clean", "staged", "unstaged"):
|
||
with self.subTest(state=state):
|
||
wt = self.root / f"worktree-{state}"
|
||
self.git(self.repo, "worktree", "add", "-b", f"feature/{state}", str(wt), "origin/main")
|
||
self.git(wt, "submodule", "update", "--init", "--recursive")
|
||
child = wt / "component"
|
||
sibling = self.root / f"component-{state}"
|
||
self.git(child, "worktree", "add", "--detach", str(sibling), "HEAD")
|
||
head = self.git(sibling, "rev-parse", "HEAD").stdout
|
||
if state != "clean":
|
||
(sibling / "content").write_text("unique sibling work\n")
|
||
if state == "staged":
|
||
self.git(sibling, "add", "content")
|
||
status = self.git(sibling, "status", "--porcelain").stdout
|
||
result = self.task(wt, "finish", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertTrue(wt.exists(), result.stdout)
|
||
self.assertEqual(self.git(child, "rev-parse", "HEAD").stdout, head)
|
||
self.assertEqual(self.git(child, "status", "--porcelain").stdout, "")
|
||
self.assertEqual(self.git(sibling, "rev-parse", "HEAD").stdout, head)
|
||
self.assertEqual(self.git(sibling, "status", "--porcelain").stdout, status)
|
||
self.assertEqual((sibling / "content").read_text(),
|
||
"v1\n" if state == "clean" else "unique sibling work\n")
|
||
|
||
def nested_component_update(self, attached):
|
||
child = self.add_submodule()
|
||
inner = self.root / "inner"
|
||
self.git(self.root, "init", "--initial-branch=main", str(inner))
|
||
(inner / "inner-content").write_text("inner v1\n")
|
||
self.git(inner, "add", ".")
|
||
self.git(inner, "commit", "-m", "Inner v1")
|
||
self.git(child, "submodule", "add", str(inner), "inner")
|
||
self.git(child, "commit", "-m", "Pin inner")
|
||
self.git(self.repo / "component", "fetch", "origin")
|
||
self.git(self.repo / "component", "checkout", "--detach", "origin/main")
|
||
self.git(self.repo, "commit", "-am", "Pin nested component")
|
||
self.git(self.repo, "push", "origin", "main")
|
||
wt = self.worktree()
|
||
self.git(wt, "submodule", "update", "--init", "--recursive")
|
||
nested = wt / "component/inner"
|
||
if attached:
|
||
self.git(nested, "switch", "-c", "feature/owned-inner")
|
||
before = self.git(nested, "rev-parse", "HEAD").stdout.strip()
|
||
(inner / "inner-content").write_text("inner v2\n")
|
||
self.git(inner, "commit", "-am", "Inner v2")
|
||
latest = self.git(inner, "rev-parse", "HEAD").stdout.strip()
|
||
self.git(child / "inner", "fetch", "origin")
|
||
self.git(child / "inner", "checkout", "--detach", latest)
|
||
self.git(child, "commit", "-am", "Update inner pin")
|
||
self.git(self.repo / "component", "fetch", "origin")
|
||
self.git(self.repo / "component", "checkout", "--detach", "origin/main")
|
||
self.git(self.repo, "commit", "-am", "Update nested component")
|
||
self.git(self.repo, "push", "origin", "main")
|
||
return wt, nested, before, latest
|
||
|
||
def test_sync_preserves_nested_feature_branch_when_parent_pin_changes(self):
|
||
wt, nested, before, _ = self.nested_component_update(attached=True)
|
||
result = self.task(wt, "sync", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertNotIn("[synced]", result.stdout)
|
||
self.assertEqual(self.git(nested, "rev-parse", "HEAD").stdout.strip(), before)
|
||
self.assertEqual(self.git(nested, "branch", "--show-current").stdout.strip(), "feature/owned-inner")
|
||
|
||
def test_sync_materializes_nested_detached_components(self):
|
||
wt, nested, _, latest = self.nested_component_update(attached=False)
|
||
self.task(wt, "sync")
|
||
self.assertEqual(self.git(nested, "rev-parse", "HEAD").stdout.strip(), latest)
|
||
self.assertEqual((nested / "inner-content").read_text(), "inner v2\n")
|
||
|
||
def test_finish_preserves_linked_nested_component_worktree(self):
|
||
wt, nested, before, _ = self.nested_component_update(attached=False)
|
||
sibling = self.root / "nested-sibling"
|
||
self.git(nested, "worktree", "add", "--detach", str(sibling), "HEAD")
|
||
result = self.task(wt, "finish", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertTrue(wt.exists(), result.stdout)
|
||
self.assertEqual(self.git(nested, "rev-parse", "HEAD").stdout.strip(), before)
|
||
self.assertEqual(self.git(sibling, "rev-parse", "HEAD").stdout.strip(), before)
|
||
self.assertEqual(self.git(sibling, "status", "--porcelain").stdout, "")
|
||
self.assertEqual((nested / "inner-content").read_text(), "inner v1\n")
|
||
self.assertEqual((sibling / "inner-content").read_text(), "inner v1\n")
|
||
|
||
def test_finish_clean_worktree_has_accurate_remote_result(self):
|
||
wt = self.worktree()
|
||
# A missing remote branch is not a successful remote deletion.
|
||
result = self.task(wt, "finish")
|
||
self.assertFalse(wt.exists(), result.stdout)
|
||
self.assertNotIn("本地+远端", result.stdout)
|
||
|
||
def test_finish_recycles_clean_submodule_without_shared_config_changes(self):
|
||
self.add_submodule()
|
||
wt = self.worktree()
|
||
self.git(wt, "submodule", "update", "--init", "--recursive")
|
||
result = self.task(wt, "finish")
|
||
self.assertFalse(wt.exists(), result.stdout)
|
||
self.assertTrue((self.repo / "component/content").is_file(), result.stdout)
|
||
self.assertEqual(self.git(self.repo, "config", "--get", "submodule.component.url").stdout.strip(), str(self.root / "component"))
|
||
|
||
def test_finish_recycles_submodule_when_index_listing_exceeds_pipe_buffer(self):
|
||
self.add_submodule()
|
||
# Entries sorting after the gitlink must outlast a 64 KiB pipe once an early-exit reader stops.
|
||
bulk = self.repo / "zz-bulk"
|
||
bulk.mkdir()
|
||
for i in range(2000):
|
||
(bulk / f"entry-{i:05d}-{'p' * 40}").write_text("")
|
||
self.git(self.repo, "add", "zz-bulk")
|
||
self.git(self.repo, "commit", "-m", "Add bulk entries")
|
||
self.git(self.repo, "push", "origin", "main")
|
||
wt = self.worktree()
|
||
self.git(wt, "submodule", "update", "--init", "--recursive")
|
||
listing = self.git(wt, "ls-files", "--stage").stdout
|
||
before_gitlink, after_gitlink = listing.split("\tcomponent\n", 1)
|
||
self.assertTrue(before_gitlink.rsplit("\n", 1)[-1].startswith("160000 "), listing[:200])
|
||
self.assertGreater(len(listing.encode()), 64 * 1024)
|
||
self.assertGreater(len(after_gitlink.encode()), 2 * 64 * 1024)
|
||
result = self.task(wt, "finish")
|
||
self.assertFalse(wt.exists(), result.stdout)
|
||
|
||
def test_finish_requires_writer_exit_confirmation(self):
|
||
wt = self.worktree()
|
||
result = self.task(wt, "finish", check=False, env=self.env)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertTrue(wt.exists(), result.stdout)
|
||
|
||
def unmerged_task(self):
|
||
wt = self.worktree()
|
||
(wt / "tracked").write_text("unmerged task work\n")
|
||
self.git(wt, "commit", "-am", "Keep unmerged work")
|
||
self.git(wt, "push", "-u", "origin", "HEAD")
|
||
return wt, self.git(wt, "rev-parse", "HEAD").stdout.strip()
|
||
|
||
def test_finish_abandon_retains_local_remote_tips_and_reason(self):
|
||
wt, head = self.unmerged_task()
|
||
reason = "superseded by a smaller task"
|
||
result = self.task(wt, "finish", "--abandon", reason)
|
||
self.assertFalse(wt.exists(), result.stdout)
|
||
self.assertEqual(self.git(self.repo, "rev-parse", "refs/heads/feature/test").stdout.strip(), head)
|
||
self.assertEqual(self.git(self.remote, "rev-parse", "refs/heads/feature/test").stdout.strip(), head)
|
||
note = self.git(self.repo, "notes", "--ref=refs/notes/task-abandon", "show", head).stdout
|
||
self.assertIn(reason, note)
|
||
self.assertIn("feature/test", note)
|
||
self.assertEqual(self.git(self.repo, "show", f"{head}:tracked").stdout, "unmerged task work\n")
|
||
|
||
def test_finish_abandon_without_reason_preserves_checkout_and_refs(self):
|
||
wt, head = self.unmerged_task()
|
||
result = self.task(wt, "finish", "--abandon", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertTrue(wt.exists(), result.stdout)
|
||
self.assertEqual(self.git(wt, "rev-parse", "HEAD").stdout.strip(), head)
|
||
self.assertEqual(self.git(self.repo, "rev-parse", "refs/heads/feature/test").stdout.strip(), head)
|
||
self.assertEqual(self.git(self.remote, "rev-parse", "refs/heads/feature/test").stdout.strip(), head)
|
||
self.assertEqual(self.git(wt, "status", "--porcelain").stdout, "")
|
||
self.assertNotEqual(self.git(self.repo, "notes", "--ref=refs/notes/task-abandon", "show", head,
|
||
check=False).returncode, 0)
|
||
|
||
def test_finish_deletes_only_merged_remote_branch(self):
|
||
wt = self.worktree()
|
||
self.git(wt, "push", "-u", "origin", "HEAD")
|
||
result = self.task(wt, "finish")
|
||
self.assertIn("remote=DELETED", result.stdout)
|
||
self.assertEqual(self.git(self.repo, "ls-remote", "--heads", "origin", "refs/heads/feature/test").stdout, "")
|
||
|
||
def test_finish_preserves_remote_unmerged_tip(self):
|
||
wt = self.worktree()
|
||
self.git(self.repo, "switch", "-c", "feature/remote-update")
|
||
(self.repo / "remote-work").write_text("another writer")
|
||
self.git(self.repo, "add", "remote-work")
|
||
self.git(self.repo, "commit", "-m", "Unmerged work")
|
||
self.git(self.repo, "push", "origin", "HEAD:refs/heads/feature/test")
|
||
self.git(self.repo, "switch", "main")
|
||
result = self.task(wt, "finish", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertTrue(wt.exists(), result.stdout)
|
||
self.assertTrue(self.git(self.repo, "ls-remote", "--heads", "origin", "refs/heads/feature/test").stdout)
|
||
|
||
def test_finish_remote_deletion_lease_preserves_concurrent_update(self):
|
||
wt = self.worktree()
|
||
old = self.git(wt, "rev-parse", "HEAD").stdout.strip()
|
||
self.git(wt, "push", "-u", "origin", "HEAD")
|
||
self.git(self.repo, "switch", "-c", "feature/racing")
|
||
(self.repo / "race-work").write_text("concurrent remote work")
|
||
self.git(self.repo, "add", "race-work")
|
||
self.git(self.repo, "commit", "-m", "Concurrent remote work")
|
||
latest = self.git(self.repo, "rev-parse", "HEAD").stdout.strip()
|
||
self.git(self.repo, "push", "origin", "feature/racing")
|
||
self.git(self.repo, "switch", "main")
|
||
hook = self.repo / ".git/hooks/pre-push"
|
||
hook.write_text("#!/bin/sh\n" + shlex.join(["git", "--git-dir", str(self.remote), "update-ref",
|
||
"refs/heads/feature/test", latest, old]) + "\n")
|
||
hook.chmod(0o755)
|
||
result = self.task(wt, "finish", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("[partial]", result.stdout)
|
||
current = self.git(self.repo, "ls-remote", "--heads", "origin", "refs/heads/feature/test").stdout
|
||
self.assertTrue(current.startswith(latest), current)
|
||
|
||
def test_sync_commit_failure_never_reports_synced(self):
|
||
wt = self.worktree()
|
||
(wt / ".task").mkdir()
|
||
(wt / ".task/feature-test.md").write_text("# Task\n")
|
||
self.git(wt, "add", ".task")
|
||
self.git(wt, "commit", "-m", "Create task card")
|
||
self.advance_remote()
|
||
hook = self.repo / ".git/hooks/pre-commit"
|
||
hook.write_text("#!/bin/sh\nexit 1\n")
|
||
hook.chmod(0o755)
|
||
result = self.task(wt, "sync", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertNotIn("[synced]", result.stdout)
|
||
self.assertIn(".task/feature-test.md", self.git(wt, "status", "--porcelain").stdout)
|
||
|
||
def task_card(self, wt):
|
||
card = wt / ".task/feature-test.md"
|
||
card.parent.mkdir(exist_ok=True)
|
||
card.write_text("# Task\n")
|
||
self.git(wt, "add", ".task")
|
||
self.git(wt, "commit", "-m", "Create task card")
|
||
return card
|
||
|
||
def capture_evidence(self, wt, command="printf 'checked\\n'"):
|
||
card = self.task_card(wt)
|
||
self.run_cmd(wt, "bash", "scripts/evidence.sh", env=self.env | {"CMD": command})
|
||
return card
|
||
|
||
def commit_evidence(self, wt):
|
||
self.git(wt, "add", ".task")
|
||
self.git(wt, "commit", "-m", "Record evidence")
|
||
|
||
def test_sync_warns_when_json_evidence_predates_noncard_changes(self):
|
||
wt = self.worktree()
|
||
self.capture_evidence(wt)
|
||
self.commit_evidence(wt)
|
||
self.advance_remote()
|
||
result = self.task(wt, "sync")
|
||
self.assertIn("[evidence] WARN", result.stdout)
|
||
self.assertIn("重跑", result.stdout)
|
||
|
||
def test_sync_keeps_json_evidence_fresh_after_metadata_only_changes(self):
|
||
wt = self.worktree()
|
||
self.capture_evidence(wt)
|
||
# A second run sees only the first run's uncommitted card receipt.
|
||
self.run_cmd(wt, "bash", "scripts/evidence.sh", env=self.env | {"CMD": "true"})
|
||
self.commit_evidence(wt)
|
||
(self.repo / ".task").mkdir()
|
||
(self.repo / ".task/other.md").write_text("Other task metadata\n")
|
||
self.git(self.repo, "add", ".task")
|
||
self.git(self.repo, "commit", "-m", "Record other task")
|
||
self.git(self.repo, "push", "origin", "main")
|
||
result = self.task(wt, "sync")
|
||
self.assertNotIn("[evidence] WARN", result.stdout)
|
||
|
||
def test_sync_warns_when_json_before_state_contains_uncommitted_code(self):
|
||
wt = self.worktree()
|
||
(wt / "tracked").write_text("uncommitted code\n")
|
||
self.capture_evidence(wt, command="git restore tracked")
|
||
self.commit_evidence(wt)
|
||
result = self.task(wt, "sync")
|
||
self.assertIn("[evidence] WARN", result.stdout)
|
||
self.assertIn("unverified", result.stdout)
|
||
|
||
def test_sync_warns_when_json_after_state_contains_uncommitted_code(self):
|
||
wt = self.worktree()
|
||
self.capture_evidence(wt, command="printf 'uncommitted code\\n' > tracked")
|
||
self.git(wt, "restore", "tracked")
|
||
self.commit_evidence(wt)
|
||
result = self.task(wt, "sync")
|
||
self.assertIn("[evidence] WARN", result.stdout)
|
||
self.assertIn("unverified", result.stdout)
|
||
|
||
def test_sync_does_not_execute_json_receipt_command(self):
|
||
wt = self.worktree()
|
||
sentinel = self.root / "command-was-executed"
|
||
self.capture_evidence(wt, command="printf ran > " + shlex.quote(str(sentinel)))
|
||
self.assertTrue(sentinel.is_file())
|
||
sentinel.unlink()
|
||
self.commit_evidence(wt)
|
||
self.task(wt, "sync")
|
||
self.assertFalse(sentinel.exists())
|
||
|
||
def rewrite_receipt(self, card, change):
|
||
prefix, block = card.read_text().split("```json\n", 1)
|
||
record, suffix = block.split("\n```", 1)
|
||
# The card holds a summary; rewrite its full receipt back into the card as a legacy block.
|
||
receipt = json.loads(Path(json.loads(record)["receipt_path"]).read_text())
|
||
change(receipt)
|
||
receipt.pop("receipt_sha256")
|
||
serialized = json.dumps(receipt, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
|
||
receipt["receipt_sha256"] = hashlib.sha256(serialized.encode()).hexdigest()
|
||
card.write_text(prefix + "```json\n" + json.dumps(receipt) + "\n```" + suffix)
|
||
|
||
def test_sync_warns_for_unresolved_json_evidence_head(self):
|
||
wt = self.worktree()
|
||
card = self.capture_evidence(wt)
|
||
self.rewrite_receipt(card, lambda receipt: receipt["before"].update(head="0" * 40))
|
||
self.commit_evidence(wt)
|
||
result = self.task(wt, "sync")
|
||
self.assertIn("[evidence] WARN", result.stdout)
|
||
self.assertIn("unverified", result.stdout)
|
||
|
||
def test_sync_warns_for_missing_json_receipt_fields_even_with_valid_checksum(self):
|
||
wt = self.worktree()
|
||
card = self.capture_evidence(wt)
|
||
self.rewrite_receipt(card, lambda receipt: receipt.pop("command"))
|
||
self.commit_evidence(wt)
|
||
result = self.task(wt, "sync")
|
||
self.assertIn("[evidence] WARN", result.stdout)
|
||
self.assertIn("unverified", result.stdout)
|
||
|
||
def test_sync_warns_for_malformed_json_evidence(self):
|
||
wt = self.worktree()
|
||
card = self.task_card(wt)
|
||
card.write_text(card.read_text() + '\n```json\n{"schema": 1, invalid}\n```\n')
|
||
self.commit_evidence(wt)
|
||
result = self.task(wt, "sync")
|
||
self.assertIn("[evidence] WARN", result.stdout)
|
||
self.assertIn("unverified", result.stdout)
|
||
|
||
def test_sync_warns_for_unresolved_legacy_evidence_head(self):
|
||
wt = self.worktree()
|
||
card = self.task_card(wt)
|
||
card.write_text(card.read_text() + "\n| 2026-10-05T00:00:00Z | check | `0000000000` | 0 | SELF | abc |\n")
|
||
self.commit_evidence(wt)
|
||
result = self.task(wt, "sync")
|
||
self.assertIn("[evidence] WARN", result.stdout)
|
||
self.assertIn("unverified", result.stdout)
|
||
|
||
def test_sync_still_warns_for_stale_legacy_table_evidence(self):
|
||
wt = self.worktree()
|
||
card = self.task_card(wt)
|
||
head = self.git(wt, "rev-parse", "--short=10", "HEAD").stdout.strip()
|
||
card.write_text(card.read_text() + f"\n| 2026-10-05T00:00:00Z | check | `{head}` | 0 | SELF | abc |\n")
|
||
self.commit_evidence(wt)
|
||
self.advance_remote()
|
||
result = self.task(wt, "sync")
|
||
self.assertIn("[evidence] WARN", result.stdout)
|
||
|
||
def test_sync_reads_legacy_commands_with_escaped_pipes(self):
|
||
wt = self.worktree()
|
||
card = self.task_card(wt)
|
||
head = self.git(wt, "rev-parse", "--short=10", "HEAD").stdout.strip()
|
||
card.write_text(card.read_text() +
|
||
f"\n| 2026-10-05T00:00:00Z | `false \\|\\| true` | `{head}` | 0 | SELF | abc |\n")
|
||
self.commit_evidence(wt)
|
||
self.advance_remote()
|
||
result = self.task(wt, "sync")
|
||
self.assertIn("[synced]", result.stdout)
|
||
self.assertIn("[evidence] WARN", result.stdout)
|
||
self.assertNotIn("unverified", result.stdout)
|
||
|
||
def test_finish_preserves_main_and_active_index_lock(self):
|
||
result = self.task(self.repo, "finish", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertTrue((self.repo / "tracked").is_file())
|
||
wt = self.worktree()
|
||
index_lock = Path(self.git(wt, "rev-parse", "--path-format=absolute", "--git-path", "index.lock").stdout.strip())
|
||
index_lock.write_text("another writer")
|
||
result = self.task(wt, "finish", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertTrue(wt.exists(), result.stdout)
|
||
self.assertEqual(index_lock.read_text(), "another writer")
|
||
|
||
def test_verify_rejects_recursive_drift_even_when_root_has_local_edits(self):
|
||
child = self.add_submodule()
|
||
wt = self.worktree()
|
||
self.git(wt, "submodule", "update", "--init", "--recursive")
|
||
(wt / "tracked").write_text("root implementation edit\n")
|
||
self.run_cmd(wt, "python3", "scripts/check-worktree.py", "verify")
|
||
(child / "content").write_text("v2\n")
|
||
self.git(child, "commit", "-am", "Component v2")
|
||
self.git(wt / "component", "fetch", "origin")
|
||
self.git(wt / "component", "checkout", "--detach", "origin/main")
|
||
result = self.run_cmd(wt, "python3", "scripts/check-worktree.py", "verify", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
|
||
def capture_precheck(self, cwd=None):
|
||
receipt = self.root / f"precheck-{len(list(self.root.glob('precheck*.json')))}.json"
|
||
env = self.env | {"RECEIPT": str(receipt), "CMD": "make precheck"}
|
||
self.run_cmd(cwd or self.repo, "bash", str(self.repo / "scripts/evidence.sh"), env=env)
|
||
return receipt
|
||
|
||
def triage_receipt(self, join=None, reason="fixture decision"):
|
||
"""经真实 task.sh triage 生成分流收据（对 Gitea 桩）。候选存在时按 join（分支名）或 NEW 显式决策。"""
|
||
receipt = self.root / f"triage-{len(list(self.root.glob('triage*.json')))}.json"
|
||
result = self.task(self.repo, "triage", "fixture requirement", check=False,
|
||
env=self.env | {"TRIAGE_RECEIPT": str(receipt)})
|
||
if result.returncode == 5:
|
||
decide_env = self.env | {"TRIAGE_RECEIPT": str(receipt), "REASON": reason,
|
||
"VERDICT": "JOIN" if join else "NEW"}
|
||
if join:
|
||
decide_env["INITIATIVE"] = join
|
||
self.task(self.repo, "triage-decide", env=decide_env)
|
||
else:
|
||
self.assertEqual(result.returncode, 0, result.stdout)
|
||
return receipt
|
||
|
||
def publish_initiative(self, branch):
|
||
"""把 initiative 分支推到 origin 并登记 open PR（模拟远端接纳中的大任务）。"""
|
||
self.git(self.repo, "push", "origin", f"refs/heads/{branch}:refs/heads/{branch}")
|
||
sha = self.git(self.repo, "rev-parse", f"refs/heads/{branch}").stdout.strip()
|
||
known = [pr for pr in self.gitea.prs if pr["head"]["ref"] == branch]
|
||
if known:
|
||
known[0]["head"]["sha"] = sha
|
||
else:
|
||
self.gitea.prs.append(FakeGitea.pr(len(self.gitea.prs) + 1, branch, sha))
|
||
|
||
def test_start_without_precheck_receipt_runs_and_imports_auto_precheck(self):
|
||
triage = self.triage_receipt()
|
||
result = self.task(self.repo, "start", "feature/new",
|
||
env=self.env | {"TRIAGE_RECEIPT": str(triage)})
|
||
line = next(row for row in result.stdout.splitlines() if row.startswith("[precheck] auto receipt="))
|
||
receipt = Path(line.split("=", 1)[1])
|
||
self.assertEqual(receipt.parent, self.root)
|
||
self.assertIn("feature-new", receipt.name)
|
||
data = json.loads(receipt.read_text())
|
||
self.assertEqual(data["command"], "make precheck")
|
||
self.assertIn("[verdict] READY", data["output"])
|
||
card = (self.root / "repo-wt-feature-new/.task/feature-new.md").read_text()
|
||
self.assertIn(data["receipt_sha256"], card)
|
||
|
||
def test_start_refuses_busy_auto_precheck_without_side_effects(self):
|
||
triage = self.triage_receipt()
|
||
(self.repo / "stray").write_text("unfinished\n")
|
||
result = self.task(self.repo, "start", "feature/new", check=False,
|
||
env=self.env | {"TRIAGE_RECEIPT": str(triage)})
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("SHARED_WORKSPACE_BUSY", result.stdout)
|
||
self.assertIn("auto precheck is not READY", result.stdout)
|
||
self.assertFalse((self.root / "repo-wt-feature-new").exists(), result.stdout)
|
||
self.assertEqual(self.git(self.repo, "branch", "--list", "feature/new").stdout.strip(), "")
|
||
|
||
def test_start_preserves_merged_local_and_remote_branch_names(self):
|
||
receipt = self.capture_precheck()
|
||
triage = self.triage_receipt()
|
||
head = self.git(self.repo, "rev-parse", "HEAD").stdout.strip()
|
||
for kind in ("local", "remote"):
|
||
with self.subTest(kind=kind):
|
||
name = "feature/merged-" + kind
|
||
if kind == "local":
|
||
self.git(self.repo, "branch", name, head)
|
||
ref = "refs/heads/" + name
|
||
else:
|
||
self.git(self.repo, "push", "origin", head + ":refs/heads/" + name)
|
||
ref = "refs/remotes/origin/" + name
|
||
self.git(self.repo, "fetch", "origin", "refs/heads/" + name + ":" + ref)
|
||
result = self.task(self.repo, "start", name, check=False,
|
||
env=self.env | {"PRECHECK_RECEIPT": str(receipt),
|
||
"TRIAGE_RECEIPT": str(triage)})
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("already merged", result.stdout)
|
||
self.assertFalse((self.root / ("repo-wt-" + name.replace("/", "-"))).exists())
|
||
self.assertEqual(self.git(self.repo, "rev-parse", ref).stdout.strip(), head)
|
||
|
||
def test_start_imports_precheck_and_creates_four_part_commit(self):
|
||
receipt = self.capture_precheck()
|
||
triage = self.triage_receipt()
|
||
self.task(self.repo, "start", "feature/new",
|
||
env=self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(triage)})
|
||
wt = self.root / "repo-wt-feature-new"
|
||
self.assertTrue((wt / ".task/feature-new.md").is_file())
|
||
card = (wt / ".task/feature-new.md").read_text()
|
||
self.assertIn("make precheck", card)
|
||
self.assertIn("- 需求分流：verdict=NEW；selected_initiative=无；requirement_sha256=", card)
|
||
self.assertIn(f"triage_receipt={triage.resolve()}；receipt_sha256=", card)
|
||
branch_line = card.index("- 分支类型：standalone")
|
||
self.assertEqual(card.index("- 需求分流："), card.index("\n", branch_line) + 1)
|
||
self.assertIn("交付模式：goal-first（目标、范围和交付终点一次确认；范围内连续推进）", card)
|
||
self.assertIn("实施执行：agent-per-unit（配置：AGENT-EXECUTION.md）", card)
|
||
self.assertIn("人工决策点：仅限目标/范围/验收变更、重大取舍、高后果操作或无法自动判定的业务验收", card)
|
||
self.assertNotIn("authorized-continuous", card)
|
||
self.assertNotIn("cursor-per-unit", card)
|
||
self.assertNotIn("Cursor unavailable means blocker only", card)
|
||
self.assertNotIn("不得交回主模型代写", card)
|
||
self.assertNotIn("never return to host", card)
|
||
self.assertIn("流程依据：仓库根目录 policy/delivery.md", card)
|
||
self.assertIn("## Agent 实施收据\n", card)
|
||
self.assertNotIn("## Cursor 实施收据\n", card)
|
||
agent_receipts = card.split("## Agent 实施收据\n", 1)[1].split("\n## ", 1)[0]
|
||
self.assertIn(
|
||
"| 单元 ID | 预期范围 | 配置摘要 | 角色 | 候选顺序 | attempt | CLI/模型 | 终态 | transport | fallback | 源码外收据路径 |",
|
||
agent_receipts,
|
||
)
|
||
self.assertIn(
|
||
"| 待定义 | — | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | — |",
|
||
agent_receipts,
|
||
)
|
||
self.assertIn(ROUTING_LINE, agent_receipts)
|
||
self.assertNotIn("Cursor、Antigravity（agy）和 PI", agent_receipts)
|
||
self.assertNotIn("## 子任务登记", card)
|
||
for field in ("当前阶段", "当前增量", "方案引用", "实施计划引用", "测试计划引用"):
|
||
self.assertIn(field + "：", card)
|
||
self.assertIn("## 交付调度\n", card)
|
||
queue = card.split("## 交付调度\n", 1)[1].split("\n## ", 1)[0]
|
||
self.assertIn("调度状态：未入队", queue)
|
||
for field in ("调度记录", "entry_id", "attempt_id", "窗口执行者", "固定候选"):
|
||
self.assertIn(field + "：无", queue)
|
||
self.assertIn("## 关联方向与规划\n", card)
|
||
planning = card.split("## 关联方向与规划\n", 1)[1].split("\n## ", 1)[0]
|
||
for field in ("适用状态", "主记录路径/版本", "对应里程碑"):
|
||
self.assertIn(field + "：待核对", planning)
|
||
self.assertIn("无适用规划时说明原因", planning)
|
||
self.assertIn("路径按仓库根目录定位，归档后不变", planning)
|
||
self.assertIn("只引用唯一正文", planning)
|
||
self.assertIn("| ID | 描述 | 项目符合度及依据 | 优先级 | 验收条件 | 依赖 | 结论 |", card)
|
||
reviews = card.split("## 阶段记录\n", 1)[1].split("\n## ", 1)[0]
|
||
self.assertIn("| 阶段 | 对象版本 | 状态 | 决策/执行者 | 时间 | 依据 |", reviews)
|
||
for stage in ("需求拆解", "需求评估", "需求表确认", "方案设计", "实施计划", "实施与自检",
|
||
"正式测试与验收", "代码 Review", "合入", "合入核验与收口"):
|
||
row = next(line for line in reviews.splitlines() if line.startswith(f"| {stage} |"))
|
||
self.assertEqual(row.split("|")[3].strip(), "待执行")
|
||
self.assertNotIn("通过", reviews)
|
||
self.assertIn("## 探索试验与迭代", card)
|
||
for field in ("假设", "范围", "判断和停止条件", "授权/决策依据", "结果", "变更影响", "需重审补测"):
|
||
self.assertIn(field, card.split("## 探索试验与迭代\n", 1)[1].split("\n## ", 1)[0])
|
||
self.assertIn("## 规划进展与下一步\n", card)
|
||
next_step = card.split("## 规划进展与下一步\n", 1)[1].split("\n## ", 1)[0]
|
||
for field in ("本次推进/剩余", "首选下一任务", "前置条件/待确认", "重复任务核对"):
|
||
self.assertIn(field + "：待核对", next_step)
|
||
self.assertIn("推荐依据：待填写", next_step)
|
||
self.assertIn("预期产物/验收条件：待定义", next_step)
|
||
self.assertIn("无适用规划、已完成或受阻时如实说明，不虚构任务", next_step)
|
||
self.assertIn("已有/在途任务、PR 与规划记录", next_step)
|
||
self.assertIn("启动授权：未授权", next_step)
|
||
self.assertIn("建议不自动启动新任务，仍须明确授权", next_step)
|
||
self.assertIn("## 合入与清理", card)
|
||
for field in ("合入记录", "清理记录"):
|
||
self.assertIn(field + "：NOT_RUN", card)
|
||
body = self.git(wt, "log", "-1", "--format=%B").stdout
|
||
for field in ("Why:", "Evidence:", "Impact:", "Refs:"):
|
||
self.assertIn(field, body)
|
||
|
||
def install_post_merge_hook(self):
|
||
hook = self.repo / ".git/hooks/post-merge"
|
||
shutil.copyfile(SOURCE / "scripts/hooks/post-merge", hook)
|
||
hook.chmod(0o755)
|
||
|
||
def publish_merged_task(self):
|
||
publisher = self.root / "publisher"
|
||
self.git(self.root, "clone", str(self.remote), str(publisher))
|
||
self.git(publisher, "switch", "-c", "feature/merged")
|
||
card = publisher / ".task/feature-merged.md"
|
||
card.parent.mkdir()
|
||
card.write_text("# Merged task\n任务主记录：待回填；状态：进行中；owner：test\n"
|
||
"- 工作区：fixture；分支：feature/merged\n")
|
||
self.git(publisher, "add", ".task")
|
||
self.git(publisher, "commit", "-m", "Record merged task")
|
||
self.git(publisher, "switch", "main")
|
||
self.git(publisher, "merge", "--no-ff", "feature/merged", "-m",
|
||
"Merge pull request 'Task' (#7) from feature/merged into main")
|
||
self.git(publisher, "push", "origin", "main")
|
||
return self.git(publisher, "rev-parse", "HEAD").stdout.strip(), card.read_text()
|
||
|
||
def test_long_merge_history_keeps_merged_card_visible_and_closable(self):
|
||
# A recent match makes grep -q/head exit before git log has finished
|
||
# writing a large history. With pipefail, its SIGPIPE used to hide the
|
||
# card from precheck and leave task-close with nothing to archive.
|
||
base = self.git(self.repo, "rev-parse", "HEAD").stdout.strip()
|
||
tree = self.git(self.repo, "rev-parse", "HEAD^{tree}").stdout.strip()
|
||
side = self.git(self.repo, "commit-tree", tree, "-p", base,
|
||
"-m", "Fixture side commit").stdout.strip()
|
||
tip = base
|
||
for i in range(220):
|
||
tip = self.git(self.repo, "commit-tree", tree, "-p", tip, "-p", side,
|
||
"-m", f"Older fixture merge {i} " + "x" * 512).stdout.strip()
|
||
self.git(self.repo, "update-ref", "refs/heads/main", tip)
|
||
self.git(self.repo, "push", "origin", "main")
|
||
actual_merge, _ = self.publish_merged_task()
|
||
self.git(self.repo, "fetch", "origin")
|
||
self.git(self.repo, "merge", "--ff-only", "origin/main")
|
||
|
||
precheck = self.task(self.repo, "precheck")
|
||
self.assertIn("[cards] 已合入未闭合任务卡:feature-merged.md", precheck.stdout)
|
||
closed = self.task(self.repo, "close")
|
||
self.assertIn("[closed] 1 张已合入卡片", closed.stdout)
|
||
refs = self.git(self.repo, "for-each-ref", "--format=%(refname:short)",
|
||
"refs/heads/feature/card-close-*").stdout.splitlines()
|
||
self.assertEqual(len(refs), 1)
|
||
archived = self.git(self.repo, "show", f"{refs[0]}:.task/archive/feature-merged.md").stdout
|
||
self.assertIn("状态：已完成（回填）", archived)
|
||
self.assertIn(actual_merge, archived)
|
||
|
||
def _stub_gitea(self, get_reply, post_reply=None):
|
||
"""Stub Gitea API：GET 按路径返回 get_reply(path)，POST 返回建 PR 应答。"""
|
||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||
import threading
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
def log_message(self, *args):
|
||
pass
|
||
|
||
def _send(self, obj):
|
||
body = json.dumps(obj).encode()
|
||
self.send_response(200)
|
||
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):
|
||
self._send(get_reply(self.path))
|
||
|
||
def do_POST(self):
|
||
self.rfile.read(int(self.headers.get("Content-Length") or 0))
|
||
self._send(post_reply or {"html_url": "http://stub.example/pr/1"})
|
||
|
||
server = HTTPServer(("127.0.0.1", 0), Handler)
|
||
threading.Thread(target=server.serve_forever, daemon=True).start()
|
||
self.addCleanup(server.shutdown)
|
||
return f"http://127.0.0.1:{server.server_port}", server
|
||
|
||
def test_close_backfills_card_landed_as_squash(self):
|
||
# squash 落地没有 "from <branch> into main" 合并提交：card_merge 须经 Gitea API 识别，
|
||
# 否则任务卡永远留在 .task/ 根目录（2026-10-14 #457 实测漏检）。
|
||
wt = self.worktree("feature/squashy")
|
||
self.record_task_card(wt)
|
||
branch = self.git(wt, "branch", "--show-current").stdout.strip()
|
||
self.git(wt, "push", "-u", "origin", branch)
|
||
self.git(self.repo, "fetch", "origin", branch)
|
||
self.git(self.repo, "merge", "--squash", "FETCH_HEAD")
|
||
self.git(self.repo, "commit", "-m", "Squash the fixture task")
|
||
squash_sha = self.git(self.repo, "rev-parse", "HEAD").stdout.strip()
|
||
self.git(self.repo, "push", "origin", "main")
|
||
|
||
base, _ = self._stub_gitea(
|
||
lambda path: [{
|
||
"number": 41, "title": "Decoy other branch", "state": "closed",
|
||
"merged": True, "head": {"ref": "feature/other", "label": "feature/other"},
|
||
"merge_commit_sha": squash_sha,
|
||
}, {
|
||
# head 分支已删的真实形态：ref 换成 pull 引用，原始分支名在 label
|
||
"number": 42, "title": "Fixture squash landing", "state": "closed",
|
||
"merged": True, "head": {"ref": "refs/pull/42/head", "label": branch},
|
||
"merge_commit_sha": squash_sha,
|
||
}] if "/pulls" in path else [],
|
||
post_reply={"html_url": "http://stub.example/pr/42"})
|
||
env = self.env | {"GITEA_TOKEN": "stub-token", "TASK_CARD_WEB_BASE": base}
|
||
closed = self.task(self.repo, "close", env=env)
|
||
self.assertIn("[closed] 1 张已合入卡片", closed.stdout)
|
||
refs = self.git(self.repo, "for-each-ref", "--format=%(refname:short)",
|
||
"refs/heads/feature/card-close-*").stdout.splitlines()
|
||
self.assertEqual(len(refs), 1)
|
||
archived = self.git(self.repo, "show", f"{refs[0]}:.task/archive/feature-squashy.md").stdout
|
||
self.assertIn("状态：已完成（回填）", archived)
|
||
self.assertIn(squash_sha, archived)
|
||
self.assertIn("PR #42", archived)
|
||
commit_msg = self.git(self.repo, "log", "-1", "--format=%B", refs[0]).stdout
|
||
self.assertIn("Gitea API squash lookup", commit_msg)
|
||
|
||
def test_close_ignores_squash_claim_when_sha_not_on_main(self):
|
||
# API 声称 merged 但 merge_commit_sha 不在 origin/main（伪造或陈旧）：不可核即不闭合。
|
||
wt = self.worktree("feature/phantom")
|
||
self.record_task_card(wt)
|
||
branch = self.git(wt, "branch", "--show-current").stdout.strip()
|
||
self.git(wt, "push", "-u", "origin", branch)
|
||
tip = self.git(wt, "rev-parse", "HEAD").stdout.strip()
|
||
base, _ = self._stub_gitea(
|
||
lambda path: [{
|
||
"number": 43, "title": "Phantom squash", "state": "closed",
|
||
"merged": True, "head": {"ref": branch},
|
||
"merge_commit_sha": tip,
|
||
}] if "/pulls" in path else [])
|
||
env = self.env | {"GITEA_TOKEN": "stub-token", "TASK_CARD_WEB_BASE": base}
|
||
closed = self.task(self.repo, "close", env=env)
|
||
self.assertIn("无已合入未闭合的任务卡", closed.stdout)
|
||
|
||
def assert_post_merge_only_reminds_owner(self, operation):
|
||
expected_head, expected_card = self.publish_merged_task()
|
||
self.git(self.repo, "fetch", "origin")
|
||
old_head = self.git(self.repo, "rev-parse", "HEAD").stdout.strip()
|
||
local_heads = self.git(self.repo, "show-ref", "--heads").stdout
|
||
worktrees = self.git(self.repo, "worktree", "list", "--porcelain").stdout
|
||
remote_refs = self.git(self.remote, "show-ref").stdout
|
||
self.install_post_merge_hook()
|
||
result = self.git(self.repo, *operation)
|
||
self.assertIn("make task-close", result.stdout)
|
||
self.assertIn("授权", result.stdout)
|
||
self.assertNotIn("[close]", result.stdout)
|
||
self.assertEqual(self.git(self.repo, "branch", "--show-current").stdout.strip(), "main")
|
||
self.assertEqual(self.git(self.repo, "rev-parse", "HEAD").stdout.strip(), expected_head)
|
||
self.assertEqual(self.git(self.repo, "show-ref", "--heads").stdout,
|
||
local_heads.replace(old_head, expected_head))
|
||
self.assertEqual(self.git(self.repo, "worktree", "list", "--porcelain").stdout,
|
||
worktrees.replace(old_head, expected_head))
|
||
self.assertEqual(self.git(self.remote, "show-ref").stdout, remote_refs)
|
||
self.assertEqual(self.git(self.repo, "status", "--porcelain").stdout, "")
|
||
self.assertEqual((self.repo / ".task/feature-merged.md").read_text(), expected_card)
|
||
self.assertFalse((self.repo / ".task/archive").exists())
|
||
|
||
def test_post_merge_main_merge_only_reminds_owner(self):
|
||
self.assert_post_merge_only_reminds_owner(("merge", "--ff-only", "origin/main"))
|
||
|
||
def test_post_merge_main_pull_only_reminds_owner(self):
|
||
self.assert_post_merge_only_reminds_owner(("pull", "--ff-only"))
|
||
|
||
def test_post_merge_is_silent_outside_main_checkout(self):
|
||
wt = self.worktree()
|
||
self.install_post_merge_hook()
|
||
self.advance_remote()
|
||
result = self.git(wt, "merge", "--ff-only", "origin/main")
|
||
self.assertNotIn("make task-close", result.stdout)
|
||
self.assertNotIn("[close]", result.stdout)
|
||
self.git(self.repo, "switch", "-c", "feature/primary")
|
||
self.git(wt, "switch", "-c", "feature/advance")
|
||
(wt / "feature-work").write_text("another task\n")
|
||
self.git(wt, "add", "feature-work")
|
||
self.git(wt, "commit", "-m", "Advance feature")
|
||
result = self.git(self.repo, "merge", "--ff-only", "feature/advance")
|
||
self.assertNotIn("make task-close", result.stdout)
|
||
self.assertNotIn("[close]", result.stdout)
|
||
|
||
def test_start_stacked_branch_records_parent_and_main_baseline(self):
|
||
base = self.git(self.repo, "rev-parse", "origin/main").stdout.strip()
|
||
parent = self.worktree("feature/parent")
|
||
(parent / "parent-change").write_text("parent feature")
|
||
self.git(parent, "add", "parent-change")
|
||
self.git(parent, "commit", "-m", "Parent change")
|
||
parent_head = self.git(parent, "rev-parse", "HEAD").stdout.strip()
|
||
receipt = self.capture_precheck(parent)
|
||
triage = self.triage_receipt()
|
||
env = self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(triage),
|
||
"TASK_PARENT_BRANCH": "feature/parent",
|
||
"TASK_PARENT_PR": "https://example.invalid/repo/pulls/42"}
|
||
self.task(parent, "start", "feature/parent-child", env=env)
|
||
wt = self.root / "repo-wt-feature-parent-child"
|
||
card = (wt / ".task/feature-parent-child.md").read_text()
|
||
self.assertIn(f"initial_base_sha）= {base}", card)
|
||
self.assertIn(f"parent_head_sha={parent_head}", card)
|
||
self.assertIn("https://example.invalid/repo/pulls/42", card)
|
||
self.assertEqual((wt / "parent-change").read_text(), "parent feature")
|
||
|
||
def test_start_offline_does_not_create_worktree(self):
|
||
triage = self.triage_receipt()
|
||
self.git(self.repo, "remote", "set-url", "origin", str(self.root / "missing.git"))
|
||
result = self.task(self.repo, "start", "feature/offline", check=False,
|
||
env=self.env | {"TRIAGE_RECEIPT": str(triage)})
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("REMOTE_UNVERIFIED", result.stdout)
|
||
self.assertFalse((self.root / "repo-wt-feature-offline").exists())
|
||
|
||
def test_start_rejects_receipt_when_authoritative_base_changed(self):
|
||
receipt = self.capture_precheck()
|
||
triage = self.triage_receipt()
|
||
self.advance_remote()
|
||
result = self.task(self.repo, "start", "feature/stale", check=False,
|
||
env=self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(triage)})
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertFalse((self.root / "repo-wt-feature-stale").exists())
|
||
|
||
def test_verify_rejects_uninitialized_component(self):
|
||
self.add_submodule()
|
||
wt = self.worktree()
|
||
result = self.run_cmd(wt, "python3", "scripts/check-worktree.py", "verify", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
|
||
# ---- 两级分支：initiative -> child -> main 与新需求强制分诊 ----
|
||
|
||
def write_initiative_card(self, branch, status="进行中", accepting="true", root="main",
|
||
kind="initiative", commit=True):
|
||
slug = branch.replace("/", "-")
|
||
card = self.repo / ".task" / f"{slug}.md"
|
||
card.parent.mkdir(exist_ok=True)
|
||
card.write_text(
|
||
f"# {branch}\n任务主记录：fixture；状态：{status}；owner：Lifecycle Test\n"
|
||
f"- 工作区：fixture（worktree）；分支：{branch}\n"
|
||
f"- 分支类型：{kind}；initiative_id：{slug}；root_target：{root}；"
|
||
f"immediate_target：{root}；accepting_requirements：{accepting}\n"
|
||
)
|
||
if commit:
|
||
self.git(self.repo, "add", ".task")
|
||
self.git(self.repo, "commit", "-m", f"Record {branch} card")
|
||
return card
|
||
|
||
def card_helper(self, func, card):
|
||
source = (self.repo / "scripts/task.sh").read_text()
|
||
lib = self.root / "task-lib.sh"
|
||
lib.write_text(source[:source.rindex('\ncase "${1:-}" in')] + "\n")
|
||
return self.run_cmd(self.repo, "bash", "-c", f'source {shlex.quote(str(lib))}; {func} "$1"',
|
||
"_", str(card)).stdout.strip()
|
||
|
||
def start_initiative(self, name="feature/alpha"):
|
||
receipt = self.capture_precheck()
|
||
triage = self.triage_receipt()
|
||
self.task(self.repo, "start", name,
|
||
env=self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(triage),
|
||
"TASK_BRANCH_KIND": "initiative"})
|
||
return self.root / ("repo-wt-" + name.replace("/", "-"))
|
||
|
||
def start_child(self, name="feature/alpha-login", initiative="feature/alpha", check=True):
|
||
# 远端须已登记该 initiative（PR+卡片），子任务才能经真实 JOIN 收据启动。
|
||
if initiative:
|
||
self.publish_initiative(initiative)
|
||
triage = self.triage_receipt(join=initiative)
|
||
else:
|
||
triage = self.root / "missing-triage.json"
|
||
receipt = self.capture_precheck()
|
||
env = self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(triage),
|
||
"TASK_BRANCH_KIND": "child", "TASK_INITIATIVE": initiative,
|
||
"TASK_PARENT_PR": "https://example.invalid/repo/pulls/7"}
|
||
result = self.task(self.repo, "start", name, env=env, check=check)
|
||
return self.root / ("repo-wt-" + name.replace("/", "-")), result
|
||
|
||
def test_triage_without_receipt_is_refused(self):
|
||
result = self.task(self.repo, "triage", "add login", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("TRIAGE_RECEIPT required", result.stdout)
|
||
|
||
def test_triage_ignores_local_cards_and_records_new_without_remote_initiative(self):
|
||
# main 上的本地卡片不是权威来源：远端没有 PR/分支时必须是 NEW，而不是按本地计数 JOIN。
|
||
self.write_initiative_card("feature/local-only")
|
||
receipt = self.triage_receipt()
|
||
data = json.loads(receipt.read_text())
|
||
self.assertEqual((data["discovery_verdict"], data["final_verdict"]), ("NEW", "NEW"))
|
||
self.assertEqual(data["candidates"], [])
|
||
self.assertIsNone(data["decision"]["selected_initiative"])
|
||
|
||
def test_remote_initiative_needs_explicit_decision_and_never_auto_joins(self):
|
||
self.start_initiative("feature/alpha")
|
||
self.publish_initiative("feature/alpha")
|
||
receipt = self.root / "single-candidate.json"
|
||
result = self.task(self.repo, "triage", "add login", check=False,
|
||
env=self.env | {"TRIAGE_RECEIPT": str(receipt)})
|
||
self.assertEqual(result.returncode, 5, result.stdout)
|
||
self.assertIn("[candidate] branch=feature/alpha", result.stdout)
|
||
self.assertIn("[verdict] DECISION_REQUIRED", result.stdout)
|
||
self.assertNotIn("[verdict] JOIN", result.stdout)
|
||
data = json.loads(receipt.read_text())
|
||
self.assertEqual(data["final_verdict"], "DECISION_REQUIRED")
|
||
self.assertIsNone(data["decision"])
|
||
|
||
def test_multiple_remote_candidates_stop_at_decision_required(self):
|
||
self.start_initiative("feature/alpha")
|
||
self.publish_initiative("feature/alpha")
|
||
self.start_initiative("feature/beta")
|
||
self.publish_initiative("feature/beta")
|
||
receipt = self.root / "multi-candidate.json"
|
||
result = self.task(self.repo, "triage", "add login", check=False,
|
||
env=self.env | {"TRIAGE_RECEIPT": str(receipt)})
|
||
self.assertEqual(result.returncode, 5, result.stdout)
|
||
self.assertIn("[candidate] branch=feature/alpha", result.stdout)
|
||
self.assertIn("[candidate] branch=feature/beta", result.stdout)
|
||
self.assertIn("[verdict] DECISION_REQUIRED", result.stdout)
|
||
self.assertNotIn("[verdict] JOIN", result.stdout)
|
||
self.assertEqual(json.loads(receipt.read_text())["final_verdict"], "DECISION_REQUIRED")
|
||
|
||
def test_precheck_reports_active_initiatives(self):
|
||
result = self.task(self.repo, "precheck", check=False)
|
||
self.assertIn("[initiatives] local cards (not authoritative): none", result.stdout)
|
||
self.write_initiative_card("feature/alpha")
|
||
self.write_initiative_card("feature/beta")
|
||
self.git(self.repo, "push", "origin", "main")
|
||
result = self.task(self.repo, "precheck", check=False)
|
||
self.assertIn("[initiatives] local cards (not authoritative): feature/alpha feature/beta", result.stdout)
|
||
|
||
def test_start_initiative_records_branch_metadata(self):
|
||
wt = self.start_initiative()
|
||
card = wt / ".task/feature-alpha.md"
|
||
text = card.read_text()
|
||
self.assertIn("- 分支类型：initiative；initiative_id：feature-alpha；root_target：main；"
|
||
"immediate_target：main；accepting_requirements：true", text)
|
||
self.assertEqual(self.git(wt, "merge-base", "HEAD", "origin/main").stdout.strip(),
|
||
self.git(self.repo, "rev-parse", "origin/main").stdout.strip())
|
||
self.assertEqual(self.card_helper("card_branch_kind", card), "initiative")
|
||
self.assertEqual(self.card_helper("card_accepting_requirements", card), "true")
|
||
|
||
def test_start_standalone_records_default_metadata(self):
|
||
receipt = self.capture_precheck()
|
||
triage = self.triage_receipt()
|
||
self.task(self.repo, "start", "feature/solo",
|
||
env=self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(triage)})
|
||
text = (self.root / "repo-wt-feature-solo/.task/feature-solo.md").read_text()
|
||
self.assertIn("- 分支类型：standalone；root_target：main；immediate_target：main；"
|
||
"accepting_requirements：false", text)
|
||
|
||
def test_start_child_records_parent_head_and_immediate_target(self):
|
||
parent = self.start_initiative()
|
||
(parent / "parent-change").write_text("initiative work\n")
|
||
self.git(parent, "add", "parent-change")
|
||
self.git(parent, "commit", "-m", "Initiative work")
|
||
parent_head = self.git(parent, "rev-parse", "HEAD").stdout.strip()
|
||
wt, _ = self.start_child()
|
||
card = wt / ".task/feature-alpha-login.md"
|
||
text = card.read_text()
|
||
self.assertIn("- 分支类型：child；initiative_id：feature-alpha；root_target：main；"
|
||
"immediate_target：feature/alpha；accepting_requirements：false；"
|
||
f"parent_branch：feature/alpha；parent_head_sha：{parent_head}；"
|
||
"parent_pr：https://example.invalid/repo/pulls/7", text)
|
||
self.assertEqual((wt / "parent-change").read_text(), "initiative work\n")
|
||
self.git(wt, "merge-base", "--is-ancestor", parent_head, "HEAD")
|
||
self.assertEqual(self.card_helper("card_immediate_target", card), "feature/alpha")
|
||
self.assertEqual(self.card_helper("card_initiative_id", card), "feature-alpha")
|
||
self.assertEqual(self.card_helper("card_root_target", card), "main")
|
||
self.assertIn("- 需求分流：verdict=JOIN；selected_initiative=feature/alpha；", text)
|
||
self.assert_lite_card(text, "feature/alpha-login")
|
||
|
||
def test_start_child_rejects_missing_initiative_or_wrong_slug(self):
|
||
self.start_initiative()
|
||
wt, result = self.start_child(name="feature/other-login", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("parent initiative slug", result.stdout)
|
||
self.assertFalse(wt.exists())
|
||
wt, result = self.start_child(initiative="", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertFalse(wt.exists())
|
||
|
||
def register_child(self, parent, child, initiative="feature/alpha", publish=True):
|
||
card = parent / ".task" / (initiative.replace("/", "-") + ".md")
|
||
text = card.read_text()
|
||
self.assertIn("| 待登记 | 否 | — | — |", text)
|
||
card.write_text(text.replace("| 待登记 | 否 | — | — |", f"| `{child}` | 是 | 进行中 | — |", 1))
|
||
self.git(parent, "add", ".task")
|
||
self.git(parent, "commit", "-m", f"Register {child}")
|
||
if publish:
|
||
self.publish_initiative(initiative)
|
||
return self.git(parent, "rev-parse", "HEAD").stdout.strip()
|
||
|
||
def planned_child_env(self):
|
||
return self.env | {"TASK_BRANCH_KIND": "child", "TASK_INITIATIVE": "feature/alpha",
|
||
"TASK_PARENT_PR": "https://example.invalid/repo/pulls/7"}
|
||
|
||
def assert_lite_card(self, text, name):
|
||
self.assertEqual([line for line in text.splitlines() if line.startswith("## ")], LITE_SECTIONS)
|
||
self.assertEqual(text.splitlines()[0], f"# {name} 任务卡")
|
||
self.assertRegex(text, r"(?m)^任务主记录：<PR 链接，交付前填写>；状态：进行中；owner：.+$")
|
||
self.assertRegex(text, r"(?m)^- 权威仓：.+；目标分支：main；R0（initial_base_sha）= [0-9a-f]{40}$")
|
||
self.assertRegex(text, rf"(?m)^- 工作区：.+（worktree）；分支：{re.escape(name)}$")
|
||
for prefix in ("- 分支类型：", "- 需求分流：", "- 授权与交付终点：",
|
||
"- 交付模式：goal-first（目标、范围和交付终点一次确认；范围内连续推进）",
|
||
"- 实施执行：agent-per-unit（配置：AGENT-EXECUTION.md）",
|
||
"- 合入执行：agent（",
|
||
"- 流程依据：仓库根目录 policy/delivery.md（归档后仍按仓库根目录定位）"):
|
||
self.assertRegex(text, "(?m)^" + re.escape(prefix))
|
||
self.assertIn(ROUTING_LINE, text)
|
||
self.assertIn("| 待定义 | — | NOT_RUN | NOT_RUN |", text)
|
||
self.assertRegex(text, r"(?m)^\| 启动 \| [0-9a-f]{40} \| [0-9a-f]{40} \| ")
|
||
self.assertIn("| 时间(UTC) | 命令 | 被测HEAD | exit | 来源 | sha256(16) |", text)
|
||
for field in ("Review 记录", "冲突记录", "CI 记录", "合入记录", "清理记录"):
|
||
self.assertIn(field + "：NOT_RUN", text)
|
||
self.assertNotIn("## 阶段记录", text)
|
||
|
||
def test_start_planned_child_without_triage_when_registered_on_origin(self):
|
||
parent = self.start_initiative()
|
||
parent_head = self.register_child(parent, "feature/alpha-login")
|
||
result = self.task(self.repo, "start", "feature/alpha-login", env=self.planned_child_env())
|
||
self.assertIn("[precheck] auto receipt=", result.stdout)
|
||
wt = self.root / "repo-wt-feature-alpha-login"
|
||
card = wt / ".task/feature-alpha-login.md"
|
||
text = card.read_text()
|
||
self.assertIn("- 需求分流：免（父 initiative 子任务登记已列出 feature/alpha-login；"
|
||
f"parent_head_sha={parent_head}）", text)
|
||
self.assertIn(f"parent_head_sha：{parent_head}；parent_pr：https://example.invalid/repo/pulls/7；"
|
||
"parent_source：origin", text)
|
||
self.git(wt, "merge-base", "--is-ancestor", parent_head, "HEAD")
|
||
self.assert_lite_card(text, "feature/alpha-login")
|
||
self.assertEqual(self.card_helper("card_immediate_target", card), "feature/alpha")
|
||
|
||
def test_start_planned_child_refuses_unregistered_child_without_side_effects(self):
|
||
parent = self.start_initiative()
|
||
self.publish_initiative("feature/alpha")
|
||
for case in ("placeholder-only", "registered-locally-only"):
|
||
with self.subTest(case=case):
|
||
if case == "registered-locally-only":
|
||
self.register_child(parent, "feature/alpha-login", publish=False)
|
||
result = self.task(self.repo, "start", "feature/alpha-login", check=False,
|
||
env=self.planned_child_env())
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("not registered in feature/alpha", result.stdout)
|
||
self.assertIn("provide TRIAGE_RECEIPT", result.stdout)
|
||
self.assertFalse((self.root / "repo-wt-feature-alpha-login").exists())
|
||
self.assertEqual(self.git(self.repo, "branch", "--list", "feature/alpha-login").stdout.strip(), "")
|
||
self.assertEqual(self.git(self.repo, "ls-remote", "--heads", "origin",
|
||
"feature/alpha-login").stdout.strip(), "")
|
||
|
||
def test_start_initiative_card_is_charter_with_child_registry(self):
|
||
wt = self.start_initiative()
|
||
card = wt / ".task/feature-alpha.md"
|
||
text = card.read_text()
|
||
goal = text.split("## 目标与非目标\n", 1)[1].split("\n## ", 1)[0]
|
||
for item in ("用途与目标", "接纳范围与非目标", "方案/计划链接", "分级测试策略", "验收条件",
|
||
"追加领域规则", "生命周期"):
|
||
self.assertIn(item, goal)
|
||
sections = [line for line in text.splitlines() if line.startswith("## ")]
|
||
self.assertEqual(sections[sections.index("## 目标与非目标") + 1], "## 子任务登记")
|
||
self.assertIn("## 阶段记录", sections)
|
||
self.assertIn(ROUTING_LINE, text)
|
||
rows = self.run_cmd(self.repo, "python3", "-c",
|
||
"import json, sys; sys.path.insert(0, 'scripts');"
|
||
"from task_pr_check import child_registry;"
|
||
"print(json.dumps(child_registry(open(sys.argv[1]).read()), ensure_ascii=False))",
|
||
str(card)).stdout
|
||
self.assertEqual(json.loads(rows), [{"子任务分支": "待登记", "必需": "否", "状态": "—", "PR": "—"}])
|
||
|
||
def test_start_fix_and_hotfix_cards_are_lite(self):
|
||
triage = self.triage_receipt()
|
||
for name in ("fix/typo", "hotfix/urgent"):
|
||
with self.subTest(name=name):
|
||
self.task(self.repo, "start", name, env=self.env | {"TRIAGE_RECEIPT": str(triage)})
|
||
slug = name.replace("/", "-")
|
||
text = (self.root / f"repo-wt-{slug}/.task/{slug}.md").read_text()
|
||
self.assert_lite_card(text, name)
|
||
self.assertIn("- 分支类型：standalone；root_target：main；", text)
|
||
self.assertIn("- 需求分流：verdict=NEW；", text)
|
||
|
||
def test_sync_child_merges_new_parent_initiative_commits(self):
|
||
parent = self.start_initiative()
|
||
wt, _ = self.start_child()
|
||
(parent / "parent-later").write_text("later initiative work\n")
|
||
self.git(parent, "add", "parent-later")
|
||
self.git(parent, "commit", "-m", "Later initiative work")
|
||
self.publish_initiative("feature/alpha") # delivery 只认远端父分支头
|
||
parent_head = self.git(parent, "rev-parse", "HEAD").stdout.strip()
|
||
result = self.task(wt, "sync")
|
||
self.assertIn("changed=1", result.stdout)
|
||
self.assertIn("sync_target=feature/alpha", result.stdout)
|
||
self.assertIn(f"sync_base={parent_head}", result.stdout)
|
||
self.git(wt, "merge-base", "--is-ancestor", parent_head, "HEAD")
|
||
self.assertEqual((wt / "parent-later").read_text(), "later initiative work\n")
|
||
self.assertIn(f"| 同步（记录前对象） | {parent_head} |",
|
||
(wt / ".task/feature-alpha-login.md").read_text())
|
||
again = self.task(wt, "sync")
|
||
self.assertIn("changed=0", again.stdout)
|
||
|
||
def test_finish_recycles_child_merged_into_parent_initiative(self):
|
||
parent = self.start_initiative()
|
||
wt, _ = self.start_child()
|
||
(wt / "child-work").write_text("child result\n")
|
||
self.git(wt, "add", "child-work")
|
||
self.git(wt, "commit", "-m", "Child work")
|
||
refused = self.task(wt, "finish", check=False)
|
||
self.assertNotEqual(refused.returncode, 0, refused.stdout)
|
||
self.assertIn("has not merged into feature/alpha", refused.stdout)
|
||
self.assertTrue(wt.exists())
|
||
self.git(parent, "merge", "--no-ff", "feature/alpha-login", "-m", "Merge child into initiative")
|
||
self.publish_initiative("feature/alpha") # delivery 的 finish 只认远端父分支已含子分支
|
||
result = self.task(wt, "finish")
|
||
self.assertIn("[verified] feature/alpha-login is included in parent initiative feature/alpha",
|
||
result.stdout)
|
||
self.assertFalse(wt.exists(), result.stdout)
|
||
self.assertEqual(self.git(self.repo, "branch", "--list", "feature/alpha-login").stdout.strip(), "")
|
||
self.assertTrue(parent.exists())
|
||
self.assertEqual(self.git(parent, "branch", "--show-current").stdout.strip(), "feature/alpha")
|
||
|
||
def test_legacy_card_defaults_to_standalone_main(self):
|
||
wt = self.worktree()
|
||
card = self.record_task_card(wt)
|
||
self.assertEqual(self.card_helper("card_branch_kind", card), "standalone")
|
||
self.assertEqual(self.card_helper("card_initiative_id", card), "")
|
||
self.assertEqual(self.card_helper("card_root_target", card), "main")
|
||
self.assertEqual(self.card_helper("card_immediate_target", card), "main")
|
||
self.assertEqual(self.card_helper("card_accepting_requirements", card), "false")
|
||
receipt = self.triage_receipt()
|
||
self.assertEqual(json.loads(receipt.read_text())["final_verdict"], "NEW")
|
||
|
||
def test_start_without_triage_receipt_creates_no_worktree_or_branch(self):
|
||
receipt = self.capture_precheck()
|
||
for kind in ("standalone", "initiative"):
|
||
with self.subTest(kind=kind):
|
||
result = self.task(self.repo, "start", "feature/untriaged", check=False,
|
||
env=self.env | {"PRECHECK_RECEIPT": str(receipt), "TASK_BRANCH_KIND": kind})
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("TRIAGE_RECEIPT required", result.stdout)
|
||
self.assertFalse((self.root / "repo-wt-feature-untriaged").exists())
|
||
self.assertEqual(self.git(self.repo, "branch", "--list", "feature/untriaged").stdout.strip(), "")
|
||
self.assertEqual(self.git(self.repo, "ls-remote", "--heads", "origin",
|
||
"feature/untriaged").stdout.strip(), "")
|
||
|
||
def test_start_child_refuses_non_join_receipt_without_side_effects(self):
|
||
self.start_initiative("feature/alpha")
|
||
self.publish_initiative("feature/alpha")
|
||
new_receipt = self.triage_receipt() # 远端存在候选，显式 NEW 决策（非适配）
|
||
receipt = self.capture_precheck()
|
||
env = self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(new_receipt),
|
||
"TASK_BRANCH_KIND": "child", "TASK_INITIATIVE": "feature/alpha",
|
||
"TASK_PARENT_PR": "https://example.invalid/repo/pulls/7"}
|
||
result = self.task(self.repo, "start", "feature/alpha-login", check=False, env=env)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("triage receipt rejected", result.stdout)
|
||
self.assertFalse((self.root / "repo-wt-feature-alpha-login").exists())
|
||
self.assertEqual(self.git(self.repo, "branch", "--list", "feature/alpha-login").stdout.strip(), "")
|
||
|
||
def test_start_refuses_receipt_when_remote_initiative_moved_after_triage(self):
|
||
self.start_initiative("feature/alpha")
|
||
self.publish_initiative("feature/alpha")
|
||
join = self.triage_receipt(join="feature/alpha")
|
||
parent = self.root / "repo-wt-feature-alpha"
|
||
(parent / "moved").write_text("later initiative work\n")
|
||
self.git(parent, "add", "moved")
|
||
self.git(parent, "commit", "-m", "Move initiative after triage")
|
||
self.publish_initiative("feature/alpha")
|
||
receipt = self.capture_precheck()
|
||
env = self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(join),
|
||
"TASK_BRANCH_KIND": "child", "TASK_INITIATIVE": "feature/alpha",
|
||
"TASK_PARENT_PR": "https://example.invalid/repo/pulls/7"}
|
||
result = self.task(self.repo, "start", "feature/alpha-login", check=False, env=env)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("remote initiative candidates changed", result.stdout)
|
||
self.assertFalse((self.root / "repo-wt-feature-alpha-login").exists())
|
||
|
||
def close_env(self):
|
||
# 测试 token 与本地 Gitea 桩：建 PR 只能落到桩上，绝不触达真实 Gitea。
|
||
return self.env | {"GITEA_TOKEN": self.gitea.token, "TASK_CARD_WEB_BASE": self.gitea.web}
|
||
|
||
def child_env(self, receipt, triage):
|
||
return self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(triage),
|
||
"TASK_BRANCH_KIND": "child", "TASK_INITIATIVE": "feature/alpha",
|
||
"TASK_PARENT_PR": "https://example.invalid/repo/pulls/7"}
|
||
|
||
def test_delivery_child_ignores_ahead_local_parent(self):
|
||
parent = self.start_initiative()
|
||
self.publish_initiative("feature/alpha")
|
||
remote_head = self.git(parent, "rev-parse", "HEAD").stdout.strip()
|
||
(parent / "unadmitted").write_text("local only\n")
|
||
self.git(parent, "add", "unadmitted")
|
||
self.git(parent, "commit", "-m", "Unpublished initiative work")
|
||
triage = self.triage_receipt(join="feature/alpha")
|
||
receipt = self.capture_precheck()
|
||
self.task(self.repo, "start", "feature/alpha-login", env=self.child_env(receipt, triage))
|
||
wt = self.root / "repo-wt-feature-alpha-login"
|
||
card = (wt / ".task/feature-alpha-login.md").read_text()
|
||
self.assertIn(f"parent_head_sha：{remote_head}", card)
|
||
self.assertIn("parent_source：origin", card)
|
||
self.assertFalse((wt / "unadmitted").exists())
|
||
|
||
def test_delivery_child_ignores_stale_local_parent(self):
|
||
self.start_initiative()
|
||
self.publish_initiative("feature/alpha")
|
||
other = self.root / "other"
|
||
self.git(self.root, "clone", "-q", str(self.remote), str(other))
|
||
self.git(other, "checkout", "-q", "feature/alpha")
|
||
(other / "remote-admitted").write_text("admitted\n")
|
||
self.git(other, "add", "remote-admitted")
|
||
self.git(other, "commit", "-m", "Admitted initiative work")
|
||
self.git(other, "push", "-q", "origin", "feature/alpha")
|
||
remote_head = self.git(other, "rev-parse", "HEAD").stdout.strip()
|
||
triage = self.triage_receipt(join="feature/alpha")
|
||
receipt = self.capture_precheck()
|
||
self.task(self.repo, "start", "feature/alpha-login", env=self.child_env(receipt, triage))
|
||
wt = self.root / "repo-wt-feature-alpha-login"
|
||
self.assertIn(f"parent_head_sha：{remote_head}", (wt / ".task/feature-alpha-login.md").read_text())
|
||
self.assertEqual((wt / "remote-admitted").read_text(), "admitted\n")
|
||
|
||
def test_delivery_child_fails_when_remote_parent_is_deleted_after_triage(self):
|
||
self.start_initiative()
|
||
self.publish_initiative("feature/alpha")
|
||
triage = self.triage_receipt(join="feature/alpha")
|
||
self.git(self.repo, "push", "origin", "--delete", "feature/alpha")
|
||
receipt = self.capture_precheck()
|
||
result = self.task(self.repo, "start", "feature/alpha-login", check=False,
|
||
env=self.child_env(receipt, triage))
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertFalse((self.root / "repo-wt-feature-alpha-login").exists())
|
||
self.assertEqual(self.git(self.repo, "branch", "--list", "feature/alpha-login").stdout.strip(), "")
|
||
|
||
def test_local_mode_child_uses_local_parent_and_marks_local_source(self):
|
||
parent = self.start_initiative()
|
||
self.publish_initiative("feature/alpha")
|
||
(parent / "local-admission").write_text("local only\n")
|
||
self.git(parent, "add", "local-admission")
|
||
self.git(parent, "commit", "-m", "Local initiative work")
|
||
local_head = self.git(parent, "rev-parse", "HEAD").stdout.strip()
|
||
triage = self.triage_receipt(join="feature/alpha")
|
||
receipt = self.capture_precheck()
|
||
self.task(self.repo, "start", "feature/alpha-login",
|
||
env=self.child_env(receipt, triage) | {"TASK_WORK_MODE": "local"})
|
||
wt = self.root / "repo-wt-feature-alpha-login"
|
||
card = (wt / ".task/feature-alpha-login.md").read_text()
|
||
self.assertIn(f"parent_head_sha：{local_head}", card)
|
||
self.assertIn("parent_source：local（非正式准入）", card)
|
||
self.assertEqual((wt / "local-admission").read_text(), "local only\n")
|
||
|
||
def test_resume_child_reports_parent_as_immediate_target(self):
|
||
parent = self.start_initiative()
|
||
wt, _ = self.start_child()
|
||
parent_head = self.git(parent, "rev-parse", "HEAD").stdout.strip()
|
||
receipt = self.capture_precheck(cwd=wt)
|
||
result = self.task(wt, "resume", env=self.env | {"PRECHECK_RECEIPT": str(receipt)})
|
||
self.assertIn(f"observed_target={parent_head} immediate_target=feature/alpha", result.stdout)
|
||
|
||
def test_resume_child_rejects_rewound_immediate_target(self):
|
||
parent = self.start_initiative()
|
||
wt, _ = self.start_child()
|
||
receipt = self.capture_precheck(cwd=wt)
|
||
self.git(parent, "reset", "--hard", "origin/main")
|
||
self.git(parent, "push", "-f", "origin", "feature/alpha")
|
||
result = self.task(wt, "resume", check=False, env=self.env | {"PRECHECK_RECEIPT": str(receipt)})
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn("no longer contains recorded parent head", result.stdout)
|
||
self.assertNotIn("[resumed]", result.stdout)
|
||
|
||
def test_close_archives_child_merged_into_initiative_and_names_it(self):
|
||
parent = self.start_initiative()
|
||
wt, _ = self.start_child()
|
||
(wt / "child-work").write_text("child result\n")
|
||
self.git(wt, "add", "child-work")
|
||
self.git(wt, "commit", "-m", "Child work")
|
||
self.git(parent, "merge", "--no-ff", "feature/alpha-login", "-m",
|
||
"Merge pull request 'feature/alpha-login' (#5) from feature/alpha-login into feature/alpha")
|
||
self.publish_initiative("feature/alpha")
|
||
closed = self.task(parent, "close", "--fold", env=self.close_env())
|
||
self.assertIn("[closed] 1 张已合入卡片", closed.stdout)
|
||
self.assertEqual(self.gitea.posted[-1]["head"], "feature/alpha")
|
||
archived = self.git(parent, "show", "HEAD:.task/archive/feature-alpha-login.md").stdout
|
||
self.assertIn("已包含于 origin/feature/alpha", archived)
|
||
self.assertIn("状态：已完成（回填）", archived)
|
||
self.assertIn("PR #5", archived)
|
||
self.assertFalse((parent / ".task/feature-alpha-login.md").exists())
|
||
|
||
def test_close_child_still_closable_after_initiative_merged_and_deleted(self):
|
||
parent = self.start_initiative()
|
||
wt, _ = self.start_child()
|
||
(wt / "child-work").write_text("child result\n")
|
||
self.git(wt, "add", "child-work")
|
||
self.git(wt, "commit", "-m", "Child work")
|
||
self.git(parent, "merge", "--no-ff", "feature/alpha-login", "-m",
|
||
"Merge pull request 'feature/alpha-login' (#5) from feature/alpha-login into feature/alpha")
|
||
self.publish_initiative("feature/alpha")
|
||
self.git(self.repo, "fetch", "-q", "origin")
|
||
self.git(self.repo, "merge", "--no-ff", "origin/feature/alpha", "-m",
|
||
"Merge pull request 'feature/alpha' (#3) from feature/alpha into main")
|
||
self.git(self.repo, "push", "origin", "main")
|
||
self.git(self.repo, "push", "origin", "--delete", "feature/alpha")
|
||
closed = self.task(self.repo, "close", env=self.close_env())
|
||
self.assertIn("[closed] 2 张已合入卡片", closed.stdout)
|
||
self.assertIn("[pr] " + self.gitea.web, closed.stdout)
|
||
self.assertEqual(self.gitea.posted[-1]["base"], "main")
|
||
refs = self.git(self.repo, "for-each-ref", "--format=%(refname:short)",
|
||
"refs/heads/feature/card-close-*").stdout.splitlines()
|
||
self.assertEqual(len(refs), 1)
|
||
archived = self.git(self.repo, "show", f"{refs[0]}:.task/archive/feature-alpha-login.md").stdout
|
||
self.assertIn("已包含于 origin/main", archived)
|
||
self.assertIn("PR #5", archived)
|
||
|
||
# ---- close --self：合入前在任务分支内闭合本卡 ----
|
||
|
||
def self_close_env(self):
|
||
return self.close_env() | {"GITEA_API_URL": self.gitea.url, "GITEA_REPO": "fixture/repo"}
|
||
|
||
def start_standalone(self, name="feature/solo"):
|
||
receipt = self.capture_precheck()
|
||
triage = self.triage_receipt()
|
||
self.task(self.repo, "start", name,
|
||
env=self.env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(triage)})
|
||
return self.root / ("repo-wt-" + name.replace("/", "-"))
|
||
|
||
def open_pr(self, wt, branch, base="main"):
|
||
self.git(wt, "push", "-q", "-u", "origin", branch)
|
||
number = len(self.gitea.prs) + 1
|
||
sha = self.git(wt, "rev-parse", "HEAD").stdout.strip()
|
||
self.gitea.prs.append(FakeGitea.pr(number, branch, sha, base=base))
|
||
return number
|
||
|
||
def test_close_self_archives_own_card_inside_its_pr(self):
|
||
wt = self.start_standalone()
|
||
(wt / "work").write_text("task result\n")
|
||
self.git(wt, "add", "work")
|
||
self.git(wt, "commit", "-m", "Task work")
|
||
number = self.open_pr(wt, "feature/solo")
|
||
result = self.task(wt, "close", "--self", env=self.self_close_env())
|
||
self.assertIn(f"[closed] .task/feature-solo.md 已在 PR #{number} 合入前闭合并归档；合入后无需 task-close",
|
||
result.stdout)
|
||
self.assertNotIn(self.gitea.token, result.stdout)
|
||
self.assertFalse((wt / ".task/feature-solo.md").exists())
|
||
text = (wt / ".task/archive/feature-solo.md").read_text()
|
||
self.assertIn("；状态：已完成（合入前闭合）；", text)
|
||
self.assertIn(f"任务主记录：PR #{number} {self.gitea.web}/{self.root.name}/remote/pulls/{number}；", text)
|
||
self.assertIn("## 结果与验证\n", text)
|
||
self.assertNotIn("## 结果与验证（完成后填写）", text)
|
||
self.assertIn("## 闭合记录（合入前折叠）", text)
|
||
self.assertIn(f"PR #{number}", text.split("## 闭合记录（合入前折叠）", 1)[1])
|
||
self.assertEqual(self.git(wt, "log", "-1", "--format=%s").stdout.strip(),
|
||
"Fold-close the task card before merge")
|
||
body = self.git(wt, "log", "-1", "--format=%B").stdout
|
||
for field in ("Why:", "Evidence:", "Impact:", "Refs: task.sh close --self"):
|
||
self.assertIn(field, body)
|
||
self.assertEqual(self.git(wt, "status", "--porcelain").stdout, "")
|
||
head = self.git(wt, "rev-parse", "HEAD").stdout.strip()
|
||
self.assertEqual(self.git(self.remote, "rev-parse", "refs/heads/feature/solo").stdout.strip(), head)
|
||
self.assertEqual(self.git(self.remote, "show", "feature/solo:.task/archive/feature-solo.md").stdout, text)
|
||
again = self.task(wt, "close", "--self", check=False, env=self.self_close_env())
|
||
self.assertNotEqual(again.returncode, 0, again.stdout)
|
||
self.assertEqual(self.git(wt, "rev-parse", "HEAD").stdout.strip(), head)
|
||
|
||
def test_close_self_accepts_explicit_pr_number_without_lookup(self):
|
||
wt = self.start_standalone()
|
||
env = self.close_env() | {"TASK_PR": "42", "GITEA_API_URL": "http://127.0.0.1:9/api/v1"}
|
||
result = self.task(wt, "close", "--self", env=env)
|
||
self.assertIn("PR #42 合入前闭合", result.stdout)
|
||
self.assertIn("任务主记录：PR #42 ", (wt / ".task/archive/feature-solo.md").read_text())
|
||
|
||
def test_close_self_refuses_main_detached_dirty_and_missing_pr(self):
|
||
main_head = self.git(self.repo, "rev-parse", "HEAD").stdout.strip()
|
||
on_main = self.task(self.repo, "close", "--self", check=False, env=self.self_close_env())
|
||
self.assertNotEqual(on_main.returncode, 0, on_main.stdout)
|
||
self.assertIn("不能在 main 上运行", on_main.stdout)
|
||
self.assertEqual(self.git(self.repo, "rev-parse", "HEAD").stdout.strip(), main_head)
|
||
wt = self.start_standalone()
|
||
head = self.git(wt, "rev-parse", "HEAD").stdout.strip()
|
||
card = wt / ".task/feature-solo.md"
|
||
original = card.read_text()
|
||
|
||
def assert_refused(result, reason):
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertIn(reason, result.stdout)
|
||
self.assertNotIn(self.gitea.token, result.stdout)
|
||
self.assertEqual(self.git(wt, "rev-parse", "HEAD").stdout.strip(), head)
|
||
self.assertEqual(card.read_text(), original)
|
||
self.assertFalse((wt / ".task/archive").exists())
|
||
|
||
(wt / "stray").write_text("uncommitted\n")
|
||
assert_refused(self.task(wt, "close", "--self", check=False, env=self.self_close_env()), "不干净")
|
||
(wt / "stray").unlink()
|
||
assert_refused(self.task(wt, "close", "--self", check=False, env=self.self_close_env()),
|
||
"需要恰好一个 head 为 feature/solo 的开放 PR")
|
||
assert_refused(self.task(wt, "close", "--self", check=False,
|
||
env=self.self_close_env() | {"TASK_PR": "7; rm"}), "TASK_PR")
|
||
self.git(wt, "checkout", "-q", "--detach")
|
||
assert_refused(self.task(wt, "close", "--self", check=False, env=self.self_close_env()), "detached HEAD")
|
||
|
||
def test_close_self_child_keeps_initiative_target_from_archive(self):
|
||
parent = self.start_initiative()
|
||
wt, _ = self.start_child()
|
||
number = self.open_pr(wt, "feature/alpha-login", base="feature/alpha")
|
||
self.task(wt, "close", "--self", env=self.self_close_env())
|
||
archived = wt / ".task/archive/feature-alpha-login.md"
|
||
self.assertIn("immediate_target：feature/alpha", archived.read_text())
|
||
self.assertIn(f"PR #{number}", archived.read_text())
|
||
(parent / "parent-later").write_text("later initiative work\n")
|
||
self.git(parent, "add", "parent-later")
|
||
self.git(parent, "commit", "-m", "Later initiative work")
|
||
self.publish_initiative("feature/alpha")
|
||
parent_head = self.git(parent, "rev-parse", "HEAD").stdout.strip()
|
||
synced = self.task(wt, "sync")
|
||
self.assertIn("sync_target=feature/alpha", synced.stdout)
|
||
self.assertIn(f"sync_base={parent_head}", synced.stdout)
|
||
self.assertIn(f"| 同步（记录前对象） | {parent_head} |", archived.read_text())
|
||
self.git(parent, "merge", "--no-ff", "feature/alpha-login", "-m", "Merge child into initiative")
|
||
self.publish_initiative("feature/alpha")
|
||
finished = self.task(wt, "finish")
|
||
self.assertIn("[verified] feature/alpha-login is included in parent initiative feature/alpha",
|
||
finished.stdout)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|
||
|
||
|
||
class FinishGuardFixTests(unittest.TestCase):
|
||
"""2026-10-05 两处 finish 守卫误报的回归：IDE 产物锁不再误报、真实 Git 锁仍拦；
|
||
子仓远端跟踪引用不再卡 pinned-history，本地独有工作仍保全。"""
|
||
|
||
def setUp(self):
|
||
self.root = tempfile.mkdtemp(prefix="finish-guard-")
|
||
self.addCleanup(shutil.rmtree, self.root, ignore_errors=True)
|
||
self.repo = Path(self.root) / "repo"
|
||
shutil.copytree(SOURCE / "scripts", self.repo / "scripts")
|
||
(self.repo / "Makefile").write_text("precheck:\n\t@bash scripts/task.sh precheck\n")
|
||
self.env = os.environ.copy()
|
||
self.env.update({"GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": "/dev/null",
|
||
"GIT_ALLOW_PROTOCOL": "file", "GIT_TERMINAL_PROMPT": "0",
|
||
"GIT_AUTHOR_NAME": "Guard Test", "GIT_AUTHOR_EMAIL": "guard@example.invalid",
|
||
"GIT_COMMITTER_NAME": "Guard Test", "GIT_COMMITTER_EMAIL": "guard@example.invalid",
|
||
"TASK_WRITERS_STOPPED": "1",
|
||
"AGENT_EVIDENCE_DIR": self.root + "/evidence",
|
||
"NO_PROXY": "127.0.0.1,localhost", "no_proxy": "127.0.0.1,localhost"})
|
||
for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GITEA_TOKEN", "GITEA_API_URL",
|
||
"GITEA_REPO", "TRIAGE_RECEIPT", "PRECHECK_RECEIPT"):
|
||
self.env.pop(key, None)
|
||
self.gitea = FakeGitea()
|
||
self.addCleanup(self.gitea.close)
|
||
self.git(self.root, "init", "--bare", "-b", "main", self.root + "/origin.git")
|
||
self.git(self.root, "init", "-b", "main", str(self.repo))
|
||
(self.repo / "tracked").write_text("base\n")
|
||
self.git(self.repo, "add", ".")
|
||
self.git(self.repo, "commit", "-q", "-m", "base")
|
||
self.git(self.repo, "remote", "add", "origin", self.root + "/origin.git")
|
||
self.git(self.repo, "push", "-q", "origin", "main")
|
||
|
||
def git(self, cwd, *args):
|
||
return subprocess.run(["git", "-C", str(cwd), *args], check=True,
|
||
capture_output=True, text=True, env=self.env)
|
||
|
||
def task(self, cwd, *args, check=True, env=None):
|
||
return subprocess.run(["bash", str(self.repo / "scripts" / "task.sh"), *args],
|
||
cwd=str(cwd), capture_output=True, text=True,
|
||
env=env or self.env, check=check)
|
||
|
||
def worktree(self):
|
||
receipt = Path(self.root) / "precheck.json"
|
||
subprocess.run(
|
||
["bash", str(self.repo / "scripts" / "evidence.sh")], cwd=str(self.repo),
|
||
capture_output=True, text=True,
|
||
env=self.env | {"RECEIPT": str(receipt), "CMD": "make precheck"}, check=True)
|
||
triage = Path(self.root) / "triage.json"
|
||
triage_env = self.env | {"GITEA_API_URL": self.gitea.url, "GITEA_REPO": "fixture/repo",
|
||
"GITEA_TOKEN": self.gitea.token}
|
||
self.task(self.repo, "triage", "guard fixture", env=triage_env | {"TRIAGE_RECEIPT": str(triage)})
|
||
self.task(self.repo, "start", "feature/guard",
|
||
env=triage_env | {"PRECHECK_RECEIPT": str(receipt), "TRIAGE_RECEIPT": str(triage)})
|
||
wt = self.root + "/repo-wt-feature-guard"
|
||
assert Path(wt).is_dir(), "worktree not created"
|
||
return Path(wt)
|
||
|
||
def add_submodule_pinned_off_main(self):
|
||
"""pin 在 feature 顶端（origin/main 不在 pin 历史内）——修复前的永久卡死场景。"""
|
||
child = Path(self.root) / "component"
|
||
self.git(self.root, "init", "-b", "main", str(child))
|
||
(child / "content").write_text("v1\n")
|
||
self.git(child, "add", ".")
|
||
self.git(child, "-c", "user.name=T", "-c", "user.email=t@t", "commit", "-q", "-m", "v1")
|
||
self.git(self.root, "init", "--bare", "-b", "main", self.root + "/component-origin.git")
|
||
self.git(child, "remote", "add", "origin", self.root + "/component-origin.git")
|
||
self.git(child, "push", "-q", "origin", "main")
|
||
self.git(child, "switch", "-c", "feature/pin-tip")
|
||
(child / "content").write_text("v2 pin tip\n")
|
||
self.git(child, "commit", "-am", "v2")
|
||
pin_tip = subprocess.run(["git", "-C", str(child), "rev-parse", "HEAD"],
|
||
capture_output=True, text=True, check=True, env=self.env).stdout.strip()
|
||
# 把 pin-tip 推上远端，子仓 add 时按该分支检出
|
||
self.git(child, "push", "-q", "origin", "feature/pin-tip")
|
||
self.git(self.repo, "-c", "protocol.file.allow=always",
|
||
"submodule", "add", "-b", "feature/pin-tip",
|
||
self.root + "/component-origin.git", "component")
|
||
self.git(self.repo, "commit", "-q", "-am", "Pin component at feature tip")
|
||
self.git(self.repo, "push", "-q", "origin", "main")
|
||
# 让子仓本地带上 origin/main 的远端跟踪引用（不在 pin 历史内）
|
||
self.git(child, "fetch", "origin")
|
||
return child, pin_tip
|
||
|
||
def test_ide_lock_in_git_dir_does_not_block_finish(self):
|
||
wt = self.worktree()
|
||
(self.repo / "tracked").write_text("implementation done\n")
|
||
self.git(self.repo, "add", "-A")
|
||
self.git(self.repo, "commit", "-q", "-m", "done")
|
||
self.git(self.repo, "push", "-q", "origin", "feature/guard")
|
||
self.git(self.repo, "merge", "--no-ff", "-q", "-m", "Merge feature/guard", "feature/guard")
|
||
self.git(self.repo, "push", "-q", "origin", "main")
|
||
ide = Path(subprocess.run(["git", "-C", str(self.repo), "rev-parse",
|
||
"--path-format=absolute", "--git-common-dir"],
|
||
capture_output=True, text=True, check=True).stdout.strip()) / "cursor" / "crepe"
|
||
ide.mkdir(parents=True, exist_ok=True)
|
||
stale = ide / "build.lock"
|
||
stale.write_text("stale IDE artifact")
|
||
result = self.task(wt, "finish", check=False)
|
||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||
self.assertFalse(wt.exists())
|
||
self.assertTrue(stale.is_file(), "IDE artifact must be preserved untouched")
|
||
|
||
def test_real_git_lock_still_blocks_finish(self):
|
||
wt = self.worktree()
|
||
(self.repo / "tracked").write_text("implementation done\n")
|
||
self.git(self.repo, "add", "-A")
|
||
self.git(self.repo, "commit", "-q", "-m", "done")
|
||
self.git(self.repo, "push", "-q", "origin", "feature/guard")
|
||
self.git(self.repo, "merge", "--no-ff", "-q", "-m", "Merge feature/guard", "feature/guard")
|
||
self.git(self.repo, "push", "-q", "origin", "main")
|
||
common = Path(subprocess.run(["git", "-C", str(self.repo), "rev-parse",
|
||
"--path-format=absolute", "--git-common-dir"],
|
||
capture_output=True, text=True, check=True).stdout.strip())
|
||
(common / "config.lock").write_text("live writer")
|
||
result = self.task(wt, "finish", check=False)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertTrue(wt.exists())
|
||
|
||
def test_remote_tracking_refs_do_not_block_finish(self):
|
||
self.add_submodule_pinned_off_main()
|
||
wt = self.worktree()
|
||
self.git(wt, "submodule", "update", "--init", "--recursive")
|
||
(self.repo / "tracked").write_text("implementation done\n")
|
||
self.git(self.repo, "add", "-A")
|
||
self.git(self.repo, "commit", "-q", "-m", "done")
|
||
self.git(self.repo, "push", "-q", "origin", "feature/guard")
|
||
self.git(self.repo, "merge", "--no-ff", "-q", "-m", "Merge feature/guard", "feature/guard")
|
||
self.git(self.repo, "push", "-q", "origin", "main")
|
||
# 模拟实际场景：子仓检出于 pin 顶端且带有远端跟踪引用与克隆 reflog
|
||
result = self.task(wt, "finish", check=False)
|
||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||
self.assertFalse(wt.exists())
|
||
|
||
|
||
def test_local_default_branch_at_origin_main_does_not_block(self):
|
||
"""克隆默认分支停在 origin/main 顶端（pin 为 feature 顶端时不在其历史内）
|
||
且 HEAD reflog 含克隆/检出轨迹——均为远端有副本的状态，不得阻断 finish。"""
|
||
self.add_submodule_pinned_off_main()
|
||
wt = self.worktree()
|
||
self.git(wt, "submodule", "update", "--init", "--recursive")
|
||
(self.repo / "tracked").write_text("implementation done\n")
|
||
self.git(self.repo, "add", "-A")
|
||
self.git(self.repo, "commit", "-q", "-m", "done")
|
||
self.git(self.repo, "push", "-q", "origin", "feature/guard")
|
||
self.git(self.repo, "merge", "--no-ff", "-q", "-m", "Merge feature/guard", "feature/guard")
|
||
self.git(self.repo, "push", "-q", "origin", "main")
|
||
result = self.task(wt, "finish", check=False)
|
||
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
||
self.assertFalse(wt.exists())
|
||
|
||
def test_local_unique_commit_still_preserved(self):
|
||
self.add_submodule_pinned_off_main()
|
||
wt = self.worktree()
|
||
self.git(wt, "submodule", "update", "--init", "--recursive")
|
||
child = wt / "component"
|
||
pinned = subprocess.run(["git", "-C", str(child), "rev-parse", "HEAD"],
|
||
capture_output=True, text=True, check=True).stdout.strip()
|
||
self.git(child, "switch", "-c", "feature/unique")
|
||
(child / "content").write_text("unique local work\n")
|
||
self.git(child, "commit", "-am", "unique")
|
||
unique = subprocess.run(["git", "-C", str(child), "rev-parse", "HEAD"],
|
||
capture_output=True, text=True, check=True).stdout.strip()
|
||
self.git(child, "checkout", "--detach", pinned)
|
||
(self.repo / "tracked").write_text("done\n")
|
||
self.git(self.repo, "add", "-A")
|
||
self.git(self.repo, "commit", "-q", "-m", "done")
|
||
self.git(self.repo, "push", "-q", "origin", "feature/guard")
|
||
self.git(self.repo, "merge", "--no-ff", "-q", "-m", "Merge feature/guard", "feature/guard")
|
||
self.git(self.repo, "push", "-q", "origin", "main")
|
||
result = self.task(wt, "finish", check=False)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertTrue(wt.exists())
|
||
kept = subprocess.run(["git", "-C", str(child), "rev-parse", "feature/unique"],
|
||
capture_output=True, text=True, check=True).stdout.strip()
|
||
self.assertEqual(kept, unique)
|