weichen/Agent-System-Dev
1
0
Fork 0
Code Issues 1 Pull Requests 7 Actions 4 Packages Projects Releases Wiki Activity
Files
976d2e84434f6a089cb8ccd1da33e151a6cc375d
Agent-System-Dev/tests/test_task.py
T
weichen 8af945a1d5 Make development gates preserve work and verifiable evidence 
Why: cleanup, component synchronization, receipt recording and credential handling could contradict the documented development boundaries.

Evidence: SELF 75 isolated governance tests and source checks are captured in .task/feature-governance-reliability.md; combination and CI verification follow.

Impact: hardens local task and CI tooling, aligns policy and component documentation, and adds continuous regression tests without changing product behavior.

Refs: .task/feature-governance-reliability.md; policy/governance.md; policy/platform-admission.md
2026-10-05 12:36:02 +00:00

407 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Lifecycle regressions against real disposable Git repositories and remotes."""
import os
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 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_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 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 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_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()
Reference in New Issue View Git Blame Copy Permalink