"""Tests for MPack — the MPack wire format replacing PackBundle. Coverage tiers -------------- - MPack TypedDict: fields, total=False (partial bundles valid) - MPackSummary TypedDict: all advisory summary fields - build_mpack: produces MPack with commits, snapshots, objects - build_mpack: have-set exclusion (only sends delta) - build_mpack: summary field populated correctly - build_mpack: raises on missing snapshot - apply_mpack: writes objects → snapshots → commits in order - apply_mpack: idempotent (safe to call twice) - apply_mpack: pack-bomb guard respected - apply_mpack: returns MPackApplyResult with counts - Round-trip: build_mpack → apply_mpack recovers all data - MPack replaces PackBundle everywhere — PackBundle no longer exists - Phase 2 — MPackMeta: mode, base_commits, created_at always present """ from __future__ import annotations from collections.abc import Mapping import datetime import json import pathlib import pytest from muse.core.ids import hash_commit as compute_commit_id, hash_snapshot as compute_snapshot_id from muse.core.commits import ( CommitRecord, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, write_snapshot, ) from muse.core.object_store import object_path, write_object from muse.core.types import blob_id, long_id from muse.core.paths import heads_dir, muse_dir, snapshots_dir from muse.core.commits import read_commit _DT = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) def _init_repo(tmp_path: pathlib.Path) -> pathlib.Path: muse = muse_dir(tmp_path) for d in ("commits", "snapshots", "objects", "refs/heads"): (muse / d).mkdir(parents=True, exist_ok=True) (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8") (muse / "repo.json").write_text( json.dumps({"repo_id": "mpack-mpack-test", "domain": "code"}), encoding="utf-8", ) return tmp_path def _commit(root: pathlib.Path, files: Mapping[str, bytes]) -> str: manifest = {} for path, content in files.items(): oid = blob_id(content) write_object(root, oid, content) manifest[path] = oid snap_id = compute_snapshot_id(manifest) write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest, created_at=_DT)) parent_ref = heads_dir(root) / "main" parent = parent_ref.read_text().strip() if parent_ref.exists() else None parent_ids = [parent] if parent else [] cid = compute_commit_id( parent_ids=parent_ids, snapshot_id=snap_id, message="test", committed_at_iso=_DT.isoformat(), ) write_commit(root, CommitRecord( commit_id=cid, branch="main", snapshot_id=snap_id, message="test", committed_at=_DT, parent_commit_id=parent, )) parent_ref.write_text(cid, encoding="utf-8") return cid # --------------------------------------------------------------------------- # MPack TypedDict # --------------------------------------------------------------------------- class TestMPackBundleTypedDict: def test_exists(self) -> None: from muse.core.mpack import MPack assert MPack is not None def test_has_commits_field(self) -> None: from muse.core.mpack import MPack from typing import get_type_hints hints = get_type_hints(MPack) assert "commits" in hints def test_has_snapshots_field(self) -> None: from muse.core.mpack import MPack from typing import get_type_hints hints = get_type_hints(MPack) assert "snapshots" in hints def test_has_objects_field(self) -> None: from muse.core.mpack import MPack from typing import get_type_hints hints = get_type_hints(MPack) assert "objects" in hints def test_has_tags_field(self) -> None: from muse.core.mpack import MPack from typing import get_type_hints hints = get_type_hints(MPack) assert "tags" in hints def test_pack_bundle_no_longer_exists(self) -> None: """PackBundle name is gone — MPack is the only public type.""" with pytest.raises(ImportError): from muse.core.mpack import PackBundle # noqa: F401 # --------------------------------------------------------------------------- # MPackSummary TypedDict # --------------------------------------------------------------------------- class TestMPackSummary: def test_exists(self) -> None: from muse.core.mpack import MPackSummary assert MPackSummary is not None def test_has_required_advisory_fields(self) -> None: from muse.core.mpack import MPackSummary from typing import get_type_hints hints = get_type_hints(MPackSummary) for field in ("commits_count", "objects_count", "objects_bytes", "agent_ids"): assert field in hints, f"MPackSummary missing field {field!r}" def test_has_branch_fields(self) -> None: from muse.core.mpack import MPackSummary from typing import get_type_hints hints = get_type_hints(MPackSummary) assert "branches" in hints # --------------------------------------------------------------------------- # build_mpack # --------------------------------------------------------------------------- class TestBuildMpack: def test_single_commit(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack root = _init_repo(tmp_path) cid = _commit(root, {"a.py": b"# a\n"}) mpack = build_mpack(root, [cid]) assert len(mpack["commits"]) == 1 assert len(mpack["snapshots"]) >= 1 assert len(mpack["objects"]) >= 1 def test_have_exclusion(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack root = _init_repo(tmp_path) c1 = _commit(root, {"a.py": b"# a\n"}) c2 = _commit(root, {"b.py": b"# b\n"}) mpack = build_mpack(root, [c2], have=[c1]) commit_ids = [c["commit_id"] for c in mpack["commits"]] assert c2 in commit_ids assert c1 not in commit_ids def test_raises_on_missing_snapshot(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack root = _init_repo(tmp_path) cid = _commit(root, {"a.py": b"# a\n"}) # Delete the snapshot from the unified object store to simulate corruption rec = read_commit(root, cid) assert rec is not None snap_file = object_path(root, rec.snapshot_id) if snap_file.exists(): snap_file.unlink() with pytest.raises(ValueError, match="snapshot"): build_mpack(root, [cid]) def test_empty_commit_ids_returns_empty_bundle(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack root = _init_repo(tmp_path) mpack = build_mpack(root, []) assert mpack.get("commits", []) == [] assert mpack.get("objects", []) == [] def test_objects_have_sha256_prefixed_ids(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack root = _init_repo(tmp_path) cid = _commit(root, {"main.py": b"print('hello')\n"}) mpack = build_mpack(root, [cid]) for obj in mpack["objects"]: assert obj["object_id"].startswith("sha256:"), ( f"object_id not sha256-prefixed: {obj['object_id']!r}" ) def test_multi_commit_chain(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack root = _init_repo(tmp_path) c1 = _commit(root, {"a.py": b"v1\n"}) c2 = _commit(root, {"a.py": b"v2\n"}) c3 = _commit(root, {"a.py": b"v3\n"}) mpack = build_mpack(root, [c3]) commit_ids = {c["commit_id"] for c in mpack["commits"]} assert c1 in commit_ids assert c2 in commit_ids assert c3 in commit_ids def test_summary_populated(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack root = _init_repo(tmp_path) _commit(root, {"a.py": b"# a\n"}) cid = _commit(root, {"b.py": b"# b\n"}) mpack = build_mpack(root, [cid]) summary = mpack.get("summary") assert summary is not None assert summary["commits_count"] >= 1 assert summary["objects_count"] >= 1 assert summary["objects_bytes"] >= 0 # --------------------------------------------------------------------------- # apply_mpack # --------------------------------------------------------------------------- class TestApplyMpack: def test_round_trip(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack, apply_mpack src = _init_repo(tmp_path / "src") dst = _init_repo(tmp_path / "dst") cid = _commit(src, {"a.py": b"# hello\n"}) mpack = build_mpack(src, [cid]) result = apply_mpack(dst, mpack) assert result["commits_written"] >= 1 assert result["objects_written"] >= 1 def test_idempotent(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack, apply_mpack root = _init_repo(tmp_path) cid = _commit(root, {"x.py": b"# x\n"}) mpack = build_mpack(root, [cid]) apply_mpack(root, mpack) result2 = apply_mpack(root, mpack) assert result2["commits_written"] == 0 assert result2["objects_skipped"] >= 1 def test_empty_bundle_is_noop(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import MPack, apply_mpack root = _init_repo(tmp_path) empty: MPack = {} result = apply_mpack(root, empty) assert result["commits_written"] == 0 assert result["objects_written"] == 0 def test_pack_bomb_rejected(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import MPack, apply_mpack root = _init_repo(tmp_path) # Create a mpack claiming 100k objects (far exceeds MAX_PACK_OBJECTS) fake_objects = [{"object_id": long_id('a'*64), "content": b"x"}] * 100_001 mpack: MPack = {"objects": fake_objects} # type: ignore[typeddict-item] with pytest.raises(ValueError, match="limit"): apply_mpack(root, mpack) def test_returns_apply_result(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack, apply_mpack src = _init_repo(tmp_path / "src") dst = _init_repo(tmp_path / "dst") cid = _commit(src, {"f.py": b"# f\n"}) mpack = build_mpack(src, [cid]) result = apply_mpack(dst, mpack) for field in ("commits_written", "snapshots_written", "objects_written", "objects_skipped"): assert field in result, f"apply_mpack result missing {field!r}" # --------------------------------------------------------------------------- # Phase 2 — MPackMeta # --------------------------------------------------------------------------- class TestMPackMeta: """build_mpack always writes a self-describing ``meta`` field.""" # ----------------------------------------------------------------- # MPackMeta TypedDict exists and is annotated # ----------------------------------------------------------------- def test_mpack_meta_typeddict_exists(self) -> None: from muse.core.mpack import MPackMeta assert MPackMeta is not None def test_mpack_meta_has_mode_annotation(self) -> None: from muse.core.mpack import MPackMeta from typing import get_type_hints hints = get_type_hints(MPackMeta) assert "mode" in hints def test_mpack_meta_has_base_commits_annotation(self) -> None: from muse.core.mpack import MPackMeta from typing import get_type_hints hints = get_type_hints(MPackMeta) assert "base_commits" in hints def test_mpack_meta_has_created_at_annotation(self) -> None: from muse.core.mpack import MPackMeta from typing import get_type_hints hints = get_type_hints(MPackMeta) assert "created_at" in hints def test_mpack_bundle_has_meta_field(self) -> None: from muse.core.mpack import MPack from typing import get_type_hints hints = get_type_hints(MPack) assert "meta" in hints # ----------------------------------------------------------------- # Full mpack (no have) — mode == "full", base_commits == [] # ----------------------------------------------------------------- def test_full_mpack_has_meta(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack repo = _init_repo(tmp_path) cid = _commit(repo, {"a.py": b"content"}) mpack = build_mpack(repo, [cid]) assert "meta" in mpack def test_full_mpack_mode_is_full(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack repo = _init_repo(tmp_path) cid = _commit(repo, {"a.py": b"content"}) mpack = build_mpack(repo, [cid]) assert mpack["meta"]["mode"] == "full" def test_full_mpack_base_commits_empty(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack repo = _init_repo(tmp_path) cid = _commit(repo, {"a.py": b"content"}) mpack = build_mpack(repo, [cid]) assert mpack["meta"]["base_commits"] == [] def test_full_mpack_created_at_is_str(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack repo = _init_repo(tmp_path) cid = _commit(repo, {"a.py": b"content"}) mpack = build_mpack(repo, [cid]) assert isinstance(mpack["meta"]["created_at"], str) assert len(mpack["meta"]["created_at"]) > 10 # not empty def test_full_mpack_created_at_is_iso(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack import datetime repo = _init_repo(tmp_path) cid = _commit(repo, {"a.py": b"content"}) mpack = build_mpack(repo, [cid]) # Must be parseable as ISO 8601 dt = datetime.datetime.fromisoformat(mpack["meta"]["created_at"].rstrip("Z")) assert dt.year >= 2024 # ----------------------------------------------------------------- # Incremental mpack (have set) — mode == "incremental" # ----------------------------------------------------------------- def test_incremental_mpack_mode_is_incremental(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack repo = _init_repo(tmp_path) base_cid = _commit(repo, {"a.py": b"v1"}) tip_cid = _commit(repo, {"a.py": b"v2"}) mpack = build_mpack(repo, [tip_cid], have=[base_cid]) assert mpack["meta"]["mode"] == "incremental" def test_incremental_mpack_base_commits_populated(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack repo = _init_repo(tmp_path) base_cid = _commit(repo, {"a.py": b"v1"}) tip_cid = _commit(repo, {"a.py": b"v2"}) mpack = build_mpack(repo, [tip_cid], have=[base_cid]) assert base_cid in mpack["meta"]["base_commits"] def test_incremental_mpack_multiple_have(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack repo = _init_repo(tmp_path) c1 = _commit(repo, {"a.py": b"v1"}) c2 = _commit(repo, {"a.py": b"v2"}) c3 = _commit(repo, {"a.py": b"v3"}) mpack = build_mpack(repo, [c3], have=[c1, c2]) assert set(mpack["meta"]["base_commits"]) == {c1, c2} def test_incremental_mpack_has_created_at(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack repo = _init_repo(tmp_path) base = _commit(repo, {"a.py": b"base"}) tip = _commit(repo, {"a.py": b"tip"}) mpack = build_mpack(repo, [tip], have=[base]) assert isinstance(mpack["meta"]["created_at"], str) # ----------------------------------------------------------------- # Empty have list treated as full # ----------------------------------------------------------------- def test_empty_have_list_means_full(self, tmp_path: pathlib.Path) -> None: from muse.core.mpack import build_mpack repo = _init_repo(tmp_path) cid = _commit(repo, {"f.py": b"data"}) mpack = build_mpack(repo, [cid], have=[]) assert mpack["meta"]["mode"] == "full" assert mpack["meta"]["base_commits"] == []