"""Hardening tests for ``muse bundle``. Covers: Unit — _iter_branches (symlink guard, size cap), _reachable_from, _load_bundle (narrow except), _resolve_refs, TypeGuards Security — symlink traversal in _iter_branches, ANSI injection in branch names and failure messages, oversized bundle rejection Perf — reachable set is pre-computed once (not per branch) JSON — _BundleCreateJson, _BundleUnbundleJson, _BundleVerifyJson, list-heads dict schema Flags — --json for create / unbundle / verify / list-heads Integration — create → unbundle round-trip with branch ref updates, --have pruning reduces bundle size, verify catches corruption and missing snapshot objects E2E — --help output for all subcommands Stress — 200-commit chain, concurrent unbundle reads """ from __future__ import annotations import datetime import hashlib import json import pathlib import threading from typing import TypedDict 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, long_id, blob_id cli = None runner = CliRunner() _invoke_lock = threading.Lock() _REPO_ID = "bundle-hardening-test" # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- class _CreateOut(TypedDict): file: str commits: int blobs: int size_bytes: int branches: list[str] class _UnbundleOut(TypedDict): commits_written: int snapshots_written: int blobs_written: int blobs_skipped: int refs_updated: list[str] class _VerifyOut(TypedDict): blobs_checked: int snapshots_checked: int all_ok: bool failures: list[str] 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 _branch_heads_map: dict[tuple[str, str], str] = {} def _make_commit( root: pathlib.Path, parent_id: str | None = None, content: bytes = b"data", branch: str = "main", ) -> str: global _counter _counter += 1 c = content + str(_counter).encode() obj_id = long_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) resolved_parent = parent_id if resolved_parent is None: key = (str(root), branch) resolved_parent = _branch_heads_map.get(key) parent_ids = [resolved_parent] if resolved_parent else [] commit_id = hash_commit( parent_ids=parent_ids, snapshot_id=snap_id, message=f"commit {_counter}", committed_at_iso=committed_at.isoformat(), ) write_commit( root, CommitRecord( commit_id=commit_id, branch=branch, snapshot_id=snap_id, message=f"commit {_counter}", committed_at=committed_at, parent_commit_id=resolved_parent, ), ) branch_ref = ref_path(root, branch) branch_ref.parent.mkdir(parents=True, exist_ok=True) branch_ref.write_text(commit_id, encoding="utf-8") _branch_heads_map[(str(root), branch)] = commit_id return commit_id def _invoke(args: list[str], env: Manifest | None = None) -> InvokeResult: with _invoke_lock: return runner.invoke(cli, args, env=env) def _parse_create(result: InvokeResult) -> _CreateOut: raw: _CreateOut = json.loads(result.output) return raw def _parse_unbundle(result: InvokeResult) -> _UnbundleOut: raw: _UnbundleOut = json.loads(result.output) return raw def _parse_verify(result: InvokeResult) -> _VerifyOut: raw: _VerifyOut = json.loads(result.output) return raw # --------------------------------------------------------------------------- # Unit: _iter_branches — symlink guard # --------------------------------------------------------------------------- def test_iter_branches_skips_symlinks(tmp_path: pathlib.Path) -> None: """A symlink inside refs/heads must be silently skipped.""" from muse.cli.commands.bundle import _iter_branches _init_repo(tmp_path) target = tmp_path / "outside.txt" target.write_text("malicious-sha" * 8, encoding="utf-8") # 64 chars h_dir = heads_dir(tmp_path) real_ref = h_dir / "main" real_ref.write_text("a" * 64, encoding="utf-8") link = h_dir / "malicious" link.symlink_to(target) result = _iter_branches(tmp_path) branch_names = [name for name, _ in result] assert "malicious" not in branch_names assert "main" in branch_names def test_iter_branches_size_cap(tmp_path: pathlib.Path) -> None: """Ref files larger than 65 bytes are read but will be invalid after strip.""" from muse.cli.commands.bundle import _iter_branches, _MAX_REF_BYTES _init_repo(tmp_path) h_dir = heads_dir(tmp_path) oversized = h_dir / "main" oversized.write_bytes(b"x" * (_MAX_REF_BYTES + 100)) result = _iter_branches(tmp_path) # Should still return one entry; the content is capped — the commit ID # won't be valid hex but _iter_branches returns it; validation is downstream. assert len(result) == 1 _, cid = result[0] assert len(cid) <= _MAX_REF_BYTES # capped at read time def test_iter_branches_empty_dir(tmp_path: pathlib.Path) -> None: from muse.cli.commands.bundle import _iter_branches _init_repo(tmp_path) result = _iter_branches(tmp_path) assert result == [] def test_iter_branches_multiple(tmp_path: pathlib.Path) -> None: from muse.cli.commands.bundle import _iter_branches _init_repo(tmp_path) h_dir = heads_dir(tmp_path) for name in ("main", "dev", "feat/foo"): p = h_dir / name p.parent.mkdir(parents=True, exist_ok=True) p.write_text("a" * 64, encoding="utf-8") result = _iter_branches(tmp_path) names = [n for n, _ in result] assert "main" in names assert "dev" in names assert "feat/foo" in names # --------------------------------------------------------------------------- # Unit: _reachable_from — correctness # --------------------------------------------------------------------------- def test_reachable_from_single(tmp_path: pathlib.Path) -> None: from muse.cli.commands.bundle import _reachable_from _init_repo(tmp_path) c1 = _make_commit(tmp_path, content=b"r1") result = _reachable_from(tmp_path, [c1]) assert c1 in result def test_reachable_from_chain(tmp_path: pathlib.Path) -> None: from muse.cli.commands.bundle import _reachable_from _init_repo(tmp_path) c1 = _make_commit(tmp_path, content=b"rc1") c2 = _make_commit(tmp_path, parent_id=c1, content=b"rc2") c3 = _make_commit(tmp_path, parent_id=c2, content=b"rc3") result = _reachable_from(tmp_path, [c3]) assert c1 in result assert c2 in result assert c3 in result def test_reachable_from_empty_tips(tmp_path: pathlib.Path) -> None: from muse.cli.commands.bundle import _reachable_from _init_repo(tmp_path) assert _reachable_from(tmp_path, []) == set() # --------------------------------------------------------------------------- # Unit: _load_bundle — narrow except # --------------------------------------------------------------------------- def test_load_bundle_not_found(tmp_path: pathlib.Path) -> None: result = _invoke( ["bundle", "verify", str(tmp_path / "missing.bundle")], env=_env(tmp_path), ) assert result.exit_code != 0 def test_load_bundle_invalid_msgpack(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) bad = tmp_path / "bad.bundle" bad.write_bytes(b"\xff\xfe this is not msgpack") result = _invoke(["bundle", "verify", str(bad)], env=_env(tmp_path)) assert result.exit_code != 0 def test_load_bundle_not_dict(tmp_path: pathlib.Path) -> None: """A valid msgpack list instead of dict must be rejected cleanly.""" _init_repo(tmp_path) bad = tmp_path / "list.bundle" bad.write_bytes(msgpack.packb([1, 2, 3], use_bin_type=True)) result = _invoke(["bundle", "verify", str(bad)], env=_env(tmp_path)) assert result.exit_code != 0 # --------------------------------------------------------------------------- # Security: ANSI injection in branch names # --------------------------------------------------------------------------- def test_list_heads_ansi_injection(tmp_path: pathlib.Path) -> None: """Branch names with ANSI escapes must be stripped in text output.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"ansi-branch") out = tmp_path / "ansi.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) # Inject a crafted branch_heads entry with ANSI escape in branch name. raw = msgpack.unpackb(out.read_bytes(), raw=False) raw["branch_heads"] = {"\x1b[31mmalicious\x1b[0m": "a" * 64} out.write_bytes(msgpack.packb(raw, use_bin_type=True)) result = _invoke(["bundle", "list-heads", str(out)], env=_env(tmp_path)) assert result.exit_code == 0 assert "\x1b" not in result.output def test_verify_failure_ansi_injection(tmp_path: pathlib.Path) -> None: """Failure messages must not allow ANSI injection through object_id fields.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"ansi-verify") out = tmp_path / "ansi-v.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) # Tamper content to trigger a hash mismatch failure. raw = msgpack.unpackb(out.read_bytes(), raw=False) if raw.get("blobs"): raw["blobs"][0]["content"] = b"\x1b[31mTAMPERED\x1b[0m" out.write_bytes(msgpack.packb(raw, use_bin_type=True)) result = _invoke(["bundle", "verify", str(out)], env=_env(tmp_path)) # Text output must strip ANSI from the failure description. assert "\x1b" not in result.output assert result.exit_code != 0 # --------------------------------------------------------------------------- # JSON schema: bundle create --json # --------------------------------------------------------------------------- def test_create_json_schema(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"cj1") out = tmp_path / "cj.bundle" result = _invoke(["bundle", "create", str(out), "--json"], env=_env(tmp_path)) assert result.exit_code == 0 data = _parse_create(result) assert data["file"] == str(out) assert data["commits"] >= 1 assert data["blobs"] >= 1 assert data["size_bytes"] > 0 assert isinstance(data["branches"], list) assert "main" in data["branches"] def test_create_json_no_output_on_success_without_flag(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"cj-no-flag") out = tmp_path / "cnf.bundle" result = _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) assert result.exit_code == 0 # Text output, not JSON. assert "Bundle" in result.output or "✅" in result.output # --------------------------------------------------------------------------- # JSON schema: bundle unbundle --json # --------------------------------------------------------------------------- def test_unbundle_json_schema(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="dst-json") _make_commit(src, content=b"uj1") out = tmp_path / "uj.bundle" _invoke(["bundle", "create", str(out)], env=_env(src)) result = _invoke(["bundle", "unbundle", str(out), "--json"], env=_env(dst)) assert result.exit_code == 0 data = _parse_unbundle(result) assert data["commits_written"] >= 1 assert isinstance(data["snapshots_written"], int) assert isinstance(data["blobs_written"], int) assert isinstance(data["blobs_skipped"], int) assert isinstance(data["refs_updated"], list) def test_unbundle_json_refs_updated(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="dst-ru") _make_commit(src, content=b"ru1") out = tmp_path / "ru.bundle" _invoke(["bundle", "create", str(out)], env=_env(src)) result = _invoke(["bundle", "unbundle", str(out), "--json"], env=_env(dst)) assert result.exit_code == 0 data = _parse_unbundle(result) assert "main" in data["refs_updated"] def test_unbundle_json_no_update_refs(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="dst-nur") _make_commit(src, content=b"nur1") out = tmp_path / "nur.bundle" _invoke(["bundle", "create", str(out)], env=_env(src)) result = _invoke( ["bundle", "unbundle", str(out), "--no-update-refs", "--json"], env=_env(dst), ) assert result.exit_code == 0 data = _parse_unbundle(result) assert data["refs_updated"] == [] # --------------------------------------------------------------------------- # JSON schema: bundle verify --json # --------------------------------------------------------------------------- def test_verify_json_schema_clean(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"vjs1") out = tmp_path / "vjs.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) result = _invoke(["bundle", "verify", str(out), "--json"], env=_env(tmp_path)) assert result.exit_code == 0 data = _parse_verify(result) assert data["all_ok"] is True assert data["blobs_checked"] >= 1 assert "snapshots_checked" in data assert data["failures"] == [] def test_verify_json_schema_corrupt(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"corrupt-j") out = tmp_path / "cj2.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) raw = msgpack.unpackb(out.read_bytes(), raw=False) if raw.get("blobs"): raw["blobs"][0]["content"] = b"tampered!" out.write_bytes(msgpack.packb(raw, use_bin_type=True)) result = _invoke(["bundle", "verify", str(out), "--json"], env=_env(tmp_path)) assert result.exit_code != 0 data = _parse_verify(result) assert data["all_ok"] is False assert len(data["failures"]) > 0 def test_verify_json_snapshots_checked(tmp_path: pathlib.Path) -> None: """``snapshots_checked`` must count non-zero when snapshots are present.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"snap-counted") out = tmp_path / "sc.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) result = _invoke(["bundle", "verify", str(out), "--json"], env=_env(tmp_path)) assert result.exit_code == 0 data = _parse_verify(result) # At least one snapshot should have been included in the bundle. assert data["snapshots_checked"] >= 1 # --------------------------------------------------------------------------- # JSON schema: bundle list-heads --json # --------------------------------------------------------------------------- def test_list_heads_json_schema(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"lhjs1") out = tmp_path / "lhjs.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) result = _invoke(["bundle", "list-heads", str(out), "--json"], env=_env(tmp_path)) assert result.exit_code == 0 data = json.loads(result.output) assert isinstance(data, dict) heads = data["heads"] assert "main" in heads for _branch, cid in heads.items(): assert isinstance(cid, str) and cid.startswith("sha256:") assert len(cid) == len("sha256:") + 64 # --------------------------------------------------------------------------- # Flags: --json rejects old --format arg # --------------------------------------------------------------------------- def test_verify_rejects_format_flag(tmp_path: pathlib.Path) -> None: """The old ``--format json`` pattern must not be accepted.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"old-fmt") out = tmp_path / "old.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) result = _invoke( ["bundle", "verify", str(out), "--format", "json"], env=_env(tmp_path) ) # --format is no longer a registered flag, so argparse returns exit 2. assert result.exit_code == 2 def test_list_heads_rejects_format_flag(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"lh-old") out = tmp_path / "lh-old.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) result = _invoke( ["bundle", "list-heads", str(out), "--format", "json"], env=_env(tmp_path) ) assert result.exit_code == 2 # --------------------------------------------------------------------------- # Integration: --have pruning # --------------------------------------------------------------------------- def test_create_have_prunes_bundle(tmp_path: pathlib.Path) -> None: """Passing --have should produce a smaller bundle than the full chain.""" _init_repo(tmp_path) c1 = _make_commit(tmp_path, content=b"have-base") _make_commit(tmp_path, parent_id=c1, content=b"have-tip") out_full = tmp_path / "full.bundle" out_pruned = tmp_path / "pruned.bundle" _invoke(["bundle", "create", str(out_full)], env=_env(tmp_path)) _invoke( ["bundle", "create", str(out_pruned), "--have", c1], env=_env(tmp_path), ) # Pruned bundle must be smaller (fewer commits packed). assert out_pruned.stat().st_size < out_full.stat().st_size def test_create_have_json_smaller_commits(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) c1 = _make_commit(tmp_path, content=b"hjp-base") _make_commit(tmp_path, parent_id=c1, content=b"hjp-tip") out_full = tmp_path / "hjp-full.bundle" out_pruned = tmp_path / "hjp-pruned.bundle" r_full = _invoke( ["bundle", "create", str(out_full), "--json"], env=_env(tmp_path) ) r_pruned = _invoke( ["bundle", "create", str(out_pruned), "--have", c1, "--json"], env=_env(tmp_path), ) full_data = _parse_create(r_full) pruned_data = _parse_create(r_pruned) assert pruned_data["commits"] < full_data["commits"] # --------------------------------------------------------------------------- # Integration: multi-branch bundle # --------------------------------------------------------------------------- def test_create_multiple_branches(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"mb-main", branch="main") _make_commit(tmp_path, content=b"mb-feat", branch="feat/x") out = tmp_path / "mb.bundle" result = _invoke( ["bundle", "create", str(out), "--json"], env=_env(tmp_path) ) assert result.exit_code == 0 data = _parse_create(result) assert "main" in data["branches"] or "feat/x" in data["branches"] def test_round_trip_with_json_summary(tmp_path: pathlib.Path) -> None: """Full create → verify → unbundle pipeline with JSON at each step.""" src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="dst-rt-json") prev: str | None = None for i in range(5): prev = _make_commit(src, parent_id=prev, content=f"rt-{i}".encode()) out = tmp_path / "rt-json.bundle" create_result = _invoke( ["bundle", "create", str(out), "--json"], env=_env(src) ) assert create_result.exit_code == 0 create_data = _parse_create(create_result) assert create_data["commits"] == 5 verify_result = _invoke( ["bundle", "verify", str(out), "--json"], env=_env(src) ) assert verify_result.exit_code == 0 verify_data = _parse_verify(verify_result) assert verify_data["all_ok"] is True unbundle_result = _invoke( ["bundle", "unbundle", str(out), "--json"], env=_env(dst) ) assert unbundle_result.exit_code == 0 unbundle_data = _parse_unbundle(unbundle_result) assert unbundle_data["commits_written"] == 5 # --------------------------------------------------------------------------- # Integration: verify detects missing snapshot objects # --------------------------------------------------------------------------- def test_verify_missing_snapshot_object(tmp_path: pathlib.Path) -> None: """Removing an object from the bundle should cause snapshot verification to fail.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"snap-miss") out = tmp_path / "snap-miss.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) raw = msgpack.unpackb(out.read_bytes(), raw=False) # Remove all objects so snapshots cannot find theirs. raw["blobs"] = [] out.write_bytes(msgpack.packb(raw, use_bin_type=True)) result = _invoke(["bundle", "verify", str(out), "--json"], env=_env(tmp_path)) data = _parse_verify(result) assert data["all_ok"] is False # Some failure should mention missing objects. assert any("missing" in f for f in data["failures"]) # --------------------------------------------------------------------------- # E2E: help output for all subcommands # --------------------------------------------------------------------------- def test_create_help_mentions_json() -> None: result = _invoke(["bundle", "create", "--help"]) assert result.exit_code == 0 assert "--json" in result.output def test_unbundle_help_mentions_json() -> None: result = _invoke(["bundle", "unbundle", "--help"]) assert result.exit_code == 0 assert "--json" in result.output def test_verify_help_mentions_json() -> None: result = _invoke(["bundle", "verify", "--help"]) assert result.exit_code == 0 assert "--json" in result.output assert "--format" not in result.output def test_list_heads_help_mentions_json() -> None: result = _invoke(["bundle", "list-heads", "--help"]) assert result.exit_code == 0 assert "--json" in result.output assert "--format" not in result.output def test_bundle_help_top_level() -> None: result = _invoke(["bundle", "--help"]) assert result.exit_code == 0 assert "create" in result.output assert "unbundle" in result.output assert "verify" in result.output assert "list-heads" in result.output # --------------------------------------------------------------------------- # Stress: 200-commit bundle # --------------------------------------------------------------------------- def test_stress_200_commit_chain(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) prev: str | None = None for i in range(200): prev = _make_commit(tmp_path, parent_id=prev, content=f"stress-{i}".encode()) out = tmp_path / "stress200.bundle" create_result = _invoke( ["bundle", "create", str(out), "--json"], env=_env(tmp_path) ) assert create_result.exit_code == 0 data = _parse_create(create_result) assert data["commits"] == 200 verify_result = _invoke(["bundle", "verify", str(out), "-q"], env=_env(tmp_path)) assert verify_result.exit_code == 0 # --------------------------------------------------------------------------- # Stress: concurrent list-heads reads # --------------------------------------------------------------------------- def test_stress_concurrent_list_heads(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"concurrent-bundle") out = tmp_path / "concurrent.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) errors: list[str] = [] def _read() -> None: r = _invoke(["bundle", "list-heads", str(out), "--json"], env=_env(tmp_path)) if r.exit_code != 0: errors.append(f"exit {r.exit_code}: {r.output}") else: try: data = json.loads(r.output) if not isinstance(data, dict): errors.append("not a dict") except json.JSONDecodeError 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 list-heads failures: {errors}" # =========================================================================== # TestBundleCreateExtended — 18 tests # =========================================================================== class TestBundleCreateExtended: def test_exits_0_basic(self, tmp_path: pathlib.Path) -> None: """Single commit → create exits 0.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"ext-basic") out = tmp_path / "basic.bundle" result = _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) assert result.exit_code == 0 def test_creates_file_on_disk(self, tmp_path: pathlib.Path) -> None: """Output file must exist after a successful create.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"ext-file") out = tmp_path / "check.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) assert out.exists() def test_text_output_mentions_commits(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"ext-txt-c") out = tmp_path / "tc.bundle" result = _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) assert "commits" in result.output def test_text_output_mentions_kib(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"ext-txt-kib") out = tmp_path / "kib.bundle" result = _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) assert "KiB" in result.output def test_text_output_contains_bundle_path(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"ext-txt-path") out = tmp_path / "pathcheck.bundle" result = _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) assert "pathcheck.bundle" in result.output def test_empty_repo_exits_1(self, tmp_path: pathlib.Path) -> None: """Repo with no commits → exit 1 (no commits to bundle).""" _init_repo(tmp_path) out = tmp_path / "empty.bundle" result = _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) assert result.exit_code == 1 def test_bad_ref_exits_1(self, tmp_path: pathlib.Path) -> None: """Unknown ref → exit 1.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"ext-bad-ref") out = tmp_path / "bad-ref.bundle" result = _invoke( ["bundle", "create", str(out), "nonexistent-branch"], env=_env(tmp_path), ) assert result.exit_code == 1 def test_json_branches_sorted(self, tmp_path: pathlib.Path) -> None: """Branches list in JSON output must be sorted.""" _init_repo(tmp_path) # Create a commit on main, then make z-branch and a-branch point to # the same commit so they are all reachable when bundling HEAD. c1 = _make_commit(tmp_path, content=b"ext-br-base", branch="main") for br in ("z-branch", "a-branch"): ref_file = ref_path(tmp_path, br) ref_file.write_text(c1, encoding="utf-8") out = tmp_path / "sorted.bundle" result = _invoke( ["bundle", "create", str(out), "--json"], env=_env(tmp_path) ) assert result.exit_code == 0 data = _parse_create(result) assert data["branches"] == sorted(data["branches"]) def test_json_size_matches_file(self, tmp_path: pathlib.Path) -> None: """size_bytes in JSON must equal the actual file size on disk.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"ext-size") out = tmp_path / "size.bundle" result = _invoke( ["bundle", "create", str(out), "--json"], env=_env(tmp_path) ) assert result.exit_code == 0 data = _parse_create(result) assert data["size_bytes"] == out.stat().st_size def test_j_alias(self, tmp_path: pathlib.Path) -> None: """-j must produce identical JSON to --json.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"ext-j-alias") out1 = tmp_path / "j1.bundle" out2 = tmp_path / "j2.bundle" r1 = _invoke(["bundle", "create", str(out1), "--json"], env=_env(tmp_path)) r2 = _invoke(["bundle", "create", str(out2), "-j"], env=_env(tmp_path)) assert r1.exit_code == 0 assert r2.exit_code == 0 d1 = json.loads(r1.output) d2 = json.loads(r2.output) # Both should have the same structural keys and counts. assert d1["commits"] == d2["commits"] assert d1["blobs"] == d2["blobs"] assert d1["branches"] == d2["branches"] def test_default_ref_is_head(self, tmp_path: pathlib.Path) -> None: """When no refs are given, HEAD is used — bundle contains the HEAD commit.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"ext-head") out = tmp_path / "head.bundle" result = _invoke( ["bundle", "create", str(out), "--json"], env=_env(tmp_path) ) assert result.exit_code == 0 data = _parse_create(result) assert data["commits"] >= 1 def test_explicit_head_ref(self, tmp_path: pathlib.Path) -> None: """Passing 'HEAD' explicitly is equivalent to the default.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"ext-head-explicit") out_default = tmp_path / "head-default.bundle" out_explicit = tmp_path / "head-explicit.bundle" _invoke(["bundle", "create", str(out_default)], env=_env(tmp_path)) result = _invoke( ["bundle", "create", str(out_explicit), "HEAD", "--json"], env=_env(tmp_path), ) assert result.exit_code == 0 data = _parse_create(result) assert data["commits"] >= 1 # Both bundles should contain the same number of commits. raw_default = __import__("msgpack").unpackb( out_default.read_bytes(), raw=False ) assert len(raw_default.get("commits", [])) == data["commits"] def test_explicit_commit_id(self, tmp_path: pathlib.Path) -> None: """A raw commit ID passed as ref is resolved correctly.""" _init_repo(tmp_path) cid = _make_commit(tmp_path, content=b"ext-cid") out = tmp_path / "cid.bundle" result = _invoke( ["bundle", "create", str(out), cid, "--json"], env=_env(tmp_path), ) assert result.exit_code == 0 data = _parse_create(result) assert data["commits"] >= 1 def test_output_is_valid_msgpack(self, tmp_path: pathlib.Path) -> None: """The output file must be valid msgpack.""" import msgpack as _mp _init_repo(tmp_path) _make_commit(tmp_path, content=b"ext-msgpack") out = tmp_path / "mp.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) raw = _mp.unpackb(out.read_bytes(), raw=False) assert isinstance(raw, dict) assert "commits" in raw def test_help_mentions_agent_quickstart(self) -> None: result = _invoke(["bundle", "create", "--help"]) assert result.exit_code == 0 assert "Agent quickstart" in result.output def test_help_mentions_exit_codes(self) -> None: result = _invoke(["bundle", "create", "--help"]) assert result.exit_code == 0 assert "Exit codes" in result.output def test_help_mentions_json_schema(self) -> None: result = _invoke(["bundle", "create", "--help"]) assert result.exit_code == 0 assert "JSON output schema" in result.output def test_multiple_have_ids_reduce_bundle(self, tmp_path: pathlib.Path) -> None: """Multiple --have IDs each reduce what is bundled.""" _init_repo(tmp_path) c1 = _make_commit(tmp_path, content=b"ext-have-1") c2 = _make_commit(tmp_path, parent_id=c1, content=b"ext-have-2") _make_commit(tmp_path, parent_id=c2, content=b"ext-have-3") out_full = tmp_path / "have-full.bundle" out_pruned = tmp_path / "have-pruned.bundle" r_full = _invoke( ["bundle", "create", str(out_full), "--json"], env=_env(tmp_path) ) r_pruned = _invoke( ["bundle", "create", str(out_pruned), "--have", c1, c2, "--json"], env=_env(tmp_path), ) assert r_full.exit_code == 0 assert r_pruned.exit_code == 0 full_data = _parse_create(r_full) pruned_data = _parse_create(r_pruned) assert pruned_data["commits"] < full_data["commits"] # =========================================================================== # TestBundleCreateSecurity — 6 tests # =========================================================================== class TestBundleCreateSecurity: def test_ansi_in_file_path_stripped_text_output( self, tmp_path: pathlib.Path ) -> None: """ANSI escape in the output file path must be stripped in text output.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"sec-ansi-path") # Build an output path whose filename component contains an ANSI escape. out = tmp_path / "\x1b[31mmalicious\x1b[0m.bundle" result = _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) assert result.exit_code == 0 assert "\x1b" not in result.output def test_control_char_in_file_path_stripped( self, tmp_path: pathlib.Path ) -> None: """Control characters in the output file path must not reach stdout.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"sec-ctrl-path") out = tmp_path / "foo\x07bar.bundle" result = _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) assert result.exit_code == 0 assert "\x07" not in result.output def test_outside_repo_exits_2(self, tmp_path: pathlib.Path) -> None: """Without a .muse directory, create must exit 2 (REPO_NOT_FOUND).""" out = tmp_path / "norepo.bundle" result = _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) assert result.exit_code == 2 def test_bad_ref_ansi_stripped_from_error( self, tmp_path: pathlib.Path ) -> None: """ANSI in an unknown ref name must not appear in the error output.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"sec-ref-ansi") out = tmp_path / "ref-ansi.bundle" malicious_ref = "\x1b[31mbadref\x1b[0m" result = _invoke( ["bundle", "create", str(out), malicious_ref], env=_env(tmp_path) ) assert result.exit_code == 1 assert "\x1b" not in result.output def test_ansi_in_have_no_injection(self, tmp_path: pathlib.Path) -> None: """ANSI characters in a --have value must not appear in any output.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"sec-have-ansi") out = tmp_path / "have-ansi.bundle" malicious_have = f"\x1b[31m{'a' * 64}\x1b[0m" result = _invoke( ["bundle", "create", str(out), "--have", malicious_have], env=_env(tmp_path), ) # The have ID won't match anything — bundle succeeds with full history. assert "\x1b" not in result.output def test_no_json_on_error(self, tmp_path: pathlib.Path) -> None: """On error (no commits), stdout must not contain JSON.""" _init_repo(tmp_path) out = tmp_path / "err-json.bundle" result = _invoke( ["bundle", "create", str(out), "--json"], env=_env(tmp_path) ) assert result.exit_code != 0 assert not result.output.strip().startswith("{") # =========================================================================== # TestBundleCreateStress — 3 tests # =========================================================================== class TestBundleCreateStress: def test_50_commit_chain(self, tmp_path: pathlib.Path) -> None: """50-commit linear chain is bundled correctly.""" _init_repo(tmp_path) prev: str | None = None for i in range(50): prev = _make_commit( tmp_path, parent_id=prev, content=f"stress50-{i}".encode() ) out = tmp_path / "stress50.bundle" result = _invoke( ["bundle", "create", str(out), "--json"], env=_env(tmp_path) ) assert result.exit_code == 0 data = _parse_create(result) assert data["commits"] == 50 assert data["size_bytes"] > 0 def test_create_with_large_have_list(self, tmp_path: pathlib.Path) -> None: """Passing 15 --have IDs on a 20-commit chain produces a smaller bundle.""" _init_repo(tmp_path) ids: list[str] = [] prev: str | None = None for i in range(20): prev = _make_commit( tmp_path, parent_id=prev, content=f"have-list-{i}".encode() ) ids.append(prev) out_full = tmp_path / "have-full-20.bundle" out_pruned = tmp_path / "have-pruned-20.bundle" r_full = _invoke( ["bundle", "create", str(out_full), "--json"], env=_env(tmp_path) ) # Pass the first 15 as --have to exclude them. have_args = ["--have"] + ids[:15] r_pruned = _invoke( ["bundle", "create", str(out_pruned)] + have_args + ["--json"], env=_env(tmp_path), ) assert r_full.exit_code == 0 assert r_pruned.exit_code == 0 full_data = _parse_create(r_full) pruned_data = _parse_create(r_pruned) assert pruned_data["commits"] < full_data["commits"] def test_many_branches(self, tmp_path: pathlib.Path) -> None: """10 branches pointing to reachable commits all appear in the bundle.""" _init_repo(tmp_path) # Build a 10-commit chain on main, then create a feature branch ref # pointing to each commit — all are reachable from HEAD. prev: str | None = None commit_ids: list[str] = [] for i in range(10): prev = _make_commit( tmp_path, parent_id=prev, content=f"stress-br-{i}".encode() ) commit_ids.append(prev) branch_names = [f"feat/stress-br-{i}" for i in range(10)] for br, cid in zip(branch_names, commit_ids): ref_file = ref_path(tmp_path, br) ref_file.parent.mkdir(parents=True, exist_ok=True) ref_file.write_text(cid, encoding="utf-8") out = tmp_path / "many-branches.bundle" result = _invoke( ["bundle", "create", str(out), "--json"], env=_env(tmp_path) ) assert result.exit_code == 0 data = _parse_create(result) for br in branch_names: assert br in data["branches"] # =========================================================================== # TestBundleUnbundleExtended — 18 tests # =========================================================================== def _make_bundle(src: pathlib.Path, dst_file: pathlib.Path) -> None: """Helper: create a bundle from src repo into dst_file.""" _invoke(["bundle", "create", str(dst_file)], env=_env(src)) class TestBundleUnbundleExtended: def _src_dst(self, tmp_path: pathlib.Path) -> 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="ub-dst") return src, dst def test_exits_0_basic(self, tmp_path: pathlib.Path) -> None: src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"ub-basic") bundle = tmp_path / "basic.bundle" _make_bundle(src, bundle) result = _invoke(["bundle", "unbundle", str(bundle)], env=_env(dst)) assert result.exit_code == 0 def test_commits_written_count(self, tmp_path: pathlib.Path) -> None: src, dst = self._src_dst(tmp_path) prev: str | None = None for i in range(3): prev = _make_commit(src, parent_id=prev, content=f"ub-cnt-{i}".encode()) bundle = tmp_path / "cnt.bundle" _make_bundle(src, bundle) result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst)) assert result.exit_code == 0 data = _parse_unbundle(result) assert data["commits_written"] == 3 def test_snapshots_written_count(self, tmp_path: pathlib.Path) -> None: src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"ub-snap") bundle = tmp_path / "snap.bundle" _make_bundle(src, bundle) result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst)) assert result.exit_code == 0 data = _parse_unbundle(result) assert data["snapshots_written"] >= 1 def test_blobs_written_count(self, tmp_path: pathlib.Path) -> None: src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"ub-obj") bundle = tmp_path / "obj.bundle" _make_bundle(src, bundle) result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst)) assert result.exit_code == 0 data = _parse_unbundle(result) assert data["blobs_written"] >= 1 def test_blobs_skipped_idempotent(self, tmp_path: pathlib.Path) -> None: """Unbundling twice: second pass skips all already-present blobs.""" src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"ub-idem") bundle = tmp_path / "idem.bundle" _make_bundle(src, bundle) _invoke(["bundle", "unbundle", str(bundle)], env=_env(dst)) result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst)) assert result.exit_code == 0 data = _parse_unbundle(result) assert data["commits_written"] == 0 assert data["blobs_written"] == 0 assert data["blobs_skipped"] >= 1 def test_text_output_mentions_commits(self, tmp_path: pathlib.Path) -> None: src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"ub-txt-c") bundle = tmp_path / "txt-c.bundle" _make_bundle(src, bundle) result = _invoke(["bundle", "unbundle", str(bundle)], env=_env(dst)) assert result.exit_code == 0 assert "commit(s)" in result.output def test_text_output_mentions_applied(self, tmp_path: pathlib.Path) -> None: src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"ub-txt-a") bundle = tmp_path / "txt-a.bundle" _make_bundle(src, bundle) result = _invoke(["bundle", "unbundle", str(bundle)], env=_env(dst)) assert result.exit_code == 0 assert "Bundle applied" in result.output def test_refs_updated_by_default(self, tmp_path: pathlib.Path) -> None: """By default, branch refs in the destination are updated.""" src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"ub-ref-up") bundle = tmp_path / "ref-up.bundle" _make_bundle(src, bundle) result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst)) assert result.exit_code == 0 data = _parse_unbundle(result) assert "main" in data["refs_updated"] def test_no_update_refs_skips_refs(self, tmp_path: pathlib.Path) -> None: src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"ub-no-ref") bundle = tmp_path / "no-ref.bundle" _make_bundle(src, bundle) result = _invoke( ["bundle", "unbundle", str(bundle), "--no-update-refs", "--json"], env=_env(dst), ) assert result.exit_code == 0 data = _parse_unbundle(result) assert data["refs_updated"] == [] def test_refs_updated_branch_file_exists(self, tmp_path: pathlib.Path) -> None: """After unbundle, the branch ref file must exist in the destination.""" src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"ub-ref-file") bundle = tmp_path / "ref-file.bundle" _make_bundle(src, bundle) _invoke(["bundle", "unbundle", str(bundle)], env=_env(dst)) ref_file = heads_dir(dst) / "main" assert ref_file.exists() cid = ref_file.read_text(encoding="utf-8").strip() # Ref files store canonical "sha256:<64hex>" format (71 chars). assert cid.startswith("sha256:") assert len(cid) == 71 def test_j_alias(self, tmp_path: pathlib.Path) -> None: """-j must produce identical JSON to --json.""" src1, dst1 = self._src_dst(tmp_path) src2 = tmp_path / "src2" dst2 = tmp_path / "dst2" src2.mkdir() dst2.mkdir() _init_repo(src2) _init_repo(dst2, repo_id="ub-j2") _make_commit(src1, content=b"ub-j-a1") _make_commit(src2, content=b"ub-j-a2") b1 = tmp_path / "j1.bundle" b2 = tmp_path / "j2.bundle" _make_bundle(src1, b1) _make_bundle(src2, b2) r1 = _invoke(["bundle", "unbundle", str(b1), "--json"], env=_env(dst1)) r2 = _invoke(["bundle", "unbundle", str(b2), "-j"], env=_env(dst2)) assert r1.exit_code == 0 assert r2.exit_code == 0 d1 = _parse_unbundle(r1) d2 = _parse_unbundle(r2) assert set(d1.keys()) == set(d2.keys()) assert d1["commits_written"] == d2["commits_written"] def test_json_refs_updated_sorted(self, tmp_path: pathlib.Path) -> None: """refs_updated in JSON output must be sorted.""" src, dst = self._src_dst(tmp_path) c1 = _make_commit(src, content=b"ub-sort-base") # Add extra branch refs pointing at c1 so the bundle has multiple heads. for br in ("z-br", "a-br"): ref = ref_path(src, br) ref.write_text(c1, encoding="utf-8") bundle = tmp_path / "sort.bundle" _make_bundle(src, bundle) result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst)) assert result.exit_code == 0 data = _parse_unbundle(result) assert data["refs_updated"] == sorted(data["refs_updated"]) def test_empty_bundle_no_crash(self, tmp_path: pathlib.Path) -> None: """An empty dict bundle (no commits/objects) must exit 0 cleanly.""" _init_repo(tmp_path) empty_bundle = tmp_path / "empty.bundle" empty_bundle.write_bytes(msgpack.packb({}, use_bin_type=True)) result = _invoke(["bundle", "unbundle", str(empty_bundle)], env=_env(tmp_path)) assert result.exit_code == 0 def test_bundle_without_branch_heads_no_refs(self, tmp_path: pathlib.Path) -> None: """A bundle missing the branch_heads key → refs_updated must be empty.""" src, dst = self._src_dst(tmp_path) _make_commit(src, content=b"ub-no-heads") bundle = tmp_path / "no-heads.bundle" _make_bundle(src, bundle) # Strip branch_heads from the bundle. raw = msgpack.unpackb(bundle.read_bytes(), raw=False) raw.pop("branch_heads", None) bundle.write_bytes(msgpack.packb(raw, use_bin_type=True)) result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst)) assert result.exit_code == 0 data = _parse_unbundle(result) assert data["refs_updated"] == [] def test_help_mentions_agent_quickstart(self) -> None: result = _invoke(["bundle", "unbundle", "--help"]) assert result.exit_code == 0 assert "Agent quickstart" in result.output def test_help_mentions_exit_codes(self) -> None: result = _invoke(["bundle", "unbundle", "--help"]) assert result.exit_code == 0 assert "Exit codes" in result.output def test_help_mentions_json_schema(self) -> None: result = _invoke(["bundle", "unbundle", "--help"]) assert result.exit_code == 0 assert "JSON output schema" in result.output def test_no_update_refs_flag_in_help(self) -> None: result = _invoke(["bundle", "unbundle", "--help"]) assert result.exit_code == 0 assert "--no-update-refs" in result.output # =========================================================================== # TestBundleUnbundleSecurity — 6 tests # =========================================================================== class TestBundleUnbundleSecurity: def test_outside_repo_exits_2(self, tmp_path: pathlib.Path) -> None: """Without a .muse directory, unbundle must exit 2 (REPO_NOT_FOUND).""" bundle = tmp_path / "norepo.bundle" bundle.write_bytes(msgpack.packb({}, use_bin_type=True)) result = _invoke(["bundle", "unbundle", str(bundle)], env=_env(tmp_path)) assert result.exit_code == 2 def test_missing_bundle_file_exits_1(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke( ["bundle", "unbundle", str(tmp_path / "missing.bundle")], env=_env(tmp_path), ) assert result.exit_code == 1 def test_invalid_msgpack_exits_1(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) corrupt = tmp_path / "corrupt.bundle" corrupt.write_bytes(b"\xff\xfe not msgpack at all") result = _invoke(["bundle", "unbundle", str(corrupt)], env=_env(tmp_path)) assert result.exit_code == 1 def test_ansi_branch_name_skipped_no_injection( self, tmp_path: pathlib.Path ) -> None: """ANSI escape in a bundle branch name is skipped; no escape in output.""" src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="sec-ansi-br") _make_commit(src, content=b"sec-ansi-br") bundle = tmp_path / "ansi-br.bundle" _make_bundle(src, bundle) # Inject an ANSI-poisoned branch name into branch_heads. raw = msgpack.unpackb(bundle.read_bytes(), raw=False) raw["branch_heads"] = {"\x1b[31mmalicious\x1b[0m": "a" * 64} bundle.write_bytes(msgpack.packb(raw, use_bin_type=True)) result = _invoke(["bundle", "unbundle", str(bundle)], env=_env(dst)) assert result.exit_code == 0 assert "\x1b" not in result.output def test_invalid_commit_id_branch_ref_skipped( self, tmp_path: pathlib.Path ) -> None: """A commit ID shorter than 64 chars in branch_heads must be skipped.""" src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="sec-short-cid") _make_commit(src, content=b"sec-short-cid") bundle = tmp_path / "short-cid.bundle" _make_bundle(src, bundle) raw = msgpack.unpackb(bundle.read_bytes(), raw=False) # Replace the commit IDs with a too-short value. raw["branch_heads"] = {"main": "tooshort"} bundle.write_bytes(msgpack.packb(raw, use_bin_type=True)) result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst)) assert result.exit_code == 0 data = _parse_unbundle(result) assert "main" not in data["refs_updated"] def test_no_json_on_missing_file(self, tmp_path: pathlib.Path) -> None: """Error path (file not found) must not emit JSON to stdout.""" _init_repo(tmp_path) result = _invoke( ["bundle", "unbundle", str(tmp_path / "ghost.bundle"), "--json"], env=_env(tmp_path), ) assert result.exit_code != 0 assert not result.output.strip().startswith("{") # =========================================================================== # TestBundleUnbundleStress — 3 tests # =========================================================================== class TestBundleUnbundleStress: def test_50_commit_chain(self, tmp_path: pathlib.Path) -> None: """50-commit chain is fully unpacked into the destination.""" src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="stress-ub-dst") prev: str | None = None for i in range(50): prev = _make_commit(src, parent_id=prev, content=f"ub50-{i}".encode()) bundle = tmp_path / "ub50.bundle" _make_bundle(src, bundle) result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst)) assert result.exit_code == 0 data = _parse_unbundle(result) assert data["commits_written"] == 50 assert data["blobs_written"] >= 50 def test_idempotent_multiple_applications(self, tmp_path: pathlib.Path) -> None: """Applying the same bundle 5 times: only the first writes anything.""" src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="stress-idem-dst") prev: str | None = None for i in range(5): prev = _make_commit(src, parent_id=prev, content=f"idem-{i}".encode()) bundle = tmp_path / "idem5.bundle" _make_bundle(src, bundle) first = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst)) assert first.exit_code == 0 first_data = _parse_unbundle(first) assert first_data["commits_written"] == 5 for _ in range(4): repeat = _invoke( ["bundle", "unbundle", str(bundle), "--json"], env=_env(dst) ) assert repeat.exit_code == 0 repeat_data = _parse_unbundle(repeat) assert repeat_data["commits_written"] == 0 assert repeat_data["blobs_written"] == 0 def test_many_branch_refs_updated(self, tmp_path: pathlib.Path) -> None: """10 branch heads in the bundle → all 10 appear in refs_updated.""" src = tmp_path / "src" dst = tmp_path / "dst" src.mkdir() dst.mkdir() _init_repo(src) _init_repo(dst, repo_id="stress-refs-dst") # Build a 10-commit chain on main. prev: str | None = None cids: list[str] = [] for i in range(10): prev = _make_commit(src, parent_id=prev, content=f"br-ref-{i}".encode()) cids.append(prev) # Create 10 feature branch refs pointing to reachable commits. br_names = [f"feat/br-{i}" for i in range(10)] for br, cid in zip(br_names, cids): ref = ref_path(src, br) ref.parent.mkdir(parents=True, exist_ok=True) ref.write_text(cid, encoding="utf-8") bundle = tmp_path / "many-refs.bundle" _make_bundle(src, bundle) result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst)) assert result.exit_code == 0 data = _parse_unbundle(result) for br in br_names: assert br in data["refs_updated"] # =========================================================================== # TestBundleVerifyExtended — 18 tests # =========================================================================== class TestBundleVerifyExtended: def _clean_bundle(self, tmp_path: pathlib.Path) -> pathlib.Path: """Create a repo with one commit and return a clean bundle path.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"vext-clean") out = tmp_path / "clean.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) return out def _corrupt_bundle(self, tmp_path: pathlib.Path) -> pathlib.Path: """Create a bundle then tamper one object's content.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"vext-corrupt") out = tmp_path / "corrupt.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) raw = msgpack.unpackb(out.read_bytes(), raw=False) if raw.get("blobs"): raw["blobs"][0]["content"] = b"TAMPERED" out.write_bytes(msgpack.packb(raw, use_bin_type=True)) return out def test_exits_0_on_clean_bundle(self, tmp_path: pathlib.Path) -> None: bundle = self._clean_bundle(tmp_path) result = _invoke(["bundle", "verify", str(bundle)], env=_env(tmp_path)) assert result.exit_code == 0 def test_exits_1_on_corrupt_object(self, tmp_path: pathlib.Path) -> None: bundle = self._corrupt_bundle(tmp_path) result = _invoke(["bundle", "verify", str(bundle)], env=_env(tmp_path)) assert result.exit_code == 1 def test_all_ok_true_on_clean(self, tmp_path: pathlib.Path) -> None: bundle = self._clean_bundle(tmp_path) result = _invoke(["bundle", "verify", str(bundle), "--json"], env=_env(tmp_path)) assert result.exit_code == 0 data = _parse_verify(result) assert data["all_ok"] is True def test_all_ok_false_on_corrupt(self, tmp_path: pathlib.Path) -> None: bundle = self._corrupt_bundle(tmp_path) result = _invoke(["bundle", "verify", str(bundle), "--json"], env=_env(tmp_path)) assert result.exit_code == 1 data = _parse_verify(result) assert data["all_ok"] is False def test_objects_checked_count(self, tmp_path: pathlib.Path) -> None: """blobs_checked must equal the number of blobs in the bundle.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"vext-cnt") out = tmp_path / "cnt.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) raw = msgpack.unpackb(out.read_bytes(), raw=False) n_objects = len(raw.get("blobs", [])) result = _invoke(["bundle", "verify", str(out), "--json"], env=_env(tmp_path)) assert result.exit_code == 0 data = _parse_verify(result) assert data["blobs_checked"] == n_objects def test_snapshots_checked_count(self, tmp_path: pathlib.Path) -> None: bundle = self._clean_bundle(tmp_path) result = _invoke(["bundle", "verify", str(bundle), "--json"], env=_env(tmp_path)) assert result.exit_code == 0 data = _parse_verify(result) assert data["snapshots_checked"] >= 1 def test_failures_empty_on_clean(self, tmp_path: pathlib.Path) -> None: bundle = self._clean_bundle(tmp_path) result = _invoke(["bundle", "verify", str(bundle), "--json"], env=_env(tmp_path)) data = _parse_verify(result) assert data["failures"] == [] def test_failures_nonempty_on_corrupt(self, tmp_path: pathlib.Path) -> None: bundle = self._corrupt_bundle(tmp_path) result = _invoke(["bundle", "verify", str(bundle), "--json"], env=_env(tmp_path)) data = _parse_verify(result) assert len(data["failures"]) >= 1 def test_quiet_clean_exits_0_no_output(self, tmp_path: pathlib.Path) -> None: bundle = self._clean_bundle(tmp_path) result = _invoke(["bundle", "verify", str(bundle), "--quiet"], env=_env(tmp_path)) assert result.exit_code == 0 assert result.output.strip() == "" def test_quiet_corrupt_exits_1_no_output(self, tmp_path: pathlib.Path) -> None: bundle = self._corrupt_bundle(tmp_path) result = _invoke(["bundle", "verify", str(bundle), "-q"], env=_env(tmp_path)) assert result.exit_code == 1 assert result.output.strip() == "" def test_json_output_is_single_line(self, tmp_path: pathlib.Path) -> None: """JSON output must be compact (no indent=2), matching all other commands.""" bundle = self._clean_bundle(tmp_path) result = _invoke(["bundle", "verify", str(bundle), "--json"], env=_env(tmp_path)) assert result.exit_code == 0 # Compact JSON has no interior newlines. assert "\n" not in result.output.strip() def test_j_alias(self, tmp_path: pathlib.Path) -> None: bundle = self._clean_bundle(tmp_path) r1 = _invoke(["bundle", "verify", str(bundle), "--json"], env=_env(tmp_path)) r2 = _invoke(["bundle", "verify", str(bundle), "-j"], env=_env(tmp_path)) assert r1.exit_code == 0 assert r2.exit_code == 0 _volatile = {"timestamp", "duration_ms"} d1 = {k: v for k, v in json.loads(r1.output).items() if k not in _volatile} d2 = {k: v for k, v in json.loads(r2.output).items() if k not in _volatile} assert d1 == d2 def test_text_output_mentions_objects_checked(self, tmp_path: pathlib.Path) -> None: bundle = self._clean_bundle(tmp_path) result = _invoke(["bundle", "verify", str(bundle)], env=_env(tmp_path)) assert "Blobs checked" in result.output def test_text_output_mentions_snapshots_checked(self, tmp_path: pathlib.Path) -> None: bundle = self._clean_bundle(tmp_path) result = _invoke(["bundle", "verify", str(bundle)], env=_env(tmp_path)) assert "Snapshots checked" in result.output def test_text_output_clean_checkmark(self, tmp_path: pathlib.Path) -> None: bundle = self._clean_bundle(tmp_path) result = _invoke(["bundle", "verify", str(bundle)], env=_env(tmp_path)) assert "Bundle is clean" in result.output def test_text_output_failures_listed(self, tmp_path: pathlib.Path) -> None: bundle = self._corrupt_bundle(tmp_path) result = _invoke(["bundle", "verify", str(bundle)], env=_env(tmp_path)) assert result.exit_code == 1 assert "hash mismatch" in result.output def test_help_mentions_agent_quickstart(self) -> None: result = _invoke(["bundle", "verify", "--help"]) assert result.exit_code == 0 assert "Agent quickstart" in result.output def test_help_mentions_exit_codes(self) -> None: result = _invoke(["bundle", "verify", "--help"]) assert result.exit_code == 0 assert "Exit codes" in result.output # =========================================================================== # TestBundleVerifySecurity — 6 tests # =========================================================================== class TestBundleVerifySecurity: def _bundle_with_ansi_object_id(self, tmp_path: pathlib.Path) -> pathlib.Path: """Bundle where an object_id contains an ANSI escape sequence.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"sec-ansi-oid") out = tmp_path / "ansi-oid.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) raw = msgpack.unpackb(out.read_bytes(), raw=False) if raw.get("blobs"): # Inject ANSI into the object_id — will trigger hash mismatch failure. raw["blobs"][0]["object_id"] = "\x1b[31mmalicious_oid_xxx\x1b[0m" out.write_bytes(msgpack.packb(raw, use_bin_type=True)) return out def _bundle_with_ansi_rel_path(self, tmp_path: pathlib.Path) -> pathlib.Path: """Bundle where a snapshot manifest key contains an ANSI escape.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"sec-ansi-path") out = tmp_path / "ansi-path.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) raw = msgpack.unpackb(out.read_bytes(), raw=False) if raw.get("snapshots"): snap = raw["snapshots"][0] # Replace manifest keys with ANSI-poisoned path. old_manifest = snap.get("manifest", {}) snap["manifest"] = { "\x1b[31mmalicious/path\x1b[0m": v for v in old_manifest.values() } out.write_bytes(msgpack.packb(raw, use_bin_type=True)) return out def test_ansi_in_object_id_failure_stripped_text( self, tmp_path: pathlib.Path ) -> None: """ANSI in object_id within a failure message must be stripped in text output.""" bundle = self._bundle_with_ansi_object_id(tmp_path) result = _invoke(["bundle", "verify", str(bundle)], env=_env(tmp_path)) assert "\x1b" not in result.output def test_ansi_in_rel_path_failure_stripped_text( self, tmp_path: pathlib.Path ) -> None: """ANSI in a manifest rel_path within a failure must be stripped in text output.""" bundle = self._bundle_with_ansi_rel_path(tmp_path) result = _invoke(["bundle", "verify", str(bundle)], env=_env(tmp_path)) assert "\x1b" not in result.output def test_ansi_in_failures_sanitized_json(self, tmp_path: pathlib.Path) -> None: """failures list in JSON output must not contain raw ANSI escapes.""" bundle = self._bundle_with_ansi_object_id(tmp_path) result = _invoke(["bundle", "verify", str(bundle), "--json"], env=_env(tmp_path)) assert "\x1b" not in result.output def test_no_repo_required(self, tmp_path: pathlib.Path) -> None: """verify must work outside any .muse repository (no require_repo call).""" work = tmp_path / "no_repo" work.mkdir() _init_repo(tmp_path) _make_commit(tmp_path, content=b"sec-no-repo") bundle = tmp_path / "no-repo.bundle" _invoke(["bundle", "create", str(bundle)], env=_env(tmp_path)) # Run verify from a directory with no .muse — must NOT exit 2. result = _invoke(["bundle", "verify", str(bundle)], env={"MUSE_REPO_ROOT": str(work)}) assert result.exit_code != 2 def test_missing_file_exits_1(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke( ["bundle", "verify", str(tmp_path / "ghost.bundle")], env=_env(tmp_path), ) assert result.exit_code == 1 def test_invalid_msgpack_exits_1(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) corrupt = tmp_path / "bad.bundle" corrupt.write_bytes(b"\xff\xfe not msgpack") result = _invoke(["bundle", "verify", str(corrupt)], env=_env(tmp_path)) assert result.exit_code == 1 # =========================================================================== # TestBundleVerifyStress — 3 tests # =========================================================================== class TestBundleVerifyStress: def test_200_commit_bundle_verify_clean(self, tmp_path: pathlib.Path) -> None: """200-commit bundle verifies clean with correct counts.""" _init_repo(tmp_path) prev: str | None = None for i in range(200): prev = _make_commit(tmp_path, parent_id=prev, content=f"vstress-{i}".encode()) out = tmp_path / "vstress200.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) result = _invoke(["bundle", "verify", str(out), "--json"], env=_env(tmp_path)) assert result.exit_code == 0 data = _parse_verify(result) assert data["all_ok"] is True assert data["blobs_checked"] >= 200 assert data["snapshots_checked"] >= 200 def test_multiple_corrupt_objects_all_detected( self, tmp_path: pathlib.Path ) -> None: """Multiple corrupted objects must each produce a failure entry.""" _init_repo(tmp_path) prev: str | None = None for i in range(5): prev = _make_commit(tmp_path, parent_id=prev, content=f"multi-corrupt-{i}".encode()) out = tmp_path / "multi-corrupt.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) raw = msgpack.unpackb(out.read_bytes(), raw=False) # Corrupt every object. for obj in raw.get("blobs", []): obj["content"] = b"TAMPERED" out.write_bytes(msgpack.packb(raw, use_bin_type=True)) result = _invoke(["bundle", "verify", str(out), "--json"], env=_env(tmp_path)) assert result.exit_code == 1 data = _parse_verify(result) assert data["all_ok"] is False assert len(data["failures"]) >= 5 def test_empty_bundle_verifies_clean(self, tmp_path: pathlib.Path) -> None: """An empty dict bundle has nothing to check and must exit 0.""" _init_repo(tmp_path) empty = tmp_path / "empty.bundle" empty.write_bytes(msgpack.packb({}, use_bin_type=True)) result = _invoke(["bundle", "verify", str(empty), "--json"], env=_env(tmp_path)) assert result.exit_code == 0 data = _parse_verify(result) assert data["all_ok"] is True assert data["blobs_checked"] == 0 assert data["failures"] == [] # =========================================================================== # TestBundleListHeadsExtended — 18 tests # =========================================================================== class TestBundleListHeadsExtended: def _bundle_with_head(self, tmp_path: pathlib.Path, branch: str = "main") -> pathlib.Path: _init_repo(tmp_path) _make_commit(tmp_path, content=b"lhe-base", branch=branch) out = tmp_path / "lhe.bundle" _invoke(["bundle", "create", str(out)], env=_env(tmp_path)) return out def test_exits_0_with_heads(self, tmp_path: pathlib.Path) -> None: bundle = self._bundle_with_head(tmp_path) result = _invoke(["bundle", "list-heads", str(bundle)], env=_env(tmp_path)) assert result.exit_code == 0 def test_exits_0_no_heads(self, tmp_path: pathlib.Path) -> None: """A bundle with no branch_heads key must still exit 0.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"lhe-noheads") bundle = tmp_path / "noheads.bundle" _invoke(["bundle", "create", str(bundle)], env=_env(tmp_path)) raw = msgpack.unpackb(bundle.read_bytes(), raw=False) raw.pop("branch_heads", None) bundle.write_bytes(msgpack.packb(raw, use_bin_type=True)) result = _invoke(["bundle", "list-heads", str(bundle)], env=_env(tmp_path)) assert result.exit_code == 0 def test_text_shows_branch_and_cid(self, tmp_path: pathlib.Path) -> None: bundle = self._bundle_with_head(tmp_path) result = _invoke(["bundle", "list-heads", str(bundle)], env=_env(tmp_path)) assert result.exit_code == 0 assert "main" in result.output def test_text_no_heads_message(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"lhe-nomsg") bundle = tmp_path / "nomsg.bundle" _invoke(["bundle", "create", str(bundle)], env=_env(tmp_path)) raw = msgpack.unpackb(bundle.read_bytes(), raw=False) raw.pop("branch_heads", None) bundle.write_bytes(msgpack.packb(raw, use_bin_type=True)) result = _invoke(["bundle", "list-heads", str(bundle)], env=_env(tmp_path)) assert "No branch heads" in result.output def test_json_returns_dict(self, tmp_path: pathlib.Path) -> None: bundle = self._bundle_with_head(tmp_path) result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path)) assert result.exit_code == 0 data = json.loads(result.output) assert isinstance(data["heads"], dict) def test_json_contains_main(self, tmp_path: pathlib.Path) -> None: bundle = self._bundle_with_head(tmp_path) result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path)) data = json.loads(result.output) assert "main" in data["heads"] def test_json_commit_id_has_sha256_prefix(self, tmp_path: pathlib.Path) -> None: bundle = self._bundle_with_head(tmp_path) result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path)) data = json.loads(result.output) for cid in data["heads"].values(): assert cid.startswith("sha256:") assert len(cid) == len("sha256:") + 64 def test_json_is_single_line(self, tmp_path: pathlib.Path) -> None: """JSON output must be compact (no indent=2).""" bundle = self._bundle_with_head(tmp_path) result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path)) assert "\n" not in result.output.strip() def test_j_alias(self, tmp_path: pathlib.Path) -> None: bundle = self._bundle_with_head(tmp_path) r1 = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path)) r2 = _invoke(["bundle", "list-heads", str(bundle), "-j"], env=_env(tmp_path)) assert r1.exit_code == 0 and r2.exit_code == 0 assert json.loads(r1.output)["heads"] == json.loads(r2.output)["heads"] def test_json_empty_on_no_heads(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, content=b"lhe-empty-json") bundle = tmp_path / "ej.bundle" _invoke(["bundle", "create", str(bundle)], env=_env(tmp_path)) raw = msgpack.unpackb(bundle.read_bytes(), raw=False) raw.pop("branch_heads", None) bundle.write_bytes(msgpack.packb(raw, use_bin_type=True)) result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path)) assert result.exit_code == 0 assert json.loads(result.output)["heads"] == {} def test_multiple_branches_all_listed_text(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) c1 = _make_commit(tmp_path, content=b"lhe-multi-base") for br in ("feat/a", "feat/b", "feat/c"): ref = ref_path(tmp_path, br) ref.parent.mkdir(parents=True, exist_ok=True) ref.write_text(c1, encoding="utf-8") bundle = tmp_path / "multi.bundle" _invoke(["bundle", "create", str(bundle)], env=_env(tmp_path)) result = _invoke(["bundle", "list-heads", str(bundle)], env=_env(tmp_path)) assert result.exit_code == 0 for br in ("feat/a", "feat/b", "feat/c"): assert br in result.output def test_multiple_branches_all_in_json(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) c1 = _make_commit(tmp_path, content=b"lhe-multi-json") for br in ("feat/x", "feat/y"): ref = ref_path(tmp_path, br) ref.parent.mkdir(parents=True, exist_ok=True) ref.write_text(c1, encoding="utf-8") bundle = tmp_path / "multij.bundle" _invoke(["bundle", "create", str(bundle)], env=_env(tmp_path)) result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path)) data = json.loads(result.output) assert "feat/x" in data["heads"] assert "feat/y" in data["heads"] def test_text_cid_shows_sha256_prefix_plus_12(self, tmp_path: pathlib.Path) -> None: """Text output shows sha256: prefix + 12 hex chars abbreviated commit ID.""" bundle = self._bundle_with_head(tmp_path) result = _invoke(["bundle", "list-heads", str(bundle)], env=_env(tmp_path)) # Each non-empty line should start with sha256:<12hex>. for line in result.output.strip().splitlines(): parts = line.split() assert parts[0].startswith("sha256:") assert len(parts[0]) == len("sha256:") + 12 def test_no_repo_required(self, tmp_path: pathlib.Path) -> None: """list-heads must work outside any .muse repository.""" work = tmp_path / "no_repo_dir" work.mkdir() _init_repo(tmp_path) _make_commit(tmp_path, content=b"lhe-no-repo") bundle = tmp_path / "norepo.bundle" _invoke(["bundle", "create", str(bundle)], env=_env(tmp_path)) result = _invoke(["bundle", "list-heads", str(bundle)], env={"MUSE_REPO_ROOT": str(work)}) assert result.exit_code != 2 def test_help_mentions_agent_quickstart(self) -> None: result = _invoke(["bundle", "list-heads", "--help"]) assert result.exit_code == 0 assert "Agent quickstart" in result.output def test_help_mentions_exit_codes(self) -> None: result = _invoke(["bundle", "list-heads", "--help"]) assert result.exit_code == 0 assert "Exit codes" in result.output def test_help_mentions_json_schema(self) -> None: result = _invoke(["bundle", "list-heads", "--help"]) assert result.exit_code == 0 assert "JSON output schema" in result.output def test_missing_file_exits_1(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke( ["bundle", "list-heads", str(tmp_path / "ghost.bundle")], env=_env(tmp_path), ) assert result.exit_code == 1 # =========================================================================== # TestBundleListHeadsSecurity — 6 tests # =========================================================================== class TestBundleListHeadsSecurity: def test_ansi_branch_name_stripped_text(self, tmp_path: pathlib.Path) -> None: """ANSI escape in branch name must not appear in text output.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"sec-lh-ansi") bundle = tmp_path / "ansi-br.bundle" _invoke(["bundle", "create", str(bundle)], env=_env(tmp_path)) raw = msgpack.unpackb(bundle.read_bytes(), raw=False) raw["branch_heads"] = {"\x1b[31mmalicious\x1b[0m": "a" * 64} bundle.write_bytes(msgpack.packb(raw, use_bin_type=True)) result = _invoke(["bundle", "list-heads", str(bundle)], env=_env(tmp_path)) assert result.exit_code == 0 assert "\x1b" not in result.output def test_ansi_commit_id_stripped_text(self, tmp_path: pathlib.Path) -> None: """ANSI escape in a commit ID must not appear in text output (cid[:12]).""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"sec-lh-cid") bundle = tmp_path / "ansi-cid.bundle" _invoke(["bundle", "create", str(bundle)], env=_env(tmp_path)) raw = msgpack.unpackb(bundle.read_bytes(), raw=False) raw["branch_heads"] = {"main": f"\x1b[31m{'a' * 64}\x1b[0m"} bundle.write_bytes(msgpack.packb(raw, use_bin_type=True)) result = _invoke(["bundle", "list-heads", str(bundle)], env=_env(tmp_path)) assert result.exit_code == 0 assert "\x1b" not in result.output def test_ansi_branch_name_stripped_json(self, tmp_path: pathlib.Path) -> None: """ANSI escape in branch name must not appear in JSON output.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"sec-lh-ansi-json") bundle = tmp_path / "ansi-br-json.bundle" _invoke(["bundle", "create", str(bundle)], env=_env(tmp_path)) raw = msgpack.unpackb(bundle.read_bytes(), raw=False) raw["branch_heads"] = {"\x1b[31mmalicious\x1b[0m": "b" * 64} bundle.write_bytes(msgpack.packb(raw, use_bin_type=True)) result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path)) assert result.exit_code == 0 assert "\x1b" not in result.output def test_ansi_commit_id_stripped_json(self, tmp_path: pathlib.Path) -> None: """ANSI escape in commit ID value must not appear in JSON output.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"sec-lh-cid-json") bundle = tmp_path / "ansi-cid-json.bundle" _invoke(["bundle", "create", str(bundle)], env=_env(tmp_path)) raw = msgpack.unpackb(bundle.read_bytes(), raw=False) raw["branch_heads"] = {"main": f"\x1b[32m{'c' * 64}\x1b[0m"} bundle.write_bytes(msgpack.packb(raw, use_bin_type=True)) result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path)) assert result.exit_code == 0 assert "\x1b" not in result.output def test_invalid_msgpack_exits_1(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) bad = tmp_path / "bad.bundle" bad.write_bytes(b"\xff\xfe garbage") result = _invoke(["bundle", "list-heads", str(bad)], env=_env(tmp_path)) assert result.exit_code == 1 def test_no_json_on_missing_file(self, tmp_path: pathlib.Path) -> None: """Missing file error must not emit JSON to stdout.""" _init_repo(tmp_path) result = _invoke( ["bundle", "list-heads", str(tmp_path / "missing.bundle"), "--json"], env=_env(tmp_path), ) assert result.exit_code != 0 assert not result.output.strip().startswith("{") # =========================================================================== # TestBundleListHeadsStress — 3 tests # =========================================================================== class TestBundleListHeadsStress: def test_50_branches_all_listed(self, tmp_path: pathlib.Path) -> None: """50 branch heads are all present in the JSON output.""" _init_repo(tmp_path) c1 = _make_commit(tmp_path, content=b"lhstress-base") branch_names = [f"feat/stress-{i}" for i in range(50)] for br in branch_names: ref = ref_path(tmp_path, br) ref.parent.mkdir(parents=True, exist_ok=True) ref.write_text(c1, encoding="utf-8") bundle = tmp_path / "stress50.bundle" _invoke(["bundle", "create", str(bundle)], env=_env(tmp_path)) result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path)) assert result.exit_code == 0 data = json.loads(result.output) for br in branch_names: assert br in data["heads"] def test_concurrent_reads_consistent(self, tmp_path: pathlib.Path) -> None: """Concurrent list-heads reads on the same bundle must all succeed.""" _init_repo(tmp_path) _make_commit(tmp_path, content=b"lhstress-concurrent") bundle = tmp_path / "concurrent.bundle" _invoke(["bundle", "create", str(bundle)], env=_env(tmp_path)) errors: list[str] = [] def _read() -> None: r = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path)) if r.exit_code != 0: errors.append(f"exit {r.exit_code}") else: try: if not isinstance(json.loads(r.output), dict): errors.append("not a dict") except json.JSONDecodeError as exc: errors.append(str(exc)) threads = [threading.Thread(target=_read) for _ in range(10)] for t in threads: t.start() for t in threads: t.join() assert not errors, f"Concurrent failures: {errors}" def test_large_bundle_list_heads_fast(self, tmp_path: pathlib.Path) -> None: """list-heads on a 200-commit bundle returns quickly (I/O, not compute).""" import time _init_repo(tmp_path) prev: str | None = None for i in range(200): prev = _make_commit(tmp_path, parent_id=prev, content=f"lhfast-{i}".encode()) bundle = tmp_path / "fast200.bundle" _invoke(["bundle", "create", str(bundle)], env=_env(tmp_path)) t0 = time.monotonic() result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path)) elapsed = time.monotonic() - t0 assert result.exit_code == 0 assert elapsed < 5.0, f"list-heads took {elapsed:.2f}s on 200-commit bundle" # --------------------------------------------------------------------------- # Flag registration tests # --------------------------------------------------------------------------- import argparse as _argparse from muse.cli.commands.bundle import register as _register_bundle from muse.core.paths import heads_dir, muse_dir, ref_path def _parse_bundle(*args: str) -> _argparse.Namespace: """Build an argument parser via register() and parse args.""" root_p = _argparse.ArgumentParser() subs = root_p.add_subparsers(dest="cmd") _register_bundle(subs) return root_p.parse_args(["bundle", *args]) class TestRegisterFlags: def test_create_default_json_out_is_false(self) -> None: ns = _parse_bundle("create", "out.bundle") assert ns.json_out is False def test_create_json_flag(self) -> None: ns = _parse_bundle("create", "out.bundle", "--json") assert ns.json_out is True def test_create_j_shorthand(self) -> None: ns = _parse_bundle("create", "out.bundle", "-j") assert ns.json_out is True def test_inspect_default_json_out_is_false(self) -> None: ns = _parse_bundle("inspect", "bundle.muse") assert ns.json_out is False def test_inspect_j_shorthand(self) -> None: ns = _parse_bundle("inspect", "bundle.muse", "-j") assert ns.json_out is True def test_verify_default_json_out_is_false(self) -> None: ns = _parse_bundle("verify", "bundle.muse") assert ns.json_out is False def test_verify_j_shorthand(self) -> None: ns = _parse_bundle("verify", "bundle.muse", "-j") assert ns.json_out is True