"""Synthetic regression coverage for dirty orphaned worktree recovery (#860). Modeled on the #850 / #855 shape without mutating their real state. """ from __future__ import annotations import json import os import shutil import tempfile import unittest from unittest import mock import dirty_orphan_worktree_recovery as dorec import issue_lock_store DEAD_PID = 999_999_999 LIVE_PID = os.getpid() BRANCH = "fix/issue-901-dirty-orphan" SOURCE_WT = "/repo/branches/issue-901-dirty-orphan" RECOVERY_WT_NAME = "recovery-issue-901-dirty-orphan" LOCAL_HEAD = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" REMOTE_HEAD = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" OTHER_HEAD = "cccccccccccccccccccccccccccccccccccccccc" FP_A = dorec.sha256_bytes(b"dirty-a") FP_B = dorec.sha256_bytes(b"dirty-b") FP_C = dorec.sha256_bytes(b"dirty-c-conflict") def durable_lock(**overrides): """#850-shaped PID-less malformed same-claimant lock.""" lock = { "issue_number": 901, "branch_name": BRANCH, "worktree_path": SOURCE_WT, "remote": "prgs", "org": "Example-Org", "repo": "Example-Repo", # intentionally no pid / session_pid / work_lease expiry "claimant": {"username": "author-user", "profile": "prgs-author"}, } lock.update(overrides) return lock def base_kwargs(**overrides): kwargs = { "issue_number": 901, "branch_name": BRANCH, "source_worktree_path": SOURCE_WT, "remote": "prgs", "org": "Example-Org", "repo": "Example-Repo", "identity": "author-user", "profile": "prgs-author", "expected_local_head": LOCAL_HEAD, "expected_remote_head": REMOTE_HEAD, "expected_dirty_fingerprints": {"a.py": FP_A, "b.py": FP_B}, "current_branch": BRANCH, "porcelain_status": " M a.py\n M b.py\n", "observed_local_head": LOCAL_HEAD, "observed_remote_head": REMOTE_HEAD, "observed_dirty_fingerprints": {"a.py": FP_A, "b.py": FP_B}, "competing_live_locks": [], "competing_live_sessions": [], "workflow_lease_active": False, "workflow_lease_expired": True, "canonical_repo_root": "/repo", "worktree_registered": True, "current_pid": LIVE_PID, } kwargs.update(overrides) return kwargs def assess(lock=None, **overrides): return dorec.assess_dirty_orphan_recovery( durable_lock() if lock is None else lock, **base_kwargs(**overrides) ) class FreshnessPidLess(unittest.TestCase): def test_pid_less_lock_is_not_live(self): freshness = issue_lock_store.assess_lock_freshness(durable_lock()) self.assertFalse(freshness["live"]) self.assertTrue(freshness.get("pid_missing")) self.assertEqual(freshness["status"], "malformed") def test_pid_less_with_far_future_expiry_still_not_live(self): lock = durable_lock( work_lease={ "operation_type": "author_issue_work", "expires_at": "2999-01-01T00:00:00Z", "last_heartbeat_at": "2999-01-01T00:00:00Z", } ) freshness = issue_lock_store.assess_lock_freshness(lock) self.assertFalse(freshness["live"]) self.assertTrue(freshness.get("pid_missing")) class EligibilityGranted(unittest.TestCase): def test_dead_same_claimant_pid_less_dirty(self): result = assess() self.assertEqual(result["outcome"], dorec.ELIGIBLE) self.assertTrue(result["eligible"]) def test_expired_workflow_lease_corroboration(self): result = assess(workflow_lease_active=False, workflow_lease_expired=True) self.assertTrue(result["eligible"]) def test_older_local_newer_remote_heads(self): result = assess() self.assertTrue(result["evidence"].get("heads_diverged")) self.assertTrue(result["eligible"]) class EligibilityRefused(unittest.TestCase): def test_active_owner_with_pid(self): lock = durable_lock(pid=LIVE_PID, session_pid=LIVE_PID) result = assess(lock=lock, owner_process_alive_override=True) self.assertEqual(result["outcome"], dorec.REFUSED) self.assertFalse(result["eligible"]) self.assertTrue(any("alive" in r for r in result["reasons"])) def test_foreign_claimant(self): result = assess(identity="other-user") self.assertEqual(result["outcome"], dorec.REFUSED) self.assertTrue(any("foreign claimant identity" in r for r in result["reasons"])) def test_foreign_profile(self): result = assess(profile="prgs-reviewer") self.assertEqual(result["outcome"], dorec.REFUSED) def test_fingerprint_mismatch(self): result = assess(observed_dirty_fingerprints={"a.py": "0" * 64, "b.py": FP_B}) self.assertEqual(result["outcome"], dorec.REFUSED) self.assertTrue(any("fingerprint mismatch" in r for r in result["reasons"])) def test_head_mismatch(self): result = assess(observed_local_head=OTHER_HEAD) self.assertEqual(result["outcome"], dorec.REFUSED) def test_remote_head_mismatch(self): result = assess(observed_remote_head=OTHER_HEAD) self.assertEqual(result["outcome"], dorec.REFUSED) def test_path_not_under_branches(self): result = assess( source_worktree_path="/tmp/branches/evil", # lock path also changed so worktree agreement holds lock=durable_lock(worktree_path="/tmp/branches/evil"), ) self.assertEqual(result["outcome"], dorec.REFUSED) self.assertTrue(any("canonical branches" in r for r in result["reasons"])) def test_unregistered_worktree(self): result = assess(worktree_registered=False) self.assertEqual(result["outcome"], dorec.REFUSED) def test_active_workflow_lease(self): result = assess(workflow_lease_active=True, workflow_lease_expired=False) self.assertEqual(result["outcome"], dorec.REFUSED) def test_unsafe_dirty_path_pin(self): result = assess( expected_dirty_fingerprints={"../etc/passwd": FP_A}, observed_dirty_fingerprints={"../etc/passwd": FP_A}, ) self.assertEqual(result["outcome"], dorec.REFUSED) def test_symlink_escape_rejected_by_ancestry(self): ok, reasons = dorec.is_path_under_canonical_branches( "/tmp/branches/evil", canonical_repo_root="/repo" ) self.assertFalse(ok) self.assertTrue(reasons) class ConflictDetection(unittest.TestCase): def test_overlapping_upstream_change(self): conflicts = dorec.detect_path_conflicts( dirty_paths=["c.py"], local_head_contents={"c.py": b"local-base"}, remote_head_contents={"c.py": b"remote-changed"}, dirty_contents={"c.py": b"dirty-c-conflict"}, ) self.assertEqual(len(conflicts), 1) self.assertEqual(conflicts[0]["path"], "c.py") def test_unchanged_upstream_no_conflict(self): conflicts = dorec.detect_path_conflicts( dirty_paths=["a.py"], local_head_contents={"a.py": b"same"}, remote_head_contents={"a.py": b"same"}, dirty_contents={"a.py": b"dirty-a"}, ) self.assertEqual(conflicts, []) class CrashSafeRecovery(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp(prefix="dirty-orphan-") self.repo = os.path.join(self.tmp, "repo") self.branches = os.path.join(self.repo, "branches") self.source = os.path.join(self.branches, "issue-901-dirty-orphan") self.recovery = os.path.join(self.branches, RECOVERY_WT_NAME) os.makedirs(self.source, exist_ok=True) os.makedirs(self.branches, exist_ok=True) # seed dirty files in source with open(os.path.join(self.source, "a.py"), "wb") as fh: fh.write(b"dirty-a") with open(os.path.join(self.source, "b.py"), "wb") as fh: fh.write(b"dirty-b") self.journal_dir = os.path.join(self.tmp, "journals") self.lock = durable_lock(worktree_path=self.source) self.assessment = dorec.assess_dirty_orphan_recovery( self.lock, **base_kwargs( source_worktree_path=self.source, canonical_repo_root=self.repo, ), ) class FakeGit(dorec.GitOps): def __init__(self, recovery_path, head): self.recovery_path = recovery_path self.head = head self.calls = [] def run(self, args, *, cwd): self.calls.append((args, cwd)) if args[:3] == ["git", "worktree", "add"]: os.makedirs(self.recovery_path, exist_ok=True) return mock.Mock(returncode=0, stdout="", stderr="") if args[:2] == ["git", "checkout"]: return mock.Mock(returncode=0, stdout="", stderr="") if args[:2] == ["git", "rev-parse"]: return mock.Mock(returncode=0, stdout=self.head + "\n", stderr="") return mock.Mock(returncode=0, stdout="", stderr="") self.git = FakeGit(self.recovery, REMOTE_HEAD) self.written_locks = [] def lock_writer(record): self.written_locks.append(record) self.lock_writer = lock_writer def tearDown(self): shutil.rmtree(self.tmp, ignore_errors=True) def _run(self, **overrides): kwargs = { "assessment": self.assessment, "existing_lock": self.lock, "issue_number": 901, "branch_name": BRANCH, "source_worktree_path": self.source, "recovery_worktree_path": self.recovery, "remote": "prgs", "org": "Example-Org", "repo": "Example-Repo", "identity": "author-user", "profile": "prgs-author", "expected_local_head": LOCAL_HEAD, "expected_remote_head": REMOTE_HEAD, "expected_dirty_fingerprints": {"a.py": FP_A, "b.py": FP_B}, "dirty_contents": {"a.py": b"dirty-a", "b.py": b"dirty-b"}, "local_head_contents": {"a.py": b"base-a", "b.py": b"base-b"}, "remote_head_contents": {"a.py": b"base-a", "b.py": b"base-b"}, "canonical_repo_root": self.repo, "bind_lock": True, "lock_writer": self.lock_writer, "git_ops": self.git, "journal_dir": self.journal_dir, "session_pid": LIVE_PID, } kwargs.update(overrides) return dorec.run_dirty_orphan_recovery(**kwargs) def test_success_preserves_dirty_bytes_and_source(self): result = self._run() self.assertTrue(result["success"]) self.assertEqual(result["outcome"], dorec.RECOVERY_COMPLETED) self.assertTrue(os.path.isdir(self.source)) with open(os.path.join(self.source, "a.py"), "rb") as fh: self.assertEqual(fh.read(), b"dirty-a") with open(os.path.join(self.recovery, "a.py"), "rb") as fh: self.assertEqual(fh.read(), b"dirty-a") with open(os.path.join(self.recovery, "b.py"), "rb") as fh: self.assertEqual(fh.read(), b"dirty-b") self.assertEqual(len(self.written_locks), 1) rec = self.written_locks[0] self.assertEqual(rec["session_pid"], LIVE_PID) self.assertTrue(rec["dirty_orphan_recovery"]["recovered"]) self.assertTrue(rec["dirty_orphan_recovery"]["source_frozen"]) def test_conflict_leaves_governed_state(self): result = self._run( expected_dirty_fingerprints={"c.py": FP_C}, dirty_contents={"c.py": b"dirty-c-conflict"}, local_head_contents={"c.py": b"local-base"}, remote_head_contents={"c.py": b"remote-changed"}, ) self.assertTrue(result["success"]) self.assertEqual(result["outcome"], dorec.CONFLICTS_PRESENT) sidecar = os.path.join(self.recovery, "c.py.recovered-dirty") self.assertTrue(os.path.isfile(sidecar)) state = os.path.join( self.recovery, dorec.CONFLICT_STATE_DIR, dorec.CONFLICT_STATE_FILE ) self.assertTrue(os.path.isfile(state)) with open(state, "r", encoding="utf-8") as fh: payload = json.load(fh) self.assertEqual(payload["resolution"], "author_edit_required") def test_interrupt_before_journal_no_artifacts(self): result = self._run(interrupt_after_phase=dorec.PHASE_ELIGIBILITY) self.assertFalse(result["success"]) self.assertEqual(result["outcome"], "INTERRUPTED") self.assertFalse(os.path.isdir(self.recovery)) def test_interrupt_after_journal_then_retry_idempotent(self): first = self._run(interrupt_after_phase=dorec.PHASE_JOURNAL_PERSISTED) self.assertEqual(first["outcome"], "INTERRUPTED") self.assertTrue(first["journal"]["artifacts_created"]["journal"]) second = self._run() self.assertTrue(second["success"]) # source still recoverable with open(os.path.join(self.source, "a.py"), "rb") as fh: self.assertEqual(fh.read(), b"dirty-a") def test_interrupt_after_worktree_then_retry(self): first = self._run(interrupt_after_phase=dorec.PHASE_RECOVERY_WORKTREE) self.assertEqual(first["outcome"], "INTERRUPTED") self.assertTrue(os.path.isdir(self.recovery)) second = self._run() self.assertTrue(second["success"]) def test_interrupt_after_binding_then_retry_complete(self): first = self._run(interrupt_after_phase=dorec.PHASE_BINDING) self.assertEqual(first["outcome"], "INTERRUPTED") second = self._run() self.assertTrue(second["success"]) # completed journal makes further retries no-ops third = self._run() self.assertEqual(third["outcome"], dorec.RECOVERY_RESUMED) def test_source_worktree_never_deleted(self): self._run() self.assertTrue(os.path.isdir(self.source)) self.assertTrue(os.path.isfile(os.path.join(self.source, "a.py"))) def test_fingerprint_drift_refuses_without_mutation(self): result = self._run(dirty_contents={"a.py": b"CHANGED", "b.py": b"dirty-b"}) self.assertFalse(result["success"]) self.assertFalse(os.path.isdir(self.recovery)) class SessionBindingPreflight(unittest.TestCase): def test_canonical_session_binding_recognized(self): lock = { "worktree_path": "/repo/branches/recovery", "session_pid": LIVE_PID, "dirty_orphan_recovery": { "recovered": True, "conflicts": [], "recovery_worktree_path": "/repo/branches/recovery", "source_worktree_path": SOURCE_WT, "accepted_head": REMOTE_HEAD, }, } result = dorec.preflight_recognizes_recovered_provenance(lock) self.assertTrue(result["recognized"]) def test_conflicts_block_commit_preflight(self): lock = { "worktree_path": "/repo/branches/recovery", "session_pid": LIVE_PID, "dirty_orphan_recovery": { "recovered": True, "conflicts": [{"path": "c.py"}], }, } result = dorec.preflight_recognizes_recovered_provenance(lock) self.assertFalse(result["recognized"]) def test_active_foreign_does_not_mutate(self): # assess-only path: foreign refused before run result = assess(identity="intruder") self.assertFalse(result["eligible"]) class JournalSymlinkRefusal(unittest.TestCase): def test_symlink_journal_path_refused_on_load(self): tmp = tempfile.mkdtemp() try: real = os.path.join(tmp, "real.json") with open(real, "w", encoding="utf-8") as fh: fh.write("{}") link = os.path.join(tmp, "link.json") os.symlink(real, link) # Point journal path helper via env key = "symlink-test" jdir = tmp # Craft path that is a symlink by saving then replacing path = dorec._journal_path(key, journal_dir=jdir) with open(path, "w", encoding="utf-8") as fh: json.dump({"idempotency_key": key}, fh) os.remove(path) os.symlink(real, path) with self.assertRaises(ValueError): dorec.load_journal(key, journal_dir=jdir) finally: shutil.rmtree(tmp, ignore_errors=True) if __name__ == "__main__": unittest.main()