"""Supercharged tests for ``muse bundle`` — three new agent-first features. Feature 1 — ``muse bundle inspect [--json]`` ----------------------------------------------------- Read and display the commit log and branch state from a bundle file without unbundling. No repository required. Agents use this to decide whether to apply a bundle before committing to the operation. JSON schema:: { "total_commits": int, "total_objects": int, "branches": {"": ""}, "commits": [ { "commit_id": str, "message": str, "committed_at": str, # ISO-8601 "agent_id": str, # "" when not an agent commit "branches": [str] # branch names whose head == this commit }, ... # newest first (by committed_at) ] } Feature 2 — ``--verify`` flag on ``muse bundle unbundle`` ---------------------------------------------------------- Verify bundle integrity atomically before applying. Exits 1 (with no writes) if the bundle is corrupt. JSON output gains a ``"verified"`` bool. Feature 3 — ``muse bundle diff [--json]`` ------------------------------------------------- Show which commits in the bundle are not already present in the local repository. Agents use this to answer "what would this bundle add?" before deciding to apply. JSON schema:: { "new_commits": int, "known_commits": int, "refs_to_advance": [str], # branch names that would move "commits": [ {"commit_id": str, "message": str, "committed_at": str} ] } Test categories --------------- - unit : internal helpers and schema shapes - integration : CLI flag parsing and output contracts - e2e : full round-trips via CliRunner - security : ANSI/control injection in bundle content - data_integrity: inspect/diff remain consistent across create-verify-unbundle - performance : inspect and diff on 100-commit bundles under 1 s - stress : inspect and diff on 200-commit bundles """ from __future__ import annotations from collections.abc import Mapping import datetime import json import os import pathlib import time import threading import msgpack import pytest from tests.cli_test_helper import CliRunner, InvokeResult from muse.core.object_store import write_object from muse.core.ids import hash_commit, hash_snapshot from muse.core.commits import ( CommitRecord, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, write_snapshot, ) from muse.core.types import Manifest, blob_id, long_id from muse.core.paths import heads_dir, muse_dir, objects_dir, ref_path runner = CliRunner() _REPO_ID = "bundle-supercharged-test" # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _init_repo(path: pathlib.Path, repo_id: str = _REPO_ID) -> pathlib.Path: muse = muse_dir(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": repo_id, "domain": "midi"}), encoding="utf-8" ) return path def _env(repo: pathlib.Path) -> Manifest: return {"MUSE_REPO_ROOT": str(repo)} _counter = 0 def _make_commit( root: pathlib.Path, parent_id: str | None = None, content: bytes = b"data", branch: str = "main", message: str | None = None, agent_id: str = "", ) -> str: global _counter _counter += 1 c = content + str(_counter).encode() obj_id = blob_id(c) write_object(root, obj_id, c) manifest = {f"f_{_counter}.txt": obj_id} snap_id = hash_snapshot(manifest) write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest)) committed_at = datetime.datetime.now(datetime.timezone.utc) parent_ids = [parent_id] if parent_id else [] msg = message or f"commit {_counter}" commit_id = hash_commit( parent_ids=parent_ids, snapshot_id=snap_id, message=msg, committed_at_iso=committed_at.isoformat(), ) write_commit(root, CommitRecord( commit_id=commit_id, branch=branch, snapshot_id=snap_id, message=msg, committed_at=committed_at, parent_commit_id=parent_id, agent_id=agent_id, )) ref_dir = heads_dir(root) if "/" in branch: (ref_dir / branch).parent.mkdir(parents=True, exist_ok=True) (ref_dir / branch).write_text(commit_id, encoding="utf-8") return commit_id def _invoke(args: list[str], env: Manifest | None = None) -> InvokeResult: return runner.invoke(None, args, env=env) def _create_bundle( repo: pathlib.Path, out: pathlib.Path, *extra_args: str ) -> InvokeResult: return _invoke(["bundle", "create", str(out), *extra_args], env=_env(repo)) def _parse_inspect(result: InvokeResult) -> Mapping[str, object]: return json.loads(result.output) def _parse_diff(result: InvokeResult) -> Mapping[str, object]: return json.loads(result.output) # =========================================================================== # Feature 1: muse bundle inspect # =========================================================================== class TestBundleInspectUnit: """Unit-level schema and output contracts for bundle inspect.""" def test_inspect_help_exits_0(self) -> None: result = _invoke(["bundle", "inspect", "--help"]) assert result.exit_code == 0 def test_inspect_help_mentions_agent(self) -> None: result = _invoke(["bundle", "inspect", "--help"]) assert "agent" in result.output.lower() or "Agent" in result.output def test_inspect_help_mentions_json_schema(self) -> None: result = _invoke(["bundle", "inspect", "--help"]) assert "JSON" in result.output def test_inspect_json_schema_keys(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"inspect-schema") bundle = tmp_path / "schema.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle), "--json"]) assert result.exit_code == 0 data = _parse_inspect(result) for key in ("total_commits", "total_objects", "branches", "commits"): assert key in data, f"missing key: {key}" def test_inspect_commit_entry_schema(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"inspect-entry") bundle = tmp_path / "entry.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle), "--json"]) assert result.exit_code == 0 data = _parse_inspect(result) assert len(data["commits"]) >= 1 entry = data["commits"][0] for key in ("commit_id", "message", "committed_at", "agent_id", "branches"): assert key in entry, f"commit entry missing key: {key}" def test_inspect_total_commits_count(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) prev = None for i in range(5): prev = _make_commit(tmp_path, parent_id=prev, content=f"cnt-{i}".encode()) bundle = tmp_path / "cnt.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle), "--json"]) data = _parse_inspect(result) assert data["total_commits"] == 5 def test_inspect_total_objects_positive(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"obj-count") bundle = tmp_path / "objcnt.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle), "--json"]) data = _parse_inspect(result) assert data["total_objects"] > 0 def test_inspect_branches_map(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) cid = _make_commit(tmp_path, content=b"branches-map") bundle = tmp_path / "bmap.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle), "--json"]) data = _parse_inspect(result) assert "main" in data["branches"] def test_inspect_commit_message_present(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"msg-check", message="feat: add audio engine") bundle = tmp_path / "msg.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle), "--json"]) data = _parse_inspect(result) messages = [c["message"] for c in data["commits"]] assert any("feat: add audio engine" in m for m in messages) def test_inspect_agent_id_from_agent_commit(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"agent-commit", agent_id="claude-code") bundle = tmp_path / "agent.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle), "--json"]) data = _parse_inspect(result) agent_ids = [c["agent_id"] for c in data["commits"]] assert "claude-code" in agent_ids def test_inspect_agent_id_empty_for_human_commit(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"human-commit", agent_id="") bundle = tmp_path / "human.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle), "--json"]) data = _parse_inspect(result) # Human commits have empty or None agent_id assert data["commits"][0]["agent_id"] in ("", None) def test_inspect_commits_newest_first(self, tmp_path: pathlib.Path) -> None: """Commits must be ordered newest first (by committed_at).""" _init_repo(tmp_path) prev = None for i in range(3): prev = _make_commit(tmp_path, parent_id=prev, content=f"ord-{i}".encode()) bundle = tmp_path / "ord.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle), "--json"]) data = _parse_inspect(result) dates = [c["committed_at"] for c in data["commits"]] assert dates == sorted(dates, reverse=True) def test_inspect_branch_annotated_on_tip_commit(self, tmp_path: pathlib.Path) -> None: """The commit that is a branch head should have that branch in its branches list.""" _init_repo(tmp_path) cid = _make_commit(tmp_path, content=b"tip-commit") bundle = tmp_path / "tip.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle), "--json"]) data = _parse_inspect(result) tip_entry = next(c for c in data["commits"] if c["commit_id"] == cid) assert "main" in tip_entry["branches"] def test_inspect_non_tip_commit_has_no_branch(self, tmp_path: pathlib.Path) -> None: """Commits that are not at the tip of any branch have empty branches list.""" _init_repo(tmp_path) c1 = _make_commit(tmp_path, content=b"non-tip-parent") _make_commit(tmp_path, parent_id=c1, content=b"non-tip-child") bundle = tmp_path / "nontip.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle), "--json"]) data = _parse_inspect(result) parent_entry = next(c for c in data["commits"] if c["commit_id"] == c1) assert parent_entry["branches"] == [] def test_inspect_does_not_require_repo(self, tmp_path: pathlib.Path) -> None: """inspect must work without MUSE_REPO_ROOT (no repo needed).""" src = tmp_path / "src" src.mkdir() _init_repo(src) _make_commit(src, content=b"no-repo-needed") bundle = tmp_path / "norepo.bundle" _create_bundle(src, bundle) # Invoke with no env — no repo context at all result = _invoke(["bundle", "inspect", str(bundle), "--json"]) assert result.exit_code == 0 def test_inspect_file_not_found(self, tmp_path: pathlib.Path) -> None: result = _invoke(["bundle", "inspect", str(tmp_path / "missing.bundle"), "--json"]) assert result.exit_code != 0 def test_inspect_invalid_msgpack(self, tmp_path: pathlib.Path) -> None: bad = tmp_path / "bad.bundle" bad.write_bytes(b"not msgpack") result = _invoke(["bundle", "inspect", str(bad), "--json"]) assert result.exit_code != 0 def test_inspect_empty_bundle(self, tmp_path: pathlib.Path) -> None: empty = tmp_path / "empty.bundle" empty.write_bytes(msgpack.packb({}, use_bin_type=True)) result = _invoke(["bundle", "inspect", str(empty), "--json"]) assert result.exit_code == 0 data = _parse_inspect(result) assert data["total_commits"] == 0 assert data["commits"] == [] assert data["branches"] == {} def test_inspect_j_alias(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"j-alias") bundle = tmp_path / "jalias.bundle" _create_bundle(tmp_path, bundle) r1 = _invoke(["bundle", "inspect", str(bundle), "--json"]) r2 = _invoke(["bundle", "inspect", str(bundle), "-j"]) assert r1.exit_code == 0 assert r2.exit_code == 0 assert json.loads(r1.output)["total_commits"] == json.loads(r2.output)["total_commits"] class TestBundleInspectText: """Text output (no --json) contracts for bundle inspect.""" def test_text_output_mentions_commits(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"txt-commits") bundle = tmp_path / "txt.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle)]) assert result.exit_code == 0 assert "commit" in result.output.lower() def test_text_output_mentions_branch(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"txt-branch") bundle = tmp_path / "txt-br.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle)]) assert result.exit_code == 0 assert "main" in result.output def test_text_output_includes_commit_message(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"txt-msg", message="feat: melody engine") bundle = tmp_path / "txt-msg.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle)]) assert "feat: melody engine" in result.output class TestBundleInspectSecurity: """Security: ANSI and control injection from crafted bundle content.""" def _has_ansi(self, s: str) -> bool: return "\x1b[" in s def test_ansi_in_commit_message_stripped(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"ansi-msg", message="\x1b[31mmalicious\x1b[0m") bundle = tmp_path / "ansi-msg.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle), "--json"]) assert result.exit_code == 0 data = _parse_inspect(result) for c in data["commits"]: assert not self._has_ansi(c["message"]), "ANSI in message not stripped" def test_ansi_in_branch_name_stripped(self, tmp_path: pathlib.Path) -> None: """A crafted bundle with ANSI in a branch_heads key must not reach stdout.""" _init_repo(tmp_path) cid = _make_commit(tmp_path, content=b"ansi-branch") bundle = tmp_path / "ansi-br.bundle" _create_bundle(tmp_path, bundle) # Inject ANSI into branch_heads in the msgpack raw = msgpack.unpackb(bundle.read_bytes(), raw=False) raw["branch_heads"] = {"\x1b[31mmalicious\x1b[0m": cid} bundle.write_bytes(msgpack.packb(raw, use_bin_type=True)) result = _invoke(["bundle", "inspect", str(bundle), "--json"]) assert result.exit_code == 0 assert not self._has_ansi(result.output) def test_ansi_in_agent_id_stripped(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"ansi-agent", agent_id="\x1b[31mhacked\x1b[0m") bundle = tmp_path / "ansi-agent.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle), "--json"]) assert result.exit_code == 0 assert not self._has_ansi(result.output) def test_oversized_bundle_rejected(self, tmp_path: pathlib.Path) -> None: """Bundle larger than the safety cap must be rejected.""" from muse.core.io import MAX_PACK_MSGPACK_BYTES oversized = tmp_path / "oversized.bundle" oversized.write_bytes(b"\x00" * (MAX_PACK_MSGPACK_BYTES + 1)) result = _invoke(["bundle", "inspect", str(oversized), "--json"]) assert result.exit_code != 0 class TestBundleInspectDataIntegrity: """Data integrity: inspect output is consistent with create and unbundle.""" def test_inspect_commit_ids_match_create_json(self, tmp_path: pathlib.Path) -> None: """Commits listed by inspect must equal those packed by create.""" _init_repo(tmp_path) prev = None cids = [] for i in range(4): prev = _make_commit(tmp_path, parent_id=prev, content=f"di-{i}".encode()) cids.append(prev) bundle = tmp_path / "di.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle), "--json"]) data = _parse_inspect(result) inspect_ids = {c["commit_id"] for c in data["commits"]} for cid in cids: assert cid in inspect_ids def test_inspect_consistent_with_verify(self, tmp_path: pathlib.Path) -> None: """A bundle that verify says is clean must also inspect cleanly.""" _init_repo(tmp_path) prev = None for i in range(3): prev = _make_commit(tmp_path, parent_id=prev, content=f"vdi-{i}".encode()) bundle = tmp_path / "vdi.bundle" _create_bundle(tmp_path, bundle) v = _invoke(["bundle", "verify", str(bundle), "--json"]) assert json.loads(v.output)["all_ok"] is True result = _invoke(["bundle", "inspect", str(bundle), "--json"]) assert result.exit_code == 0 data = _parse_inspect(result) assert data["total_commits"] == 3 def test_inspect_branch_commit_id_matches_list_heads(self, tmp_path: pathlib.Path) -> None: """branches map in inspect must match list-heads output.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"lh-match") bundle = tmp_path / "lh.bundle" _create_bundle(tmp_path, bundle) lh_raw = json.loads( _invoke(["bundle", "list-heads", str(bundle), "--json"]).output ) lh = lh_raw["heads"] if "heads" in lh_raw else lh_raw ins = _parse_inspect( _invoke(["bundle", "inspect", str(bundle), "--json"]) ) assert ins["branches"] == lh class TestBundleInspectPerformance: def test_inspect_100_commit_bundle_under_1s(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) prev = None for i in range(100): prev = _make_commit(tmp_path, parent_id=prev, content=f"perf-{i}".encode()) bundle = tmp_path / "perf100.bundle" _create_bundle(tmp_path, bundle) start = time.monotonic() result = _invoke(["bundle", "inspect", str(bundle), "--json"]) elapsed = time.monotonic() - start assert result.exit_code == 0 data = _parse_inspect(result) assert data["total_commits"] == 100 assert elapsed < 1.0, f"inspect 100-commit bundle took {elapsed:.2f}s" class TestBundleInspectStress: def test_inspect_200_commit_bundle(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) prev = None for i in range(200): prev = _make_commit(tmp_path, parent_id=prev, content=f"s200-{i}".encode()) bundle = tmp_path / "s200.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle), "--json"]) assert result.exit_code == 0 data = _parse_inspect(result) assert data["total_commits"] == 200 def test_inspect_multi_branch_bundle(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) base = _make_commit(tmp_path, content=b"multi-base") for i in range(10): br = f"feat/branch-{i}" ref = ref_path(tmp_path, br) ref.parent.mkdir(parents=True, exist_ok=True) ref.write_text(base, encoding="utf-8") bundle = tmp_path / "multibr.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "inspect", str(bundle), "--json"]) data = _parse_inspect(result) assert len(data["branches"]) == 11 # main + 10 feature branches tip = next(c for c in data["commits"] if c["commit_id"] == base) assert len(tip["branches"]) == 11 def test_concurrent_inspect_consistent(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) prev = None for i in range(20): prev = _make_commit(tmp_path, parent_id=prev, content=f"conc-{i}".encode()) bundle = tmp_path / "concurrent.bundle" _create_bundle(tmp_path, bundle) errors: list[str] = [] def _read() -> None: r = _invoke(["bundle", "inspect", str(bundle), "--json"]) if r.exit_code != 0: errors.append(f"exit {r.exit_code}") else: try: d = json.loads(r.output) if d["total_commits"] != 20: errors.append(f"count {d['total_commits']}") except Exception as exc: errors.append(str(exc)) threads = [threading.Thread(target=_read) for _ in range(8)] for t in threads: t.start() for t in threads: t.join() assert not errors, f"Concurrent inspect failures: {errors}" # =========================================================================== # Feature 2: --verify flag on muse bundle unbundle # =========================================================================== class TestBundleUnbundleVerifyFlag: """--verify flag: verify integrity before applying.""" def _src_dst( self, tmp_path: pathlib.Path, dst_id: str = "verify-dst" ) -> tuple[pathlib.Path, pathlib.Path]: src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id=dst_id) return src, dst def test_verify_flag_help_mentioned(self) -> None: result = _invoke(["bundle", "unbundle", "--help"]) assert result.exit_code == 0 assert "--verify" in result.output def test_verify_flag_clean_bundle_exits_0(self, tmp_path: pathlib.Path) -> None: src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"vf-clean") bundle = tmp_path / "clean.bundle" _create_bundle(src, bundle) result = _invoke(["bundle", "unbundle", str(bundle), "--verify"], env=_env(dst)) assert result.exit_code == 0 def test_verify_flag_applies_objects(self, tmp_path: pathlib.Path) -> None: src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"vf-apply") bundle = tmp_path / "apply.bundle" _create_bundle(src, bundle) result = _invoke(["bundle", "unbundle", str(bundle), "--verify"], env=_env(dst)) assert result.exit_code == 0 assert "unpacked" in result.output.lower() or "commit" in result.output.lower() def test_verify_flag_corrupt_bundle_exits_1(self, tmp_path: pathlib.Path) -> None: src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"vf-corrupt") bundle = tmp_path / "corrupt.bundle" _create_bundle(src, bundle) # Corrupt an object raw = msgpack.unpackb(bundle.read_bytes(), raw=False) if raw.get("blobs"): raw["blobs"][0]["content"] = b"TAMPERED" bundle.write_bytes(msgpack.packb(raw, use_bin_type=True)) result = _invoke(["bundle", "unbundle", str(bundle), "--verify"], env=_env(dst)) assert result.exit_code != 0 def test_verify_flag_corrupt_does_not_write(self, tmp_path: pathlib.Path) -> None: """When --verify fails, no objects must be written to the destination.""" src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"vf-no-write") bundle = tmp_path / "nowrite.bundle" _create_bundle(src, bundle) raw = msgpack.unpackb(bundle.read_bytes(), raw=False) obj_ids_before = set(raw.get("branch_heads", {}).values()) if raw.get("blobs"): raw["blobs"][0]["content"] = b"CORRUPTED" bundle.write_bytes(msgpack.packb(raw, use_bin_type=True)) _invoke(["bundle", "unbundle", str(bundle), "--verify"], env=_env(dst)) # Destination object store must be empty obj_dir = objects_dir(dst) written = list(obj_dir.rglob("*")) if obj_dir.exists() else [] written_files = [p for p in written if p.is_file()] assert len(written_files) == 0, "Objects were written despite corrupt bundle" def test_verify_flag_json_output_has_verified_field( self, tmp_path: pathlib.Path ) -> None: src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"vf-json") bundle = tmp_path / "json.bundle" _create_bundle(src, bundle) result = _invoke( ["bundle", "unbundle", str(bundle), "--verify", "--json"], env=_env(dst) ) assert result.exit_code == 0 data = json.loads(result.output) assert "verified" in data assert data["verified"] is True def test_verify_flag_json_corrupt_verified_false( self, tmp_path: pathlib.Path ) -> None: src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"vf-json-corrupt") bundle = tmp_path / "json-corrupt.bundle" _create_bundle(src, bundle) raw = msgpack.unpackb(bundle.read_bytes(), raw=False) if raw.get("blobs"): raw["blobs"][0]["content"] = b"CORRUPT" bundle.write_bytes(msgpack.packb(raw, use_bin_type=True)) result = _invoke( ["bundle", "unbundle", str(bundle), "--verify", "--json"], env=_env(dst) ) assert result.exit_code != 0 def test_no_verify_flag_still_works(self, tmp_path: pathlib.Path) -> None: """Without --verify the old behavior is unchanged.""" src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"vf-no-flag") bundle = tmp_path / "noflag.bundle" _create_bundle(src, bundle) result = _invoke(["bundle", "unbundle", str(bundle)], env=_env(dst)) assert result.exit_code == 0 def test_verify_and_no_update_refs_combined(self, tmp_path: pathlib.Path) -> None: """--verify and --no-update-refs must be combinable.""" src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"vf-no-refs") bundle = tmp_path / "norefs.bundle" _create_bundle(src, bundle) result = _invoke( ["bundle", "unbundle", str(bundle), "--verify", "--no-update-refs", "--json"], env=_env(dst), ) assert result.exit_code == 0 data = json.loads(result.output) assert data["verified"] is True assert data["refs_updated"] == [] def test_verify_flag_security_corrupt_before_parse( self, tmp_path: pathlib.Path ) -> None: """Bytes-level corruption (not msgpack) is caught before any write.""" src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"vf-bytes-corrupt") bundle = tmp_path / "bytes-corrupt.bundle" _create_bundle(src, bundle) raw = bundle.read_bytes() # Flip bytes in the middle to corrupt msgpack framing mid = len(raw) // 2 corrupted = raw[:mid] + bytes(b ^ 0xFF for b in raw[mid:mid + 20]) + raw[mid + 20:] bundle.write_bytes(corrupted) result = _invoke(["bundle", "unbundle", str(bundle), "--verify"], env=_env(dst)) assert result.exit_code != 0 class TestBundleUnbundleVerifyStress: def test_verify_flag_100_commit_bundle(self, tmp_path: pathlib.Path) -> None: src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="stress-verify-dst") prev = None for i in range(100): prev = _make_commit(src, parent_id=prev, content=f"sv-{i}".encode()) bundle = tmp_path / "sv100.bundle" _create_bundle(src, bundle) start = time.monotonic() result = _invoke( ["bundle", "unbundle", str(bundle), "--verify", "--json"], env=_env(dst) ) elapsed = time.monotonic() - start assert result.exit_code == 0 data = json.loads(result.output) assert data["verified"] is True assert data["commits_written"] == 100 assert elapsed < 5.0, f"verify+unbundle 100 commits took {elapsed:.2f}s" # =========================================================================== # Feature 3: muse bundle diff # =========================================================================== class TestBundleDiffUnit: """Unit-level schema and output contracts for bundle diff.""" def test_diff_help_exits_0(self) -> None: result = _invoke(["bundle", "diff", "--help"]) assert result.exit_code == 0 def test_diff_help_mentions_agent(self) -> None: result = _invoke(["bundle", "diff", "--help"]) assert "agent" in result.output.lower() or "Agent" in result.output def test_diff_json_schema_keys(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"diff-schema") bundle = tmp_path / "dschema.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "diff", str(bundle), "--json"], env=_env(tmp_path)) assert result.exit_code == 0 data = _parse_diff(result) for key in ("new_commits", "known_commits", "refs_to_advance", "commits"): assert key in data, f"diff JSON missing key: {key}" def test_diff_known_commits_when_already_applied( self, tmp_path: pathlib.Path ) -> None: """If the repo already has all bundle commits, new_commits == 0.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"diff-known") bundle = tmp_path / "known.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "diff", str(bundle), "--json"], env=_env(tmp_path)) data = _parse_diff(result) assert data["new_commits"] == 0 assert data["known_commits"] >= 1 def test_diff_new_commits_in_fresh_repo(self, tmp_path: pathlib.Path) -> None: """Diff against a fresh repo with no commits: all bundle commits are new.""" src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="diff-fresh-dst") prev = None for i in range(3): prev = _make_commit(src, parent_id=prev, content=f"df-{i}".encode()) bundle = tmp_path / "fresh.bundle" _create_bundle(src, bundle) result = _invoke(["bundle", "diff", str(bundle), "--json"], env=_env(dst)) data = _parse_diff(result) assert data["new_commits"] == 3 assert data["known_commits"] == 0 def test_diff_refs_to_advance_populated(self, tmp_path: pathlib.Path) -> None: """refs_to_advance must contain branch names that would move.""" src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="diff-refs-dst") _make_commit(src, content=b"diff-refs") bundle = tmp_path / "refs.bundle" _create_bundle(src, bundle) result = _invoke(["bundle", "diff", str(bundle), "--json"], env=_env(dst)) data = _parse_diff(result) assert "main" in data["refs_to_advance"] def test_diff_refs_to_advance_empty_when_known( self, tmp_path: pathlib.Path ) -> None: """When the repo is already up-to-date, refs_to_advance is empty.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"diff-upto-date") bundle = tmp_path / "upto.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "diff", str(bundle), "--json"], env=_env(tmp_path)) data = _parse_diff(result) assert data["refs_to_advance"] == [] def test_diff_commits_list_contains_new_entries( self, tmp_path: pathlib.Path ) -> None: src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="diff-commits-dst") prev = None cids = [] for i in range(3): prev = _make_commit(src, parent_id=prev, content=f"dc-{i}".encode()) cids.append(prev) bundle = tmp_path / "commits.bundle" _create_bundle(src, bundle) result = _invoke(["bundle", "diff", str(bundle), "--json"], env=_env(dst)) data = _parse_diff(result) listed_ids = {c["commit_id"] for c in data["commits"]} for cid in cids: assert cid in listed_ids def test_diff_commit_entry_schema(self, tmp_path: pathlib.Path) -> None: src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="diff-entry-dst") _make_commit(src, content=b"diff-entry") bundle = tmp_path / "entry.bundle" _create_bundle(src, bundle) result = _invoke(["bundle", "diff", str(bundle), "--json"], env=_env(dst)) data = _parse_diff(result) if data["commits"]: entry = data["commits"][0] for key in ("commit_id", "message", "committed_at"): assert key in entry def test_diff_partial_known_commits(self, tmp_path: pathlib.Path) -> None: """When the dst repo has some but not all commits, count matches.""" src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="diff-partial-dst") # Build 5-commit chain; write first 2 to dst manually prev = None all_ids: list[str] = [] for i in range(5): prev = _make_commit(src, parent_id=prev, content=f"partial-{i}".encode()) all_ids.append(prev) # Copy first 2 commits into dst so they are "known" from muse.core.commits import read_commit for cid in all_ids[:2]: rec = read_commit(src, cid) if rec: write_commit(dst, rec) bundle = tmp_path / "partial.bundle" _create_bundle(src, bundle) result = _invoke(["bundle", "diff", str(bundle), "--json"], env=_env(dst)) data = _parse_diff(result) assert data["new_commits"] == 3 assert data["known_commits"] == 2 def test_diff_requires_repo(self, tmp_path: pathlib.Path) -> None: """diff requires a repository (unlike inspect/verify/list-heads).""" src = tmp_path / "src" src.mkdir() _init_repo(src) _make_commit(src, content=b"diff-needs-repo") bundle = tmp_path / "needsrepo.bundle" _create_bundle(src, bundle) # Point MUSE_REPO_ROOT at a directory with no .muse → require_repo() fails. no_repo = tmp_path / "no_repo" no_repo.mkdir() result = _invoke(["bundle", "diff", str(bundle), "--json"], env=_env(no_repo)) assert result.exit_code != 0 def test_diff_file_not_found(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke( ["bundle", "diff", str(tmp_path / "missing.bundle"), "--json"], env=_env(tmp_path), ) assert result.exit_code != 0 def test_diff_j_alias(self, tmp_path: pathlib.Path) -> None: src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="diff-j-alias-dst") _make_commit(src, content=b"diff-jalias") bundle = tmp_path / "jalias.bundle" _create_bundle(src, bundle) r1 = _invoke(["bundle", "diff", str(bundle), "--json"], env=_env(dst)) r2 = _invoke(["bundle", "diff", str(bundle), "-j"], env=_env(dst)) assert r1.exit_code == 0 assert r2.exit_code == 0 d1 = json.loads(r1.output) d2 = json.loads(r2.output) assert d1["new_commits"] == d2["new_commits"] class TestBundleDiffText: def test_text_output_mentions_new_commits(self, tmp_path: pathlib.Path) -> None: src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="diff-txt-dst") _make_commit(src, content=b"diff-txt") bundle = tmp_path / "txt.bundle" _create_bundle(src, bundle) result = _invoke(["bundle", "diff", str(bundle)], env=_env(dst)) assert result.exit_code == 0 assert "new" in result.output.lower() or "commit" in result.output.lower() def test_text_output_up_to_date_message(self, tmp_path: pathlib.Path) -> None: """When nothing is new, output should say so.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"diff-uptodate-txt") bundle = tmp_path / "uptodate.bundle" _create_bundle(tmp_path, bundle) result = _invoke(["bundle", "diff", str(bundle)], env=_env(tmp_path)) assert result.exit_code == 0 # Should mention up-to-date or 0 new commits assert "0" in result.output or "up-to-date" in result.output.lower() class TestBundleDiffSecurity: def _has_ansi(self, s: str) -> bool: return "\x1b[" in s def test_ansi_in_bundle_message_stripped(self, tmp_path: pathlib.Path) -> None: src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="diff-sec-ansi-dst") _make_commit(src, content=b"diff-sec-ansi", message="\x1b[31mmalicious\x1b[0m") bundle = tmp_path / "ansi.bundle" _create_bundle(src, bundle) result = _invoke(["bundle", "diff", str(bundle), "--json"], env=_env(dst)) assert result.exit_code == 0 assert not self._has_ansi(result.output) class TestBundleDiffDataIntegrity: def test_diff_then_unbundle_gives_zero_new(self, tmp_path: pathlib.Path) -> None: """After unbundling, a second diff should show 0 new commits.""" src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="diff-di-dst") prev = None for i in range(3): prev = _make_commit(src, parent_id=prev, content=f"di-dt-{i}".encode()) bundle = tmp_path / "di.bundle" _create_bundle(src, bundle) # Before unbundle: 3 new r1 = _invoke(["bundle", "diff", str(bundle), "--json"], env=_env(dst)) assert json.loads(r1.output)["new_commits"] == 3 # Unbundle _invoke(["bundle", "unbundle", str(bundle)], env=_env(dst)) # After unbundle: 0 new r2 = _invoke(["bundle", "diff", str(bundle), "--json"], env=_env(dst)) assert json.loads(r2.output)["new_commits"] == 0 def test_diff_new_count_matches_actual_writes( self, tmp_path: pathlib.Path ) -> None: """new_commits from diff must equal commits_written from unbundle --json.""" src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="diff-di2-dst") prev = None for i in range(5): prev = _make_commit(src, parent_id=prev, content=f"match-{i}".encode()) bundle = tmp_path / "match.bundle" _create_bundle(src, bundle) diff_data = json.loads( _invoke(["bundle", "diff", str(bundle), "--json"], env=_env(dst)).output ) unbundle_data = json.loads( _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst)).output ) assert diff_data["new_commits"] == unbundle_data["commits_written"] class TestBundleDiffPerformance: def test_diff_100_commit_bundle_under_1s(self, tmp_path: pathlib.Path) -> None: src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="diff-perf-dst") prev = None for i in range(100): prev = _make_commit(src, parent_id=prev, content=f"dp-{i}".encode()) bundle = tmp_path / "dp100.bundle" _create_bundle(src, bundle) start = time.monotonic() result = _invoke(["bundle", "diff", str(bundle), "--json"], env=_env(dst)) elapsed = time.monotonic() - start assert result.exit_code == 0 data = _parse_diff(result) assert data["new_commits"] == 100 assert elapsed < 1.0, f"diff 100-commit bundle took {elapsed:.2f}s" class TestBundleDiffStress: def test_diff_200_commit_bundle(self, tmp_path: pathlib.Path) -> None: src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="diff-stress-dst") prev = None for i in range(200): prev = _make_commit(src, parent_id=prev, content=f"ds-{i}".encode()) bundle = tmp_path / "ds200.bundle" _create_bundle(src, bundle) result = _invoke(["bundle", "diff", str(bundle), "--json"], env=_env(dst)) assert result.exit_code == 0 data = _parse_diff(result) assert data["new_commits"] == 200