"""Supercharge tests for ``muse pack-objects``, ``unpack-objects``, and ``verify-pack``. TDD — [RED] tests fail until the feature lands; [GREEN] tests fill existing gaps. New features under test ----------------------- - ``duration_ms`` [RED] — wall-clock ms in every JSON output path - ``exit_code`` [RED] — always present in every JSON output path - ``object_bytes`` [RED] — total raw bytes in ``pack-objects --dry-run`` Gap-fill coverage [GREEN] -------------------------- - dry-run keys validated exhaustively (want, have, commits, snapshots, blobs) - unpack round-trip output fields present (commits_written, blobs_written, …) - verify-pack all_ok field and failures list - stat mode counts correct """ from __future__ import annotations from collections.abc import Mapping import datetime import json import pathlib import msgpack import pytest from muse.core.errors import ExitCode from muse.core.object_store import write_object 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 tests.cli_test_helper import CliRunner, InvokeResult from muse.core.types import long_id, blob_id from muse.core.paths import config_toml_path, heads_dir, muse_dir, ref_path runner = CliRunner() _TS = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) # --------------------------------------------------------------------------- # Shared helpers # --------------------------------------------------------------------------- def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path: repo = tmp_path / "repo" muse = muse_dir(repo) for sub in ("objects", "commits", "snapshots", "refs/heads"): (muse / sub).mkdir(parents=True) (muse / "HEAD").write_text("ref: refs/heads/main") (muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"})) return repo def _write_obj(repo: pathlib.Path, content: bytes) -> str: oid = blob_id(content) write_object(repo, oid, content) return oid def _commit( repo: pathlib.Path, msg: str, manifest: dict[str, str], *, branch: str = "main", parent: str | None = None, ) -> str: sid = compute_snapshot_id(manifest) write_snapshot(repo, SnapshotRecord(snapshot_id=sid, manifest=manifest, created_at=_TS)) parent_ids = [parent] if parent else [] cid = compute_commit_id( parent_ids=parent_ids, snapshot_id=sid, message=msg, committed_at_iso=_TS.isoformat(), author="gabriel",) write_commit(repo, CommitRecord( commit_id=cid, branch=branch, snapshot_id=sid, message=msg, committed_at=_TS, author="gabriel", parent_commit_id=parent, parent2_commit_id=None, )) ref = ref_path(repo, branch) ref.parent.mkdir(parents=True, exist_ok=True) ref.write_text(cid) return cid def _pack(repo: pathlib.Path, *args: str) -> InvokeResult: return runner.invoke(None, ["pack-objects", *args], env={"MUSE_REPO_ROOT": str(repo)}) def _unpack(repo: pathlib.Path, mpack: bytes, *args: str) -> InvokeResult: extra = [] if "--json" in args else ["--json"] return runner.invoke( None, ["unpack-objects", *extra, *args], env={"MUSE_REPO_ROOT": str(repo)}, input=mpack, ) def _verify(repo: pathlib.Path, mpack: bytes, *args: str) -> InvokeResult: extra = [] if "--json" in args else ["--json"] return runner.invoke( None, ["verify-pack", *extra, *args], env={"MUSE_REPO_ROOT": str(repo)}, input=mpack, ) def _make_bundle(repo: pathlib.Path) -> bytes: """Pack HEAD and return raw msgpack bytes.""" oid = _write_obj(repo, b"hello") _commit(repo, "init", {"f.py": oid}) r = _pack(repo, "HEAD") assert r.exit_code == 0, r.output return r.stdout_bytes # raw binary from stdout.buffer def _json_out(r: InvokeResult) -> Mapping[str, object]: for line in r.output.splitlines(): line = line.strip() if line.startswith("{"): return json.loads(line) raise ValueError(f"No JSON in output:\n{r.output!r}") # --------------------------------------------------------------------------- # pack-objects --dry-run: duration_ms, exit_code, object_bytes [RED] # --------------------------------------------------------------------------- class TestPackObjectsDryRunSupercharge: """[RED] New fields in --dry-run JSON output.""" def test_dry_run_has_duration_ms(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) r = _pack(repo, "HEAD", "--dry-run") assert r.exit_code == 0 d = _json_out(r) assert "duration_ms" in d def test_dry_run_duration_ms_non_negative(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) r = _pack(repo, "HEAD", "--dry-run") d = _json_out(r) assert d["duration_ms"] >= 0.0 def test_dry_run_has_exit_code(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) r = _pack(repo, "HEAD", "--dry-run") d = _json_out(r) assert "exit_code" in d def test_dry_run_exit_code_is_zero_on_success(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) r = _pack(repo, "HEAD", "--dry-run") d = _json_out(r) assert d["exit_code"] == 0 def test_dry_run_has_object_bytes(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) content = b"some content here" oid = _write_obj(repo, content) _commit(repo, "init", {"f.py": oid}) r = _pack(repo, "HEAD", "--dry-run") d = _json_out(r) assert "object_bytes" in d def test_dry_run_object_bytes_matches_content_size(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) content = b"x" * 256 oid = _write_obj(repo, content) _commit(repo, "init", {"f.py": oid}) r = _pack(repo, "HEAD", "--dry-run") d = _json_out(r) assert d["object_bytes"] == 256 def test_dry_run_object_bytes_sums_multiple_objects(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid_a = _write_obj(repo, b"a" * 100) oid_b = _write_obj(repo, b"b" * 200) _commit(repo, "init", {"a.py": oid_a, "b.py": oid_b}) r = _pack(repo, "HEAD", "--dry-run") d = _json_out(r) assert d["object_bytes"] == 300 def test_dry_run_object_bytes_is_int(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_obj(repo, b"y") _commit(repo, "init", {"f.py": oid}) r = _pack(repo, "HEAD", "--dry-run") d = _json_out(r) assert isinstance(d["object_bytes"], int) def test_dry_run_object_bytes_zero_for_empty_pack(self, tmp_path: pathlib.Path) -> None: """--have HEAD means nothing new to pack → 0 objects → 0 bytes.""" repo = _make_repo(tmp_path) oid = _write_obj(repo, b"z") cid = _commit(repo, "init", {"f.py": oid}) r = _pack(repo, cid, "--have", cid, "--dry-run") d = _json_out(r) assert d["object_bytes"] == 0 # --------------------------------------------------------------------------- # pack-objects --dry-run: existing fields still present [GREEN] # --------------------------------------------------------------------------- class TestPackObjectsDryRunGreen: """[GREEN] Existing dry-run fields remain after adding new ones.""" def test_want_field_present(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) d = _json_out(_pack(repo, "HEAD", "--dry-run")) assert "want" in d def test_have_field_present(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) d = _json_out(_pack(repo, "HEAD", "--dry-run")) assert "have" in d def test_commits_field_present(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) d = _json_out(_pack(repo, "HEAD", "--dry-run")) assert "commits" in d def test_snapshots_field_present(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) d = _json_out(_pack(repo, "HEAD", "--dry-run")) assert "snapshots" in d def test_blobs_field_present(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) d = _json_out(_pack(repo, "HEAD", "--dry-run")) assert "blobs" in d def test_have_pruning_reduces_blobs(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid = _write_obj(repo, b"v1") c1 = _commit(repo, "c1", {"f.py": oid}) oid2 = _write_obj(repo, b"v2") _commit(repo, "c2", {"f.py": oid2}, parent=c1) full = _json_out(_pack(repo, "HEAD", "--dry-run")) pruned = _json_out(_pack(repo, "HEAD", "--have", c1, "--dry-run")) assert pruned["blobs"] < full["blobs"] # --------------------------------------------------------------------------- # unpack-objects: duration_ms and exit_code [RED] # --------------------------------------------------------------------------- class TestUnpackObjectsSupercharge: """[RED] duration_ms and exit_code in unpack-objects JSON output.""" def test_unpack_json_has_duration_ms(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) dest = _make_repo(tmp_path / "dest") r = _unpack(dest, mpack) assert r.exit_code == 0 d = _json_out(r) assert "duration_ms" in d def test_unpack_duration_ms_non_negative(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) dest = _make_repo(tmp_path / "dest") d = _json_out(_unpack(dest, mpack)) assert d["duration_ms"] >= 0.0 def test_unpack_json_has_exit_code(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) dest = _make_repo(tmp_path / "dest") d = _json_out(_unpack(dest, mpack)) assert "exit_code" in d def test_unpack_exit_code_zero_on_success(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) dest = _make_repo(tmp_path / "dest") d = _json_out(_unpack(dest, mpack)) assert d["exit_code"] == 0 def test_unpack_duration_ms_under_two_seconds(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) for i in range(20): oid = _write_obj(repo, f"content {i}".encode() * 50) _commit(repo, f"c{i}", {f"f{i}.py": oid}, parent=None if i == 0 else None) # single chain not needed for pack mpack = _make_bundle(repo) dest = _make_repo(tmp_path / "dest") d = _json_out(_unpack(dest, mpack)) assert d["duration_ms"] < 2000.0 # --------------------------------------------------------------------------- # unpack-objects: existing output fields still present [GREEN] # --------------------------------------------------------------------------- class TestUnpackObjectsGreen: def test_commits_written_field(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) dest = _make_repo(tmp_path / "dest") d = _json_out(_unpack(dest, mpack)) assert "commits_written" in d def test_snapshots_written_field(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) dest = _make_repo(tmp_path / "dest") d = _json_out(_unpack(dest, mpack)) assert "snapshots_written" in d def test_blobs_written_field(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) dest = _make_repo(tmp_path / "dest") d = _json_out(_unpack(dest, mpack)) assert "blobs_written" in d def test_blobs_skipped_field(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) dest = _make_repo(tmp_path / "dest") d = _json_out(_unpack(dest, mpack)) assert "blobs_skipped" in d def test_idempotent_second_unpack_skips_all(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) dest = _make_repo(tmp_path / "dest") _unpack(dest, mpack) d = _json_out(_unpack(dest, mpack)) assert d["blobs_written"] == 0 # --------------------------------------------------------------------------- # verify-pack: duration_ms and exit_code [RED] # --------------------------------------------------------------------------- class TestVerifyPackSupercharge: """[RED] duration_ms and exit_code in verify-pack JSON output.""" def test_verify_json_has_duration_ms(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) r = _verify(repo, mpack) assert r.exit_code == 0 d = _json_out(r) assert "duration_ms" in d def test_verify_duration_ms_non_negative(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) d = _json_out(_verify(repo, mpack)) assert d["duration_ms"] >= 0.0 def test_verify_json_has_exit_code(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) d = _json_out(_verify(repo, mpack)) assert "exit_code" in d def test_verify_exit_code_zero_on_clean(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) d = _json_out(_verify(repo, mpack)) assert d["exit_code"] == 0 def test_verify_exit_code_nonzero_on_corrupt(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) # Corrupt the mpack: tamper the blob content inside the parsed structure raw = msgpack.unpackb(mpack, raw=False) raw["blobs"][0]["content"] = b"tampered!" corrupted = msgpack.packb(raw, use_bin_type=True) r = _verify(repo, corrupted) # Either fails to parse or reports integrity failure assert r.exit_code != 0 or not _json_out(r).get("all_ok", True) def test_verify_stat_has_duration_ms(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) r = _verify(repo, mpack, "--stat") assert r.exit_code == 0 d = _json_out(r) assert "duration_ms" in d def test_verify_stat_has_exit_code(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) d = _json_out(_verify(repo, mpack, "--stat")) assert "exit_code" in d def test_verify_duration_ms_under_two_seconds(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) for i in range(50): _write_obj(repo, f"obj {i}".encode() * 100) mpack = _make_bundle(repo) d = _json_out(_verify(repo, mpack)) assert d["duration_ms"] < 2000.0 # --------------------------------------------------------------------------- # verify-pack: existing fields still present [GREEN] # --------------------------------------------------------------------------- class TestVerifyPackGreen: def test_all_ok_field_clean_bundle(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) d = _json_out(_verify(repo, mpack)) assert d["all_ok"] is True def test_failures_empty_on_clean_bundle(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) d = _json_out(_verify(repo, mpack)) assert d["failures"] == [] def test_objects_checked_field(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) d = _json_out(_verify(repo, mpack)) assert "blobs_checked" in d def test_stat_objects_count(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) oid_a = _write_obj(repo, b"a") oid_b = _write_obj(repo, b"b") _commit(repo, "init", {"a.py": oid_a, "b.py": oid_b}) mpack = _pack(repo, "HEAD").stdout_bytes d = _json_out(_verify(repo, mpack, "--stat")) assert d["blobs"] >= 2 def test_stat_commits_count(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) mpack = _make_bundle(repo) d = _json_out(_verify(repo, mpack, "--stat")) assert d["commits"] >= 1 # --------------------------------------------------------------------------- # Phase 3 — build_mpack fails loudly on MISSING objects [RED] # --------------------------------------------------------------------------- def _write_promisor_config(repo: pathlib.Path, remote_name: str = "origin") -> None: config_path = config_toml_path(repo) config_path.write_text( f"[remotes.{remote_name}]\n" f'url = "https://localhost:1337/test/repo"\n', encoding="utf-8", ) class TestPackObjectsMissingObjectValidation: """pack-objects fails loudly when a snapshot references a MISSING object.""" def test_missing_object_exits_nonzero(self, tmp_path: pathlib.Path) -> None: """pack-objects exits nonzero when a snapshot refs an object absent with no promisor.""" repo = _make_repo(tmp_path) # Write snapshot that refs an object we deliberately do NOT write from muse.core.ids import hash_snapshot as compute_snapshot_id from muse.core.commits import ( CommitRecord, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, write_snapshot, ) missing_oid = long_id("a" * 64) sid = compute_snapshot_id({"missing.py": missing_oid}) write_snapshot(repo, SnapshotRecord(snapshot_id=sid, manifest={"missing.py": missing_oid}, created_at=_TS)) cid = _commit.__wrapped__(repo, "broken commit", {"missing.py": missing_oid}) if hasattr(_commit, "__wrapped__") else None # Use the helpers directly from muse.core.ids import hash_commit as compute_commit_id cid = compute_commit_id( parent_ids=[], snapshot_id=sid, message="broken commit", committed_at_iso=_TS.isoformat(), author="gabriel",) write_commit(repo, CommitRecord( commit_id=cid, branch="main", snapshot_id=sid, message="broken commit", committed_at=_TS, author="gabriel", parent_commit_id=None, parent2_commit_id=None, )) (heads_dir(repo) / "main").write_text(cid) r = _pack(repo, "HEAD") assert r.exit_code != 0 def test_missing_object_error_mentions_object_id(self, tmp_path: pathlib.Path) -> None: """Error output names the missing object so the user knows what to fix.""" repo = _make_repo(tmp_path) missing_oid = long_id("b" * 64) from muse.core.ids import hash_snapshot as compute_snapshot_id, hash_commit as compute_commit_id from muse.core.commits import ( CommitRecord, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, write_snapshot, ) sid = compute_snapshot_id({"gone.py": missing_oid}) write_snapshot(repo, SnapshotRecord(snapshot_id=sid, manifest={"gone.py": missing_oid}, created_at=_TS)) cid = compute_commit_id( parent_ids=[], snapshot_id=sid, message="gone", committed_at_iso=_TS.isoformat(), author="gabriel",) write_commit(repo, CommitRecord( commit_id=cid, branch="main", snapshot_id=sid, message="gone", committed_at=_TS, author="gabriel", parent_commit_id=None, parent2_commit_id=None, )) (heads_dir(repo) / "main").write_text(cid) r = _pack(repo, "HEAD") assert r.exit_code != 0 # Error should mention the missing object or "missing" assert "missing" in (r.output + r.stderr).lower() or "absent" in (r.output + r.stderr).lower() def test_promised_object_does_not_fail(self, tmp_path: pathlib.Path) -> None: """PROMISED objects (promisor remote configured) are skipped, not failures.""" repo = _make_repo(tmp_path) _write_promisor_config(repo) missing_oid = long_id("c" * 64) from muse.core.ids import hash_snapshot as compute_snapshot_id, hash_commit as compute_commit_id from muse.core.commits import ( CommitRecord, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, write_snapshot, ) sid = compute_snapshot_id({"remote.py": missing_oid}) write_snapshot(repo, SnapshotRecord(snapshot_id=sid, manifest={"remote.py": missing_oid}, created_at=_TS)) cid = compute_commit_id( parent_ids=[], snapshot_id=sid, message="partial clone", committed_at_iso=_TS.isoformat(), author="gabriel",) write_commit(repo, CommitRecord( commit_id=cid, branch="main", snapshot_id=sid, message="partial clone", committed_at=_TS, author="gabriel", parent_commit_id=None, parent2_commit_id=None, )) (heads_dir(repo) / "main").write_text(cid) r = _pack(repo, "HEAD") assert r.exit_code == 0 def test_present_object_always_passes(self, tmp_path: pathlib.Path) -> None: """Fully self-contained mpack with all objects present passes.""" repo = _make_repo(tmp_path) oid = _write_obj(repo, b"complete content") _commit(repo, "good", {"file.py": oid}) r = _pack(repo, "HEAD") assert r.exit_code == 0 class TestRegisterFlags: def test_json_short_flag(self) -> None: import argparse from muse.cli.commands.pack_objects import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) args = p.parse_args(['pack-objects', 'HEAD', '-j']) assert args.json_out is True def test_json_long_flag(self) -> None: import argparse from muse.cli.commands.pack_objects import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) args = p.parse_args(['pack-objects', 'HEAD', '--json']) assert args.json_out is True def test_default_no_json(self) -> None: import argparse from muse.cli.commands.pack_objects import register p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) # Command-specific required args may differ; just check dest exists when possible try: args = p.parse_args(['pack-objects', 'HEAD']) assert args.json_out is False except SystemExit: pass # required positional args missing — flag default still correct