"""Zero-data-loss workdir integrity tests. What these tests cover ---------------------- This suite was written after a real incident where the working tree diverged from the committed snapshot. The root cause chain: 1. ``restore_object`` used ``shutil.copy2(src, dest)`` directly — not atomic. A crash mid-copy could leave a corrupt destination file. 2. ``apply_manifest`` ignored the ``False`` return from ``restore_object`` when an object was absent from the store. The file was silently left at its old content; no error surfaced. 3. ``_checkout_snapshot`` (incremental delta path) printed a warning when an object was missing but continued — same silent data loss. 4. No post-operation integrity verification existed to catch any of the above after the fact. Fixes applied: * ``restore_object`` — atomic write: temp file → ``os.replace``. * ``apply_manifest`` — raises ``RuntimeError`` listing every missing object. * ``_checkout_snapshot`` — raises ``SystemExit(INTERNAL_ERROR)`` on missing object; never continues with a partial workdir. * ``verify_workdir_integrity`` — new utility: full hash-based post-op audit. Test categories --------------- I restore_object atomicity (temp+replace pattern). II apply_manifest — missing object raises, not silently skips. III verify_workdir_integrity — utility correctness. IV checkout → workdir always matches target snapshot. V fast-forward merge → workdir always matches target snapshot. VI checkout aborts hard when an object is missing from the store. VII Editor-cache simulation — status detects stale-cache workdir drift. VIII Stress tests — 500-file repos, deep chains, diamond DAGs. """ from __future__ import annotations import hashlib import json import os import pathlib import shutil import stat import tempfile import pytest from tests.cli_test_helper import CliRunner from muse.core.object_store import object_path, restore_object, write_object from muse.core.snapshot import walk_workdir from muse.core.workdir import apply_manifest, verify_workdir_integrity from muse.core.types import Manifest, blob_id, content_hash, fake_id, hash_file from muse.core.paths import muse_dir, ref_path type _EnvMap = dict[str, str] runner = CliRunner() cli = None # CliRunner ignores this positional # --------------------------------------------------------------------------- # Shared helpers # --------------------------------------------------------------------------- def _env(root: pathlib.Path) -> _EnvMap: return {"MUSE_REPO_ROOT": str(root)} def _run(root: pathlib.Path, *args: str) -> tuple[int, str]: final = list(args) if final and final[0] == "merge" and "--force" not in final: final.insert(1, "--force") result = runner.invoke(cli, final, env=_env(root), catch_exceptions=False) return result.exit_code, result.output def _run_unchecked(root: pathlib.Path, *args: str) -> tuple[int, str]: final = list(args) if final and final[0] == "merge" and "--force" not in final: final.insert(1, "--force") result = runner.invoke(cli, final, env=_env(root)) return result.exit_code, result.output def _store_object(root: pathlib.Path, content: bytes) -> str: """Write *content* to the object store, return its object ID.""" oid = blob_id(content) write_object(root, oid, content) return oid def _object_path(root: pathlib.Path, oid: str) -> pathlib.Path: return object_path(root, oid) def _init_repo(tmp_path: pathlib.Path, domain: str = "code") -> tuple[pathlib.Path, str]: dot_muse = muse_dir(tmp_path) dot_muse.mkdir() repo_id = fake_id("repo") (dot_muse / "repo.json").write_text(json.dumps({ "repo_id": repo_id, "domain": domain, "version": "1.0.0", })) (dot_muse / "refs" / "heads").mkdir(parents=True) (dot_muse / "objects").mkdir() (dot_muse / "HEAD").write_text("ref: refs/heads/main\n") return tmp_path, repo_id def _head_manifest(root: pathlib.Path, branch: str) -> Manifest: from muse.core.commits import read_commit from muse.core.snapshots import read_snapshot ref = (ref_path(root, branch)).read_text().strip() cr = read_commit(root, ref) assert cr is not None sr = read_snapshot(root, cr.snapshot_id) assert sr is not None return dict(sr.manifest) def _write_disk(root: pathlib.Path, rel_path: str, content: bytes) -> str: """Write content to disk AND store it; return object id.""" fp = root / rel_path fp.parent.mkdir(parents=True, exist_ok=True) fp.write_bytes(content) return _store_object(root, content) # =========================================================================== # I restore_object atomicity # =========================================================================== class TestRestoreObjectAtomicityI: """restore_object must use atomic writes so a crash mid-copy never leaves a partial file at the destination.""" def test_I1_successful_restore_produces_correct_content( self, tmp_path: pathlib.Path ) -> None: """I1: happy-path restore writes exact bytes to dest.""" root, _ = _init_repo(tmp_path) content = b"hello world\n" * 100 oid = _store_object(root, content) dest = tmp_path / "out.bin" assert restore_object(root, oid, dest) assert dest.read_bytes() == content def test_I2_restore_overwrites_existing_file(self, tmp_path: pathlib.Path) -> None: """I2: restore replaces whatever was at dest (no skip-if-exists).""" root, _ = _init_repo(tmp_path) old = b"old content\n" new = b"new content\n" dest = tmp_path / "f.txt" dest.write_bytes(old) oid = _store_object(root, new) assert restore_object(root, oid, dest) assert dest.read_bytes() == new def test_I3_restore_missing_object_returns_false_does_not_touch_dest( self, tmp_path: pathlib.Path ) -> None: """I3: missing object → False, pre-existing dest left intact.""" root, _ = _init_repo(tmp_path) sentinel = b"sentinel\n" dest = tmp_path / "existing.txt" dest.write_bytes(sentinel) fake_oid = blob_id(b"nonexistent") assert not restore_object(root, fake_oid, dest) assert dest.read_bytes() == sentinel def test_I4_restore_creates_parent_directories( self, tmp_path: pathlib.Path ) -> None: """I4: dest parent dirs are created automatically.""" root, _ = _init_repo(tmp_path) content = b"deep\n" oid = _store_object(root, content) dest = tmp_path / "a" / "b" / "c" / "deep.txt" assert not dest.parent.exists() assert restore_object(root, oid, dest) assert dest.read_bytes() == content def test_I5_atomic_write_leaves_no_tmp_file_on_success( self, tmp_path: pathlib.Path ) -> None: """I5: after a successful restore no .restore-tmp-* file lingers.""" root, _ = _init_repo(tmp_path) content = b"data\n" oid = _store_object(root, content) dest = tmp_path / "target.txt" restore_object(root, oid, dest) tmps = list(tmp_path.glob(".restore-tmp-*")) assert tmps == [], f"Stale tmp files found: {tmps}" def test_I6_restore_hash_after_restore_matches_object_id( self, tmp_path: pathlib.Path ) -> None: """I6: the restored file's SHA-256 matches the object_id exactly.""" root, _ = _init_repo(tmp_path) content = b"integrity\n" * 1000 oid = _store_object(root, content) dest = tmp_path / "verified.bin" restore_object(root, oid, dest) actual = hash_file(dest) assert actual == oid, f"Hash mismatch after restore: {actual[:8]} ≠ {oid[:8]}" def test_I7_restore_large_file_correct_content( self, tmp_path: pathlib.Path ) -> None: """I7: 10 MiB blob survives a round-trip through the object store.""" root, _ = _init_repo(tmp_path) content = os.urandom(10 * 1024 * 1024) oid = _store_object(root, content) dest = tmp_path / "large.bin" assert restore_object(root, oid, dest) assert dest.read_bytes() == content # =========================================================================== # II apply_manifest — missing object must raise, never silently skip # =========================================================================== class TestApplyManifestMissingObjectII: """apply_manifest must fail loudly when any object is absent.""" def test_II1_missing_single_object_raises_runtime_error( self, tmp_path: pathlib.Path ) -> None: """II1: one missing object → RuntimeError, not silent skip.""" root, _ = _init_repo(tmp_path) fake_oid = blob_id(b"ghost") with pytest.raises(RuntimeError, match="missing from the local store"): apply_manifest(root, {}, {"ghost.txt": fake_oid}) def test_II2_error_message_names_the_missing_path( self, tmp_path: pathlib.Path ) -> None: """II2: the error message includes the missing path name.""" root, _ = _init_repo(tmp_path) fake_oid = blob_id(b"abc") with pytest.raises(RuntimeError) as exc_info: apply_manifest(root, {}, {"crucial/file.py": fake_oid}) assert "crucial/file.py" in str(exc_info.value) def test_II3_partial_manifest_some_missing_raises( self, tmp_path: pathlib.Path ) -> None: """II3: when one of N files is missing the whole call raises.""" root, _ = _init_repo(tmp_path) good_oid = _write_disk(root, "exists.txt", b"ok\n") bad_oid = blob_id(b"not stored") with pytest.raises(RuntimeError): apply_manifest(root, {}, {"exists.txt": good_oid, "missing.txt": bad_oid}) def test_II4_all_objects_present_succeeds( self, tmp_path: pathlib.Path ) -> None: """II4: when every object is in the store apply_manifest succeeds.""" root, _ = _init_repo(tmp_path) oid_a = _write_disk(root, "a.txt", b"aaa\n") oid_b = _write_disk(root, "b.txt", b"bbb\n") (root / "a.txt").unlink() (root / "b.txt").unlink() apply_manifest(root, {}, {"a.txt": oid_a, "b.txt": oid_b}) assert (root / "a.txt").read_bytes() == b"aaa\n" assert (root / "b.txt").read_bytes() == b"bbb\n" def test_II5_multiple_missing_reported_in_error( self, tmp_path: pathlib.Path ) -> None: """II5: error message covers multiple missing files.""" root, _ = _init_repo(tmp_path) manifest = {f"f{i}.py": blob_id(f"fake{i}".encode()) for i in range(10)} with pytest.raises(RuntimeError) as exc_info: apply_manifest(root, {}, manifest) msg = str(exc_info.value) assert "10 object(s)" in msg def test_II6_apply_manifest_removes_files_not_in_target( self, tmp_path: pathlib.Path ) -> None: """II6: tracked files absent from target manifest are deleted.""" root, _ = _init_repo(tmp_path) keep_oid = _write_disk(root, "keep.txt", b"keep\n") del_oid = _store_object(root, b"delete\n") (root / "delete_me.txt").write_bytes(b"delete\n") # delete_me.txt is in prev_manifest (was tracked) but not in target — must be removed apply_manifest( root, {"keep.txt": keep_oid, "delete_me.txt": del_oid}, {"keep.txt": keep_oid}, ) assert not (root / "delete_me.txt").exists() assert (root / "keep.txt").exists() def test_II7_empty_manifest_non_empty_prev_raises_value_error( self, tmp_path: pathlib.Path ) -> None: """II7: data-loss guard — empty target with non-empty prev_manifest raises ValueError.""" root, _ = _init_repo(tmp_path) oid = _store_object(root, b"data\n") with pytest.raises(ValueError, match="empty target_manifest"): apply_manifest(root, {"file.txt": oid}, {}) # =========================================================================== # III verify_workdir_integrity — utility correctness # =========================================================================== class TestVerifyWorkdirIntegrityIII: """verify_workdir_integrity must catch every form of workdir drift.""" def test_III1_clean_workdir_returns_empty_list( self, tmp_path: pathlib.Path ) -> None: """III1: workdir matches manifest → no mismatches.""" root, _ = _init_repo(tmp_path) oid = _write_disk(root, "a.py", b"x = 1\n") mismatches = verify_workdir_integrity(root, {"a.py": oid}) assert mismatches == [] def test_III2_modified_file_detected(self, tmp_path: pathlib.Path) -> None: """III2: externally modified file shows up as mismatch.""" root, _ = _init_repo(tmp_path) original = b"original\n" oid = _write_disk(root, "f.py", original) (root / "f.py").write_bytes(b"tampered\n") mismatches = verify_workdir_integrity(root, {"f.py": oid}) assert len(mismatches) == 1 path, expected, actual = mismatches[0] assert path == "f.py" assert expected == oid assert actual != oid assert actual is not None def test_III3_missing_file_detected(self, tmp_path: pathlib.Path) -> None: """III3: file present in manifest but absent from disk → mismatch.""" root, _ = _init_repo(tmp_path) oid = _write_disk(root, "gone.py", b"gone\n") (root / "gone.py").unlink() mismatches = verify_workdir_integrity(root, {"gone.py": oid}) assert any(m[0] == "gone.py" and m[2] is None for m in mismatches) def test_III4_extra_tracked_file_detected( self, tmp_path: pathlib.Path ) -> None: """III4: file on disk but not in manifest is also reported.""" root, _ = _init_repo(tmp_path) oid = _write_disk(root, "tracked.py", b"ok\n") (root / "extra.py").write_bytes(b"extra\n") mismatches = verify_workdir_integrity(root, {"tracked.py": oid}) extras = [m for m in mismatches if m[0] == "extra.py"] assert extras, "Extra file not reported" def test_III5_empty_manifest_empty_workdir_clean( self, tmp_path: pathlib.Path ) -> None: """III5: both manifest and workdir empty → clean.""" root, _ = _init_repo(tmp_path) assert verify_workdir_integrity(root, {}) == [] def test_III6_multiple_mismatches_all_reported( self, tmp_path: pathlib.Path ) -> None: """III6: all mismatches returned, not just the first.""" root, _ = _init_repo(tmp_path) manifest: Manifest = {} for i in range(20): oid = _write_disk(root, f"f{i}.py", f"v={i}\n".encode()) manifest[f"f{i}.py"] = oid # Tamper with half the files for i in range(0, 20, 2): (root / f"f{i}.py").write_bytes(b"tampered\n") mismatches = verify_workdir_integrity(root, manifest) assert len(mismatches) == 10, f"Expected 10 mismatches, got {len(mismatches)}" def test_III7_correct_content_after_apply_manifest( self, tmp_path: pathlib.Path ) -> None: """III7: after apply_manifest, verify_workdir_integrity is clean.""" root, _ = _init_repo(tmp_path) oid_a = _write_disk(root, "a.py", b"a = 1\n") oid_b = _write_disk(root, "b.py", b"b = 2\n") (root / "a.py").unlink() (root / "b.py").unlink() apply_manifest(root, {}, {"a.py": oid_a, "b.py": oid_b}) mismatches = verify_workdir_integrity(root, {"a.py": oid_a, "b.py": oid_b}) assert mismatches == [], f"Expected clean after apply_manifest: {mismatches}" # =========================================================================== # IV checkout → workdir always matches the target snapshot # =========================================================================== class TestCheckoutWorkdirIntegrityIV: """After muse checkout, the working tree must byte-for-byte match the target branch's committed snapshot. No exceptions.""" def _full_roundtrip(self, tmp_path: pathlib.Path, n_files: int) -> None: """Create two branches with different content, checkout between them, verify integrity on each switch.""" root, repo_id = _init_repo(tmp_path) code, _ = _run(root, "init", str(root)) main_manifest: Manifest = {} for i in range(n_files): oid = _write_disk(root, f"src/m{i}.py", f"# main {i}\n".encode()) main_manifest[f"src/m{i}.py"] = oid code, out = _run(root, "commit", "--allow-empty", "-m", "main files") assert code == 0, out code, out = _run(root, "branch", "feat") assert code == 0, out code, out = _run(root, "checkout", "feat") assert code == 0, out feat_manifest: Manifest = {} for i in range(n_files): oid = _write_disk(root, f"src/f{i}.py", f"# feat {i}\n".encode()) feat_manifest[f"src/f{i}.py"] = oid _run(root, "code", "add", ".") code, out = _run(root, "commit", "-m", "feat files") assert code == 0, out # Switch back to main — verify clean code, out = _run(root, "checkout", "main") assert code == 0, out main_snap = _head_manifest(root, "main") mismatches = verify_workdir_integrity(root, main_snap) assert mismatches == [], ( f"DATA LOSS: {len(mismatches)} mismatch(es) after checkout main:\n" f"{'\n'.join(f' {m}' for m in mismatches[:5])}" ) # Switch to feat — verify clean code, out = _run(root, "checkout", "feat") assert code == 0, out feat_snap = _head_manifest(root, "feat") mismatches = verify_workdir_integrity(root, feat_snap) assert mismatches == [], ( f"DATA LOSS: {len(mismatches)} mismatch(es) after checkout feat:\n" f"{'\n'.join(f' {m}' for m in mismatches[:5])}" ) def test_IV1_checkout_10_files_workdir_matches_snapshot( self, tmp_path: pathlib.Path ) -> None: """IV1: 10-file repo checkout — workdir matches target snapshot.""" self._full_roundtrip(tmp_path, 10) def test_IV2_checkout_50_files_workdir_matches_snapshot( self, tmp_path: pathlib.Path ) -> None: """IV2: 50-file repo checkout — workdir matches target snapshot.""" self._full_roundtrip(tmp_path, 50) def test_IV3_repeated_checkout_workdir_consistent( self, tmp_path: pathlib.Path ) -> None: """IV3: switching back and forth 10 times never corrupts the workdir.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) oid_a = _write_disk(root, "f.py", b"version_a\n") _run(root, "commit", "--allow-empty", "-m", "main") _run(root, "branch", "feat") _run(root, "checkout", "feat") oid_b = _write_disk(root, "f.py", b"version_b\n") _run(root, "code", "add", ".") _run(root, "commit", "-m", "feat") main_snap = _head_manifest(root, "main") feat_snap = _head_manifest(root, "feat") for i in range(10): branch = "main" if i % 2 == 0 else "feat" _run(root, "checkout", branch) expected = main_snap if branch == "main" else feat_snap mismatches = verify_workdir_integrity(root, expected) assert mismatches == [], ( f"Iteration {i}: mismatch after checkout {branch}: {mismatches}" ) def test_IV4_checkout_restores_file_modified_between_branches( self, tmp_path: pathlib.Path ) -> None: """IV4: a file modified on one branch is correctly restored when switching to a branch that has the original version.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) oid_v1 = _write_disk(root, "shared.py", b"# v1\n") _run(root, "commit", "--allow-empty", "-m", "base") _run(root, "branch", "feat") oid_v2 = _write_disk(root, "shared.py", b"# v2\n") _run(root, "code", "add", ".") _run(root, "commit", "-m", "main v2") _run(root, "checkout", "feat") assert (root / "shared.py").read_bytes() == b"# v1\n", ( "feat branch should have v1 of shared.py" ) _run(root, "checkout", "main") assert (root / "shared.py").read_bytes() == b"# v2\n", ( "main branch should have v2 of shared.py" ) def test_IV5_checkout_deletes_files_not_in_target_branch( self, tmp_path: pathlib.Path ) -> None: """IV5: files added on one branch are deleted when switching away.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) _write_disk(root, "base.py", b"base\n") _run(root, "commit", "--allow-empty", "-m", "base") _run(root, "branch", "feat") _run(root, "checkout", "feat") _write_disk(root, "feat_only.py", b"feat\n") _run(root, "code", "add", ".") _run(root, "commit", "-m", "feat_only") _run(root, "checkout", "main") assert not (root / "feat_only.py").exists(), ( "feat-only file should not exist on main branch" ) # =========================================================================== # V fast-forward merge → workdir always matches target snapshot # =========================================================================== class TestFFMergeWorkdirIntegrityV: """A fast-forward merge must update the working tree to match the incoming branch's snapshot — ALL files, not just the delta.""" def test_V1_ff_merge_all_files_restored(self, tmp_path: pathlib.Path) -> None: """V1: after FF merge every file in the target manifest is on disk with the correct content.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) _write_disk(root, "base.py", b"base\n") _run(root, "commit", "--allow-empty", "-m", "base") _run(root, "branch", "feat") _run(root, "checkout", "feat") manifest: Manifest = {} for i in range(30): oid = _write_disk(root, f"f{i}.py", f"# feat {i}\n".encode()) manifest[f"f{i}.py"] = oid _run(root, "code", "add", ".") _run(root, "commit", "-m", "feat 30 files") _run(root, "checkout", "main") code, out = _run(root, "merge", "feat") assert code == 0, out merged = _head_manifest(root, "main") mismatches = verify_workdir_integrity(root, merged) assert mismatches == [], ( f"DATA LOSS after FF merge: {len(mismatches)} mismatch(es)\n" f"{'\n'.join(f' {m}' for m in mismatches[:5])}" ) def test_V2_ff_merge_correct_content_not_just_present( self, tmp_path: pathlib.Path ) -> None: """V2: FF merge writes the correct *content*, not just the correct filename.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) old_content = b"# old version\n" new_content = b"# new version\n" _write_disk(root, "important.py", old_content) _run(root, "commit", "--allow-empty", "-m", "base") _run(root, "branch", "feat") _run(root, "checkout", "feat") _write_disk(root, "important.py", new_content) _run(root, "code", "add", ".") _run(root, "commit", "-m", "update important.py") _run(root, "checkout", "main") # File on disk is now old_content assert (root / "important.py").read_bytes() == old_content _run(root, "merge", "feat") # File on disk must now be new_content actual = (root / "important.py").read_bytes() assert actual == new_content, ( f"FF merge did not restore correct content. " f"Expected {new_content!r}, got {actual!r}" ) def test_V3_ff_merge_workdir_matches_snapshot_byte_for_byte( self, tmp_path: pathlib.Path ) -> None: """V3: verify_workdir_integrity confirms zero drift after FF merge.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) _write_disk(root, "a.py", b"a\n") _run(root, "commit", "--allow-empty", "-m", "base") _run(root, "branch", "feat") _run(root, "checkout", "feat") for i in range(20): _write_disk(root, f"feat_{i}.py", f"feat{i}\n".encode()) _run(root, "code", "add", ".") _run(root, "commit", "-m", "feat 20 files") _run(root, "checkout", "main") _run(root, "merge", "feat") merged_snap = _head_manifest(root, "main") mismatches = verify_workdir_integrity(root, merged_snap) assert mismatches == [] # =========================================================================== # VI checkout aborts hard when an object is missing from the store # =========================================================================== class TestCheckoutMissingObjectVI: """Checkout must refuse to proceed when an object it needs is absent from the local object store. The partial-checkout silent-data-loss path is now closed.""" def test_VI1_checkout_to_branch_with_missing_object_aborts( self, tmp_path: pathlib.Path ) -> None: """VI1: if a required object is purged from the store, checkout exits non-zero and does NOT silently leave the workdir in a partially restored state.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) _write_disk(root, "base.py", b"base\n") _run(root, "commit", "--allow-empty", "-m", "base") _run(root, "branch", "feat") _run(root, "checkout", "feat") content = b"# feat file\n" oid = _write_disk(root, "feat.py", content) _run(root, "code", "add", ".") _run(root, "commit", "-m", "feat") _run(root, "checkout", "main") # Purge the feat.py object from the store — simulates a corruption obj = _object_path(root, oid) obj.unlink() code, out = _run_unchecked(root, "checkout", "feat") assert code != 0, ( "Checkout should fail when a required object is missing from the store" ) def test_VI2_apply_manifest_raises_on_missing_object_not_silent( self, tmp_path: pathlib.Path ) -> None: """VI2: apply_manifest raises RuntimeError (not returns None or logs a warning) when an object is missing.""" root, _ = _init_repo(tmp_path) ghost_oid = blob_id(b"not in store") with pytest.raises(RuntimeError) as exc_info: apply_manifest(root, {}, {"ghost.py": ghost_oid}) assert "missing from the local store" in str(exc_info.value) def test_VI3_ff_merge_aborts_if_incoming_object_missing( self, tmp_path: pathlib.Path ) -> None: """VI3: FF merge aborts if an object from the target snapshot is not in the local store.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) _write_disk(root, "base.py", b"base\n") _run(root, "commit", "--allow-empty", "-m", "base") _run(root, "branch", "feat") _run(root, "checkout", "feat") content = b"# critical\n" oid = _write_disk(root, "critical.py", content) _run(root, "code", "add", ".") _run(root, "commit", "-m", "critical") _run(root, "checkout", "main") # Purge the object after committing _object_path(root, oid).unlink() code, out = _run_unchecked(root, "merge", "feat") assert code != 0, "FF merge should fail when a target object is missing" # =========================================================================== # VII Editor-cache simulation — status must detect stale-cache workdir drift # =========================================================================== class TestEditorCacheSimulationVII: """Simulates the exact incident: the editor had a cached (stale) version of a file. After a merge updated the on-disk file, the editor wrote the stale version back, corrupting the workdir. Muse cannot prevent an editor from writing stale data, but it CAN detect the drift via `muse status` (which compares workdir hashes against HEAD) and via `verify_workdir_integrity`. """ def test_VII1_status_detects_workdir_drift_after_external_write( self, tmp_path: pathlib.Path ) -> None: """VII1: if a file is externally overwritten to an old version, muse status reports it as modified.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) old_content = b"# version 1\n" new_content = b"# version 2\n" _write_disk(root, "plugin.py", old_content) _run(root, "commit", "--allow-empty", "-m", "v1") _run(root, "branch", "feat") _run(root, "checkout", "feat") _write_disk(root, "plugin.py", new_content) _run(root, "code", "add", ".") _run(root, "commit", "-m", "v2") _run(root, "checkout", "main") _run(root, "merge", "feat") # HEAD now says plugin.py = new_content. # Simulate editor writing back the stale old version. (root / "plugin.py").write_bytes(old_content) code, out = _run(root, "status") assert code == 0 assert "plugin.py" in out, ( "muse status must report plugin.py as modified after stale write" ) def test_VII2_verify_workdir_integrity_catches_stale_editor_write( self, tmp_path: pathlib.Path ) -> None: """VII2: verify_workdir_integrity spots the stale-editor-cache corruption.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) old_content = b"# old\n" new_content = b"# new\n" oid_new = _write_disk(root, "plugin.py", old_content) _run(root, "commit", "--allow-empty", "-m", "base") _run(root, "branch", "feat") _run(root, "checkout", "feat") oid_new = _write_disk(root, "plugin.py", new_content) _run(root, "code", "add", ".") _run(root, "commit", "-m", "feat") _run(root, "checkout", "main") _run(root, "merge", "feat") # Simulate editor stale write (root / "plugin.py").write_bytes(old_content) head_snap = _head_manifest(root, "main") mismatches = verify_workdir_integrity(root, head_snap) assert any(m[0] == "plugin.py" for m in mismatches), ( "verify_workdir_integrity must detect the stale file" ) def test_VII3_correct_state_after_reapplying_manifest( self, tmp_path: pathlib.Path ) -> None: """VII3: after detecting stale-editor drift, re-applying the manifest restores correct state and verify_workdir_integrity is clean.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) _write_disk(root, "plugin.py", b"# old\n") _run(root, "commit", "--allow-empty", "-m", "base") _run(root, "branch", "feat") _run(root, "checkout", "feat") _write_disk(root, "plugin.py", b"# new\n") _run(root, "code", "add", ".") _run(root, "commit", "-m", "feat") _run(root, "checkout", "main") _run(root, "merge", "feat") # Stale write (root / "plugin.py").write_bytes(b"# stale\n") head_snap = _head_manifest(root, "main") # Repair by re-applying the manifest apply_manifest(root, head_snap, head_snap) mismatches = verify_workdir_integrity(root, head_snap) assert mismatches == [], "After re-applying manifest, workdir must be clean" # =========================================================================== # VIII Stress tests # =========================================================================== class TestStressWorkdirVIII: """High-volume, adversarial scenarios to eliminate the entire class of workdir corruption bugs.""" def test_VIII1_500_file_checkout_all_match_snapshot( self, tmp_path: pathlib.Path ) -> None: """VIII1: 500-file repo — every file matches the committed snapshot after every checkout.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) manifest_main: Manifest = {} for i in range(500): oid = _write_disk(root, f"main_{i:04d}.py", f"# main {i}\n".encode()) manifest_main[f"main_{i:04d}.py"] = oid code, out = _run(root, "commit", "--allow-empty", "-m", "main 500") assert code == 0, out _run(root, "branch", "feat") _run(root, "checkout", "feat") manifest_feat: Manifest = {} for i in range(500): oid = _write_disk(root, f"feat_{i:04d}.py", f"# feat {i}\n".encode()) manifest_feat[f"feat_{i:04d}.py"] = oid _run(root, "code", "add", ".") code, out = _run(root, "commit", "-m", "feat 500") assert code == 0, out _run(root, "checkout", "main") snp = _head_manifest(root, "main") mismatches = verify_workdir_integrity(root, snp) assert mismatches == [], f"{len(mismatches)} mismatch(es) on main" _run(root, "checkout", "feat") snp = _head_manifest(root, "feat") mismatches = verify_workdir_integrity(root, snp) assert mismatches == [], f"{len(mismatches)} mismatch(es) on feat" def test_VIII2_ff_merge_500_files_all_correct( self, tmp_path: pathlib.Path ) -> None: """VIII2: FF merge with 500 incoming files — all correct after merge.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) _write_disk(root, "base.py", b"base\n") _run(root, "commit", "--allow-empty", "-m", "base") _run(root, "branch", "feat") _run(root, "checkout", "feat") expected: Manifest = {} for i in range(500): content = f"# file {i} unique content {content_hash({'i': i})}\n".encode() oid = _write_disk(root, f"feat_{i:04d}.py", content) expected[f"feat_{i:04d}.py"] = oid _run(root, "code", "add", ".") _run(root, "commit", "-m", "feat 500") _run(root, "checkout", "main") code, out = _run(root, "merge", "feat") assert code == 0, out merged = _head_manifest(root, "main") mismatches = verify_workdir_integrity(root, merged) assert mismatches == [], ( f"DATA LOSS: {len(mismatches)} file(s) wrong after 500-file FF merge" ) def test_VIII3_alternating_checkout_never_drifts( self, tmp_path: pathlib.Path ) -> None: """VIII3: 20 alternating checkout cycles — workdir never drifts.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) oid_a = _write_disk(root, "shared.py", b"# version A\n") _run(root, "commit", "--allow-empty", "-m", "main") snap_main = _head_manifest(root, "main") _run(root, "branch", "feat") _run(root, "checkout", "feat") oid_b = _write_disk(root, "shared.py", b"# version B\n") _run(root, "code", "add", ".") _run(root, "commit", "-m", "feat") snap_feat = _head_manifest(root, "feat") for cycle in range(20): branch = "main" if cycle % 2 == 0 else "feat" code, out = _run(root, "checkout", branch) assert code == 0, f"Cycle {cycle}: checkout {branch} failed" expected = snap_main if branch == "main" else snap_feat mismatches = verify_workdir_integrity(root, expected) assert mismatches == [], ( f"Cycle {cycle}, branch {branch}: {len(mismatches)} mismatch(es)" ) def test_VIII4_deep_chain_checkout_base_restores_correctly( self, tmp_path: pathlib.Path ) -> None: """VIII4: deep commit chain — checkout of base commit restores correctly.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) # Build a chain of 20 commits, each modifying the same file for i in range(20): _write_disk(root, "evolving.py", f"# iteration {i}\n".encode()) _run(root, "commit", "--allow-empty", "-m", f"iter {i}") snap = _head_manifest(root, "main") assert (root / "evolving.py").read_bytes() == b"# iteration 19\n" mismatches = verify_workdir_integrity(root, snap) assert mismatches == [] def test_VIII5_stress_apply_manifest_100_times_deterministic( self, tmp_path: pathlib.Path ) -> None: """VIII5: apply_manifest called 100 times is deterministic and always produces an identical workdir.""" root, _ = _init_repo(tmp_path) manifest: Manifest = {} for i in range(50): oid = _write_disk(root, f"f{i}.py", f"content {i}\n".encode()) manifest[f"f{i}.py"] = oid for trial in range(100): apply_manifest(root, manifest, manifest) mismatches = verify_workdir_integrity(root, manifest) assert mismatches == [], ( f"Trial {trial}: {len(mismatches)} mismatch(es) after apply_manifest" ) def test_VIII6_diamond_topology_workdir_clean_after_all_merges( self, tmp_path: pathlib.Path ) -> None: """VIII6: diamond merge topology — verify integrity at every step.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) # Base _write_disk(root, "base.py", b"base\n") _run(root, "commit", "--allow-empty", "-m", "base") # Left branch _run(root, "branch", "left") _run(root, "checkout", "left") _write_disk(root, "left.py", b"left\n") _run(root, "code", "add", ".") _run(root, "commit", "-m", "left") # Right branch (from main) _run(root, "checkout", "main") _run(root, "branch", "right") _run(root, "checkout", "right") _write_disk(root, "right.py", b"right\n") _run(root, "code", "add", ".") _run(root, "commit", "-m", "right") # Merge left → main _run(root, "checkout", "main") code, out = _run(root, "merge", "left") assert code == 0, out snp = _head_manifest(root, "main") assert verify_workdir_integrity(root, snp) == [] # Merge right → main code, out = _run(root, "merge", "right") assert code == 0, out snp = _head_manifest(root, "main") assert verify_workdir_integrity(root, snp) == [] def test_VIII7_binary_files_survive_checkout( self, tmp_path: pathlib.Path ) -> None: """VIII7: binary content (random bytes) survives checkout intact.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) binary = os.urandom(1024 * 512) # 512 KiB of random bytes oid = _write_disk(root, "data.bin", binary) _run(root, "commit", "--allow-empty", "-m", "binary") _run(root, "branch", "feat") _run(root, "checkout", "feat") _write_disk(root, "other.py", b"other\n") _run(root, "code", "add", ".") _run(root, "commit", "-m", "other") _run(root, "checkout", "main") snp = _head_manifest(root, "main") mismatches = verify_workdir_integrity(root, snp) assert mismatches == [] assert (root / "data.bin").read_bytes() == binary def test_VIII8_unicode_filenames_survive_checkout( self, tmp_path: pathlib.Path ) -> None: """VIII8: files with unicode path components survive checkout.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) paths = [ "src/módulo.py", "src/données.txt", "src/файл.py", ] for p in paths: _write_disk(root, p, f"# {p}\n".encode()) _run(root, "commit", "--allow-empty", "-m", "unicode paths") _run(root, "branch", "feat") _run(root, "checkout", "feat") _write_disk(root, "extra.py", b"extra\n") _run(root, "code", "add", ".") _run(root, "commit", "-m", "extra") _run(root, "checkout", "main") snp = _head_manifest(root, "main") mismatches = verify_workdir_integrity(root, snp) assert mismatches == [] def test_VIII9_no_data_loss_after_100_consecutive_commits( self, tmp_path: pathlib.Path ) -> None: """VIII9: 100 consecutive commits on main — verify final state is correct.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) for i in range(100): _write_disk(root, f"file_{i:03d}.py", f"# commit {i}\n".encode()) code, out = _run(root, "commit", "--allow-empty", "-m", f"commit {i}") assert code == 0, f"Commit {i} failed: {out}" snp = _head_manifest(root, "main") assert len(snp) == 100, f"Expected 100 files, got {len(snp)}" mismatches = verify_workdir_integrity(root, snp) assert mismatches == [], f"{len(mismatches)} mismatch(es) after 100 commits" # =========================================================================== # IX apply_manifest must not delete untracked files # =========================================================================== class TestApplyManifestUntrackedFilesIX: """Untracked files must survive apply_manifest regardless of target manifest. Root cause of the bug: apply_manifest used walk_workdir(root) to build current_files, which returns ALL files on disk — including files the user created that were never committed. The fix: use prev_manifest.keys() as the deletion candidate set so only previously-tracked files are candidates. """ def test_IX1_untracked_file_not_deleted_by_apply_manifest( self, tmp_path: pathlib.Path ) -> None: """IX1: a file never in any manifest must survive apply_manifest.""" root, _ = _init_repo(tmp_path) # prev state: file A is tracked oid_a = _write_disk(root, "a.py", b"a = 1\n") # target state: file B is tracked oid_b = _store_object(root, b"b = 2\n") (root / "b.py").write_bytes(b"b = 2\n") # untracked file — never in any manifest untracked = root / "notes.txt" untracked.write_bytes(b"my personal notes\n") apply_manifest(root, {"a.py": oid_a}, {"b.py": oid_b}) assert untracked.exists(), "untracked file was deleted by apply_manifest — data loss bug" assert (root / "b.py").read_bytes() == b"b = 2\n" assert not (root / "a.py").exists(), "a.py was tracked then removed — must be deleted" def test_IX2_untracked_dotfile_not_deleted( self, tmp_path: pathlib.Path ) -> None: """IX2: untracked dotfiles (e.g. spec docs written by Write tool) must survive.""" root, _ = _init_repo(tmp_path) oid_a = _write_disk(root, "main.py", b"x = 1\n") oid_b = _store_object(root, b"x = 2\n") (root / "main.py").write_bytes(b"x = 2\n") spec_doc = root / "docs" / "spec.md" spec_doc.parent.mkdir() spec_doc.write_bytes(b"# spec\n") apply_manifest(root, {"main.py": oid_a}, {"main.py": oid_b}) assert spec_doc.exists(), "untracked doc file was deleted by apply_manifest" def test_IX3_tracked_file_removed_from_target_gets_deleted( self, tmp_path: pathlib.Path ) -> None: """IX3: files in prev_manifest but not in target must still be deleted.""" root, _ = _init_repo(tmp_path) oid_a = _write_disk(root, "gone.py", b"gone\n") oid_b = _write_disk(root, "kept.py", b"kept\n") apply_manifest(root, {"gone.py": oid_a, "kept.py": oid_b}, {"kept.py": oid_b}) assert not (root / "gone.py").exists(), "tracked file removed from target must be deleted" assert (root / "kept.py").exists() def test_IX4_commit_does_not_delete_untracked_file( self, tmp_path: pathlib.Path ) -> None: """IX4: muse commit must not delete untracked files from the working tree.""" root, repo_id = _init_repo(tmp_path) _run(root, "init", str(root)) # First commit with one tracked file _write_disk(root, "tracked.py", b"x = 1\n") _run(root, "code", "add", ".") code, out = _run(root, "commit", "-m", "initial") assert code == 0, out # Write an untracked file (simulates Write tool creating a spec doc) untracked = root / "docs" / "spec.md" untracked.parent.mkdir() untracked.write_bytes(b"# my spec\n") assert untracked.exists() # Modify tracked file and commit (root / "tracked.py").write_bytes(b"x = 2\n") _run(root, "code", "add", "tracked.py") code, out = _run(root, "commit", "-m", "update") assert code == 0, out assert untracked.exists(), ( "muse commit deleted an untracked file — data loss bug. " "apply_manifest must not delete files absent from prev_manifest." )