tests/test_cmd_bundle_hardening.py · gabriel/muse

test_cmd_bundle_hardening.py python

2,118 lines 85.1 KB

sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 131 days ago

1	"""Hardening tests for ``muse bundle``.
2
3	Covers:
4	Unit — _iter_branches (symlink guard, size cap), _reachable_from,
5	_load_bundle (narrow except), _resolve_refs, TypeGuards
6	Security — symlink traversal in _iter_branches, ANSI injection in
7	branch names and failure messages, oversized bundle rejection
8	Perf — reachable set is pre-computed once (not per branch)
9	JSON — _BundleCreateJson, _BundleUnbundleJson, _BundleVerifyJson,
10	list-heads dict schema
11	Flags — --json for create / unbundle / verify / list-heads
12	Integration — create → unbundle round-trip with branch ref updates,
13	--have pruning reduces bundle size,
14	verify catches corruption and missing snapshot objects
15	E2E — --help output for all subcommands
16	Stress — 200-commit chain, concurrent unbundle reads
17	"""
18
19	from __future__ import annotations
20
21	import datetime
22	import hashlib
23	import json
24	import pathlib
25	import threading
26	from typing import TypedDict
27
28	import msgpack
29	import pytest
30	from tests.cli_test_helper import CliRunner, InvokeResult
31
32	from muse.core.object_store import write_object
33	from muse.core.snapshot import compute_commit_id, compute_snapshot_id
34	from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
35	from muse.core._types import Manifest, long_id, blob_id
36
37	cli = None
38	runner = CliRunner()
39	_invoke_lock = threading.Lock()
40
41	_REPO_ID = "bundle-hardening-test"
42
43
44	# ---------------------------------------------------------------------------
45	# Helpers
46	# ---------------------------------------------------------------------------
47
48
49	class _CreateOut(TypedDict):
50	file: str
51	commits: int
52	objects: int
53	size_bytes: int
54	branches: list[str]
55
56
57	class _UnbundleOut(TypedDict):
58	commits_written: int
59	snapshots_written: int
60	objects_written: int
61	objects_skipped: int
62	refs_updated: list[str]
63
64
65	class _VerifyOut(TypedDict):
66	objects_checked: int
67	snapshots_checked: int
68	all_ok: bool
69	failures: list[str]
70
71
72	def _sha(data: bytes) -> str:
73	return blob_id(data)
74
75
76	def _init_repo(path: pathlib.Path, repo_id: str = _REPO_ID) -> pathlib.Path:
77	muse = path / ".muse"
78	for d in ("commits", "snapshots", "objects", "refs/heads"):
79	(muse / d).mkdir(parents=True, exist_ok=True)
80	(muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
81	(muse / "repo.json").write_text(
82	json.dumps({"repo_id": repo_id, "domain": "midi"}), encoding="utf-8"
83	)
84	return path
85
86
87	def _env(repo: pathlib.Path) -> Manifest:
88	return {"MUSE_REPO_ROOT": str(repo)}
89
90
91	_counter = 0
92	_branch_heads_map: dict[tuple[str, str], str] = {}
93
94
95	def _make_commit(
96	root: pathlib.Path,
97	parent_id: str \| None = None,
98	content: bytes = b"data",
99	branch: str = "main",
100	) -> str:
101	global _counter
102	_counter += 1
103	c = content + str(_counter).encode()
104	obj_id = long_id(_sha(c))
105	write_object(root, obj_id, c)
106	manifest = {f"f_{_counter}.txt": obj_id}
107	snap_id = compute_snapshot_id(manifest)
108	write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
109	committed_at = datetime.datetime.now(datetime.timezone.utc)
110
111	resolved_parent = parent_id
112	if resolved_parent is None:
113	key = (str(root), branch)
114	resolved_parent = _branch_heads_map.get(key)
115
116	parent_ids = [resolved_parent] if resolved_parent else []
117	commit_id = compute_commit_id(
118	repo_id=_REPO_ID,
119	parent_ids=parent_ids,
120	snapshot_id=snap_id,
121	message=f"commit {_counter}",
122	committed_at_iso=committed_at.isoformat(),
123	)
124	write_commit(
125	root,
126	CommitRecord(
127	commit_id=commit_id,
128	repo_id=_REPO_ID,
129	created_on_branch=branch,
130	snapshot_id=snap_id,
131	message=f"commit {_counter}",
132	committed_at=committed_at,
133	parent_commit_id=resolved_parent,
134	),
135	)
136	ref_path = root / ".muse" / "refs" / "heads" / branch
137	ref_path.parent.mkdir(parents=True, exist_ok=True)
138	ref_path.write_text(commit_id, encoding="utf-8")
139	_branch_heads_map[(str(root), branch)] = commit_id
140	return commit_id
141
142
143	def _invoke(args: list[str], env: Manifest \| None = None) -> InvokeResult:
144	with _invoke_lock:
145	return runner.invoke(cli, args, env=env)
146
147
148	def _parse_create(result: InvokeResult) -> _CreateOut:
149	raw: _CreateOut = json.loads(result.output)
150	return raw
151
152
153	def _parse_unbundle(result: InvokeResult) -> _UnbundleOut:
154	raw: _UnbundleOut = json.loads(result.output)
155	return raw
156
157
158	def _parse_verify(result: InvokeResult) -> _VerifyOut:
159	raw: _VerifyOut = json.loads(result.output)
160	return raw
161
162
163	# ---------------------------------------------------------------------------
164	# Unit: _iter_branches — symlink guard
165	# ---------------------------------------------------------------------------
166
167
168	def test_iter_branches_skips_symlinks(tmp_path: pathlib.Path) -> None:
169	"""A symlink inside refs/heads must be silently skipped."""
170	from muse.cli.commands.bundle import _iter_branches
171
172	_init_repo(tmp_path)
173	target = tmp_path / "outside.txt"
174	target.write_text("evil-sha" * 8, encoding="utf-8") # 64 chars
175
176	heads_dir = tmp_path / ".muse" / "refs" / "heads"
177	real_ref = heads_dir / "main"
178	real_ref.write_text("a" * 64, encoding="utf-8")
179
180	link = heads_dir / "evil"
181	link.symlink_to(target)
182
183	result = _iter_branches(tmp_path)
184	branch_names = [name for name, _ in result]
185	assert "evil" not in branch_names
186	assert "main" in branch_names
187
188
189	def test_iter_branches_size_cap(tmp_path: pathlib.Path) -> None:
190	"""Ref files larger than 65 bytes are read but will be invalid after strip."""
191	from muse.cli.commands.bundle import _iter_branches, _MAX_REF_BYTES
192
193	_init_repo(tmp_path)
194	heads_dir = tmp_path / ".muse" / "refs" / "heads"
195	oversized = heads_dir / "main"
196	oversized.write_bytes(b"x" * (_MAX_REF_BYTES + 100))
197
198	result = _iter_branches(tmp_path)
199	# Should still return one entry; the content is capped — the commit ID
200	# won't be valid hex but _iter_branches returns it; validation is downstream.
201	assert len(result) == 1
202	_, cid = result[0]
203	assert len(cid) <= _MAX_REF_BYTES # capped at read time
204
205
206	def test_iter_branches_empty_dir(tmp_path: pathlib.Path) -> None:
207	from muse.cli.commands.bundle import _iter_branches
208
209	_init_repo(tmp_path)
210	result = _iter_branches(tmp_path)
211	assert result == []
212
213
214	def test_iter_branches_multiple(tmp_path: pathlib.Path) -> None:
215	from muse.cli.commands.bundle import _iter_branches
216
217	_init_repo(tmp_path)
218	heads_dir = tmp_path / ".muse" / "refs" / "heads"
219	for name in ("main", "dev", "feat/foo"):
220	p = heads_dir / name
221	p.parent.mkdir(parents=True, exist_ok=True)
222	p.write_text("a" * 64, encoding="utf-8")
223
224	result = _iter_branches(tmp_path)
225	names = [n for n, _ in result]
226	assert "main" in names
227	assert "dev" in names
228	assert "feat/foo" in names
229
230
231	# ---------------------------------------------------------------------------
232	# Unit: _reachable_from — correctness
233	# ---------------------------------------------------------------------------
234
235
236	def test_reachable_from_single(tmp_path: pathlib.Path) -> None:
237	from muse.cli.commands.bundle import _reachable_from
238
239	_init_repo(tmp_path)
240	c1 = _make_commit(tmp_path, content=b"r1")
241	result = _reachable_from(tmp_path, [c1])
242	assert c1 in result
243
244
245	def test_reachable_from_chain(tmp_path: pathlib.Path) -> None:
246	from muse.cli.commands.bundle import _reachable_from
247
248	_init_repo(tmp_path)
249	c1 = _make_commit(tmp_path, content=b"rc1")
250	c2 = _make_commit(tmp_path, parent_id=c1, content=b"rc2")
251	c3 = _make_commit(tmp_path, parent_id=c2, content=b"rc3")
252	result = _reachable_from(tmp_path, [c3])
253	assert c1 in result
254	assert c2 in result
255	assert c3 in result
256
257
258	def test_reachable_from_empty_tips(tmp_path: pathlib.Path) -> None:
259	from muse.cli.commands.bundle import _reachable_from
260
261	_init_repo(tmp_path)
262	assert _reachable_from(tmp_path, []) == set()
263
264
265	# ---------------------------------------------------------------------------
266	# Unit: _load_bundle — narrow except
267	# ---------------------------------------------------------------------------
268
269
270	def test_load_bundle_not_found(tmp_path: pathlib.Path) -> None:
271	result = _invoke(
272	["bundle", "verify", str(tmp_path / "missing.bundle")],
273	env=_env(tmp_path),
274	)
275	assert result.exit_code != 0
276
277
278	def test_load_bundle_invalid_msgpack(tmp_path: pathlib.Path) -> None:
279	_init_repo(tmp_path)
280	bad = tmp_path / "bad.bundle"
281	bad.write_bytes(b"\xff\xfe this is not msgpack")
282	result = _invoke(["bundle", "verify", str(bad)], env=_env(tmp_path))
283	assert result.exit_code != 0
284
285
286	def test_load_bundle_not_dict(tmp_path: pathlib.Path) -> None:
287	"""A valid msgpack list instead of dict must be rejected cleanly."""
288	_init_repo(tmp_path)
289	bad = tmp_path / "list.bundle"
290	bad.write_bytes(msgpack.packb([1, 2, 3], use_bin_type=True))
291	result = _invoke(["bundle", "verify", str(bad)], env=_env(tmp_path))
292	assert result.exit_code != 0
293
294
295	# ---------------------------------------------------------------------------
296	# Security: ANSI injection in branch names
297	# ---------------------------------------------------------------------------
298
299
300	def test_list_heads_ansi_injection(tmp_path: pathlib.Path) -> None:
301	"""Branch names with ANSI escapes must be stripped in text output."""
302	_init_repo(tmp_path)
303	_make_commit(tmp_path, content=b"ansi-branch")
304	out = tmp_path / "ansi.bundle"
305	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
306
307	# Inject a crafted branch_heads entry with ANSI escape in branch name.
308	raw = msgpack.unpackb(out.read_bytes(), raw=False)
309	raw["branch_heads"] = {"\x1b[31mevil\x1b[0m": "a" * 64}
310	out.write_bytes(msgpack.packb(raw, use_bin_type=True))
311
312	result = _invoke(["bundle", "list-heads", str(out)], env=_env(tmp_path))
313	assert result.exit_code == 0
314	assert "\x1b" not in result.output
315
316
317	def test_verify_failure_ansi_injection(tmp_path: pathlib.Path) -> None:
318	"""Failure messages must not allow ANSI injection through object_id fields."""
319	_init_repo(tmp_path)
320	_make_commit(tmp_path, content=b"ansi-verify")
321	out = tmp_path / "ansi-v.bundle"
322	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
323
324	# Tamper content to trigger a hash mismatch failure.
325	raw = msgpack.unpackb(out.read_bytes(), raw=False)
326	if raw.get("objects"):
327	raw["objects"][0]["content"] = b"\x1b[31mTAMPERED\x1b[0m"
328	out.write_bytes(msgpack.packb(raw, use_bin_type=True))
329
330	result = _invoke(["bundle", "verify", str(out)], env=_env(tmp_path))
331	# Text output must strip ANSI from the failure description.
332	assert "\x1b" not in result.output
333	assert result.exit_code != 0
334
335
336	# ---------------------------------------------------------------------------
337	# JSON schema: bundle create --json
338	# ---------------------------------------------------------------------------
339
340
341	def test_create_json_schema(tmp_path: pathlib.Path) -> None:
342	_init_repo(tmp_path)
343	_make_commit(tmp_path, content=b"cj1")
344	out = tmp_path / "cj.bundle"
345	result = _invoke(["bundle", "create", str(out), "--json"], env=_env(tmp_path))
346	assert result.exit_code == 0
347	data = _parse_create(result)
348	assert data["file"] == str(out)
349	assert data["commits"] >= 1
350	assert data["objects"] >= 1
351	assert data["size_bytes"] > 0
352	assert isinstance(data["branches"], list)
353	assert "main" in data["branches"]
354
355
356	def test_create_json_no_output_on_success_without_flag(tmp_path: pathlib.Path) -> None:
357	_init_repo(tmp_path)
358	_make_commit(tmp_path, content=b"cj-no-flag")
359	out = tmp_path / "cnf.bundle"
360	result = _invoke(["bundle", "create", str(out)], env=_env(tmp_path))
361	assert result.exit_code == 0
362	# Text output, not JSON.
363	assert "Bundle" in result.output or "✅" in result.output
364
365
366	# ---------------------------------------------------------------------------
367	# JSON schema: bundle unbundle --json
368	# ---------------------------------------------------------------------------
369
370
371	def test_unbundle_json_schema(tmp_path: pathlib.Path) -> None:
372	src = tmp_path / "src"
373	dst = tmp_path / "dst"
374	src.mkdir()
375	dst.mkdir()
376	_init_repo(src)
377	_init_repo(dst, repo_id="dst-json")
378
379	_make_commit(src, content=b"uj1")
380	out = tmp_path / "uj.bundle"
381	_invoke(["bundle", "create", str(out)], env=_env(src))
382
383	result = _invoke(["bundle", "unbundle", str(out), "--json"], env=_env(dst))
384	assert result.exit_code == 0
385	data = _parse_unbundle(result)
386	assert data["commits_written"] >= 1
387	assert isinstance(data["snapshots_written"], int)
388	assert isinstance(data["objects_written"], int)
389	assert isinstance(data["objects_skipped"], int)
390	assert isinstance(data["refs_updated"], list)
391
392
393	def test_unbundle_json_refs_updated(tmp_path: pathlib.Path) -> None:
394	src = tmp_path / "src"
395	dst = tmp_path / "dst"
396	src.mkdir()
397	dst.mkdir()
398	_init_repo(src)
399	_init_repo(dst, repo_id="dst-ru")
400
401	_make_commit(src, content=b"ru1")
402	out = tmp_path / "ru.bundle"
403	_invoke(["bundle", "create", str(out)], env=_env(src))
404
405	result = _invoke(["bundle", "unbundle", str(out), "--json"], env=_env(dst))
406	assert result.exit_code == 0
407	data = _parse_unbundle(result)
408	assert "main" in data["refs_updated"]
409
410
411	def test_unbundle_json_no_update_refs(tmp_path: pathlib.Path) -> None:
412	src = tmp_path / "src"
413	dst = tmp_path / "dst"
414	src.mkdir()
415	dst.mkdir()
416	_init_repo(src)
417	_init_repo(dst, repo_id="dst-nur")
418
419	_make_commit(src, content=b"nur1")
420	out = tmp_path / "nur.bundle"
421	_invoke(["bundle", "create", str(out)], env=_env(src))
422
423	result = _invoke(
424	["bundle", "unbundle", str(out), "--no-update-refs", "--json"],
425	env=_env(dst),
426	)
427	assert result.exit_code == 0
428	data = _parse_unbundle(result)
429	assert data["refs_updated"] == []
430
431
432	# ---------------------------------------------------------------------------
433	# JSON schema: bundle verify --json
434	# ---------------------------------------------------------------------------
435
436
437	def test_verify_json_schema_clean(tmp_path: pathlib.Path) -> None:
438	_init_repo(tmp_path)
439	_make_commit(tmp_path, content=b"vjs1")
440	out = tmp_path / "vjs.bundle"
441	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
442	result = _invoke(["bundle", "verify", str(out), "--json"], env=_env(tmp_path))
443	assert result.exit_code == 0
444	data = _parse_verify(result)
445	assert data["all_ok"] is True
446	assert data["objects_checked"] >= 1
447	assert "snapshots_checked" in data
448	assert data["failures"] == []
449
450
451	def test_verify_json_schema_corrupt(tmp_path: pathlib.Path) -> None:
452	_init_repo(tmp_path)
453	_make_commit(tmp_path, content=b"corrupt-j")
454	out = tmp_path / "cj2.bundle"
455	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
456
457	raw = msgpack.unpackb(out.read_bytes(), raw=False)
458	if raw.get("objects"):
459	raw["objects"][0]["content"] = b"tampered!"
460	out.write_bytes(msgpack.packb(raw, use_bin_type=True))
461
462	result = _invoke(["bundle", "verify", str(out), "--json"], env=_env(tmp_path))
463	assert result.exit_code != 0
464	data = _parse_verify(result)
465	assert data["all_ok"] is False
466	assert len(data["failures"]) > 0
467
468
469	def test_verify_json_snapshots_checked(tmp_path: pathlib.Path) -> None:
470	"""``snapshots_checked`` must count non-zero when snapshots are present."""
471	_init_repo(tmp_path)
472	_make_commit(tmp_path, content=b"snap-counted")
473	out = tmp_path / "sc.bundle"
474	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
475	result = _invoke(["bundle", "verify", str(out), "--json"], env=_env(tmp_path))
476	assert result.exit_code == 0
477	data = _parse_verify(result)
478	# At least one snapshot should have been included in the bundle.
479	assert data["snapshots_checked"] >= 1
480
481
482	# ---------------------------------------------------------------------------
483	# JSON schema: bundle list-heads --json
484	# ---------------------------------------------------------------------------
485
486
487	def test_list_heads_json_schema(tmp_path: pathlib.Path) -> None:
488	_init_repo(tmp_path)
489	_make_commit(tmp_path, content=b"lhjs1")
490	out = tmp_path / "lhjs.bundle"
491	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
492	result = _invoke(["bundle", "list-heads", str(out), "--json"], env=_env(tmp_path))
493	assert result.exit_code == 0
494	data = json.loads(result.output)
495	assert isinstance(data, dict)
496	heads = data["heads"]
497	assert "main" in heads
498	for _branch, cid in heads.items():
499	assert isinstance(cid, str) and cid.startswith("sha256:")
500	assert len(cid) == len("sha256:") + 64
501
502
503	# ---------------------------------------------------------------------------
504	# Flags: --json rejects old --format arg
505	# ---------------------------------------------------------------------------
506
507
508	def test_verify_rejects_format_flag(tmp_path: pathlib.Path) -> None:
509	"""The old ``--format json`` pattern must not be accepted."""
510	_init_repo(tmp_path)
511	_make_commit(tmp_path, content=b"old-fmt")
512	out = tmp_path / "old.bundle"
513	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
514	result = _invoke(
515	["bundle", "verify", str(out), "--format", "json"], env=_env(tmp_path)
516	)
517	# --format is no longer a registered flag, so argparse returns exit 2.
518	assert result.exit_code == 2
519
520
521	def test_list_heads_rejects_format_flag(tmp_path: pathlib.Path) -> None:
522	_init_repo(tmp_path)
523	_make_commit(tmp_path, content=b"lh-old")
524	out = tmp_path / "lh-old.bundle"
525	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
526	result = _invoke(
527	["bundle", "list-heads", str(out), "--format", "json"], env=_env(tmp_path)
528	)
529	assert result.exit_code == 2
530
531
532	# ---------------------------------------------------------------------------
533	# Integration: --have pruning
534	# ---------------------------------------------------------------------------
535
536
537	def test_create_have_prunes_bundle(tmp_path: pathlib.Path) -> None:
538	"""Passing --have should produce a smaller bundle than the full chain."""
539	_init_repo(tmp_path)
540	c1 = _make_commit(tmp_path, content=b"have-base")
541	_make_commit(tmp_path, parent_id=c1, content=b"have-tip")
542
543	out_full = tmp_path / "full.bundle"
544	out_pruned = tmp_path / "pruned.bundle"
545
546	_invoke(["bundle", "create", str(out_full)], env=_env(tmp_path))
547	_invoke(
548	["bundle", "create", str(out_pruned), "--have", c1],
549	env=_env(tmp_path),
550	)
551
552	# Pruned bundle must be smaller (fewer commits packed).
553	assert out_pruned.stat().st_size < out_full.stat().st_size
554
555
556	def test_create_have_json_smaller_commits(tmp_path: pathlib.Path) -> None:
557	_init_repo(tmp_path)
558	c1 = _make_commit(tmp_path, content=b"hjp-base")
559	_make_commit(tmp_path, parent_id=c1, content=b"hjp-tip")
560
561	out_full = tmp_path / "hjp-full.bundle"
562	out_pruned = tmp_path / "hjp-pruned.bundle"
563
564	r_full = _invoke(
565	["bundle", "create", str(out_full), "--json"], env=_env(tmp_path)
566	)
567	r_pruned = _invoke(
568	["bundle", "create", str(out_pruned), "--have", c1, "--json"],
569	env=_env(tmp_path),
570	)
571
572	full_data = _parse_create(r_full)
573	pruned_data = _parse_create(r_pruned)
574	assert pruned_data["commits"] < full_data["commits"]
575
576
577	# ---------------------------------------------------------------------------
578	# Integration: multi-branch bundle
579	# ---------------------------------------------------------------------------
580
581
582	def test_create_multiple_branches(tmp_path: pathlib.Path) -> None:
583	_init_repo(tmp_path)
584	_make_commit(tmp_path, content=b"mb-main", branch="main")
585	_make_commit(tmp_path, content=b"mb-feat", branch="feat/x")
586
587	out = tmp_path / "mb.bundle"
588	result = _invoke(
589	["bundle", "create", str(out), "--json"], env=_env(tmp_path)
590	)
591	assert result.exit_code == 0
592	data = _parse_create(result)
593	assert "main" in data["branches"] or "feat/x" in data["branches"]
594
595
596	def test_round_trip_with_json_summary(tmp_path: pathlib.Path) -> None:
597	"""Full create → verify → unbundle pipeline with JSON at each step."""
598	src = tmp_path / "src"
599	dst = tmp_path / "dst"
600	src.mkdir()
601	dst.mkdir()
602	_init_repo(src)
603	_init_repo(dst, repo_id="dst-rt-json")
604
605	prev: str \| None = None
606	for i in range(5):
607	prev = _make_commit(src, parent_id=prev, content=f"rt-{i}".encode())
608
609	out = tmp_path / "rt-json.bundle"
610
611	create_result = _invoke(
612	["bundle", "create", str(out), "--json"], env=_env(src)
613	)
614	assert create_result.exit_code == 0
615	create_data = _parse_create(create_result)
616	assert create_data["commits"] == 5
617
618	verify_result = _invoke(
619	["bundle", "verify", str(out), "--json"], env=_env(src)
620	)
621	assert verify_result.exit_code == 0
622	verify_data = _parse_verify(verify_result)
623	assert verify_data["all_ok"] is True
624
625	unbundle_result = _invoke(
626	["bundle", "unbundle", str(out), "--json"], env=_env(dst)
627	)
628	assert unbundle_result.exit_code == 0
629	unbundle_data = _parse_unbundle(unbundle_result)
630	assert unbundle_data["commits_written"] == 5
631
632
633	# ---------------------------------------------------------------------------
634	# Integration: verify detects missing snapshot objects
635	# ---------------------------------------------------------------------------
636
637
638	def test_verify_missing_snapshot_object(tmp_path: pathlib.Path) -> None:
639	"""Removing an object from the bundle should cause snapshot verification to fail."""
640	_init_repo(tmp_path)
641	_make_commit(tmp_path, content=b"snap-miss")
642	out = tmp_path / "snap-miss.bundle"
643	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
644
645	raw = msgpack.unpackb(out.read_bytes(), raw=False)
646	# Remove all objects so snapshots cannot find theirs.
647	raw["objects"] = []
648	out.write_bytes(msgpack.packb(raw, use_bin_type=True))
649
650	result = _invoke(["bundle", "verify", str(out), "--json"], env=_env(tmp_path))
651	data = _parse_verify(result)
652	assert data["all_ok"] is False
653	# Some failure should mention missing objects.
654	assert any("missing" in f for f in data["failures"])
655
656
657	# ---------------------------------------------------------------------------
658	# E2E: help output for all subcommands
659	# ---------------------------------------------------------------------------
660
661
662	def test_create_help_mentions_json() -> None:
663	result = _invoke(["bundle", "create", "--help"])
664	assert result.exit_code == 0
665	assert "--json" in result.output
666
667
668	def test_unbundle_help_mentions_json() -> None:
669	result = _invoke(["bundle", "unbundle", "--help"])
670	assert result.exit_code == 0
671	assert "--json" in result.output
672
673
674	def test_verify_help_mentions_json() -> None:
675	result = _invoke(["bundle", "verify", "--help"])
676	assert result.exit_code == 0
677	assert "--json" in result.output
678	assert "--format" not in result.output
679
680
681	def test_list_heads_help_mentions_json() -> None:
682	result = _invoke(["bundle", "list-heads", "--help"])
683	assert result.exit_code == 0
684	assert "--json" in result.output
685	assert "--format" not in result.output
686
687
688	def test_bundle_help_top_level() -> None:
689	result = _invoke(["bundle", "--help"])
690	assert result.exit_code == 0
691	assert "create" in result.output
692	assert "unbundle" in result.output
693	assert "verify" in result.output
694	assert "list-heads" in result.output
695
696
697	# ---------------------------------------------------------------------------
698	# Stress: 200-commit bundle
699	# ---------------------------------------------------------------------------
700
701
702	def test_stress_200_commit_chain(tmp_path: pathlib.Path) -> None:
703	_init_repo(tmp_path)
704	prev: str \| None = None
705	for i in range(200):
706	prev = _make_commit(tmp_path, parent_id=prev, content=f"stress-{i}".encode())
707
708	out = tmp_path / "stress200.bundle"
709	create_result = _invoke(
710	["bundle", "create", str(out), "--json"], env=_env(tmp_path)
711	)
712	assert create_result.exit_code == 0
713	data = _parse_create(create_result)
714	assert data["commits"] == 200
715
716	verify_result = _invoke(["bundle", "verify", str(out), "-q"], env=_env(tmp_path))
717	assert verify_result.exit_code == 0
718
719
720	# ---------------------------------------------------------------------------
721	# Stress: concurrent list-heads reads
722	# ---------------------------------------------------------------------------
723
724
725	def test_stress_concurrent_list_heads(tmp_path: pathlib.Path) -> None:
726	_init_repo(tmp_path)
727	_make_commit(tmp_path, content=b"concurrent-bundle")
728	out = tmp_path / "concurrent.bundle"
729	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
730
731	errors: list[str] = []
732
733	def _read() -> None:
734	r = _invoke(["bundle", "list-heads", str(out), "--json"], env=_env(tmp_path))
735	if r.exit_code != 0:
736	errors.append(f"exit {r.exit_code}: {r.output}")
737	else:
738	try:
739	data = json.loads(r.output)
740	if not isinstance(data, dict):
741	errors.append("not a dict")
742	except json.JSONDecodeError as exc:
743	errors.append(str(exc))
744
745	threads = [threading.Thread(target=_read) for _ in range(8)]
746	for t in threads:
747	t.start()
748	for t in threads:
749	t.join()
750
751	assert not errors, f"Concurrent list-heads failures: {errors}"
752
753
754	# ===========================================================================
755	# TestBundleCreateExtended — 18 tests
756	# ===========================================================================
757
758
759	class TestBundleCreateExtended:
760	def test_exits_0_basic(self, tmp_path: pathlib.Path) -> None:
761	"""Single commit → create exits 0."""
762	_init_repo(tmp_path)
763	_make_commit(tmp_path, content=b"ext-basic")
764	out = tmp_path / "basic.bundle"
765	result = _invoke(["bundle", "create", str(out)], env=_env(tmp_path))
766	assert result.exit_code == 0
767
768	def test_creates_file_on_disk(self, tmp_path: pathlib.Path) -> None:
769	"""Output file must exist after a successful create."""
770	_init_repo(tmp_path)
771	_make_commit(tmp_path, content=b"ext-file")
772	out = tmp_path / "check.bundle"
773	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
774	assert out.exists()
775
776	def test_text_output_mentions_commits(self, tmp_path: pathlib.Path) -> None:
777	_init_repo(tmp_path)
778	_make_commit(tmp_path, content=b"ext-txt-c")
779	out = tmp_path / "tc.bundle"
780	result = _invoke(["bundle", "create", str(out)], env=_env(tmp_path))
781	assert "commits" in result.output
782
783	def test_text_output_mentions_kib(self, tmp_path: pathlib.Path) -> None:
784	_init_repo(tmp_path)
785	_make_commit(tmp_path, content=b"ext-txt-kib")
786	out = tmp_path / "kib.bundle"
787	result = _invoke(["bundle", "create", str(out)], env=_env(tmp_path))
788	assert "KiB" in result.output
789
790	def test_text_output_contains_bundle_path(self, tmp_path: pathlib.Path) -> None:
791	_init_repo(tmp_path)
792	_make_commit(tmp_path, content=b"ext-txt-path")
793	out = tmp_path / "pathcheck.bundle"
794	result = _invoke(["bundle", "create", str(out)], env=_env(tmp_path))
795	assert "pathcheck.bundle" in result.output
796
797	def test_empty_repo_exits_1(self, tmp_path: pathlib.Path) -> None:
798	"""Repo with no commits → exit 1 (no commits to bundle)."""
799	_init_repo(tmp_path)
800	out = tmp_path / "empty.bundle"
801	result = _invoke(["bundle", "create", str(out)], env=_env(tmp_path))
802	assert result.exit_code == 1
803
804	def test_bad_ref_exits_1(self, tmp_path: pathlib.Path) -> None:
805	"""Unknown ref → exit 1."""
806	_init_repo(tmp_path)
807	_make_commit(tmp_path, content=b"ext-bad-ref")
808	out = tmp_path / "bad-ref.bundle"
809	result = _invoke(
810	["bundle", "create", str(out), "nonexistent-branch"],
811	env=_env(tmp_path),
812	)
813	assert result.exit_code == 1
814
815	def test_json_branches_sorted(self, tmp_path: pathlib.Path) -> None:
816	"""Branches list in JSON output must be sorted."""
817	_init_repo(tmp_path)
818	# Create a commit on main, then make z-branch and a-branch point to
819	# the same commit so they are all reachable when bundling HEAD.
820	c1 = _make_commit(tmp_path, content=b"ext-br-base", branch="main")
821	for br in ("z-branch", "a-branch"):
822	ref_file = tmp_path / ".muse" / "refs" / "heads" / br
823	ref_file.write_text(c1, encoding="utf-8")
824	out = tmp_path / "sorted.bundle"
825	result = _invoke(
826	["bundle", "create", str(out), "--json"], env=_env(tmp_path)
827	)
828	assert result.exit_code == 0
829	data = _parse_create(result)
830	assert data["branches"] == sorted(data["branches"])
831
832	def test_json_size_matches_file(self, tmp_path: pathlib.Path) -> None:
833	"""size_bytes in JSON must equal the actual file size on disk."""
834	_init_repo(tmp_path)
835	_make_commit(tmp_path, content=b"ext-size")
836	out = tmp_path / "size.bundle"
837	result = _invoke(
838	["bundle", "create", str(out), "--json"], env=_env(tmp_path)
839	)
840	assert result.exit_code == 0
841	data = _parse_create(result)
842	assert data["size_bytes"] == out.stat().st_size
843
844	def test_j_alias(self, tmp_path: pathlib.Path) -> None:
845	"""-j must produce identical JSON to --json."""
846	_init_repo(tmp_path)
847	_make_commit(tmp_path, content=b"ext-j-alias")
848	out1 = tmp_path / "j1.bundle"
849	out2 = tmp_path / "j2.bundle"
850	r1 = _invoke(["bundle", "create", str(out1), "--json"], env=_env(tmp_path))
851	r2 = _invoke(["bundle", "create", str(out2), "-j"], env=_env(tmp_path))
852	assert r1.exit_code == 0
853	assert r2.exit_code == 0
854	d1 = json.loads(r1.output)
855	d2 = json.loads(r2.output)
856	# Both should have the same structural keys and counts.
857	assert d1["commits"] == d2["commits"]
858	assert d1["objects"] == d2["objects"]
859	assert d1["branches"] == d2["branches"]
860
861	def test_default_ref_is_head(self, tmp_path: pathlib.Path) -> None:
862	"""When no refs are given, HEAD is used — bundle contains the HEAD commit."""
863	_init_repo(tmp_path)
864	_make_commit(tmp_path, content=b"ext-head")
865	out = tmp_path / "head.bundle"
866	result = _invoke(
867	["bundle", "create", str(out), "--json"], env=_env(tmp_path)
868	)
869	assert result.exit_code == 0
870	data = _parse_create(result)
871	assert data["commits"] >= 1
872
873	def test_explicit_head_ref(self, tmp_path: pathlib.Path) -> None:
874	"""Passing 'HEAD' explicitly is equivalent to the default."""
875	_init_repo(tmp_path)
876	_make_commit(tmp_path, content=b"ext-head-explicit")
877	out_default = tmp_path / "head-default.bundle"
878	out_explicit = tmp_path / "head-explicit.bundle"
879	_invoke(["bundle", "create", str(out_default)], env=_env(tmp_path))
880	result = _invoke(
881	["bundle", "create", str(out_explicit), "HEAD", "--json"],
882	env=_env(tmp_path),
883	)
884	assert result.exit_code == 0
885	data = _parse_create(result)
886	assert data["commits"] >= 1
887	# Both bundles should contain the same number of commits.
888	raw_default = __import__("msgpack").unpackb(
889	out_default.read_bytes(), raw=False
890	)
891	assert len(raw_default.get("commits", [])) == data["commits"]
892
893	def test_explicit_commit_id(self, tmp_path: pathlib.Path) -> None:
894	"""A raw commit ID passed as ref is resolved correctly."""
895	_init_repo(tmp_path)
896	cid = _make_commit(tmp_path, content=b"ext-cid")
897	out = tmp_path / "cid.bundle"
898	result = _invoke(
899	["bundle", "create", str(out), cid, "--json"],
900	env=_env(tmp_path),
901	)
902	assert result.exit_code == 0
903	data = _parse_create(result)
904	assert data["commits"] >= 1
905
906	def test_output_is_valid_msgpack(self, tmp_path: pathlib.Path) -> None:
907	"""The output file must be valid msgpack."""
908	import msgpack as _mp
909
910	_init_repo(tmp_path)
911	_make_commit(tmp_path, content=b"ext-msgpack")
912	out = tmp_path / "mp.bundle"
913	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
914	raw = _mp.unpackb(out.read_bytes(), raw=False)
915	assert isinstance(raw, dict)
916	assert "commits" in raw
917
918	def test_help_mentions_agent_quickstart(self) -> None:
919	result = _invoke(["bundle", "create", "--help"])
920	assert result.exit_code == 0
921	assert "Agent quickstart" in result.output
922
923	def test_help_mentions_exit_codes(self) -> None:
924	result = _invoke(["bundle", "create", "--help"])
925	assert result.exit_code == 0
926	assert "Exit codes" in result.output
927
928	def test_help_mentions_json_schema(self) -> None:
929	result = _invoke(["bundle", "create", "--help"])
930	assert result.exit_code == 0
931	assert "JSON output schema" in result.output
932
933	def test_multiple_have_ids_reduce_bundle(self, tmp_path: pathlib.Path) -> None:
934	"""Multiple --have IDs each reduce what is bundled."""
935	_init_repo(tmp_path)
936	c1 = _make_commit(tmp_path, content=b"ext-have-1")
937	c2 = _make_commit(tmp_path, parent_id=c1, content=b"ext-have-2")
938	_make_commit(tmp_path, parent_id=c2, content=b"ext-have-3")
939
940	out_full = tmp_path / "have-full.bundle"
941	out_pruned = tmp_path / "have-pruned.bundle"
942	r_full = _invoke(
943	["bundle", "create", str(out_full), "--json"], env=_env(tmp_path)
944	)
945	r_pruned = _invoke(
946	["bundle", "create", str(out_pruned), "--have", c1, c2, "--json"],
947	env=_env(tmp_path),
948	)
949	assert r_full.exit_code == 0
950	assert r_pruned.exit_code == 0
951	full_data = _parse_create(r_full)
952	pruned_data = _parse_create(r_pruned)
953	assert pruned_data["commits"] < full_data["commits"]
954
955
956	# ===========================================================================
957	# TestBundleCreateSecurity — 6 tests
958	# ===========================================================================
959
960
961	class TestBundleCreateSecurity:
962	def test_ansi_in_file_path_stripped_text_output(
963	self, tmp_path: pathlib.Path
964	) -> None:
965	"""ANSI escape in the output file path must be stripped in text output."""
966	_init_repo(tmp_path)
967	_make_commit(tmp_path, content=b"sec-ansi-path")
968	# Build an output path whose filename component contains an ANSI escape.
969	out = tmp_path / "\x1b[31mevil\x1b[0m.bundle"
970	result = _invoke(["bundle", "create", str(out)], env=_env(tmp_path))
971	assert result.exit_code == 0
972	assert "\x1b" not in result.output
973
974	def test_control_char_in_file_path_stripped(
975	self, tmp_path: pathlib.Path
976	) -> None:
977	"""Control characters in the output file path must not reach stdout."""
978	_init_repo(tmp_path)
979	_make_commit(tmp_path, content=b"sec-ctrl-path")
980	out = tmp_path / "foo\x07bar.bundle"
981	result = _invoke(["bundle", "create", str(out)], env=_env(tmp_path))
982	assert result.exit_code == 0
983	assert "\x07" not in result.output
984
985	def test_outside_repo_exits_2(self, tmp_path: pathlib.Path) -> None:
986	"""Without a .muse directory, create must exit 2 (REPO_NOT_FOUND)."""
987	out = tmp_path / "norepo.bundle"
988	result = _invoke(["bundle", "create", str(out)], env=_env(tmp_path))
989	assert result.exit_code == 2
990
991	def test_bad_ref_ansi_stripped_from_error(
992	self, tmp_path: pathlib.Path
993	) -> None:
994	"""ANSI in an unknown ref name must not appear in the error output."""
995	_init_repo(tmp_path)
996	_make_commit(tmp_path, content=b"sec-ref-ansi")
997	out = tmp_path / "ref-ansi.bundle"
998	evil_ref = "\x1b[31mbadref\x1b[0m"
999	result = _invoke(
1000	["bundle", "create", str(out), evil_ref], env=_env(tmp_path)
1001	)
1002	assert result.exit_code == 1
1003	assert "\x1b" not in result.output
1004
1005	def test_ansi_in_have_no_injection(self, tmp_path: pathlib.Path) -> None:
1006	"""ANSI characters in a --have value must not appear in any output."""
1007	_init_repo(tmp_path)
1008	_make_commit(tmp_path, content=b"sec-have-ansi")
1009	out = tmp_path / "have-ansi.bundle"
1010	evil_have = "\x1b[31m" + "a" * 64 + "\x1b[0m"
1011	result = _invoke(
1012	["bundle", "create", str(out), "--have", evil_have],
1013	env=_env(tmp_path),
1014	)
1015	# The have ID won't match anything — bundle succeeds with full history.
1016	assert "\x1b" not in result.output
1017
1018	def test_no_json_on_error(self, tmp_path: pathlib.Path) -> None:
1019	"""On error (no commits), stdout must not contain JSON."""
1020	_init_repo(tmp_path)
1021	out = tmp_path / "err-json.bundle"
1022	result = _invoke(
1023	["bundle", "create", str(out), "--json"], env=_env(tmp_path)
1024	)
1025	assert result.exit_code != 0
1026	assert not result.output.strip().startswith("{")
1027
1028
1029	# ===========================================================================
1030	# TestBundleCreateStress — 3 tests
1031	# ===========================================================================
1032
1033
1034	class TestBundleCreateStress:
1035	def test_50_commit_chain(self, tmp_path: pathlib.Path) -> None:
1036	"""50-commit linear chain is bundled correctly."""
1037	_init_repo(tmp_path)
1038	prev: str \| None = None
1039	for i in range(50):
1040	prev = _make_commit(
1041	tmp_path, parent_id=prev, content=f"stress50-{i}".encode()
1042	)
1043	out = tmp_path / "stress50.bundle"
1044	result = _invoke(
1045	["bundle", "create", str(out), "--json"], env=_env(tmp_path)
1046	)
1047	assert result.exit_code == 0
1048	data = _parse_create(result)
1049	assert data["commits"] == 50
1050	assert data["size_bytes"] > 0
1051
1052	def test_create_with_large_have_list(self, tmp_path: pathlib.Path) -> None:
1053	"""Passing 15 --have IDs on a 20-commit chain produces a smaller bundle."""
1054	_init_repo(tmp_path)
1055	ids: list[str] = []
1056	prev: str \| None = None
1057	for i in range(20):
1058	prev = _make_commit(
1059	tmp_path, parent_id=prev, content=f"have-list-{i}".encode()
1060	)
1061	ids.append(prev)
1062
1063	out_full = tmp_path / "have-full-20.bundle"
1064	out_pruned = tmp_path / "have-pruned-20.bundle"
1065	r_full = _invoke(
1066	["bundle", "create", str(out_full), "--json"], env=_env(tmp_path)
1067	)
1068	# Pass the first 15 as --have to exclude them.
1069	have_args = ["--have"] + ids[:15]
1070	r_pruned = _invoke(
1071	["bundle", "create", str(out_pruned)] + have_args + ["--json"],
1072	env=_env(tmp_path),
1073	)
1074	assert r_full.exit_code == 0
1075	assert r_pruned.exit_code == 0
1076	full_data = _parse_create(r_full)
1077	pruned_data = _parse_create(r_pruned)
1078	assert pruned_data["commits"] < full_data["commits"]
1079
1080	def test_many_branches(self, tmp_path: pathlib.Path) -> None:
1081	"""10 branches pointing to reachable commits all appear in the bundle."""
1082	_init_repo(tmp_path)
1083	# Build a 10-commit chain on main, then create a feature branch ref
1084	# pointing to each commit — all are reachable from HEAD.
1085	prev: str \| None = None
1086	commit_ids: list[str] = []
1087	for i in range(10):
1088	prev = _make_commit(
1089	tmp_path, parent_id=prev, content=f"stress-br-{i}".encode()
1090	)
1091	commit_ids.append(prev)
1092	branch_names = [f"feat/stress-br-{i}" for i in range(10)]
1093	for br, cid in zip(branch_names, commit_ids):
1094	ref_file = tmp_path / ".muse" / "refs" / "heads" / br
1095	ref_file.parent.mkdir(parents=True, exist_ok=True)
1096	ref_file.write_text(cid, encoding="utf-8")
1097	out = tmp_path / "many-branches.bundle"
1098	result = _invoke(
1099	["bundle", "create", str(out), "--json"], env=_env(tmp_path)
1100	)
1101	assert result.exit_code == 0
1102	data = _parse_create(result)
1103	for br in branch_names:
1104	assert br in data["branches"]
1105
1106
1107	# ===========================================================================
1108	# TestBundleUnbundleExtended — 18 tests
1109	# ===========================================================================
1110
1111
1112	def _make_bundle(src: pathlib.Path, dst_file: pathlib.Path) -> None:
1113	"""Helper: create a bundle from src repo into dst_file."""
1114	_invoke(["bundle", "create", str(dst_file)], env=_env(src))
1115
1116
1117	class TestBundleUnbundleExtended:
1118	def _src_dst(self, tmp_path: pathlib.Path) -> tuple[pathlib.Path, pathlib.Path]:
1119	src = tmp_path / "src"
1120	dst = tmp_path / "dst"
1121	src.mkdir()
1122	dst.mkdir()
1123	_init_repo(src)
1124	_init_repo(dst, repo_id="ub-dst")
1125	return src, dst
1126
1127	def test_exits_0_basic(self, tmp_path: pathlib.Path) -> None:
1128	src, dst = self._src_dst(tmp_path)
1129	_make_commit(src, content=b"ub-basic")
1130	bundle = tmp_path / "basic.bundle"
1131	_make_bundle(src, bundle)
1132	result = _invoke(["bundle", "unbundle", str(bundle)], env=_env(dst))
1133	assert result.exit_code == 0
1134
1135	def test_commits_written_count(self, tmp_path: pathlib.Path) -> None:
1136	src, dst = self._src_dst(tmp_path)
1137	prev: str \| None = None
1138	for i in range(3):
1139	prev = _make_commit(src, parent_id=prev, content=f"ub-cnt-{i}".encode())
1140	bundle = tmp_path / "cnt.bundle"
1141	_make_bundle(src, bundle)
1142	result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst))
1143	assert result.exit_code == 0
1144	data = _parse_unbundle(result)
1145	assert data["commits_written"] == 3
1146
1147	def test_snapshots_written_count(self, tmp_path: pathlib.Path) -> None:
1148	src, dst = self._src_dst(tmp_path)
1149	_make_commit(src, content=b"ub-snap")
1150	bundle = tmp_path / "snap.bundle"
1151	_make_bundle(src, bundle)
1152	result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst))
1153	assert result.exit_code == 0
1154	data = _parse_unbundle(result)
1155	assert data["snapshots_written"] >= 1
1156
1157	def test_objects_written_count(self, tmp_path: pathlib.Path) -> None:
1158	src, dst = self._src_dst(tmp_path)
1159	_make_commit(src, content=b"ub-obj")
1160	bundle = tmp_path / "obj.bundle"
1161	_make_bundle(src, bundle)
1162	result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst))
1163	assert result.exit_code == 0
1164	data = _parse_unbundle(result)
1165	assert data["objects_written"] >= 1
1166
1167	def test_objects_skipped_idempotent(self, tmp_path: pathlib.Path) -> None:
1168	"""Unbundling twice: second pass skips all already-present objects."""
1169	src, dst = self._src_dst(tmp_path)
1170	_make_commit(src, content=b"ub-idem")
1171	bundle = tmp_path / "idem.bundle"
1172	_make_bundle(src, bundle)
1173	_invoke(["bundle", "unbundle", str(bundle)], env=_env(dst))
1174	result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst))
1175	assert result.exit_code == 0
1176	data = _parse_unbundle(result)
1177	assert data["commits_written"] == 0
1178	assert data["objects_written"] == 0
1179	assert data["objects_skipped"] >= 1
1180
1181	def test_text_output_mentions_commits(self, tmp_path: pathlib.Path) -> None:
1182	src, dst = self._src_dst(tmp_path)
1183	_make_commit(src, content=b"ub-txt-c")
1184	bundle = tmp_path / "txt-c.bundle"
1185	_make_bundle(src, bundle)
1186	result = _invoke(["bundle", "unbundle", str(bundle)], env=_env(dst))
1187	assert result.exit_code == 0
1188	assert "commit(s)" in result.output
1189
1190	def test_text_output_mentions_applied(self, tmp_path: pathlib.Path) -> None:
1191	src, dst = self._src_dst(tmp_path)
1192	_make_commit(src, content=b"ub-txt-a")
1193	bundle = tmp_path / "txt-a.bundle"
1194	_make_bundle(src, bundle)
1195	result = _invoke(["bundle", "unbundle", str(bundle)], env=_env(dst))
1196	assert result.exit_code == 0
1197	assert "Bundle applied" in result.output
1198
1199	def test_refs_updated_by_default(self, tmp_path: pathlib.Path) -> None:
1200	"""By default, branch refs in the destination are updated."""
1201	src, dst = self._src_dst(tmp_path)
1202	_make_commit(src, content=b"ub-ref-up")
1203	bundle = tmp_path / "ref-up.bundle"
1204	_make_bundle(src, bundle)
1205	result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst))
1206	assert result.exit_code == 0
1207	data = _parse_unbundle(result)
1208	assert "main" in data["refs_updated"]
1209
1210	def test_no_update_refs_skips_refs(self, tmp_path: pathlib.Path) -> None:
1211	src, dst = self._src_dst(tmp_path)
1212	_make_commit(src, content=b"ub-no-ref")
1213	bundle = tmp_path / "no-ref.bundle"
1214	_make_bundle(src, bundle)
1215	result = _invoke(
1216	["bundle", "unbundle", str(bundle), "--no-update-refs", "--json"],
1217	env=_env(dst),
1218	)
1219	assert result.exit_code == 0
1220	data = _parse_unbundle(result)
1221	assert data["refs_updated"] == []
1222
1223	def test_refs_updated_branch_file_exists(self, tmp_path: pathlib.Path) -> None:
1224	"""After unbundle, the branch ref file must exist in the destination."""
1225	src, dst = self._src_dst(tmp_path)
1226	_make_commit(src, content=b"ub-ref-file")
1227	bundle = tmp_path / "ref-file.bundle"
1228	_make_bundle(src, bundle)
1229	_invoke(["bundle", "unbundle", str(bundle)], env=_env(dst))
1230	ref_file = dst / ".muse" / "refs" / "heads" / "main"
1231	assert ref_file.exists()
1232	cid = ref_file.read_text(encoding="utf-8").strip()
1233	# Ref files store canonical "sha256:<64hex>" format (71 chars).
1234	assert cid.startswith("sha256:")
1235	assert len(cid) == 71
1236
1237	def test_j_alias(self, tmp_path: pathlib.Path) -> None:
1238	"""-j must produce identical JSON to --json."""
1239	src1, dst1 = self._src_dst(tmp_path)
1240	src2 = tmp_path / "src2"
1241	dst2 = tmp_path / "dst2"
1242	src2.mkdir()
1243	dst2.mkdir()
1244	_init_repo(src2)
1245	_init_repo(dst2, repo_id="ub-j2")
1246
1247	_make_commit(src1, content=b"ub-j-a1")
1248	_make_commit(src2, content=b"ub-j-a2")
1249	b1 = tmp_path / "j1.bundle"
1250	b2 = tmp_path / "j2.bundle"
1251	_make_bundle(src1, b1)
1252	_make_bundle(src2, b2)
1253
1254	r1 = _invoke(["bundle", "unbundle", str(b1), "--json"], env=_env(dst1))
1255	r2 = _invoke(["bundle", "unbundle", str(b2), "-j"], env=_env(dst2))
1256	assert r1.exit_code == 0
1257	assert r2.exit_code == 0
1258	d1 = _parse_unbundle(r1)
1259	d2 = _parse_unbundle(r2)
1260	assert set(d1.keys()) == set(d2.keys())
1261	assert d1["commits_written"] == d2["commits_written"]
1262
1263	def test_json_refs_updated_sorted(self, tmp_path: pathlib.Path) -> None:
1264	"""refs_updated in JSON output must be sorted."""
1265	src, dst = self._src_dst(tmp_path)
1266	c1 = _make_commit(src, content=b"ub-sort-base")
1267	# Add extra branch refs pointing at c1 so the bundle has multiple heads.
1268	for br in ("z-br", "a-br"):
1269	ref = src / ".muse" / "refs" / "heads" / br
1270	ref.write_text(c1, encoding="utf-8")
1271	bundle = tmp_path / "sort.bundle"
1272	_make_bundle(src, bundle)
1273	result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst))
1274	assert result.exit_code == 0
1275	data = _parse_unbundle(result)
1276	assert data["refs_updated"] == sorted(data["refs_updated"])
1277
1278	def test_empty_bundle_no_crash(self, tmp_path: pathlib.Path) -> None:
1279	"""An empty dict bundle (no commits/objects) must exit 0 cleanly."""
1280	_init_repo(tmp_path)
1281	empty_bundle = tmp_path / "empty.bundle"
1282	empty_bundle.write_bytes(msgpack.packb({}, use_bin_type=True))
1283	result = _invoke(["bundle", "unbundle", str(empty_bundle)], env=_env(tmp_path))
1284	assert result.exit_code == 0
1285
1286	def test_bundle_without_branch_heads_no_refs(self, tmp_path: pathlib.Path) -> None:
1287	"""A bundle missing the branch_heads key → refs_updated must be empty."""
1288	src, dst = self._src_dst(tmp_path)
1289	_make_commit(src, content=b"ub-no-heads")
1290	bundle = tmp_path / "no-heads.bundle"
1291	_make_bundle(src, bundle)
1292	# Strip branch_heads from the bundle.
1293	raw = msgpack.unpackb(bundle.read_bytes(), raw=False)
1294	raw.pop("branch_heads", None)
1295	bundle.write_bytes(msgpack.packb(raw, use_bin_type=True))
1296	result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst))
1297	assert result.exit_code == 0
1298	data = _parse_unbundle(result)
1299	assert data["refs_updated"] == []
1300
1301	def test_help_mentions_agent_quickstart(self) -> None:
1302	result = _invoke(["bundle", "unbundle", "--help"])
1303	assert result.exit_code == 0
1304	assert "Agent quickstart" in result.output
1305
1306	def test_help_mentions_exit_codes(self) -> None:
1307	result = _invoke(["bundle", "unbundle", "--help"])
1308	assert result.exit_code == 0
1309	assert "Exit codes" in result.output
1310
1311	def test_help_mentions_json_schema(self) -> None:
1312	result = _invoke(["bundle", "unbundle", "--help"])
1313	assert result.exit_code == 0
1314	assert "JSON output schema" in result.output
1315
1316	def test_no_update_refs_flag_in_help(self) -> None:
1317	result = _invoke(["bundle", "unbundle", "--help"])
1318	assert result.exit_code == 0
1319	assert "--no-update-refs" in result.output
1320
1321
1322	# ===========================================================================
1323	# TestBundleUnbundleSecurity — 6 tests
1324	# ===========================================================================
1325
1326
1327	class TestBundleUnbundleSecurity:
1328	def test_outside_repo_exits_2(self, tmp_path: pathlib.Path) -> None:
1329	"""Without a .muse directory, unbundle must exit 2 (REPO_NOT_FOUND)."""
1330	bundle = tmp_path / "norepo.bundle"
1331	bundle.write_bytes(msgpack.packb({}, use_bin_type=True))
1332	result = _invoke(["bundle", "unbundle", str(bundle)], env=_env(tmp_path))
1333	assert result.exit_code == 2
1334
1335	def test_missing_bundle_file_exits_1(self, tmp_path: pathlib.Path) -> None:
1336	_init_repo(tmp_path)
1337	result = _invoke(
1338	["bundle", "unbundle", str(tmp_path / "missing.bundle")],
1339	env=_env(tmp_path),
1340	)
1341	assert result.exit_code == 1
1342
1343	def test_invalid_msgpack_exits_1(self, tmp_path: pathlib.Path) -> None:
1344	_init_repo(tmp_path)
1345	corrupt = tmp_path / "corrupt.bundle"
1346	corrupt.write_bytes(b"\xff\xfe not msgpack at all")
1347	result = _invoke(["bundle", "unbundle", str(corrupt)], env=_env(tmp_path))
1348	assert result.exit_code == 1
1349
1350	def test_ansi_branch_name_skipped_no_injection(
1351	self, tmp_path: pathlib.Path
1352	) -> None:
1353	"""ANSI escape in a bundle branch name is skipped; no escape in output."""
1354	src = tmp_path / "src"
1355	dst = tmp_path / "dst"
1356	src.mkdir()
1357	dst.mkdir()
1358	_init_repo(src)
1359	_init_repo(dst, repo_id="sec-ansi-br")
1360	_make_commit(src, content=b"sec-ansi-br")
1361	bundle = tmp_path / "ansi-br.bundle"
1362	_make_bundle(src, bundle)
1363	# Inject an ANSI-poisoned branch name into branch_heads.
1364	raw = msgpack.unpackb(bundle.read_bytes(), raw=False)
1365	raw["branch_heads"] = {"\x1b[31mevil\x1b[0m": "a" * 64}
1366	bundle.write_bytes(msgpack.packb(raw, use_bin_type=True))
1367	result = _invoke(["bundle", "unbundle", str(bundle)], env=_env(dst))
1368	assert result.exit_code == 0
1369	assert "\x1b" not in result.output
1370
1371	def test_invalid_commit_id_branch_ref_skipped(
1372	self, tmp_path: pathlib.Path
1373	) -> None:
1374	"""A commit ID shorter than 64 chars in branch_heads must be skipped."""
1375	src = tmp_path / "src"
1376	dst = tmp_path / "dst"
1377	src.mkdir()
1378	dst.mkdir()
1379	_init_repo(src)
1380	_init_repo(dst, repo_id="sec-short-cid")
1381	_make_commit(src, content=b"sec-short-cid")
1382	bundle = tmp_path / "short-cid.bundle"
1383	_make_bundle(src, bundle)
1384	raw = msgpack.unpackb(bundle.read_bytes(), raw=False)
1385	# Replace the commit IDs with a too-short value.
1386	raw["branch_heads"] = {"main": "tooshort"}
1387	bundle.write_bytes(msgpack.packb(raw, use_bin_type=True))
1388	result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst))
1389	assert result.exit_code == 0
1390	data = _parse_unbundle(result)
1391	assert "main" not in data["refs_updated"]
1392
1393	def test_no_json_on_missing_file(self, tmp_path: pathlib.Path) -> None:
1394	"""Error path (file not found) must not emit JSON to stdout."""
1395	_init_repo(tmp_path)
1396	result = _invoke(
1397	["bundle", "unbundle", str(tmp_path / "ghost.bundle"), "--json"],
1398	env=_env(tmp_path),
1399	)
1400	assert result.exit_code != 0
1401	assert not result.output.strip().startswith("{")
1402
1403
1404	# ===========================================================================
1405	# TestBundleUnbundleStress — 3 tests
1406	# ===========================================================================
1407
1408
1409	class TestBundleUnbundleStress:
1410	def test_50_commit_chain(self, tmp_path: pathlib.Path) -> None:
1411	"""50-commit chain is fully unpacked into the destination."""
1412	src = tmp_path / "src"
1413	dst = tmp_path / "dst"
1414	src.mkdir()
1415	dst.mkdir()
1416	_init_repo(src)
1417	_init_repo(dst, repo_id="stress-ub-dst")
1418	prev: str \| None = None
1419	for i in range(50):
1420	prev = _make_commit(src, parent_id=prev, content=f"ub50-{i}".encode())
1421	bundle = tmp_path / "ub50.bundle"
1422	_make_bundle(src, bundle)
1423	result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst))
1424	assert result.exit_code == 0
1425	data = _parse_unbundle(result)
1426	assert data["commits_written"] == 50
1427	assert data["objects_written"] >= 50
1428
1429	def test_idempotent_multiple_applications(self, tmp_path: pathlib.Path) -> None:
1430	"""Applying the same bundle 5 times: only the first writes anything."""
1431	src = tmp_path / "src"
1432	dst = tmp_path / "dst"
1433	src.mkdir()
1434	dst.mkdir()
1435	_init_repo(src)
1436	_init_repo(dst, repo_id="stress-idem-dst")
1437	prev: str \| None = None
1438	for i in range(5):
1439	prev = _make_commit(src, parent_id=prev, content=f"idem-{i}".encode())
1440	bundle = tmp_path / "idem5.bundle"
1441	_make_bundle(src, bundle)
1442	first = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst))
1443	assert first.exit_code == 0
1444	first_data = _parse_unbundle(first)
1445	assert first_data["commits_written"] == 5
1446	for _ in range(4):
1447	repeat = _invoke(
1448	["bundle", "unbundle", str(bundle), "--json"], env=_env(dst)
1449	)
1450	assert repeat.exit_code == 0
1451	repeat_data = _parse_unbundle(repeat)
1452	assert repeat_data["commits_written"] == 0
1453	assert repeat_data["objects_written"] == 0
1454
1455	def test_many_branch_refs_updated(self, tmp_path: pathlib.Path) -> None:
1456	"""10 branch heads in the bundle → all 10 appear in refs_updated."""
1457	src = tmp_path / "src"
1458	dst = tmp_path / "dst"
1459	src.mkdir()
1460	dst.mkdir()
1461	_init_repo(src)
1462	_init_repo(dst, repo_id="stress-refs-dst")
1463	# Build a 10-commit chain on main.
1464	prev: str \| None = None
1465	cids: list[str] = []
1466	for i in range(10):
1467	prev = _make_commit(src, parent_id=prev, content=f"br-ref-{i}".encode())
1468	cids.append(prev)
1469	# Create 10 feature branch refs pointing to reachable commits.
1470	br_names = [f"feat/br-{i}" for i in range(10)]
1471	for br, cid in zip(br_names, cids):
1472	ref = src / ".muse" / "refs" / "heads" / br
1473	ref.parent.mkdir(parents=True, exist_ok=True)
1474	ref.write_text(cid, encoding="utf-8")
1475	bundle = tmp_path / "many-refs.bundle"
1476	_make_bundle(src, bundle)
1477	result = _invoke(["bundle", "unbundle", str(bundle), "--json"], env=_env(dst))
1478	assert result.exit_code == 0
1479	data = _parse_unbundle(result)
1480	for br in br_names:
1481	assert br in data["refs_updated"]
1482
1483
1484	# ===========================================================================
1485	# TestBundleVerifyExtended — 18 tests
1486	# ===========================================================================
1487
1488
1489	class TestBundleVerifyExtended:
1490	def _clean_bundle(self, tmp_path: pathlib.Path) -> pathlib.Path:
1491	"""Create a repo with one commit and return a clean bundle path."""
1492	_init_repo(tmp_path)
1493	_make_commit(tmp_path, content=b"vext-clean")
1494	out = tmp_path / "clean.bundle"
1495	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
1496	return out
1497
1498	def _corrupt_bundle(self, tmp_path: pathlib.Path) -> pathlib.Path:
1499	"""Create a bundle then tamper one object's content."""
1500	_init_repo(tmp_path)
1501	_make_commit(tmp_path, content=b"vext-corrupt")
1502	out = tmp_path / "corrupt.bundle"
1503	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
1504	raw = msgpack.unpackb(out.read_bytes(), raw=False)
1505	if raw.get("objects"):
1506	raw["objects"][0]["content"] = b"TAMPERED"
1507	out.write_bytes(msgpack.packb(raw, use_bin_type=True))
1508	return out
1509
1510	def test_exits_0_on_clean_bundle(self, tmp_path: pathlib.Path) -> None:
1511	bundle = self._clean_bundle(tmp_path)
1512	result = _invoke(["bundle", "verify", str(bundle)], env=_env(tmp_path))
1513	assert result.exit_code == 0
1514
1515	def test_exits_1_on_corrupt_object(self, tmp_path: pathlib.Path) -> None:
1516	bundle = self._corrupt_bundle(tmp_path)
1517	result = _invoke(["bundle", "verify", str(bundle)], env=_env(tmp_path))
1518	assert result.exit_code == 1
1519
1520	def test_all_ok_true_on_clean(self, tmp_path: pathlib.Path) -> None:
1521	bundle = self._clean_bundle(tmp_path)
1522	result = _invoke(["bundle", "verify", str(bundle), "--json"], env=_env(tmp_path))
1523	assert result.exit_code == 0
1524	data = _parse_verify(result)
1525	assert data["all_ok"] is True
1526
1527	def test_all_ok_false_on_corrupt(self, tmp_path: pathlib.Path) -> None:
1528	bundle = self._corrupt_bundle(tmp_path)
1529	result = _invoke(["bundle", "verify", str(bundle), "--json"], env=_env(tmp_path))
1530	assert result.exit_code == 1
1531	data = _parse_verify(result)
1532	assert data["all_ok"] is False
1533
1534	def test_objects_checked_count(self, tmp_path: pathlib.Path) -> None:
1535	"""objects_checked must equal the number of objects in the bundle."""
1536	_init_repo(tmp_path)
1537	_make_commit(tmp_path, content=b"vext-cnt")
1538	out = tmp_path / "cnt.bundle"
1539	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
1540	raw = msgpack.unpackb(out.read_bytes(), raw=False)
1541	n_objects = len(raw.get("objects", []))
1542	result = _invoke(["bundle", "verify", str(out), "--json"], env=_env(tmp_path))
1543	assert result.exit_code == 0
1544	data = _parse_verify(result)
1545	assert data["objects_checked"] == n_objects
1546
1547	def test_snapshots_checked_count(self, tmp_path: pathlib.Path) -> None:
1548	bundle = self._clean_bundle(tmp_path)
1549	result = _invoke(["bundle", "verify", str(bundle), "--json"], env=_env(tmp_path))
1550	assert result.exit_code == 0
1551	data = _parse_verify(result)
1552	assert data["snapshots_checked"] >= 1
1553
1554	def test_failures_empty_on_clean(self, tmp_path: pathlib.Path) -> None:
1555	bundle = self._clean_bundle(tmp_path)
1556	result = _invoke(["bundle", "verify", str(bundle), "--json"], env=_env(tmp_path))
1557	data = _parse_verify(result)
1558	assert data["failures"] == []
1559
1560	def test_failures_nonempty_on_corrupt(self, tmp_path: pathlib.Path) -> None:
1561	bundle = self._corrupt_bundle(tmp_path)
1562	result = _invoke(["bundle", "verify", str(bundle), "--json"], env=_env(tmp_path))
1563	data = _parse_verify(result)
1564	assert len(data["failures"]) >= 1
1565
1566	def test_quiet_clean_exits_0_no_output(self, tmp_path: pathlib.Path) -> None:
1567	bundle = self._clean_bundle(tmp_path)
1568	result = _invoke(["bundle", "verify", str(bundle), "--quiet"], env=_env(tmp_path))
1569	assert result.exit_code == 0
1570	assert result.output.strip() == ""
1571
1572	def test_quiet_corrupt_exits_1_no_output(self, tmp_path: pathlib.Path) -> None:
1573	bundle = self._corrupt_bundle(tmp_path)
1574	result = _invoke(["bundle", "verify", str(bundle), "-q"], env=_env(tmp_path))
1575	assert result.exit_code == 1
1576	assert result.output.strip() == ""
1577
1578	def test_json_output_is_single_line(self, tmp_path: pathlib.Path) -> None:
1579	"""JSON output must be compact (no indent=2), matching all other commands."""
1580	bundle = self._clean_bundle(tmp_path)
1581	result = _invoke(["bundle", "verify", str(bundle), "--json"], env=_env(tmp_path))
1582	assert result.exit_code == 0
1583	# Compact JSON has no interior newlines.
1584	assert "\n" not in result.output.strip()
1585
1586	def test_j_alias(self, tmp_path: pathlib.Path) -> None:
1587	bundle = self._clean_bundle(tmp_path)
1588	r1 = _invoke(["bundle", "verify", str(bundle), "--json"], env=_env(tmp_path))
1589	r2 = _invoke(["bundle", "verify", str(bundle), "-j"], env=_env(tmp_path))
1590	assert r1.exit_code == 0
1591	assert r2.exit_code == 0
1592	_volatile = {"timestamp", "duration_ms"}
1593	d1 = {k: v for k, v in json.loads(r1.output).items() if k not in _volatile}
1594	d2 = {k: v for k, v in json.loads(r2.output).items() if k not in _volatile}
1595	assert d1 == d2
1596
1597	def test_text_output_mentions_objects_checked(self, tmp_path: pathlib.Path) -> None:
1598	bundle = self._clean_bundle(tmp_path)
1599	result = _invoke(["bundle", "verify", str(bundle)], env=_env(tmp_path))
1600	assert "Objects checked" in result.output
1601
1602	def test_text_output_mentions_snapshots_checked(self, tmp_path: pathlib.Path) -> None:
1603	bundle = self._clean_bundle(tmp_path)
1604	result = _invoke(["bundle", "verify", str(bundle)], env=_env(tmp_path))
1605	assert "Snapshots checked" in result.output
1606
1607	def test_text_output_clean_checkmark(self, tmp_path: pathlib.Path) -> None:
1608	bundle = self._clean_bundle(tmp_path)
1609	result = _invoke(["bundle", "verify", str(bundle)], env=_env(tmp_path))
1610	assert "Bundle is clean" in result.output
1611
1612	def test_text_output_failures_listed(self, tmp_path: pathlib.Path) -> None:
1613	bundle = self._corrupt_bundle(tmp_path)
1614	result = _invoke(["bundle", "verify", str(bundle)], env=_env(tmp_path))
1615	assert result.exit_code == 1
1616	assert "hash mismatch" in result.output
1617
1618	def test_help_mentions_agent_quickstart(self) -> None:
1619	result = _invoke(["bundle", "verify", "--help"])
1620	assert result.exit_code == 0
1621	assert "Agent quickstart" in result.output
1622
1623	def test_help_mentions_exit_codes(self) -> None:
1624	result = _invoke(["bundle", "verify", "--help"])
1625	assert result.exit_code == 0
1626	assert "Exit codes" in result.output
1627
1628
1629	# ===========================================================================
1630	# TestBundleVerifySecurity — 6 tests
1631	# ===========================================================================
1632
1633
1634	class TestBundleVerifySecurity:
1635	def _bundle_with_ansi_object_id(self, tmp_path: pathlib.Path) -> pathlib.Path:
1636	"""Bundle where an object_id contains an ANSI escape sequence."""
1637	_init_repo(tmp_path)
1638	_make_commit(tmp_path, content=b"sec-ansi-oid")
1639	out = tmp_path / "ansi-oid.bundle"
1640	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
1641	raw = msgpack.unpackb(out.read_bytes(), raw=False)
1642	if raw.get("objects"):
1643	# Inject ANSI into the object_id — will trigger hash mismatch failure.
1644	raw["objects"][0]["object_id"] = "\x1b[31mevil_oid_xxx\x1b[0m"
1645	out.write_bytes(msgpack.packb(raw, use_bin_type=True))
1646	return out
1647
1648	def _bundle_with_ansi_rel_path(self, tmp_path: pathlib.Path) -> pathlib.Path:
1649	"""Bundle where a snapshot manifest key contains an ANSI escape."""
1650	_init_repo(tmp_path)
1651	_make_commit(tmp_path, content=b"sec-ansi-path")
1652	out = tmp_path / "ansi-path.bundle"
1653	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
1654	raw = msgpack.unpackb(out.read_bytes(), raw=False)
1655	if raw.get("snapshots"):
1656	snap = raw["snapshots"][0]
1657	# Replace manifest keys with ANSI-poisoned path.
1658	old_manifest = snap.get("manifest", {})
1659	snap["manifest"] = {
1660	"\x1b[31mevil/path\x1b[0m": v for v in old_manifest.values()
1661	}
1662	out.write_bytes(msgpack.packb(raw, use_bin_type=True))
1663	return out
1664
1665	def test_ansi_in_object_id_failure_stripped_text(
1666	self, tmp_path: pathlib.Path
1667	) -> None:
1668	"""ANSI in object_id within a failure message must be stripped in text output."""
1669	bundle = self._bundle_with_ansi_object_id(tmp_path)
1670	result = _invoke(["bundle", "verify", str(bundle)], env=_env(tmp_path))
1671	assert "\x1b" not in result.output
1672
1673	def test_ansi_in_rel_path_failure_stripped_text(
1674	self, tmp_path: pathlib.Path
1675	) -> None:
1676	"""ANSI in a manifest rel_path within a failure must be stripped in text output."""
1677	bundle = self._bundle_with_ansi_rel_path(tmp_path)
1678	result = _invoke(["bundle", "verify", str(bundle)], env=_env(tmp_path))
1679	assert "\x1b" not in result.output
1680
1681	def test_ansi_in_failures_sanitized_json(self, tmp_path: pathlib.Path) -> None:
1682	"""failures list in JSON output must not contain raw ANSI escapes."""
1683	bundle = self._bundle_with_ansi_object_id(tmp_path)
1684	result = _invoke(["bundle", "verify", str(bundle), "--json"], env=_env(tmp_path))
1685	assert "\x1b" not in result.output
1686
1687	def test_no_repo_required(self, tmp_path: pathlib.Path) -> None:
1688	"""verify must work outside any .muse repository (no require_repo call)."""
1689	work = tmp_path / "no_repo"
1690	work.mkdir()
1691	_init_repo(tmp_path)
1692	_make_commit(tmp_path, content=b"sec-no-repo")
1693	bundle = tmp_path / "no-repo.bundle"
1694	_invoke(["bundle", "create", str(bundle)], env=_env(tmp_path))
1695	# Run verify from a directory with no .muse — must NOT exit 2.
1696	result = _invoke(["bundle", "verify", str(bundle)], env={"MUSE_REPO_ROOT": str(work)})
1697	assert result.exit_code != 2
1698
1699	def test_missing_file_exits_1(self, tmp_path: pathlib.Path) -> None:
1700	_init_repo(tmp_path)
1701	result = _invoke(
1702	["bundle", "verify", str(tmp_path / "ghost.bundle")],
1703	env=_env(tmp_path),
1704	)
1705	assert result.exit_code == 1
1706
1707	def test_invalid_msgpack_exits_1(self, tmp_path: pathlib.Path) -> None:
1708	_init_repo(tmp_path)
1709	corrupt = tmp_path / "bad.bundle"
1710	corrupt.write_bytes(b"\xff\xfe not msgpack")
1711	result = _invoke(["bundle", "verify", str(corrupt)], env=_env(tmp_path))
1712	assert result.exit_code == 1
1713
1714
1715	# ===========================================================================
1716	# TestBundleVerifyStress — 3 tests
1717	# ===========================================================================
1718
1719
1720	class TestBundleVerifyStress:
1721	def test_200_commit_bundle_verify_clean(self, tmp_path: pathlib.Path) -> None:
1722	"""200-commit bundle verifies clean with correct counts."""
1723	_init_repo(tmp_path)
1724	prev: str \| None = None
1725	for i in range(200):
1726	prev = _make_commit(tmp_path, parent_id=prev, content=f"vstress-{i}".encode())
1727	out = tmp_path / "vstress200.bundle"
1728	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
1729	result = _invoke(["bundle", "verify", str(out), "--json"], env=_env(tmp_path))
1730	assert result.exit_code == 0
1731	data = _parse_verify(result)
1732	assert data["all_ok"] is True
1733	assert data["objects_checked"] >= 200
1734	assert data["snapshots_checked"] >= 200
1735
1736	def test_multiple_corrupt_objects_all_detected(
1737	self, tmp_path: pathlib.Path
1738	) -> None:
1739	"""Multiple corrupted objects must each produce a failure entry."""
1740	_init_repo(tmp_path)
1741	prev: str \| None = None
1742	for i in range(5):
1743	prev = _make_commit(tmp_path, parent_id=prev, content=f"multi-corrupt-{i}".encode())
1744	out = tmp_path / "multi-corrupt.bundle"
1745	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
1746	raw = msgpack.unpackb(out.read_bytes(), raw=False)
1747	# Corrupt every object.
1748	for obj in raw.get("objects", []):
1749	obj["content"] = b"TAMPERED"
1750	out.write_bytes(msgpack.packb(raw, use_bin_type=True))
1751	result = _invoke(["bundle", "verify", str(out), "--json"], env=_env(tmp_path))
1752	assert result.exit_code == 1
1753	data = _parse_verify(result)
1754	assert data["all_ok"] is False
1755	assert len(data["failures"]) >= 5
1756
1757	def test_empty_bundle_verifies_clean(self, tmp_path: pathlib.Path) -> None:
1758	"""An empty dict bundle has nothing to check and must exit 0."""
1759	_init_repo(tmp_path)
1760	empty = tmp_path / "empty.bundle"
1761	empty.write_bytes(msgpack.packb({}, use_bin_type=True))
1762	result = _invoke(["bundle", "verify", str(empty), "--json"], env=_env(tmp_path))
1763	assert result.exit_code == 0
1764	data = _parse_verify(result)
1765	assert data["all_ok"] is True
1766	assert data["objects_checked"] == 0
1767	assert data["failures"] == []
1768
1769
1770	# ===========================================================================
1771	# TestBundleListHeadsExtended — 18 tests
1772	# ===========================================================================
1773
1774
1775	class TestBundleListHeadsExtended:
1776	def _bundle_with_head(self, tmp_path: pathlib.Path, branch: str = "main") -> pathlib.Path:
1777	_init_repo(tmp_path)
1778	_make_commit(tmp_path, content=b"lhe-base", branch=branch)
1779	out = tmp_path / "lhe.bundle"
1780	_invoke(["bundle", "create", str(out)], env=_env(tmp_path))
1781	return out
1782
1783	def test_exits_0_with_heads(self, tmp_path: pathlib.Path) -> None:
1784	bundle = self._bundle_with_head(tmp_path)
1785	result = _invoke(["bundle", "list-heads", str(bundle)], env=_env(tmp_path))
1786	assert result.exit_code == 0
1787
1788	def test_exits_0_no_heads(self, tmp_path: pathlib.Path) -> None:
1789	"""A bundle with no branch_heads key must still exit 0."""
1790	_init_repo(tmp_path)
1791	_make_commit(tmp_path, content=b"lhe-noheads")
1792	bundle = tmp_path / "noheads.bundle"
1793	_invoke(["bundle", "create", str(bundle)], env=_env(tmp_path))
1794	raw = msgpack.unpackb(bundle.read_bytes(), raw=False)
1795	raw.pop("branch_heads", None)
1796	bundle.write_bytes(msgpack.packb(raw, use_bin_type=True))
1797	result = _invoke(["bundle", "list-heads", str(bundle)], env=_env(tmp_path))
1798	assert result.exit_code == 0
1799
1800	def test_text_shows_branch_and_cid(self, tmp_path: pathlib.Path) -> None:
1801	bundle = self._bundle_with_head(tmp_path)
1802	result = _invoke(["bundle", "list-heads", str(bundle)], env=_env(tmp_path))
1803	assert result.exit_code == 0
1804	assert "main" in result.output
1805
1806	def test_text_no_heads_message(self, tmp_path: pathlib.Path) -> None:
1807	_init_repo(tmp_path)
1808	_make_commit(tmp_path, content=b"lhe-nomsg")
1809	bundle = tmp_path / "nomsg.bundle"
1810	_invoke(["bundle", "create", str(bundle)], env=_env(tmp_path))
1811	raw = msgpack.unpackb(bundle.read_bytes(), raw=False)
1812	raw.pop("branch_heads", None)
1813	bundle.write_bytes(msgpack.packb(raw, use_bin_type=True))
1814	result = _invoke(["bundle", "list-heads", str(bundle)], env=_env(tmp_path))
1815	assert "No branch heads" in result.output
1816
1817	def test_json_returns_dict(self, tmp_path: pathlib.Path) -> None:
1818	bundle = self._bundle_with_head(tmp_path)
1819	result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path))
1820	assert result.exit_code == 0
1821	data = json.loads(result.output)
1822	assert isinstance(data["heads"], dict)
1823
1824	def test_json_contains_main(self, tmp_path: pathlib.Path) -> None:
1825	bundle = self._bundle_with_head(tmp_path)
1826	result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path))
1827	data = json.loads(result.output)
1828	assert "main" in data["heads"]
1829
1830	def test_json_commit_id_has_sha256_prefix(self, tmp_path: pathlib.Path) -> None:
1831	bundle = self._bundle_with_head(tmp_path)
1832	result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path))
1833	data = json.loads(result.output)
1834	for cid in data["heads"].values():
1835	assert cid.startswith("sha256:")
1836	assert len(cid) == len("sha256:") + 64
1837
1838	def test_json_is_single_line(self, tmp_path: pathlib.Path) -> None:
1839	"""JSON output must be compact (no indent=2)."""
1840	bundle = self._bundle_with_head(tmp_path)
1841	result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path))
1842	assert "\n" not in result.output.strip()
1843
1844	def test_j_alias(self, tmp_path: pathlib.Path) -> None:
1845	bundle = self._bundle_with_head(tmp_path)
1846	r1 = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path))
1847	r2 = _invoke(["bundle", "list-heads", str(bundle), "-j"], env=_env(tmp_path))
1848	assert r1.exit_code == 0 and r2.exit_code == 0
1849	assert json.loads(r1.output)["heads"] == json.loads(r2.output)["heads"]
1850
1851	def test_json_empty_on_no_heads(self, tmp_path: pathlib.Path) -> None:
1852	_init_repo(tmp_path)
1853	_make_commit(tmp_path, content=b"lhe-empty-json")
1854	bundle = tmp_path / "ej.bundle"
1855	_invoke(["bundle", "create", str(bundle)], env=_env(tmp_path))
1856	raw = msgpack.unpackb(bundle.read_bytes(), raw=False)
1857	raw.pop("branch_heads", None)
1858	bundle.write_bytes(msgpack.packb(raw, use_bin_type=True))
1859	result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path))
1860	assert result.exit_code == 0
1861	assert json.loads(result.output)["heads"] == {}
1862
1863	def test_multiple_branches_all_listed_text(self, tmp_path: pathlib.Path) -> None:
1864	_init_repo(tmp_path)
1865	c1 = _make_commit(tmp_path, content=b"lhe-multi-base")
1866	for br in ("feat/a", "feat/b", "feat/c"):
1867	ref = tmp_path / ".muse" / "refs" / "heads" / br
1868	ref.parent.mkdir(parents=True, exist_ok=True)
1869	ref.write_text(c1, encoding="utf-8")
1870	bundle = tmp_path / "multi.bundle"
1871	_invoke(["bundle", "create", str(bundle)], env=_env(tmp_path))
1872	result = _invoke(["bundle", "list-heads", str(bundle)], env=_env(tmp_path))
1873	assert result.exit_code == 0
1874	for br in ("feat/a", "feat/b", "feat/c"):
1875	assert br in result.output
1876
1877	def test_multiple_branches_all_in_json(self, tmp_path: pathlib.Path) -> None:
1878	_init_repo(tmp_path)
1879	c1 = _make_commit(tmp_path, content=b"lhe-multi-json")
1880	for br in ("feat/x", "feat/y"):
1881	ref = tmp_path / ".muse" / "refs" / "heads" / br
1882	ref.parent.mkdir(parents=True, exist_ok=True)
1883	ref.write_text(c1, encoding="utf-8")
1884	bundle = tmp_path / "multij.bundle"
1885	_invoke(["bundle", "create", str(bundle)], env=_env(tmp_path))
1886	result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path))
1887	data = json.loads(result.output)
1888	assert "feat/x" in data["heads"]
1889	assert "feat/y" in data["heads"]
1890
1891	def test_text_cid_shows_sha256_prefix_plus_12(self, tmp_path: pathlib.Path) -> None:
1892	"""Text output shows sha256: prefix + 12 hex chars abbreviated commit ID."""
1893	bundle = self._bundle_with_head(tmp_path)
1894	result = _invoke(["bundle", "list-heads", str(bundle)], env=_env(tmp_path))
1895	# Each non-empty line should start with sha256:<12hex>.
1896	for line in result.output.strip().splitlines():
1897	parts = line.split()
1898	assert parts[0].startswith("sha256:")
1899	assert len(parts[0]) == len("sha256:") + 12
1900
1901	def test_no_repo_required(self, tmp_path: pathlib.Path) -> None:
1902	"""list-heads must work outside any .muse repository."""
1903	work = tmp_path / "no_repo_dir"
1904	work.mkdir()
1905	_init_repo(tmp_path)
1906	_make_commit(tmp_path, content=b"lhe-no-repo")
1907	bundle = tmp_path / "norepo.bundle"
1908	_invoke(["bundle", "create", str(bundle)], env=_env(tmp_path))
1909	result = _invoke(["bundle", "list-heads", str(bundle)], env={"MUSE_REPO_ROOT": str(work)})
1910	assert result.exit_code != 2
1911
1912	def test_help_mentions_agent_quickstart(self) -> None:
1913	result = _invoke(["bundle", "list-heads", "--help"])
1914	assert result.exit_code == 0
1915	assert "Agent quickstart" in result.output
1916
1917	def test_help_mentions_exit_codes(self) -> None:
1918	result = _invoke(["bundle", "list-heads", "--help"])
1919	assert result.exit_code == 0
1920	assert "Exit codes" in result.output
1921
1922	def test_help_mentions_json_schema(self) -> None:
1923	result = _invoke(["bundle", "list-heads", "--help"])
1924	assert result.exit_code == 0
1925	assert "JSON output schema" in result.output
1926
1927	def test_missing_file_exits_1(self, tmp_path: pathlib.Path) -> None:
1928	_init_repo(tmp_path)
1929	result = _invoke(
1930	["bundle", "list-heads", str(tmp_path / "ghost.bundle")],
1931	env=_env(tmp_path),
1932	)
1933	assert result.exit_code == 1
1934
1935
1936	# ===========================================================================
1937	# TestBundleListHeadsSecurity — 6 tests
1938	# ===========================================================================
1939
1940
1941	class TestBundleListHeadsSecurity:
1942	def test_ansi_branch_name_stripped_text(self, tmp_path: pathlib.Path) -> None:
1943	"""ANSI escape in branch name must not appear in text output."""
1944	_init_repo(tmp_path)
1945	_make_commit(tmp_path, content=b"sec-lh-ansi")
1946	bundle = tmp_path / "ansi-br.bundle"
1947	_invoke(["bundle", "create", str(bundle)], env=_env(tmp_path))
1948	raw = msgpack.unpackb(bundle.read_bytes(), raw=False)
1949	raw["branch_heads"] = {"\x1b[31mevil\x1b[0m": "a" * 64}
1950	bundle.write_bytes(msgpack.packb(raw, use_bin_type=True))
1951	result = _invoke(["bundle", "list-heads", str(bundle)], env=_env(tmp_path))
1952	assert result.exit_code == 0
1953	assert "\x1b" not in result.output
1954
1955	def test_ansi_commit_id_stripped_text(self, tmp_path: pathlib.Path) -> None:
1956	"""ANSI escape in a commit ID must not appear in text output (cid[:12])."""
1957	_init_repo(tmp_path)
1958	_make_commit(tmp_path, content=b"sec-lh-cid")
1959	bundle = tmp_path / "ansi-cid.bundle"
1960	_invoke(["bundle", "create", str(bundle)], env=_env(tmp_path))
1961	raw = msgpack.unpackb(bundle.read_bytes(), raw=False)
1962	raw["branch_heads"] = {"main": "\x1b[31m" + "a" * 64 + "\x1b[0m"}
1963	bundle.write_bytes(msgpack.packb(raw, use_bin_type=True))
1964	result = _invoke(["bundle", "list-heads", str(bundle)], env=_env(tmp_path))
1965	assert result.exit_code == 0
1966	assert "\x1b" not in result.output
1967
1968	def test_ansi_branch_name_stripped_json(self, tmp_path: pathlib.Path) -> None:
1969	"""ANSI escape in branch name must not appear in JSON output."""
1970	_init_repo(tmp_path)
1971	_make_commit(tmp_path, content=b"sec-lh-ansi-json")
1972	bundle = tmp_path / "ansi-br-json.bundle"
1973	_invoke(["bundle", "create", str(bundle)], env=_env(tmp_path))
1974	raw = msgpack.unpackb(bundle.read_bytes(), raw=False)
1975	raw["branch_heads"] = {"\x1b[31mevil\x1b[0m": "b" * 64}
1976	bundle.write_bytes(msgpack.packb(raw, use_bin_type=True))
1977	result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path))
1978	assert result.exit_code == 0
1979	assert "\x1b" not in result.output
1980
1981	def test_ansi_commit_id_stripped_json(self, tmp_path: pathlib.Path) -> None:
1982	"""ANSI escape in commit ID value must not appear in JSON output."""
1983	_init_repo(tmp_path)
1984	_make_commit(tmp_path, content=b"sec-lh-cid-json")
1985	bundle = tmp_path / "ansi-cid-json.bundle"
1986	_invoke(["bundle", "create", str(bundle)], env=_env(tmp_path))
1987	raw = msgpack.unpackb(bundle.read_bytes(), raw=False)
1988	raw["branch_heads"] = {"main": "\x1b[32m" + "c" * 64 + "\x1b[0m"}
1989	bundle.write_bytes(msgpack.packb(raw, use_bin_type=True))
1990	result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path))
1991	assert result.exit_code == 0
1992	assert "\x1b" not in result.output
1993
1994	def test_invalid_msgpack_exits_1(self, tmp_path: pathlib.Path) -> None:
1995	_init_repo(tmp_path)
1996	bad = tmp_path / "bad.bundle"
1997	bad.write_bytes(b"\xff\xfe garbage")
1998	result = _invoke(["bundle", "list-heads", str(bad)], env=_env(tmp_path))
1999	assert result.exit_code == 1
2000
2001	def test_no_json_on_missing_file(self, tmp_path: pathlib.Path) -> None:
2002	"""Missing file error must not emit JSON to stdout."""
2003	_init_repo(tmp_path)
2004	result = _invoke(
2005	["bundle", "list-heads", str(tmp_path / "missing.bundle"), "--json"],
2006	env=_env(tmp_path),
2007	)
2008	assert result.exit_code != 0
2009	assert not result.output.strip().startswith("{")
2010
2011
2012	# ===========================================================================
2013	# TestBundleListHeadsStress — 3 tests
2014	# ===========================================================================
2015
2016
2017	class TestBundleListHeadsStress:
2018	def test_50_branches_all_listed(self, tmp_path: pathlib.Path) -> None:
2019	"""50 branch heads are all present in the JSON output."""
2020	_init_repo(tmp_path)
2021	c1 = _make_commit(tmp_path, content=b"lhstress-base")
2022	branch_names = [f"feat/stress-{i}" for i in range(50)]
2023	for br in branch_names:
2024	ref = tmp_path / ".muse" / "refs" / "heads" / br
2025	ref.parent.mkdir(parents=True, exist_ok=True)
2026	ref.write_text(c1, encoding="utf-8")
2027	bundle = tmp_path / "stress50.bundle"
2028	_invoke(["bundle", "create", str(bundle)], env=_env(tmp_path))
2029	result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path))
2030	assert result.exit_code == 0
2031	data = json.loads(result.output)
2032	for br in branch_names:
2033	assert br in data["heads"]
2034
2035	def test_concurrent_reads_consistent(self, tmp_path: pathlib.Path) -> None:
2036	"""Concurrent list-heads reads on the same bundle must all succeed."""
2037	_init_repo(tmp_path)
2038	_make_commit(tmp_path, content=b"lhstress-concurrent")
2039	bundle = tmp_path / "concurrent.bundle"
2040	_invoke(["bundle", "create", str(bundle)], env=_env(tmp_path))
2041	errors: list[str] = []
2042
2043	def _read() -> None:
2044	r = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path))
2045	if r.exit_code != 0:
2046	errors.append(f"exit {r.exit_code}")
2047	else:
2048	try:
2049	if not isinstance(json.loads(r.output), dict):
2050	errors.append("not a dict")
2051	except json.JSONDecodeError as exc:
2052	errors.append(str(exc))
2053
2054	threads = [threading.Thread(target=_read) for _ in range(10)]
2055	for t in threads:
2056	t.start()
2057	for t in threads:
2058	t.join()
2059	assert not errors, f"Concurrent failures: {errors}"
2060
2061	def test_large_bundle_list_heads_fast(self, tmp_path: pathlib.Path) -> None:
2062	"""list-heads on a 200-commit bundle returns quickly (I/O, not compute)."""
2063	import time
2064	_init_repo(tmp_path)
2065	prev: str \| None = None
2066	for i in range(200):
2067	prev = _make_commit(tmp_path, parent_id=prev, content=f"lhfast-{i}".encode())
2068	bundle = tmp_path / "fast200.bundle"
2069	_invoke(["bundle", "create", str(bundle)], env=_env(tmp_path))
2070	t0 = time.monotonic()
2071	result = _invoke(["bundle", "list-heads", str(bundle), "--json"], env=_env(tmp_path))
2072	elapsed = time.monotonic() - t0
2073	assert result.exit_code == 0
2074	assert elapsed < 5.0, f"list-heads took {elapsed:.2f}s on 200-commit bundle"
2075	# ---------------------------------------------------------------------------
2076	# Flag registration tests
2077	# ---------------------------------------------------------------------------
2078
2079	import argparse as _argparse
2080	from muse.cli.commands.bundle import register as _register_bundle
2081
2082
2083	def _parse_bundle(*args: str) -> _argparse.Namespace:
2084	"""Build an argument parser via register() and parse args."""
2085	root_p = _argparse.ArgumentParser()
2086	subs = root_p.add_subparsers(dest="cmd")
2087	_register_bundle(subs)
2088	return root_p.parse_args(["bundle", *args])
2089
2090
2091	class TestRegisterFlags:
2092	def test_create_default_json_out_is_false(self) -> None:
2093	ns = _parse_bundle("create", "out.bundle")
2094	assert ns.json_out is False
2095
2096	def test_create_json_flag(self) -> None:
2097	ns = _parse_bundle("create", "out.bundle", "--json")
2098	assert ns.json_out is True
2099
2100	def test_create_j_shorthand(self) -> None:
2101	ns = _parse_bundle("create", "out.bundle", "-j")
2102	assert ns.json_out is True
2103
2104	def test_inspect_default_json_out_is_false(self) -> None:
2105	ns = _parse_bundle("inspect", "bundle.muse")
2106	assert ns.json_out is False
2107
2108	def test_inspect_j_shorthand(self) -> None:
2109	ns = _parse_bundle("inspect", "bundle.muse", "-j")
2110	assert ns.json_out is True
2111
2112	def test_verify_default_json_out_is_false(self) -> None:
2113	ns = _parse_bundle("verify", "bundle.muse")
2114	assert ns.json_out is False
2115
2116	def test_verify_j_shorthand(self) -> None:
2117	ns = _parse_bundle("verify", "bundle.muse", "-j")
2118	assert ns.json_out is True

File History 3 commits

sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 131 days ago

sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 137 days ago

sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 140 days ago

class _CreateOut

class _UnbundleOut

class _VerifyOut

function _sha

function _init_repo

function _env

function _make_commit

function _invoke

function _parse_create

function _parse_unbundle

function _parse_verify

function test_iter_branches_skips_symlinks

function test_iter_branches_size_cap

function test_iter_branches_empty_dir

function test_iter_branches_multiple

function test_reachable_from_single

function test_reachable_from_chain

function test_reachable_from_empty_tips

function test_load_bundle_not_found

function test_load_bundle_invalid_msgpack

function test_load_bundle_not_dict

function test_list_heads_ansi_injection

function test_verify_failure_ansi_injection

function test_create_json_schema

function test_create_json_no_output_on_success_without_flag

function test_unbundle_json_schema

function test_unbundle_json_refs_updated

function test_unbundle_json_no_update_refs

function test_verify_json_schema_clean

function test_verify_json_schema_corrupt

function test_verify_json_snapshots_checked

function test_list_heads_json_schema

function test_verify_rejects_format_flag

function test_list_heads_rejects_format_flag

function test_create_have_prunes_bundle

function test_create_have_json_smaller_commits

function test_create_multiple_branches

function test_round_trip_with_json_summary

function test_verify_missing_snapshot_object

function test_create_help_mentions_json

function test_unbundle_help_mentions_json

function test_verify_help_mentions_json

function test_list_heads_help_mentions_json

function test_bundle_help_top_level

function test_stress_200_commit_chain

function test_stress_concurrent_list_heads

function _read

class TestBundleCreateExtended

function test_exits_0_basic

function test_creates_file_on_disk

function test_text_output_mentions_commits

function test_text_output_mentions_kib

function test_text_output_contains_bundle_path

function test_empty_repo_exits_1

function test_bad_ref_exits_1

function test_json_branches_sorted

function test_json_size_matches_file

function test_j_alias

function test_default_ref_is_head

function test_explicit_head_ref

function test_explicit_commit_id

function test_output_is_valid_msgpack

function test_help_mentions_agent_quickstart

function test_help_mentions_exit_codes

function test_help_mentions_json_schema

function test_multiple_have_ids_reduce_bundle

class TestBundleCreateSecurity

function test_ansi_in_file_path_stripped_text_output

function test_control_char_in_file_path_stripped

function test_outside_repo_exits_2

function test_bad_ref_ansi_stripped_from_error

function test_ansi_in_have_no_injection

function test_no_json_on_error

class TestBundleCreateStress

function test_50_commit_chain

function test_create_with_large_have_list

function test_many_branches

function _make_bundle

class TestBundleUnbundleExtended

function _src_dst

function test_exits_0_basic

function test_commits_written_count

function test_snapshots_written_count

function test_objects_written_count

function test_objects_skipped_idempotent

function test_text_output_mentions_commits

function test_text_output_mentions_applied

function test_refs_updated_by_default

function test_no_update_refs_skips_refs

function test_refs_updated_branch_file_exists

function test_j_alias

function test_json_refs_updated_sorted

function test_empty_bundle_no_crash

function test_bundle_without_branch_heads_no_refs

function test_help_mentions_agent_quickstart

function test_help_mentions_exit_codes

function test_help_mentions_json_schema

function test_no_update_refs_flag_in_help

class TestBundleUnbundleSecurity

function test_outside_repo_exits_2

function test_missing_bundle_file_exits_1

function test_invalid_msgpack_exits_1

function test_ansi_branch_name_skipped_no_injection

function test_invalid_commit_id_branch_ref_skipped

function test_no_json_on_missing_file

class TestBundleUnbundleStress

function test_50_commit_chain

function test_idempotent_multiple_applications

function test_many_branch_refs_updated

class TestBundleVerifyExtended

function _clean_bundle

function _corrupt_bundle

function test_exits_0_on_clean_bundle

function test_exits_1_on_corrupt_object

function test_all_ok_true_on_clean

function test_all_ok_false_on_corrupt

function test_objects_checked_count

function test_snapshots_checked_count

function test_failures_empty_on_clean

function test_failures_nonempty_on_corrupt

function test_quiet_clean_exits_0_no_output

function test_quiet_corrupt_exits_1_no_output

function test_json_output_is_single_line

function test_j_alias

function test_text_output_mentions_objects_checked

function test_text_output_mentions_snapshots_checked

function test_text_output_clean_checkmark

function test_text_output_failures_listed

function test_help_mentions_agent_quickstart

function test_help_mentions_exit_codes

class TestBundleVerifySecurity

function _bundle_with_ansi_object_id

function _bundle_with_ansi_rel_path

function test_ansi_in_object_id_failure_stripped_text

function test_ansi_in_rel_path_failure_stripped_text

function test_ansi_in_failures_sanitized_json

function test_no_repo_required

function test_missing_file_exits_1

function test_invalid_msgpack_exits_1

class TestBundleVerifyStress

function test_200_commit_bundle_verify_clean

function test_multiple_corrupt_objects_all_detected

function test_empty_bundle_verifies_clean

class TestBundleListHeadsExtended

function _bundle_with_head

function test_exits_0_with_heads

function test_exits_0_no_heads

function test_text_shows_branch_and_cid

function test_text_no_heads_message

function test_json_returns_dict

function test_json_contains_main

function test_json_commit_id_has_sha256_prefix

function test_json_is_single_line

function test_j_alias

function test_json_empty_on_no_heads

function test_multiple_branches_all_listed_text

function test_multiple_branches_all_in_json

function test_text_cid_shows_sha256_prefix_plus_12

function test_no_repo_required

function test_help_mentions_agent_quickstart

function test_help_mentions_exit_codes

function test_help_mentions_json_schema

function test_missing_file_exits_1

class TestBundleListHeadsSecurity

function test_ansi_branch_name_stripped_text

function test_ansi_commit_id_stripped_text

function test_ansi_branch_name_stripped_json

function test_ansi_commit_id_stripped_json

function test_invalid_msgpack_exits_1

function test_no_json_on_missing_file

class TestBundleListHeadsStress

function test_50_branches_all_listed

function test_concurrent_reads_consistent

function _read

function test_large_bundle_list_heads_fast

function _parse_bundle

class TestRegisterFlags

function test_create_default_json_out_is_false

function test_create_json_flag

function test_create_j_shorthand

function test_inspect_default_json_out_is_false

function test_inspect_j_shorthand

function test_verify_default_json_out_is_false

function test_verify_j_shorthand

Pathtests/test_cmd_bundle_hardening.py

Lines2,118

Size85.1 KB

LangPython

Refsha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9

Object ID

sha256:d15b77ac2b6bf7cd2b9dc61003c3de6fb7a2a006a8472ae8c151ccf42c08c822…

Last commit

sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9

docs: expand cache plan with all seven testing ti…

131 days ago

Quick links

Blame History