weichen
c1fdf0141c
Why: main advanced with merged-task-name and legacy evidence guards during this task. Evidence: SELF 44 lifecycle regressions pass after resolving both hunks; independent review repairs are recorded in the task card. Impact: retains both task-start protections and the stronger JSON/legacy receipt reader, without rewriting either branch history. Refs: Dev PR #211 and #212; .task/feature-governance-reliability.md
636 lines
33 KiB
Python
636 lines
33 KiB
Python
"""Lifecycle regressions against real disposable Git repositories and remotes."""
|
||
import os
|
||
import json
|
||
import hashlib
|
||
from pathlib import Path
|
||
import shutil
|
||
import shlex
|
||
import subprocess
|
||
import tempfile
|
||
import unittest
|
||
|
||
|
||
SOURCE = Path(__file__).resolve().parents[1]
|
||
|
||
|
||
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.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"})
|
||
for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GITEA_TOKEN", "PRECHECK_RECEIPT",
|
||
"TASK_PARENT_BRANCH", "TASK_PARENT_PR", "FORCE_ABANDON", "TASK_TARGET_BRANCH"):
|
||
self.env.pop(key, None)
|
||
self.git(self.root, "init", "--bare", "--initial-branch=main", str(self.remote))
|
||
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"}
|
||
return self.run_cmd(cwd, "bash", str(self.repo / "scripts/task.sh"), command, *args, **kwargs)
|
||
|
||
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 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_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_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")
|
||
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)
|
||
receipt = json.loads(record)
|
||
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 / "precheck.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 test_start_requires_precheck_receipt_before_creating_worktree(self):
|
||
result = self.task(self.repo, "start", "feature/new", check=False)
|
||
self.assertNotEqual(result.returncode, 0, result.stdout)
|
||
self.assertFalse((self.root / "repo-wt-feature-new").exists(), result.stdout)
|
||
|
||
def test_start_preserves_merged_local_and_remote_branch_names(self):
|
||
receipt = self.capture_precheck()
|
||
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)})
|
||
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()
|
||
self.task(self.repo, "start", "feature/new", env=self.env | {"PRECHECK_RECEIPT": str(receipt)})
|
||
wt = self.root / "repo-wt-feature-new"
|
||
self.assertTrue((wt / ".task/feature-new.md").is_file())
|
||
self.assertIn("make precheck", (wt / ".task/feature-new.md").read_text())
|
||
body = self.git(wt, "log", "-1", "--format=%B").stdout
|
||
for field in ("Why:", "Evidence:", "Impact:", "Refs:"):
|
||
self.assertIn(field, body)
|
||
|
||
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)
|
||
env = self.env | {"PRECHECK_RECEIPT": str(receipt), "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):
|
||
self.git(self.repo, "remote", "set-url", "origin", str(self.root / "missing.git"))
|
||
result = self.task(self.repo, "start", "feature/offline", check=False)
|
||
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()
|
||
self.advance_remote()
|
||
result = self.task(self.repo, "start", "feature/stale", check=False,
|
||
env=self.env | {"PRECHECK_RECEIPT": str(receipt)})
|
||
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)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|