"""Tests for snapshot delta encoding/decoding pure functions. compute_snapshot_delta(base, new) -> (added_or_modified, removed) apply_snapshot_delta(base, added, removed) -> full manifest Test plan: A. Identity — delta against self is empty B. Add one file C. Modify one file (same path, new hash) D. Remove one file E. Mixed — add + modify + remove in one delta F. Roundtrip — apply(compute(base, new), base) == new G. Delta is smaller than full manifest for typical commits H. Empty base manifest (first commit ever) I. Empty new manifest (all files deleted) """ from __future__ import annotations from muse.core.types import Manifest, NULL_LONG_ID, long_id def _make_manifest(n: int, prefix: str = "src/file") -> Manifest: """Build a deterministic {path: sha256_hex} manifest with n entries.""" return { f"{prefix}_{i}.py": long_id(f"{'a' * 60}{i:04d}") for i in range(n) } # --------------------------------------------------------------------------- # A. Identity # --------------------------------------------------------------------------- def test_delta_identity_is_empty() -> None: from muse.core.mpack import compute_snapshot_delta m = _make_manifest(100) added, removed = compute_snapshot_delta(m, m) assert added == {} assert removed == [] # --------------------------------------------------------------------------- # B. Add one file # --------------------------------------------------------------------------- def test_delta_add_one_file() -> None: from muse.core.mpack import compute_snapshot_delta base = _make_manifest(10) new = dict(base) new["src/new_file.py"] = long_id("b" * 64) added, removed = compute_snapshot_delta(base, new) assert added == {"src/new_file.py": long_id("b" * 64)} assert removed == [] # --------------------------------------------------------------------------- # C. Modify one file # --------------------------------------------------------------------------- def test_delta_modify_one_file() -> None: from muse.core.mpack import compute_snapshot_delta base = _make_manifest(10) new = dict(base) existing_path = "src/file_5.py" new[existing_path] = long_id("c" * 64) added, removed = compute_snapshot_delta(base, new) assert added == {existing_path: long_id("c" * 64)} assert removed == [] # --------------------------------------------------------------------------- # D. Remove one file # --------------------------------------------------------------------------- def test_delta_remove_one_file() -> None: from muse.core.mpack import compute_snapshot_delta base = _make_manifest(10) new = dict(base) del new["src/file_3.py"] added, removed = compute_snapshot_delta(base, new) assert added == {} assert "src/file_3.py" in removed # --------------------------------------------------------------------------- # E. Mixed # --------------------------------------------------------------------------- def test_delta_mixed_operations() -> None: from muse.core.mpack import compute_snapshot_delta base = _make_manifest(10) new = dict(base) new["src/file_0.py"] = long_id("d" * 64) # modify del new["src/file_9.py"] # remove new["src/brand_new.py"] = long_id("e" * 64) # add added, removed = compute_snapshot_delta(base, new) assert "src/file_0.py" in added assert added["src/file_0.py"] == long_id("d" * 64) assert "src/brand_new.py" in added assert "src/file_9.py" in removed assert len(added) == 2 assert len(removed) == 1 # --------------------------------------------------------------------------- # F. Roundtrip # --------------------------------------------------------------------------- def test_delta_roundtrip() -> None: from muse.core.mpack import compute_snapshot_delta, apply_snapshot_delta base = _make_manifest(200) # Simulate a typical code commit: 3 files changed, 1 added, 1 deleted new = dict(base) new["src/file_10.py"] = long_id("f" * 64) new["src/file_20.py"] = NULL_LONG_ID new["src/file_30.py"] = long_id("1" * 64) new["src/new_feature.py"] = long_id("2" * 64) del new["src/file_199.py"] added, removed = compute_snapshot_delta(base, new) reconstructed = apply_snapshot_delta(base, added, removed) assert reconstructed == new def test_delta_roundtrip_large_repo() -> None: from muse.core.mpack import compute_snapshot_delta, apply_snapshot_delta base = _make_manifest(1071) # muse repo scale new = dict(base) # typical commit: 2 files changed new["src/file_500.py"] = long_id("a" * 63 + "1") new["src/file_501.py"] = long_id("a" * 63 + "2") added, removed = compute_snapshot_delta(base, new) reconstructed = apply_snapshot_delta(base, added, removed) assert reconstructed == new # --------------------------------------------------------------------------- # G. Delta size vs full manifest size # --------------------------------------------------------------------------- def test_delta_smaller_than_full_for_typical_commit() -> None: import json as _json from muse.core.mpack import compute_snapshot_delta base = _make_manifest(1071) # muse repo scale new = dict(base) # Change 3 files for i in [100, 200, 300]: new[f"src/file_{i}.py"] = long_id("9" * 64) added, removed = compute_snapshot_delta(base, new) delta_bytes = len(_json.dumps({"added": added, "removed": removed}).encode()) full_bytes = len(_json.dumps(new).encode()) assert delta_bytes < full_bytes, ( f"delta ({delta_bytes}B) should be smaller than full manifest ({full_bytes}B)" ) # --------------------------------------------------------------------------- # H. Empty base (first commit) # --------------------------------------------------------------------------- def test_delta_from_empty_base() -> None: from muse.core.mpack import compute_snapshot_delta, apply_snapshot_delta base: dict[str, str] = {} new = _make_manifest(10) added, removed = compute_snapshot_delta(base, new) assert added == new assert removed == [] assert apply_snapshot_delta(base, added, removed) == new # --------------------------------------------------------------------------- # I. Empty new manifest (all files deleted) # --------------------------------------------------------------------------- def test_delta_to_empty_manifest() -> None: from muse.core.mpack import compute_snapshot_delta, apply_snapshot_delta base = _make_manifest(5) new: dict[str, str] = {} added, removed = compute_snapshot_delta(base, new) assert added == {} assert set(removed) == set(base.keys()) assert apply_snapshot_delta(base, added, removed) == {}