"""Regression test suite for native author issue worktree bootstrap (#850).""" from __future__ import annotations import json import os import shutil import subprocess import tempfile import unittest import author_issue_bootstrap import task_capability_map class TestAuthorIssueBootstrap(unittest.TestCase): """Test suite covering AC1-AC10 and comment #14959 specification.""" def setUp(self): self.tmp_dir = tempfile.mkdtemp(prefix="test_bootstrap_") self.repo_dir = os.path.join(self.tmp_dir, "repo") os.makedirs(self.repo_dir) # Initialize synthetic git repo subprocess.run(["git", "init", "-b", "master"], cwd=self.repo_dir, check=True, capture_output=True) subprocess.run(["git", "config", "user.name", "Test User"], cwd=self.repo_dir, check=True) subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=self.repo_dir, check=True) readme = os.path.join(self.repo_dir, "README.md") with open(readme, "w", encoding="utf-8") as f: f.write("# Test Repo\n") subprocess.run(["git", "add", "README.md"], cwd=self.repo_dir, check=True, capture_output=True) subprocess.run(["git", "commit", "-m", "initial commit"], cwd=self.repo_dir, check=True, capture_output=True) rev_res = subprocess.run(["git", "rev-parse", "HEAD"], cwd=self.repo_dir, capture_output=True, text=True, check=True) self.master_sha = rev_res.stdout.strip() self.branches_dir = os.path.join(self.repo_dir, "branches") os.makedirs(self.branches_dir, exist_ok=True) self.lock_dir = os.path.join(self.tmp_dir, "locks") os.makedirs(self.lock_dir, exist_ok=True) self.journal_dir = os.path.join(self.tmp_dir, "journals") os.makedirs(self.journal_dir, exist_ok=True) os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"] = self.journal_dir def tearDown(self): os.environ.pop("GITEA_BOOTSTRAP_JOURNAL_DIR", None) shutil.rmtree(self.tmp_dir, ignore_errors=True) def test_bootstrap_success_path(self): """AC1/AC3/AC8: Successful bootstrap creates branch, worktree, registration, and lock proof.""" key = "test_key_success_1" res = author_issue_bootstrap.bootstrap_author_issue_worktree( issue_number=850, canonical_repo_root=self.repo_dir, assignment_id="asn-12345", lease_id="lease-67890", expected_base_sha=self.master_sha, idempotency_key=key, remote="prgs", lock_dir=self.lock_dir, ) self.assertTrue(res.get("success"), f"Bootstrap failed: {res}") self.assertFalse(res.get("replayed")) self.assertEqual(res.get("issue_number"), 850) self.assertEqual(res.get("base_sha"), self.master_sha) self.assertIn("branches/fix-issue-850-native-mcp-bootstrap", res.get("worktree_path")) # Verify worktree directory exists and is registered worktree_path = res["worktree_path"] self.assertTrue(os.path.isdir(worktree_path)) wt_list = subprocess.run(["git", "-C", self.repo_dir, "worktree", "list"], capture_output=True, text=True, check=True) self.assertIn(worktree_path, wt_list.stdout) # Verify phase journal written journal = author_issue_bootstrap.load_phase_journal(key) self.assertIsNotNone(journal) self.assertTrue(journal.get("completed")) self.assertEqual(journal.get("current_phase"), author_issue_bootstrap.PHASE_7_TRANSITION_COMPLETED) def test_idempotent_replay(self): """Item 2: Replaying with identical key returns cached transition without duplicate creation.""" key = "test_key_idempotent_1" res1 = author_issue_bootstrap.bootstrap_author_issue_worktree( issue_number=850, canonical_repo_root=self.repo_dir, idempotency_key=key, lock_dir=self.lock_dir, ) self.assertTrue(res1["success"], f"res1 failed: {res1}") self.assertFalse(res1.get("replayed")) # Second call res2 = author_issue_bootstrap.bootstrap_author_issue_worktree( issue_number=850, canonical_repo_root=self.repo_dir, idempotency_key=key, lock_dir=self.lock_dir, ) self.assertTrue(res2["success"], f"res2 failed: {res2}") self.assertTrue(res2.get("replayed")) self.assertEqual(res1["worktree_path"], res2["worktree_path"]) def test_stale_concurrency_pin_refusal(self): """Item 3: Mismatched expected base SHA fails closed without silent rebasing.""" stale_sha = "0000000000000000000000000000000000000000" res = author_issue_bootstrap.bootstrap_author_issue_worktree( issue_number=850, canonical_repo_root=self.repo_dir, expected_base_sha=stale_sha, lock_dir=self.lock_dir, ) self.assertFalse(res["success"]) self.assertEqual(res.get("reason_code"), "stale_concurrency_pin") self.assertIn("exact_next_action", res) def test_path_outside_branches_root_refusal(self): """Item 6: Worktree path outside branches/ root is refused.""" outside_path = os.path.join(self.tmp_dir, "outside_worktree") res = author_issue_bootstrap.bootstrap_author_issue_worktree( issue_number=850, canonical_repo_root=self.repo_dir, worktree_path=outside_path, lock_dir=self.lock_dir, ) self.assertFalse(res["success"]) self.assertEqual(res.get("reason_code"), "path_outside_canonical_branches_root") def test_preexisting_dirty_worktree_preservation(self): """Item 6: Preexisting dirty worktree fails closed and is NOT modified or cleaned.""" branch = "fix/issue-850-dirty-test" wt_path = os.path.join(self.branches_dir, "fix-issue-850-dirty-test") subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", "-b", branch, wt_path], check=True, capture_output=True) # Create dirty untracked file dirty_file = os.path.join(wt_path, "dirty.txt") with open(dirty_file, "w") as f: f.write("dirty edits\n") res = author_issue_bootstrap.bootstrap_author_issue_worktree( issue_number=850, canonical_repo_root=self.repo_dir, branch_name=branch, worktree_path=wt_path, lock_dir=self.lock_dir, ) self.assertFalse(res["success"]) self.assertEqual(res.get("reason_code"), "preexisting_dirty_worktree") # Prove dirty file is preserved byte-for-byte self.assertTrue(os.path.exists(dirty_file)) with open(dirty_file, "r") as f: self.assertEqual(f.read(), "dirty edits\n") def test_compensating_recovery_on_failed_phase(self): """AC4/Item 4: Failure during transition rolls back ONLY newly created artifacts.""" key = "test_key_recovery_1" # Simulate partial progress in journal journal = { "idempotency_key": key, "issue_number": 850, "branch_name": "fix/issue-850-recovery-test", "worktree_path": os.path.join(self.branches_dir, "fix-issue-850-recovery-test"), "artifacts_created": { "branch_created": True, "worktree_dir_created": True, "worktree_registered": True, "lock_created": False, }, "failure_reason": "simulated lock failure", "current_phase": author_issue_bootstrap.PHASE_5_REGISTRATION_VERIFIED, "completed": False, } # Create the branch and worktree manually to simulate partial state subprocess.run(["git", "-C", self.repo_dir, "branch", journal["branch_name"]], check=True, capture_output=True) subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", journal["worktree_path"], journal["branch_name"]], check=True, capture_output=True) # Run compensating recovery rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir) self.assertTrue(rec["executed"]) self.assertIn(f"worktree_path:{journal['worktree_path']}", rec["rolled_back"]) self.assertIn(f"branch:{journal['branch_name']}", rec["rolled_back"]) # Prove worktree directory and branch were rolled back self.assertFalse(os.path.exists(journal["worktree_path"])) branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", journal["branch_name"]], capture_output=True, text=True, check=False) self.assertNotEqual(branch_check.returncode, 0) def test_task_capability_map_integration(self): """Verify task_capability_map has bootstrap_author_issue_worktree configured correctly.""" self.assertEqual(task_capability_map.required_role("bootstrap_author_issue_worktree"), "author") self.assertEqual(task_capability_map.required_permission("bootstrap_author_issue_worktree"), "gitea.branch.create") self.assertTrue(task_capability_map.preflight_task_matches("work_issue", "bootstrap_author_issue_worktree")) self.assertTrue(task_capability_map.preflight_task_matches("bootstrap_author_issue_worktree", "lock_issue")) if __name__ == "__main__": unittest.main()