"""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 from unittest import mock import author_issue_bootstrap import task_capability_map def _concurrent_bootstrap_worker(args: tuple[str, int, str, str, str, str]) -> dict: repo_dir, issue_num, key, lock_dir, journal_dir, master_sha = args os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"] = journal_dir return author_issue_bootstrap.bootstrap_author_issue_worktree( issue_number=issue_num, canonical_repo_root=repo_dir, expected_base_sha=master_sha, idempotency_key=key, lock_dir=lock_dir, owner_session="session-concurrent-test", active_identity="jcwalker3", active_profile="prgs-author", ) 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, owner_session="session-test-1234", ) 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, journal_dir=self.lock_dir) 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, owner_session="session-test-1234", ) 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, owner_session="session-test-1234", ) 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, owner_session="session-test-1234", ) 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, owner_session="session-test-1234", ) 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, owner_session="session-test-1234", ) 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_cross_process_concurrency(self): """Review #525 Finding 1: Genuine cross-process concurrency locking prevents corruption.""" import concurrent.futures key = "test_concurrent_key_850" args = (self.repo_dir, 850, key, self.lock_dir, self.journal_dir, self.master_sha) with concurrent.futures.ProcessPoolExecutor(max_workers=2) as executor: fut1 = executor.submit(_concurrent_bootstrap_worker, args) fut2 = executor.submit(_concurrent_bootstrap_worker, args) res1 = fut1.result(timeout=10) res2 = fut2.result(timeout=10) self.assertTrue(res1["success"], f"res1 failed: {res1}") self.assertTrue(res2["success"], f"res2 failed: {res2}") # One process performs creation, the other process receives idempotent replay replayed_count = sum(1 for r in (res1, res2) if r.get("replayed")) created_count = sum(1 for r in (res1, res2) if not r.get("replayed")) self.assertEqual(replayed_count, 1) self.assertEqual(created_count, 1) self.assertEqual(res1["worktree_path"], res2["worktree_path"]) def test_interrupted_replay_preserves_artifacts_created_provenance(self): """Review #525 Finding 2: Replaying incomplete journal preserves creation provenance monotonically.""" key = "test_key_interrupted_replay_1" branch = "fix/issue-850-interrupted-replay" wt_path = os.path.join(self.branches_dir, "fix-issue-850-interrupted-replay") # Simulate Phase 2/3 completion where branch and worktree directory were created by this transition journal = { "idempotency_key": key, "issue_number": 850, "branch_name": branch, "worktree_path": wt_path, "active_identity": "jcwalker3", "active_profile": "prgs-author", "remote": "prgs", "org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools", "phases": { author_issue_bootstrap.PHASE_1_REQUEST_ACCEPTED: {"status": "completed"}, author_issue_bootstrap.PHASE_2_BRANCH_CONFIRMED: {"status": "completed", "created": True}, }, "artifacts_created": { "branch_created": True, "worktree_dir_created": True, "worktree_registered": True, "lock_created": False, }, "current_phase": author_issue_bootstrap.PHASE_3_PATH_RESERVED, "completed": False, } # Pre-create the branch and worktree on disk to simulate partial state after crash subprocess.run(["git", "-C", self.repo_dir, "branch", branch, self.master_sha], check=True, capture_output=True) subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", wt_path, branch], check=True, capture_output=True) author_issue_bootstrap.save_phase_journal(journal, journal_dir=self.lock_dir) # Now resume/replay the transition but simulate lock binding failure during Phase 6 with mock.patch("issue_lock_store.bind_session_lock", side_effect=RuntimeError("Lock failure test")): res = author_issue_bootstrap.bootstrap_author_issue_worktree( issue_number=850, canonical_repo_root=self.repo_dir, branch_name=branch, worktree_path=wt_path, idempotency_key=key, lock_dir=self.lock_dir, owner_session="session-test-1234", ) self.assertFalse(res["success"]) self.assertEqual(res.get("reason_code"), "issue_lock_acquisition_failed") # Verify that compensating recovery correctly deleted transition-created branch & worktree # because creation provenance was preserved across replay (NOT downgraded to False!) self.assertFalse(os.path.exists(wt_path)) branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", branch], capture_output=True, text=True, check=False) self.assertNotEqual(branch_check.returncode, 0) def test_transition_created_only_compensation(self): """Review #525 Finding 4: Preexisting branch is NOT deleted by compensation when only worktree was transition-created.""" key = "test_key_preexisting_branch_compensation" preexisting_branch = "fix/issue-850-preexisting" wt_path = os.path.join(self.branches_dir, "fix-issue-850-preexisting") # Create branch BEFORE bootstrap (preexisting branch) subprocess.run(["git", "-C", self.repo_dir, "branch", preexisting_branch, self.master_sha], check=True, capture_output=True) # Call bootstrap with simulated failure during Phase 6 (lock binding) with mock.patch("issue_lock_store.bind_session_lock", side_effect=RuntimeError("Simulated lock failure")): res = author_issue_bootstrap.bootstrap_author_issue_worktree( issue_number=850, canonical_repo_root=self.repo_dir, branch_name=preexisting_branch, worktree_path=wt_path, idempotency_key=key, lock_dir=self.lock_dir, owner_session="session-test-1234", ) self.assertFalse(res["success"]) # Worktree dir was created by transition -> removed by compensation self.assertFalse(os.path.exists(wt_path)) # Preexisting branch was NOT created by transition -> MUST BE PRESERVED! branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", preexisting_branch], capture_output=True, text=True, check=False) self.assertEqual(branch_check.returncode, 0, "Preexisting branch was deleted by mistake!") def test_incompatible_idempotency_replay_refusal(self): """Review #525 Finding 4: Replaying key with incompatible parameters returns refusal.""" key = "test_key_incompatible_replay" res1 = author_issue_bootstrap.bootstrap_author_issue_worktree( issue_number=850, canonical_repo_root=self.repo_dir, branch_name="fix/issue-850-param-a", idempotency_key=key, lock_dir=self.lock_dir, owner_session="session-test-1234", ) self.assertTrue(res1["success"]) # Second call with different branch_name res2 = author_issue_bootstrap.bootstrap_author_issue_worktree( issue_number=850, canonical_repo_root=self.repo_dir, branch_name="fix/issue-850-param-b", idempotency_key=key, lock_dir=self.lock_dir, owner_session="session-test-1234", ) self.assertFalse(res2["success"]) self.assertEqual(res2.get("reason_code"), "incompatible_idempotency_replay") def test_exact_next_action_satisfiable_via_mcp(self): """Review #525 Finding 4: exact_next_action provides satisfiable MCP actions, not shell commands.""" key = "test_key_next_action_mcp" 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, owner_session="session-test-1234", ) next_action = res.get("exact_next_action", "") self.assertNotIn("scripts/worktree-start", next_action) self.assertNotIn("git worktree add", next_action) self.assertNotIn("bash", next_action.lower()) def test_missing_owner_session_refusal(self): """Finding D: Missing owner_session context fails closed with typed refusal and zero mutation.""" res = author_issue_bootstrap.bootstrap_author_issue_worktree( issue_number=850, canonical_repo_root=self.repo_dir, owner_session=None, lock_dir=self.lock_dir, ) self.assertFalse(res["success"]) self.assertEqual(res.get("reason_code"), "missing_owner_session") self.assertIn("exact_next_action", res) def test_symlink_lock_file_refusal(self): """Finding C: BootstrapTransitionLock refuses to follow symlinks.""" key = "test_symlink_lock_key" safe_key = "".join(c if c.isalnum() or c in ("-", "_", ".") else "_" for c in key) lock_path = os.path.join(self.lock_dir, f"{safe_key}.lock") target_file = os.path.join(self.tmp_dir, "fake_target") with open(target_file, "w") as f: f.write("target") os.symlink(target_file, lock_path) with self.assertRaises(RuntimeError) as ctx: with author_issue_bootstrap.BootstrapTransitionLock(key, journal_dir=self.lock_dir): pass self.assertIn("symlink", str(ctx.exception).lower()) def test_lock_directory_escape_refusal(self): """Finding C: BootstrapTransitionLock refuses keys that escape lock directory.""" with mock.patch("os.path.abspath", return_value="/tmp/outside/evil_key.lock"): with self.assertRaises(RuntimeError) as ctx: author_issue_bootstrap.BootstrapTransitionLock("key", journal_dir=self.lock_dir) self.assertIn("escapes", str(ctx.exception).lower()) def test_missing_active_identity_refusal(self): """F-5: Missing active_identity parameter fails closed.""" res = author_issue_bootstrap.bootstrap_author_issue_worktree( issue_number=850, canonical_repo_root=self.repo_dir, owner_session="session-test-1234", active_identity=None, active_profile="prgs-author", lock_dir=self.lock_dir, ) self.assertFalse(res["success"]) self.assertEqual(res.get("reason_code"), "missing_active_identity") def test_missing_active_profile_refusal(self): """F-5: Missing active_profile parameter fails closed.""" res = author_issue_bootstrap.bootstrap_author_issue_worktree( issue_number=850, canonical_repo_root=self.repo_dir, owner_session="session-test-1234", active_identity="jcwalker3", active_profile=None, lock_dir=self.lock_dir, ) self.assertFalse(res["success"]) self.assertEqual(res.get("reason_code"), "missing_active_profile") def test_dirty_worktree_preserved_during_recovery(self): """F-4: Compensating recovery does not delete dirty worktree.""" branch = "fix/issue-850-rec-dirty" wt_path = os.path.join(self.branches_dir, "fix-issue-850-rec-dirty") subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", "-b", branch, wt_path], check=True, capture_output=True) dirty_file = os.path.join(wt_path, "dirty.txt") with open(dirty_file, "w") as f: f.write("uncommitted work") journal = { "idempotency_key": "test_dirty_rec", "issue_number": 850, "branch_name": branch, "worktree_path": wt_path, "artifacts_created": { "worktree_dir_created": True, "worktree_registered": True, }, "failure_reason": "test dirty recovery", } rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir, journal_dir=self.lock_dir) self.assertTrue(os.path.exists(wt_path)) self.assertIn(f"worktree_path_preserved_dirty:{wt_path}", rec["rolled_back"]) def test_branch_with_commits_preserved_during_recovery(self): """F-4: Compensating recovery does not delete branch with author commits.""" branch = "fix/issue-850-rec-commits" subprocess.run(["git", "-C", self.repo_dir, "branch", branch, self.master_sha], check=True, capture_output=True) # Add a commit on the branch wt_path = os.path.join(self.branches_dir, "fix-issue-850-rec-commits") subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", wt_path, branch], check=True, capture_output=True) cfile = os.path.join(wt_path, "commit.txt") with open(cfile, "w") as f: f.write("author commit") subprocess.run(["git", "-C", wt_path, "add", "commit.txt"], check=True, capture_output=True) subprocess.run(["git", "-C", wt_path, "commit", "-m", "author commit"], check=True, capture_output=True) subprocess.run(["git", "-C", self.repo_dir, "worktree", "remove", "--force", wt_path], check=True, capture_output=True) journal = { "idempotency_key": "test_commits_rec", "issue_number": 850, "branch_name": branch, "resolved_base_sha": self.master_sha, "artifacts_created": { "branch_created": True, }, "failure_reason": "test commit branch recovery", } rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir, journal_dir=self.lock_dir) branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", branch], capture_output=True, text=True, check=False) self.assertEqual(branch_check.returncode, 0, "Branch with commits was deleted!") self.assertIn(f"branch_preserved_commits:{branch}", rec["rolled_back"]) 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()