"""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 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": f"sha256:{'a' * 60}{i:04d}" for i in range(n) } # --------------------------------------------------------------------------- # A. Identity # --------------------------------------------------------------------------- def test_delta_identity_is_empty(): 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(): from muse.core.mpack import compute_snapshot_delta base = _make_manifest(10) new = dict(base) new["src/new_file.py"] = "sha256:" + "b" * 64 added, removed = compute_snapshot_delta(base, new) assert added == {"src/new_file.py": "sha256:" + "b" * 64} assert removed == [] # --------------------------------------------------------------------------- # C. Modify one file # --------------------------------------------------------------------------- def test_delta_modify_one_file(): from muse.core.mpack import compute_snapshot_delta base = _make_manifest(10) new = dict(base) existing_path = "src/file_5.py" new[existing_path] = "sha256:" + "c" * 64 added, removed = compute_snapshot_delta(base, new) assert added == {existing_path: "sha256:" + "c" * 64} assert removed == [] # --------------------------------------------------------------------------- # D. Remove one file # --------------------------------------------------------------------------- def test_delta_remove_one_file(): 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(): from muse.core.mpack import compute_snapshot_delta base = _make_manifest(10) new = dict(base) new["src/file_0.py"] = "sha256:" + "d" * 64 # modify del new["src/file_9.py"] # remove new["src/brand_new.py"] = "sha256:" + "e" * 64 # add added, removed = compute_snapshot_delta(base, new) assert "src/file_0.py" in added assert added["src/file_0.py"] == "sha256:" + "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(): 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"] = "sha256:" + "f" * 64 new["src/file_20.py"] = "sha256:" + "0" * 64 new["src/file_30.py"] = "sha256:" + "1" * 64 new["src/new_feature.py"] = "sha256:" + "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(): 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"] = "sha256:" + "a" * 63 + "1" new["src/file_501.py"] = "sha256:" + "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(): from muse.core.mpack import compute_snapshot_delta import msgpack 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"] = "sha256:" + "9" * 64 added, removed = compute_snapshot_delta(base, new) delta_bytes = len(msgpack.packb({"added": added, "removed": removed}, use_bin_type=True)) full_bytes = len(msgpack.packb(new, use_bin_type=True)) 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(): 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(): 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) == {}