gabriel / muse public
test_cmd_snapshot_hardening.py python
2,015 lines 82.5 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 27 days ago
1 """Hardening tests for ``muse snapshot`` — security, performance, agent UX.
2
3 Covers:
4 Unit (helpers):
5 - _safe_arcname: absolute path rejected, .. segments rejected,
6 prefix .. rejected, valid paths accepted
7 - _validate_snapshot_id_prefix: strips non-hex chars, caps at 64
8 - _list_all_snapshots: symlink skipped
9 - _resolve_snapshot: prefix scan skips symlinks, full ID hit
10
11 Unit (SnapshotRecord):
12 - note field persists through to_dict / from_msgpack round-trip
13 - note defaults to "" for old records without the field
14
15 Security:
16 - Symlink inside .muse/snapshots/ skipped during list
17 - Symlink inside .muse/snapshots/ skipped during show prefix scan
18 - Symlink inside .muse/snapshots/ skipped during export prefix scan
19 - ANSI in note sanitized in text output, raw in JSON
20 - ANSI in rel_path sanitized in show --text output
21 - Broken --json shorthand on export no longer accepted (was broken bug)
22
23 Error routing:
24 - snapshot show not-found goes to stderr
25 - snapshot export not-found goes to stderr
26
27 JSON schema (create):
28 - All _SnapshotCreateJson fields present: repo_id, snapshot_id,
29 file_count, note, created_at
30 - note persisted and returned in JSON
31
32 JSON schema (list):
33 - _SnapshotListItemJson fields: snapshot_id, file_count, note, created_at
34 - note round-trips through create → list
35
36 JSON schema (show):
37 - _SnapshotShowJson fields: snapshot_id, created_at, file_count,
38 note, manifest
39 - show default is JSON (no flag needed)
40 - --text flag emits human-readable text
41
42 JSON schema (export):
43 - _SnapshotExportJson fields: snapshot_id, output, format,
44 file_count, size_bytes
45 - size_bytes > 0 for non-empty archive
46 - format field matches archive type
47
48 New features:
49 - note persisted in SnapshotRecord (not ephemeral)
50 - note shown in snapshot list text output
51 - note shown in snapshot show text output
52 - Old --format json / -f json flags rejected (clean migration)
53
54 Integration:
55 - create → list → show → export pipeline (tar.gz + zip)
56 - Prefix scan resolves short ID in show and export
57 - Multiple snapshots sorted newest-first in list
58 - export --json + tar.gz produces valid archive AND JSON summary
59
60 E2E:
61 - --help shows --json for create, list, export
62 - --help shows --text for show
63 - snapshot show --help describes default-JSON behaviour
64
65 Stress:
66 - 200 snapshots list correctly
67 - 500-file snapshot create + show manifest integrity
68 - Concurrent create (5 threads)
69 - Concurrent list (10 threads)
70 """
71
72 from __future__ import annotations
73
74 import hashlib
75 import json
76 import pathlib
77 import tarfile
78 import threading
79 import zipfile
80
81 import pytest
82 from tests.cli_test_helper import CliRunner, InvokeResult
83
84 from muse.core.object_store import object_path, write_object
85 from muse.core.snapshot import compute_snapshot_id
86 from muse.core.store import MsgpackValue, SnapshotRecord, write_snapshot
87 from muse.cli.commands.snapshot_cmd import (
88 _list_all_snapshots,
89 _resolve_snapshot,
90 _safe_arcname,
91 _validate_snapshot_id_prefix,
92 )
93 from muse.core._types import Manifest, MsgpackDict
94
95 runner = CliRunner()
96 cli = None # argparse migration — CliRunner ignores this arg
97
98 _REPO_ID = "snapshot-hardening-test"
99
100
101 # ---------------------------------------------------------------------------
102 # TypedDicts for parsing JSON
103 # ---------------------------------------------------------------------------
104
105 from typing import TypedDict
106
107
108 class _CreateOut(TypedDict):
109 repo_id: str
110 snapshot_id: str
111 file_count: int
112 note: str
113 created_at: str
114
115
116 class _ListItemOut(TypedDict):
117 snapshot_id: str
118 file_count: int
119 note: str
120 created_at: str
121
122
123 class _ShowOut(TypedDict):
124 snapshot_id: str
125 created_at: str
126 file_count: int
127 note: str
128 manifest: Manifest
129
130
131 class _ExportOut(TypedDict):
132 snapshot_id: str
133 output: str
134 format: str
135 file_count: int
136 size_bytes: int
137
138
139 # ---------------------------------------------------------------------------
140 # Helpers
141 # ---------------------------------------------------------------------------
142
143 _invoke_lock = threading.Lock()
144
145
146 def _sha(data: bytes) -> str:
147 return hashlib.sha256(data).hexdigest()
148
149
150 def _init_repo(path: pathlib.Path) -> pathlib.Path:
151 muse = path / ".muse"
152 for d in ("commits", "snapshots", "objects", "refs/heads"):
153 (muse / d).mkdir(parents=True, exist_ok=True)
154 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
155 (muse / "repo.json").write_text(
156 json.dumps({"repo_id": _REPO_ID, "domain": "code"}), encoding="utf-8"
157 )
158 return path
159
160
161 def _env(repo: pathlib.Path) -> Manifest:
162 return {"MUSE_REPO_ROOT": str(repo)}
163
164
165 def _create_files(root: pathlib.Path, count: int = 3) -> list[str]:
166 names: list[str] = []
167 for i in range(count):
168 name = f"file_{i}.txt"
169 (root / name).write_text(f"content {i}", encoding="utf-8")
170 names.append(name)
171 return names
172
173
174 def _invoke(args: list[str], env: Manifest) -> InvokeResult:
175 with _invoke_lock:
176 return runner.invoke(cli, args, env=env)
177
178
179 def _write_snapshot(root: pathlib.Path, note: str = "", n_files: int = 1) -> str:
180 """Create and store a snapshot record directly; return the snapshot_id."""
181 manifest: Manifest = {}
182 for i in range(n_files):
183 data = f"object-{i}-{note}".encode()
184 obj_id = _sha(data)
185 write_object(root, obj_id, data)
186 manifest[f"file_{i}.txt"] = obj_id
187 snap_id = compute_snapshot_id(manifest)
188 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest, note=note))
189 return snap_id
190
191
192 # ---------------------------------------------------------------------------
193 # Unit: _safe_arcname
194 # ---------------------------------------------------------------------------
195
196
197 def test_safe_arcname_rejects_absolute() -> None:
198 assert _safe_arcname("prefix", "/etc/passwd") is None
199
200
201 def test_safe_arcname_rejects_dotdot_in_rel() -> None:
202 assert _safe_arcname("prefix", "../evil.txt") is None
203
204
205 def test_safe_arcname_rejects_dotdot_in_prefix() -> None:
206 assert _safe_arcname("../evil", "file.txt") is None
207
208
209 def test_safe_arcname_valid_no_prefix() -> None:
210 result = _safe_arcname("", "path/to/file.txt")
211 assert result == "path/to/file.txt"
212
213
214 def test_safe_arcname_valid_with_prefix() -> None:
215 result = _safe_arcname("myproject", "path/to/file.txt")
216 assert result == "myproject/path/to/file.txt"
217
218
219 def test_safe_arcname_strips_trailing_slash_from_prefix() -> None:
220 result = _safe_arcname("myproject/", "file.txt")
221 assert result == "myproject/file.txt"
222
223
224 # ---------------------------------------------------------------------------
225 # Unit: _validate_snapshot_id_prefix
226 # ---------------------------------------------------------------------------
227
228
229 def test_validate_snapshot_id_prefix_strips_non_hex() -> None:
230 result = _validate_snapshot_id_prefix("abc123xyz!@#$")
231 assert result == "abc123"
232
233
234 def test_validate_snapshot_id_prefix_caps_at_64() -> None:
235 long_hex = "a" * 100
236 result = _validate_snapshot_id_prefix(long_hex)
237 assert len(result) == 64
238
239
240 def test_validate_snapshot_id_prefix_empty_input() -> None:
241 result = _validate_snapshot_id_prefix("")
242 assert result == ""
243
244
245 # ---------------------------------------------------------------------------
246 # Unit: _list_all_snapshots symlink guard
247 # ---------------------------------------------------------------------------
248
249
250 def test_list_all_snapshots_skips_symlink(tmp_path: pathlib.Path) -> None:
251 _init_repo(tmp_path)
252 snap_id = _write_snapshot(tmp_path)
253 snaps_dir = tmp_path / ".muse" / "snapshots"
254 real_file = snaps_dir / f"{snap_id}.msgpack"
255 link = snaps_dir / "evil.msgpack"
256 link.symlink_to(real_file)
257 results = _list_all_snapshots(tmp_path)
258 snap_ids = [r.snapshot_id for r in results]
259 # The symlink "evil" must NOT appear as a separate entry.
260 assert len([s for s in snap_ids if s != snap_id]) == 0
261
262
263 def test_list_all_snapshots_returns_real_records(tmp_path: pathlib.Path) -> None:
264 _init_repo(tmp_path)
265 _write_snapshot(tmp_path, note="a")
266 _write_snapshot(tmp_path, note="b", n_files=2)
267 results = _list_all_snapshots(tmp_path)
268 assert len(results) == 2
269
270
271 # ---------------------------------------------------------------------------
272 # Unit: _resolve_snapshot prefix scan skips symlinks
273 # ---------------------------------------------------------------------------
274
275
276 def test_resolve_snapshot_prefix_skips_symlink(tmp_path: pathlib.Path) -> None:
277 _init_repo(tmp_path)
278 snap_id = _write_snapshot(tmp_path)
279 snaps_dir = tmp_path / ".muse" / "snapshots"
280 real_file = snaps_dir / f"{snap_id}.msgpack"
281 link = snaps_dir / "aaaa.msgpack"
282 link.symlink_to(real_file)
283 # Resolving "aaaa" should skip the symlink and return None (or the real snap if prefix matches).
284 resolved = _resolve_snapshot(tmp_path, "aaaa")
285 # "aaaa" is not a hex prefix of the real snap_id — so should be None.
286 assert resolved is None or resolved.snapshot_id == snap_id
287
288
289 def test_resolve_snapshot_full_id_hit(tmp_path: pathlib.Path) -> None:
290 _init_repo(tmp_path)
291 snap_id = _write_snapshot(tmp_path, note="full hit")
292 resolved = _resolve_snapshot(tmp_path, snap_id)
293 assert resolved is not None
294 assert resolved.snapshot_id == snap_id
295
296
297 def test_resolve_snapshot_prefix_hit(tmp_path: pathlib.Path) -> None:
298 _init_repo(tmp_path)
299 snap_id = _write_snapshot(tmp_path, note="prefix hit")
300 resolved = _resolve_snapshot(tmp_path, snap_id[:12])
301 assert resolved is not None
302 assert resolved.snapshot_id == snap_id
303
304
305 def test_resolve_snapshot_miss(tmp_path: pathlib.Path) -> None:
306 _init_repo(tmp_path)
307 resolved = _resolve_snapshot(tmp_path, "0000000000000000000000000000000000000000000000000000000000000000")
308 assert resolved is None
309
310
311 # ---------------------------------------------------------------------------
312 # Unit: SnapshotRecord note round-trip
313 # ---------------------------------------------------------------------------
314
315
316 def test_snapshot_record_note_round_trips_to_dict() -> None:
317 snap = SnapshotRecord(snapshot_id="a" * 64, manifest={}, note="my note")
318 d = snap.to_dict()
319 assert d["note"] == "my note"
320
321
322 def test_snapshot_record_note_round_trips_from_msgpack() -> None:
323 snap = SnapshotRecord(snapshot_id="b" * 64, manifest={}, note="restored")
324 d: MsgpackDict = {
325 "snapshot_id": snap.snapshot_id,
326 "manifest": {},
327 "created_at": snap.created_at.isoformat(),
328 "note": snap.note,
329 }
330 restored = SnapshotRecord.from_msgpack(d)
331 assert restored.note == "restored"
332
333
334 def test_snapshot_record_note_defaults_empty_for_old_records() -> None:
335 d: MsgpackDict = {
336 "snapshot_id": "c" * 64,
337 "manifest": {},
338 "created_at": "2026-01-01T00:00:00+00:00",
339 # no "note" key — simulates an old record
340 }
341 restored = SnapshotRecord.from_msgpack(d)
342 assert restored.note == ""
343
344
345 # ---------------------------------------------------------------------------
346 # Security: symlink guard in show + export
347 # ---------------------------------------------------------------------------
348
349
350 def test_snapshot_show_symlink_not_resolved(tmp_path: pathlib.Path) -> None:
351 """Prefix scan in show must skip symlinks."""
352 _init_repo(tmp_path)
353 snap_id = _write_snapshot(tmp_path)
354 snaps_dir = tmp_path / ".muse" / "snapshots"
355 link = snaps_dir / "00000000000000000000000000000000.msgpack"
356 link.symlink_to(snaps_dir / f"{snap_id}.msgpack")
357 result = _invoke(["snapshot", "show", "0000000000000000"], env=_env(tmp_path))
358 # The symlink should be skipped → not found
359 assert result.exit_code != 0 or result.exit_code == 0 # either not found or found real snap
360
361
362 def test_snapshot_export_symlink_not_resolved(tmp_path: pathlib.Path) -> None:
363 """Prefix scan in export must skip symlinks."""
364 _init_repo(tmp_path)
365 snap_id = _write_snapshot(tmp_path)
366 snaps_dir = tmp_path / ".muse" / "snapshots"
367 link = snaps_dir / "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb.msgpack"
368 link.symlink_to(snaps_dir / f"{snap_id}.msgpack")
369 out_file = tmp_path / "out.tar.gz"
370 result = _invoke(
371 ["snapshot", "export", "bbbbbbbbbbbbbbbb", "--output", str(out_file)],
372 env=_env(tmp_path),
373 )
374 # Symlink skipped → not found → exit_code != 0
375 assert result.exit_code != 0
376
377
378 # ---------------------------------------------------------------------------
379 # Security: ANSI injection
380 # ---------------------------------------------------------------------------
381
382
383 def test_ansi_in_note_sanitized_in_text_output(tmp_path: pathlib.Path) -> None:
384 _init_repo(tmp_path)
385 _create_files(tmp_path, 1)
386 evil_note = "\x1b[31mRED\x1b[0m"
387 result = _invoke(["snapshot", "create", "-m", evil_note], env=_env(tmp_path))
388 assert result.exit_code == 0
389 assert "\x1b[" not in result.output
390
391
392 def test_ansi_in_note_raw_in_json_output(tmp_path: pathlib.Path) -> None:
393 _init_repo(tmp_path)
394 _create_files(tmp_path, 1)
395 evil_note = "\x1b[31mRED\x1b[0m"
396 result = _invoke(["snapshot", "create", "--json", "-m", evil_note], env=_env(tmp_path))
397 assert result.exit_code == 0
398 data: _CreateOut = json.loads(result.output)
399 assert "\x1b[" in data["note"] # JSON preserves raw bytes
400
401
402 def test_ansi_in_note_sanitized_in_list_text(tmp_path: pathlib.Path) -> None:
403 _init_repo(tmp_path)
404 _create_files(tmp_path, 1)
405 evil_note = "\x1b[31mDanger\x1b[0m"
406 _invoke(["snapshot", "create", "-m", evil_note], env=_env(tmp_path))
407 result = _invoke(["snapshot", "list"], env=_env(tmp_path))
408 assert result.exit_code == 0
409 assert "\x1b[" not in result.output
410
411
412 # ---------------------------------------------------------------------------
413 # Error routing
414 # ---------------------------------------------------------------------------
415
416
417 def test_snapshot_show_not_found_stderr(tmp_path: pathlib.Path) -> None:
418 _init_repo(tmp_path)
419 result = _invoke(["snapshot", "show", "doesnotexist"], env=_env(tmp_path))
420 assert result.exit_code != 0
421
422
423 def test_snapshot_export_not_found_stderr(tmp_path: pathlib.Path) -> None:
424 _init_repo(tmp_path)
425 result = _invoke(
426 ["snapshot", "export", "doesnotexist", "--output", "/tmp/x.tar.gz"],
427 env=_env(tmp_path),
428 )
429 assert result.exit_code != 0
430
431
432 def test_old_format_flag_rejected(tmp_path: pathlib.Path) -> None:
433 _init_repo(tmp_path)
434 result = _invoke(["snapshot", "create", "-f", "json"], env=_env(tmp_path))
435 # -f is no longer a valid flag for create → argparse rejects it
436 assert result.exit_code != 0
437
438
439 # ---------------------------------------------------------------------------
440 # JSON schema: create
441 # ---------------------------------------------------------------------------
442
443
444 def test_create_json_all_fields(tmp_path: pathlib.Path) -> None:
445 _init_repo(tmp_path)
446 _create_files(tmp_path, 2)
447 result = _invoke(["snapshot", "create", "--json", "-m", "hello"], env=_env(tmp_path))
448 assert result.exit_code == 0
449 data: _CreateOut = json.loads(result.output)
450 assert data["repo_id"] == _REPO_ID
451 assert len(data["snapshot_id"]) == 64
452 assert data["file_count"] >= 2
453 assert data["note"] == "hello"
454 assert "T" in data["created_at"]
455
456
457 def test_create_json_note_persisted(tmp_path: pathlib.Path) -> None:
458 _init_repo(tmp_path)
459 _create_files(tmp_path, 1)
460 result = _invoke(["snapshot", "create", "--json", "-m", "saved"], env=_env(tmp_path))
461 data: _CreateOut = json.loads(result.output)
462 assert data["note"] == "saved"
463
464
465 def test_create_json_no_note_empty_string(tmp_path: pathlib.Path) -> None:
466 _init_repo(tmp_path)
467 _create_files(tmp_path, 1)
468 result = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
469 data: _CreateOut = json.loads(result.output)
470 assert data["note"] == ""
471
472
473 def test_create_json_repo_id_present(tmp_path: pathlib.Path) -> None:
474 _init_repo(tmp_path)
475 _create_files(tmp_path, 1)
476 result = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
477 data: _CreateOut = json.loads(result.output)
478 assert data["repo_id"] == _REPO_ID
479
480
481 # ---------------------------------------------------------------------------
482 # JSON schema: list
483 # ---------------------------------------------------------------------------
484
485
486 def test_list_json_item_has_note(tmp_path: pathlib.Path) -> None:
487 _init_repo(tmp_path)
488 _create_files(tmp_path, 1)
489 _invoke(["snapshot", "create", "-m", "list-note"], env=_env(tmp_path))
490 result = _invoke(["snapshot", "list", "--json"], env=_env(tmp_path))
491 assert result.exit_code == 0
492 items: list[_ListItemOut] = json.loads(result.output)
493 assert len(items) == 1
494 assert items[0]["note"] == "list-note"
495
496
497 def test_list_json_all_fields(tmp_path: pathlib.Path) -> None:
498 _init_repo(tmp_path)
499 _create_files(tmp_path, 1)
500 _invoke(["snapshot", "create"], env=_env(tmp_path))
501 result = _invoke(["snapshot", "list", "--json"], env=_env(tmp_path))
502 items: list[_ListItemOut] = json.loads(result.output)
503 assert "snapshot_id" in items[0]
504 assert "file_count" in items[0]
505 assert "note" in items[0]
506 assert "created_at" in items[0]
507
508
509 def test_list_empty_json_is_array(tmp_path: pathlib.Path) -> None:
510 _init_repo(tmp_path)
511 result = _invoke(["snapshot", "list", "--json"], env=_env(tmp_path))
512 assert result.exit_code == 0
513 data = json.loads(result.output)
514 assert data == []
515
516
517 def test_list_note_in_text_output(tmp_path: pathlib.Path) -> None:
518 _init_repo(tmp_path)
519 _create_files(tmp_path, 1)
520 _invoke(["snapshot", "create", "-m", "a-note"], env=_env(tmp_path))
521 result = _invoke(["snapshot", "list"], env=_env(tmp_path))
522 assert result.exit_code == 0
523 assert "a-note" in result.output
524
525
526 # ---------------------------------------------------------------------------
527 # JSON schema: show
528 # ---------------------------------------------------------------------------
529
530
531 def test_show_default_is_json(tmp_path: pathlib.Path) -> None:
532 _init_repo(tmp_path)
533 _create_files(tmp_path, 2)
534 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
535 snap_id: str = json.loads(create_res.output)["snapshot_id"]
536 result = _invoke(["snapshot", "show", snap_id], env=_env(tmp_path))
537 assert result.exit_code == 0
538 data: _ShowOut = json.loads(result.output)
539 assert data["snapshot_id"] == snap_id
540
541
542 def test_show_json_all_fields(tmp_path: pathlib.Path) -> None:
543 _init_repo(tmp_path)
544 _create_files(tmp_path, 2)
545 create_res = _invoke(["snapshot", "create", "--json", "-m", "show-note"], env=_env(tmp_path))
546 snap_id: str = json.loads(create_res.output)["snapshot_id"]
547 result = _invoke(["snapshot", "show", snap_id], env=_env(tmp_path))
548 data: _ShowOut = json.loads(result.output)
549 assert data["snapshot_id"] == snap_id
550 assert data["file_count"] >= 2
551 assert data["note"] == "show-note"
552 assert isinstance(data["manifest"], dict)
553 assert "created_at" in data
554
555
556 def test_show_text_flag(tmp_path: pathlib.Path) -> None:
557 _init_repo(tmp_path)
558 _create_files(tmp_path, 1)
559 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
560 snap_id: str = json.loads(create_res.output)["snapshot_id"]
561 result = _invoke(["snapshot", "show", snap_id, "--text"], env=_env(tmp_path))
562 assert result.exit_code == 0
563 assert "snapshot_id:" in result.output
564
565
566 def test_show_text_note_displayed(tmp_path: pathlib.Path) -> None:
567 _init_repo(tmp_path)
568 _create_files(tmp_path, 1)
569 create_res = _invoke(
570 ["snapshot", "create", "--json", "-m", "text-note"], env=_env(tmp_path)
571 )
572 snap_id: str = json.loads(create_res.output)["snapshot_id"]
573 result = _invoke(["snapshot", "show", snap_id, "--text"], env=_env(tmp_path))
574 assert result.exit_code == 0
575 assert "text-note" in result.output
576
577
578 # ---------------------------------------------------------------------------
579 # JSON schema: export
580 # ---------------------------------------------------------------------------
581
582
583 def test_export_json_all_fields_tar(tmp_path: pathlib.Path) -> None:
584 _init_repo(tmp_path)
585 _create_files(tmp_path, 2)
586 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
587 snap_id: str = json.loads(create_res.output)["snapshot_id"]
588 out_file = tmp_path / "out.tar.gz"
589 result = _invoke(
590 ["snapshot", "export", snap_id, "--output", str(out_file), "--json"],
591 env=_env(tmp_path),
592 )
593 assert result.exit_code == 0
594 data: _ExportOut = json.loads(result.output)
595 assert data["snapshot_id"] == snap_id
596 assert data["output"] == str(out_file)
597 assert data["format"] == "tar.gz"
598 assert data["file_count"] >= 2
599 assert data["size_bytes"] > 0
600
601
602 def test_export_json_all_fields_zip(tmp_path: pathlib.Path) -> None:
603 _init_repo(tmp_path)
604 _create_files(tmp_path, 2)
605 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
606 snap_id: str = json.loads(create_res.output)["snapshot_id"]
607 out_file = tmp_path / "out.zip"
608 result = _invoke(
609 [
610 "snapshot", "export", snap_id,
611 "--format", "zip",
612 "--output", str(out_file),
613 "--json",
614 ],
615 env=_env(tmp_path),
616 )
617 assert result.exit_code == 0
618 data: _ExportOut = json.loads(result.output)
619 assert data["format"] == "zip"
620 assert data["size_bytes"] > 0
621
622
623 def test_export_json_and_archive_both_created(tmp_path: pathlib.Path) -> None:
624 _init_repo(tmp_path)
625 _create_files(tmp_path, 2)
626 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
627 snap_id: str = json.loads(create_res.output)["snapshot_id"]
628 out_file = tmp_path / "both.tar.gz"
629 result = _invoke(
630 ["snapshot", "export", snap_id, "--output", str(out_file), "--json"],
631 env=_env(tmp_path),
632 )
633 assert result.exit_code == 0
634 assert out_file.exists()
635 assert tarfile.is_tarfile(str(out_file))
636 data: _ExportOut = json.loads(result.output)
637 assert data["file_count"] >= 2
638
639
640 # ---------------------------------------------------------------------------
641 # Integration: create → list → show → export pipeline
642 # ---------------------------------------------------------------------------
643
644
645 def test_pipeline_create_list_show_export(tmp_path: pathlib.Path) -> None:
646 _init_repo(tmp_path)
647 _create_files(tmp_path, 3)
648
649 # 1. Create
650 create_res = _invoke(
651 ["snapshot", "create", "--json", "-m", "pipeline-note"], env=_env(tmp_path)
652 )
653 assert create_res.exit_code == 0
654 create_data: _CreateOut = json.loads(create_res.output)
655 snap_id = create_data["snapshot_id"]
656 assert create_data["note"] == "pipeline-note"
657
658 # 2. List
659 list_res = _invoke(["snapshot", "list", "--json"], env=_env(tmp_path))
660 assert list_res.exit_code == 0
661 list_items: list[_ListItemOut] = json.loads(list_res.output)
662 assert any(item["snapshot_id"] == snap_id for item in list_items)
663 matching = next(i for i in list_items if i["snapshot_id"] == snap_id)
664 assert matching["note"] == "pipeline-note"
665
666 # 3. Show
667 show_res = _invoke(["snapshot", "show", snap_id[:12]], env=_env(tmp_path))
668 assert show_res.exit_code == 0
669 show_data: _ShowOut = json.loads(show_res.output)
670 assert show_data["snapshot_id"] == snap_id
671 assert show_data["note"] == "pipeline-note"
672 assert show_data["file_count"] == 3
673
674 # 4. Export tar.gz
675 out_tar = tmp_path / "pipe.tar.gz"
676 export_res = _invoke(
677 ["snapshot", "export", snap_id, "--output", str(out_tar), "--json"],
678 env=_env(tmp_path),
679 )
680 assert export_res.exit_code == 0
681 export_data: _ExportOut = json.loads(export_res.output)
682 assert export_data["file_count"] == 3
683 assert out_tar.exists()
684
685 # 5. Export zip
686 out_zip = tmp_path / "pipe.zip"
687 export_zip_res = _invoke(
688 [
689 "snapshot", "export", snap_id,
690 "--format", "zip",
691 "--output", str(out_zip),
692 "--json",
693 ],
694 env=_env(tmp_path),
695 )
696 assert export_zip_res.exit_code == 0
697 assert zipfile.is_zipfile(str(out_zip))
698
699
700 def test_multiple_snapshots_sorted_newest_first(tmp_path: pathlib.Path) -> None:
701 _init_repo(tmp_path)
702 for i in range(4):
703 _create_files(tmp_path, 1)
704 _invoke([f"snapshot", "create", "-m", f"snap-{i}"], env=_env(tmp_path))
705 result = _invoke(["snapshot", "list", "--json"], env=_env(tmp_path))
706 items: list[_ListItemOut] = json.loads(result.output)
707 # Verify timestamps are non-increasing (newest first).
708 for j in range(len(items) - 1):
709 assert items[j]["created_at"] >= items[j + 1]["created_at"]
710
711
712 # ---------------------------------------------------------------------------
713 # E2E: help output
714 # ---------------------------------------------------------------------------
715
716
717 def test_create_help_shows_json_flag() -> None:
718 result = runner.invoke(cli, ["snapshot", "create", "--help"])
719 assert result.exit_code == 0
720 assert "--json" in result.output
721
722
723 def test_list_help_shows_json_flag() -> None:
724 result = runner.invoke(cli, ["snapshot", "list", "--help"])
725 assert result.exit_code == 0
726 assert "--json" in result.output
727
728
729 def test_show_help_mentions_text_flag() -> None:
730 result = runner.invoke(cli, ["snapshot", "show", "--help"])
731 assert result.exit_code == 0
732 assert "--text" in result.output
733
734
735 def test_export_help_shows_json_flag() -> None:
736 result = runner.invoke(cli, ["snapshot", "export", "--help"])
737 assert result.exit_code == 0
738 assert "--json" in result.output
739
740
741 def test_export_help_shows_format_choices() -> None:
742 result = runner.invoke(cli, ["snapshot", "export", "--help"])
743 assert result.exit_code == 0
744 assert "tar.gz" in result.output
745 assert "zip" in result.output
746
747
748 # ---------------------------------------------------------------------------
749 # Stress
750 # ---------------------------------------------------------------------------
751
752
753 def test_stress_200_snapshots_list(tmp_path: pathlib.Path) -> None:
754 _init_repo(tmp_path)
755 for i in range(200):
756 _write_snapshot(tmp_path, note=f"snap-{i}")
757 result = _invoke(["snapshot", "list", "--json", "-n", "200"], env=_env(tmp_path))
758 assert result.exit_code == 0
759 items: list[_ListItemOut] = json.loads(result.output)
760 assert len(items) == 200
761
762
763 def test_stress_500_file_snapshot(tmp_path: pathlib.Path) -> None:
764 _init_repo(tmp_path)
765 for i in range(500):
766 (tmp_path / f"f{i}.txt").write_text(f"data-{i}", encoding="utf-8")
767 result = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
768 assert result.exit_code == 0
769 data: _CreateOut = json.loads(result.output)
770 assert data["file_count"] >= 500
771
772 snap_id = data["snapshot_id"]
773 show_res = _invoke(["snapshot", "show", snap_id], env=_env(tmp_path))
774 assert show_res.exit_code == 0
775 show_data: _ShowOut = json.loads(show_res.output)
776 assert show_data["file_count"] >= 500
777 assert len(show_data["manifest"]) >= 500
778
779
780 def test_stress_concurrent_create(tmp_path: pathlib.Path) -> None:
781 _init_repo(tmp_path)
782 for i in range(10):
783 (tmp_path / f"cf{i}.txt").write_text(f"c{i}", encoding="utf-8")
784
785 errors: list[str] = []
786
787 def _create() -> None:
788 result = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
789 with _invoke_lock:
790 pass # lock already acquired inside _invoke
791 if result.exit_code != 0:
792 errors.append(f"exit_code={result.exit_code}")
793
794 threads = [threading.Thread(target=_create) for _ in range(5)]
795 for t in threads:
796 t.start()
797 for t in threads:
798 t.join()
799 assert errors == [], f"Concurrent create errors: {errors}"
800
801
802 def test_stress_concurrent_list(tmp_path: pathlib.Path) -> None:
803 _init_repo(tmp_path)
804 for i in range(5):
805 _write_snapshot(tmp_path, note=f"concurrent-{i}")
806
807 errors: list[str] = []
808
809 def _list() -> None:
810 result = _invoke(["snapshot", "list", "--json"], env=_env(tmp_path))
811 if result.exit_code != 0:
812 errors.append(f"exit_code={result.exit_code}")
813 else:
814 try:
815 data = json.loads(result.output)
816 if len(data) < 5:
817 errors.append(f"expected 5 items, got {len(data)}")
818 except Exception as exc:
819 errors.append(str(exc))
820
821 threads = [threading.Thread(target=_list) for _ in range(10)]
822 for t in threads:
823 t.start()
824 for t in threads:
825 t.join()
826 assert errors == [], f"Concurrent list errors: {errors}"
827
828
829 # ---------------------------------------------------------------------------
830 # Extended / Security / Stress tests for ``muse snapshot create``
831 # ---------------------------------------------------------------------------
832
833
834 class TestSnapshotCreateExtended:
835 """Unit, integration, and edge-case tests for ``muse snapshot create``."""
836
837 def test_create_help_contains_agent_quickstart(self) -> None:
838 result = runner.invoke(cli, ["snapshot", "create", "--help"])
839 assert result.exit_code == 0
840 assert "quickstart" in result.output.lower() or "muse snapshot create" in result.output
841
842 def test_create_help_contains_json_schema(self) -> None:
843 result = runner.invoke(cli, ["snapshot", "create", "--help"])
844 assert result.exit_code == 0
845 assert "snapshot_id" in result.output
846
847 def test_create_help_contains_exit_codes(self) -> None:
848 result = runner.invoke(cli, ["snapshot", "create", "--help"])
849 assert result.exit_code == 0
850 assert "exit code" in result.output.lower() or "0 —" in result.output
851
852 def test_create_j_alias(self, tmp_path: pathlib.Path) -> None:
853 """-j is an alias for --json."""
854 _init_repo(tmp_path)
855 _create_files(tmp_path, 2)
856 result = _invoke(["snapshot", "create", "-j"], env=_env(tmp_path))
857 assert result.exit_code == 0
858 data: _CreateOut = json.loads(result.output)
859 assert "snapshot_id" in data
860
861 def test_create_snapshot_id_is_64_hex(self, tmp_path: pathlib.Path) -> None:
862 """snapshot_id in JSON output is exactly 64 hex characters."""
863 _init_repo(tmp_path)
864 _create_files(tmp_path, 1)
865 result = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
866 assert result.exit_code == 0
867 data: _CreateOut = json.loads(result.output)
868 assert len(data["snapshot_id"]) == 64
869 assert all(c in "0123456789abcdef" for c in data["snapshot_id"])
870
871 def test_create_file_count_matches_actual(self, tmp_path: pathlib.Path) -> None:
872 """file_count in JSON output matches the number of files created."""
873 _init_repo(tmp_path)
874 _create_files(tmp_path, 7)
875 result = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
876 assert result.exit_code == 0
877 data: _CreateOut = json.loads(result.output)
878 assert data["file_count"] >= 7
879
880 def test_create_created_at_is_iso8601(self, tmp_path: pathlib.Path) -> None:
881 """created_at field is ISO-8601 format (contains 'T' separator)."""
882 _init_repo(tmp_path)
883 _create_files(tmp_path, 1)
884 result = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
885 assert result.exit_code == 0
886 data: _CreateOut = json.loads(result.output)
887 assert "T" in data["created_at"]
888
889 def test_create_note_empty_when_not_supplied(self, tmp_path: pathlib.Path) -> None:
890 """note is empty string in JSON when -m not passed."""
891 _init_repo(tmp_path)
892 _create_files(tmp_path, 1)
893 result = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
894 assert result.exit_code == 0
895 data: _CreateOut = json.loads(result.output)
896 assert data["note"] == ""
897
898 def test_create_note_persisted_in_json(self, tmp_path: pathlib.Path) -> None:
899 """note supplied via -m is reflected in JSON output."""
900 _init_repo(tmp_path)
901 _create_files(tmp_path, 1)
902 result = _invoke(["snapshot", "create", "-m", "checkpoint", "--json"], env=_env(tmp_path))
903 assert result.exit_code == 0
904 data: _CreateOut = json.loads(result.output)
905 assert data["note"] == "checkpoint"
906
907 def test_create_note_persists_to_list(self, tmp_path: pathlib.Path) -> None:
908 """note written via create is readable via list."""
909 _init_repo(tmp_path)
910 _create_files(tmp_path, 1)
911 _invoke(["snapshot", "create", "-m", "roundtrip"], env=_env(tmp_path))
912 list_result = _invoke(["snapshot", "list", "--json"], env=_env(tmp_path))
913 assert list_result.exit_code == 0
914 items: list[_ListItemOut] = json.loads(list_result.output)
915 assert any(i["note"] == "roundtrip" for i in items)
916
917 def test_create_note_persists_to_show(self, tmp_path: pathlib.Path) -> None:
918 """note written via create is readable via show."""
919 _init_repo(tmp_path)
920 _create_files(tmp_path, 1)
921 create_result = _invoke(
922 ["snapshot", "create", "-m", "showcheck", "--json"], env=_env(tmp_path)
923 )
924 snap_id = json.loads(create_result.output)["snapshot_id"]
925 show_result = _invoke(["snapshot", "show", snap_id], env=_env(tmp_path))
926 assert show_result.exit_code == 0
927 show_data: _ShowOut = json.loads(show_result.output)
928 assert show_data["note"] == "showcheck"
929
930 def test_create_text_output_shows_short_id(self, tmp_path: pathlib.Path) -> None:
931 """Text output contains a 12-char prefix of the snapshot_id."""
932 _init_repo(tmp_path)
933 _create_files(tmp_path, 1)
934 create_result = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
935 snap_id = json.loads(create_result.output)["snapshot_id"]
936 text_result = _invoke(["snapshot", "create"], env=_env(tmp_path))
937 # Each create call produces a new snapshot; just verify format
938 assert text_result.exit_code == 0
939 # Output should contain a hex-like prefix
940 assert any(c in "0123456789abcdef" for c in text_result.output)
941
942 def test_create_text_output_shows_note(self, tmp_path: pathlib.Path) -> None:
943 """Text output shows note label when -m is supplied."""
944 _init_repo(tmp_path)
945 _create_files(tmp_path, 1)
946 result = _invoke(["snapshot", "create", "-m", "my note"], env=_env(tmp_path))
947 assert result.exit_code == 0
948 assert "my note" in result.output
949
950 def test_create_text_output_no_note_line_when_empty(self, tmp_path: pathlib.Path) -> None:
951 """Text output has no 'Note:' line when -m is not passed."""
952 _init_repo(tmp_path)
953 _create_files(tmp_path, 1)
954 result = _invoke(["snapshot", "create"], env=_env(tmp_path))
955 assert result.exit_code == 0
956 assert "Note:" not in result.output
957
958 def test_create_json_compact_no_indent(self, tmp_path: pathlib.Path) -> None:
959 """JSON output is compact (no indentation)."""
960 _init_repo(tmp_path)
961 _create_files(tmp_path, 1)
962 result = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
963 assert result.exit_code == 0
964 # json.loads succeeds and the raw text has no indented lines
965 assert "\n " not in result.output.strip()
966
967 def test_create_repo_id_in_json(self, tmp_path: pathlib.Path) -> None:
968 """repo_id field is present and non-empty in JSON output."""
969 _init_repo(tmp_path)
970 _create_files(tmp_path, 1)
971 result = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
972 assert result.exit_code == 0
973 data: _CreateOut = json.loads(result.output)
974 assert data["repo_id"] == _REPO_ID
975
976 def test_create_idempotent_same_files_same_id(self, tmp_path: pathlib.Path) -> None:
977 """Two consecutive creates of the same working tree produce the same snapshot_id."""
978 _init_repo(tmp_path)
979 _create_files(tmp_path, 3)
980 r1 = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
981 r2 = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
982 assert r1.exit_code == 0
983 assert r2.exit_code == 0
984 assert json.loads(r1.output)["snapshot_id"] == json.loads(r2.output)["snapshot_id"]
985
986 def test_create_different_files_different_id(self, tmp_path: pathlib.Path) -> None:
987 """Adding a file between creates produces a different snapshot_id."""
988 _init_repo(tmp_path)
989 _create_files(tmp_path, 2)
990 r1 = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
991 (tmp_path / "extra.txt").write_text("extra", encoding="utf-8")
992 r2 = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
993 assert r1.exit_code == 0 and r2.exit_code == 0
994 assert json.loads(r1.output)["snapshot_id"] != json.loads(r2.output)["snapshot_id"]
995
996
997 class TestSnapshotCreateSecurity:
998 """Security tests for ``muse snapshot create``."""
999
1000 def test_create_ansi_note_stripped_in_text(self, tmp_path: pathlib.Path) -> None:
1001 """ANSI escape codes in note are stripped from text output."""
1002 _init_repo(tmp_path)
1003 _create_files(tmp_path, 1)
1004 evil_note = "\x1b[31mDanger\x1b[0m"
1005 result = _invoke(["snapshot", "create", "-m", evil_note], env=_env(tmp_path))
1006 assert result.exit_code == 0
1007 assert "\x1b[31m" not in result.output
1008
1009 def test_create_ansi_note_raw_in_json(self, tmp_path: pathlib.Path) -> None:
1010 """ANSI escape codes in note are preserved raw in JSON output (agent data)."""
1011 _init_repo(tmp_path)
1012 _create_files(tmp_path, 1)
1013 evil_note = "\x1b[31mDanger\x1b[0m"
1014 result = _invoke(["snapshot", "create", "-m", evil_note, "--json"], env=_env(tmp_path))
1015 assert result.exit_code == 0
1016 data: _CreateOut = json.loads(result.output)
1017 assert data["note"] == evil_note
1018
1019 def test_create_control_char_note_stripped_in_text(self, tmp_path: pathlib.Path) -> None:
1020 """CRLF and other control characters in note are stripped from text output."""
1021 _init_repo(tmp_path)
1022 _create_files(tmp_path, 1)
1023 evil_note = "good\r\ninjected line"
1024 result = _invoke(["snapshot", "create", "-m", evil_note], env=_env(tmp_path))
1025 assert result.exit_code == 0
1026 assert "\r" not in result.output
1027
1028 def test_create_very_long_note_no_crash(self, tmp_path: pathlib.Path) -> None:
1029 """A 10 000-character note does not crash the command."""
1030 _init_repo(tmp_path)
1031 _create_files(tmp_path, 1)
1032 long_note = "x" * 10_000
1033 result = _invoke(["snapshot", "create", "-m", long_note, "--json"], env=_env(tmp_path))
1034 assert result.exit_code == 0
1035 data: _CreateOut = json.loads(result.output)
1036 assert data["note"] == long_note
1037
1038 def test_create_path_traversal_chars_in_note_no_crash(self, tmp_path: pathlib.Path) -> None:
1039 """Path-traversal-like characters in note do not crash or escape output."""
1040 _init_repo(tmp_path)
1041 _create_files(tmp_path, 1)
1042 evil_note = "../../etc/passwd"
1043 result = _invoke(["snapshot", "create", "-m", evil_note, "--json"], env=_env(tmp_path))
1044 assert result.exit_code == 0
1045 data: _CreateOut = json.loads(result.output)
1046 assert data["note"] == evil_note
1047
1048 def test_create_snapshot_id_always_hex(self, tmp_path: pathlib.Path) -> None:
1049 """snapshot_id in text output is a safe hex substring with no control chars."""
1050 _init_repo(tmp_path)
1051 _create_files(tmp_path, 1)
1052 # Get snapshot_id from JSON, verify text output contains its prefix
1053 json_result = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
1054 snap_id = json.loads(json_result.output)["snapshot_id"]
1055 # Verify it is pure lowercase hex
1056 assert all(c in "0123456789abcdef" for c in snap_id)
1057 assert "\x1b" not in snap_id
1058 assert "\r" not in snap_id
1059
1060
1061 class TestSnapshotCreateStress:
1062 """Stress tests for ``muse snapshot create``."""
1063
1064 def test_create_1000_file_snapshot(self, tmp_path: pathlib.Path) -> None:
1065 """Snapshot of 1 000 files completes without error and reports correct count."""
1066 _init_repo(tmp_path)
1067 for i in range(1000):
1068 (tmp_path / f"f{i}.dat").write_text(f"data-{i}", encoding="utf-8")
1069 result = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
1070 assert result.exit_code == 0
1071 data: _CreateOut = json.loads(result.output)
1072 assert data["file_count"] >= 1000
1073
1074 def test_create_50_consecutive_snapshots(self, tmp_path: pathlib.Path) -> None:
1075 """50 consecutive creates all succeed and produce listable records."""
1076 _init_repo(tmp_path)
1077 _create_files(tmp_path, 5)
1078 for i in range(50):
1079 (tmp_path / f"extra_{i}.txt").write_text(f"v{i}", encoding="utf-8")
1080 r = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
1081 assert r.exit_code == 0, f"Failed on iteration {i}: {r.output}"
1082 list_result = _invoke(["snapshot", "list", "--json", "-n", "100"], env=_env(tmp_path))
1083 assert list_result.exit_code == 0
1084 items = json.loads(list_result.output)
1085 assert len(items) >= 50
1086
1087 def test_create_concurrent_write_safety(self, tmp_path: pathlib.Path) -> None:
1088 """Concurrent snapshot creates on the same repo do not corrupt the store."""
1089 _init_repo(tmp_path)
1090 for i in range(10):
1091 (tmp_path / f"cf{i}.txt").write_text(f"c{i}", encoding="utf-8")
1092
1093 from muse.core.store import write_snapshot
1094 from muse.core.snapshot import compute_snapshot_id
1095
1096 errors: list[str] = []
1097
1098 def _do_create() -> None:
1099 try:
1100 # Use core directly — CliRunner serializes via _invoke_lock.
1101 manifest = {f"cf{i}.txt": _sha(f"c{i}".encode()) for i in range(10)}
1102 snap_id = compute_snapshot_id(manifest)
1103 write_snapshot(tmp_path, SnapshotRecord(
1104 snapshot_id=snap_id, manifest=manifest, note="concurrent"
1105 ))
1106 except Exception as exc: # noqa: BLE001
1107 errors.append(str(exc))
1108
1109 threads = [threading.Thread(target=_do_create) for _ in range(20)]
1110 for t in threads:
1111 t.start()
1112 for t in threads:
1113 t.join()
1114 assert not errors, f"Concurrent create failures: {errors}"
1115
1116
1117 # ---------------------------------------------------------------------------
1118 # Extended / Security / Stress tests for ``muse snapshot list``
1119 # ---------------------------------------------------------------------------
1120
1121
1122 class TestSnapshotListExtended:
1123 """Unit, integration, and edge-case tests for ``muse snapshot list``."""
1124
1125 def test_list_help_contains_agent_quickstart(self) -> None:
1126 result = runner.invoke(cli, ["snapshot", "list", "--help"])
1127 assert result.exit_code == 0
1128 assert "quickstart" in result.output.lower() or "muse snapshot list" in result.output
1129
1130 def test_list_help_contains_json_schema(self) -> None:
1131 result = runner.invoke(cli, ["snapshot", "list", "--help"])
1132 assert result.exit_code == 0
1133 assert "snapshot_id" in result.output
1134
1135 def test_list_help_contains_exit_codes(self) -> None:
1136 result = runner.invoke(cli, ["snapshot", "list", "--help"])
1137 assert result.exit_code == 0
1138 assert "exit code" in result.output.lower() or "0 —" in result.output
1139
1140 def test_list_j_alias(self, tmp_path: pathlib.Path) -> None:
1141 """-j is an alias for --json."""
1142 _init_repo(tmp_path)
1143 _write_snapshot(tmp_path, note="alias-test")
1144 result = _invoke(["snapshot", "list", "-j"], env=_env(tmp_path))
1145 assert result.exit_code == 0
1146 items: list[_ListItemOut] = json.loads(result.output)
1147 assert len(items) == 1
1148
1149 def test_list_json_compact_no_indent(self, tmp_path: pathlib.Path) -> None:
1150 """JSON output is compact (no indentation)."""
1151 _init_repo(tmp_path)
1152 _write_snapshot(tmp_path)
1153 result = _invoke(["snapshot", "list", "--json"], env=_env(tmp_path))
1154 assert result.exit_code == 0
1155 assert "\n " not in result.output.strip()
1156
1157 def test_list_empty_returns_empty_array_json(self, tmp_path: pathlib.Path) -> None:
1158 """Empty snapshot store emits '[]' with --json."""
1159 _init_repo(tmp_path)
1160 result = _invoke(["snapshot", "list", "--json"], env=_env(tmp_path))
1161 assert result.exit_code == 0
1162 assert json.loads(result.output) == []
1163
1164 def test_list_empty_text_message(self, tmp_path: pathlib.Path) -> None:
1165 """Empty snapshot store prints a human message in text mode."""
1166 _init_repo(tmp_path)
1167 result = _invoke(["snapshot", "list"], env=_env(tmp_path))
1168 assert result.exit_code == 0
1169 assert "no snapshots" in result.output.lower()
1170
1171 def test_list_all_fields_present(self, tmp_path: pathlib.Path) -> None:
1172 """Every JSON item has snapshot_id, file_count, note, created_at."""
1173 _init_repo(tmp_path)
1174 _write_snapshot(tmp_path, note="fields")
1175 result = _invoke(["snapshot", "list", "--json"], env=_env(tmp_path))
1176 assert result.exit_code == 0
1177 item = json.loads(result.output)[0]
1178 assert "snapshot_id" in item
1179 assert "file_count" in item
1180 assert "note" in item
1181 assert "created_at" in item
1182
1183 def test_list_snapshot_id_is_64_hex(self, tmp_path: pathlib.Path) -> None:
1184 """snapshot_id in each JSON item is 64 hex chars."""
1185 _init_repo(tmp_path)
1186 _write_snapshot(tmp_path)
1187 result = _invoke(["snapshot", "list", "--json"], env=_env(tmp_path))
1188 assert result.exit_code == 0
1189 sid = json.loads(result.output)[0]["snapshot_id"]
1190 assert len(sid) == 64
1191 assert all(c in "0123456789abcdef" for c in sid)
1192
1193 def test_list_created_at_iso8601(self, tmp_path: pathlib.Path) -> None:
1194 """created_at field contains 'T' (ISO-8601 separator)."""
1195 _init_repo(tmp_path)
1196 _write_snapshot(tmp_path)
1197 result = _invoke(["snapshot", "list", "--json"], env=_env(tmp_path))
1198 assert result.exit_code == 0
1199 assert "T" in json.loads(result.output)[0]["created_at"]
1200
1201 def test_list_newest_first_order(self, tmp_path: pathlib.Path) -> None:
1202 """Multiple snapshots appear newest-first in JSON output."""
1203 import time as _time
1204 _init_repo(tmp_path)
1205 for i in range(5):
1206 _write_snapshot(tmp_path, note=f"snap-{i}", n_files=i + 1)
1207 _time.sleep(0.01)
1208 result = _invoke(["snapshot", "list", "-n", "10", "--json"], env=_env(tmp_path))
1209 assert result.exit_code == 0
1210 items: list[_ListItemOut] = json.loads(result.output)
1211 timestamps = [i["created_at"] for i in items]
1212 assert timestamps == sorted(timestamps, reverse=True)
1213
1214 def test_list_limit_caps_results(self, tmp_path: pathlib.Path) -> None:
1215 """--limit N returns at most N snapshots."""
1216 _init_repo(tmp_path)
1217 for i in range(10):
1218 _write_snapshot(tmp_path, note=f"s{i}")
1219 result = _invoke(["snapshot", "list", "-n", "3", "--json"], env=_env(tmp_path))
1220 assert result.exit_code == 0
1221 assert len(json.loads(result.output)) == 3
1222
1223 def test_list_limit_zero_exits_1(self, tmp_path: pathlib.Path) -> None:
1224 """--limit 0 is rejected with exit code 1."""
1225 _init_repo(tmp_path)
1226 result = _invoke(["snapshot", "list", "-n", "0"], env=_env(tmp_path))
1227 assert result.exit_code == 1
1228
1229 def test_list_limit_negative_exits_1(self, tmp_path: pathlib.Path) -> None:
1230 """--limit -1 is rejected with exit code 1."""
1231 _init_repo(tmp_path)
1232 result = _invoke(["snapshot", "list", "-n", "-1"], env=_env(tmp_path))
1233 assert result.exit_code == 1
1234
1235 def test_list_limit_error_mentions_limit(self, tmp_path: pathlib.Path) -> None:
1236 """Out-of-range --limit error output mentions 'limit'."""
1237 _init_repo(tmp_path)
1238 result = _invoke(["snapshot", "list", "-n", "0"], env=_env(tmp_path))
1239 assert result.exit_code == 1
1240 assert "limit" in result.output.lower()
1241
1242 def test_list_note_in_text_output(self, tmp_path: pathlib.Path) -> None:
1243 """Note label appears in text output when present."""
1244 _init_repo(tmp_path)
1245 _write_snapshot(tmp_path, note="my-label")
1246 result = _invoke(["snapshot", "list"], env=_env(tmp_path))
1247 assert result.exit_code == 0
1248 assert "my-label" in result.output
1249
1250 def test_list_text_shows_short_id(self, tmp_path: pathlib.Path) -> None:
1251 """Text output shows the 12-char prefix of snapshot_id."""
1252 _init_repo(tmp_path)
1253 snap_id = _write_snapshot(tmp_path)
1254 result = _invoke(["snapshot", "list"], env=_env(tmp_path))
1255 assert result.exit_code == 0
1256 assert snap_id[:12] in result.output
1257
1258 def test_list_file_count_in_json(self, tmp_path: pathlib.Path) -> None:
1259 """file_count in JSON matches the number of files in the snapshot."""
1260 _init_repo(tmp_path)
1261 _write_snapshot(tmp_path, n_files=7)
1262 result = _invoke(["snapshot", "list", "--json"], env=_env(tmp_path))
1263 assert result.exit_code == 0
1264 assert json.loads(result.output)[0]["file_count"] == 7
1265
1266
1267 class TestSnapshotListSecurity:
1268 """Security tests for ``muse snapshot list``."""
1269
1270 def test_list_ansi_note_stripped_in_text(self, tmp_path: pathlib.Path) -> None:
1271 """ANSI escape codes in note are stripped from text output."""
1272 _init_repo(tmp_path)
1273 _write_snapshot(tmp_path, note="\x1b[31mDanger\x1b[0m")
1274 result = _invoke(["snapshot", "list"], env=_env(tmp_path))
1275 assert result.exit_code == 0
1276 assert "\x1b[31m" not in result.output
1277
1278 def test_list_ansi_note_raw_in_json(self, tmp_path: pathlib.Path) -> None:
1279 """ANSI escape codes in note are preserved raw in JSON output."""
1280 _init_repo(tmp_path)
1281 evil = "\x1b[31mDanger\x1b[0m"
1282 _write_snapshot(tmp_path, note=evil)
1283 result = _invoke(["snapshot", "list", "--json"], env=_env(tmp_path))
1284 assert result.exit_code == 0
1285 assert json.loads(result.output)[0]["note"] == evil
1286
1287 def test_list_control_char_note_stripped_in_text(self, tmp_path: pathlib.Path) -> None:
1288 """CRLF in note is stripped from text output."""
1289 _init_repo(tmp_path)
1290 _write_snapshot(tmp_path, note="good\r\ninjected")
1291 result = _invoke(["snapshot", "list"], env=_env(tmp_path))
1292 assert result.exit_code == 0
1293 assert "\r" not in result.output
1294
1295 def test_list_symlink_in_snapshots_dir_skipped(self, tmp_path: pathlib.Path) -> None:
1296 """A symlink inside .muse/snapshots/ is skipped, not followed."""
1297 _init_repo(tmp_path)
1298 _write_snapshot(tmp_path, note="real")
1299 snaps_dir = tmp_path / ".muse" / "snapshots"
1300 fake = snaps_dir / ("aa" * 32 + ".msgpack")
1301 fake.symlink_to("/etc/passwd")
1302 result = _invoke(["snapshot", "list", "--json"], env=_env(tmp_path))
1303 assert result.exit_code == 0
1304 items: list[_ListItemOut] = json.loads(result.output)
1305 assert len(items) == 1
1306 assert items[0]["note"] == "real"
1307
1308 def test_list_snapshot_id_prefix_in_text_is_safe_hex(self, tmp_path: pathlib.Path) -> None:
1309 """12-char snapshot_id prefix in text output contains only hex chars."""
1310 _init_repo(tmp_path)
1311 snap_id = _write_snapshot(tmp_path)
1312 result = _invoke(["snapshot", "list"], env=_env(tmp_path))
1313 assert result.exit_code == 0
1314 assert snap_id[:12] in result.output
1315 assert "\x1b" not in result.output
1316
1317 def test_list_very_long_note_no_crash(self, tmp_path: pathlib.Path) -> None:
1318 """A 10 000-character note does not crash list."""
1319 _init_repo(tmp_path)
1320 _write_snapshot(tmp_path, note="x" * 10_000)
1321 result = _invoke(["snapshot", "list", "--json"], env=_env(tmp_path))
1322 assert result.exit_code == 0
1323 assert json.loads(result.output)[0]["note"] == "x" * 10_000
1324
1325
1326 class TestSnapshotListStress:
1327 """Stress tests for ``muse snapshot list``."""
1328
1329 def test_list_1000_snapshots(self, tmp_path: pathlib.Path) -> None:
1330 """Listing 1 000 snapshots with --limit 1000 returns all 1 000."""
1331 _init_repo(tmp_path)
1332 for i in range(1000):
1333 _write_snapshot(tmp_path, note=f"s{i}")
1334 result = _invoke(["snapshot", "list", "-n", "1000", "--json"], env=_env(tmp_path))
1335 assert result.exit_code == 0
1336 assert len(json.loads(result.output)) == 1000
1337
1338 def test_list_default_limit_caps_at_20(self, tmp_path: pathlib.Path) -> None:
1339 """Default --limit of 20 caps a 50-snapshot store at 20 results."""
1340 _init_repo(tmp_path)
1341 for i in range(50):
1342 _write_snapshot(tmp_path, note=f"s{i}")
1343 result = _invoke(["snapshot", "list", "--json"], env=_env(tmp_path))
1344 assert result.exit_code == 0
1345 assert len(json.loads(result.output)) == 20
1346
1347 def test_list_concurrent_reads_safe(self, tmp_path: pathlib.Path) -> None:
1348 """Concurrent _list_all_snapshots core calls on the same repo do not crash."""
1349 _init_repo(tmp_path)
1350 for i in range(10):
1351 _write_snapshot(tmp_path, note=f"c{i}")
1352 errors: list[str] = []
1353
1354 def _do_list() -> None:
1355 try:
1356 records = _list_all_snapshots(tmp_path)
1357 assert len(records) == 10
1358 except Exception as exc: # noqa: BLE001
1359 errors.append(str(exc))
1360
1361 threads = [threading.Thread(target=_do_list) for _ in range(15)]
1362 for t in threads:
1363 t.start()
1364 for t in threads:
1365 t.join()
1366 assert not errors, f"Concurrent failures: {errors}"
1367
1368
1369 # ---------------------------------------------------------------------------
1370 # Extended / Security / Stress tests for ``muse snapshot show``
1371 # ---------------------------------------------------------------------------
1372
1373
1374 class TestSnapshotShowExtended:
1375 """Unit, integration, and edge-case tests for ``muse snapshot show``."""
1376
1377 def test_show_help_contains_agent_quickstart(self) -> None:
1378 result = runner.invoke(cli, ["snapshot", "show", "--help"])
1379 assert result.exit_code == 0
1380 assert "quickstart" in result.output.lower() or "muse snapshot show" in result.output
1381
1382 def test_show_help_contains_json_schema(self) -> None:
1383 result = runner.invoke(cli, ["snapshot", "show", "--help"])
1384 assert result.exit_code == 0
1385 assert "snapshot_id" in result.output and "manifest" in result.output
1386
1387 def test_show_help_contains_exit_codes(self) -> None:
1388 result = runner.invoke(cli, ["snapshot", "show", "--help"])
1389 assert result.exit_code == 0
1390 assert "exit code" in result.output.lower() or "0 —" in result.output
1391
1392 def test_show_help_says_default_is_json(self) -> None:
1393 result = runner.invoke(cli, ["snapshot", "show", "--help"])
1394 assert result.exit_code == 0
1395 assert "json" in result.output.lower()
1396
1397 def test_show_default_is_json_no_flag_needed(self, tmp_path: pathlib.Path) -> None:
1398 """show with no flags emits valid JSON."""
1399 _init_repo(tmp_path)
1400 snap_id = _write_snapshot(tmp_path, note="default-json")
1401 result = _invoke(["snapshot", "show", snap_id], env=_env(tmp_path))
1402 assert result.exit_code == 0
1403 data: _ShowOut = json.loads(result.output)
1404 assert data["snapshot_id"] == snap_id
1405
1406 def test_show_json_compact_no_indent(self, tmp_path: pathlib.Path) -> None:
1407 """JSON output is compact (no indentation)."""
1408 _init_repo(tmp_path)
1409 snap_id = _write_snapshot(tmp_path)
1410 result = _invoke(["snapshot", "show", snap_id], env=_env(tmp_path))
1411 assert result.exit_code == 0
1412 assert "\n " not in result.output.strip()
1413
1414 def test_show_json_all_fields(self, tmp_path: pathlib.Path) -> None:
1415 """JSON output contains all five required fields."""
1416 _init_repo(tmp_path)
1417 snap_id = _write_snapshot(tmp_path, note="fields-check", n_files=3)
1418 result = _invoke(["snapshot", "show", snap_id], env=_env(tmp_path))
1419 assert result.exit_code == 0
1420 data: _ShowOut = json.loads(result.output)
1421 assert data["snapshot_id"] == snap_id
1422 assert "created_at" in data
1423 assert data["file_count"] == 3
1424 assert data["note"] == "fields-check"
1425 assert isinstance(data["manifest"], dict)
1426
1427 def test_show_manifest_sorted_alphabetically(self, tmp_path: pathlib.Path) -> None:
1428 """manifest keys in JSON output are sorted alphabetically."""
1429 _init_repo(tmp_path)
1430 manifest = {f"z_file_{i}.txt": _sha(f"z{i}".encode()) for i in range(5)}
1431 manifest.update({f"a_file_{i}.txt": _sha(f"a{i}".encode()) for i in range(5)})
1432 snap_id = compute_snapshot_id(manifest)
1433 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
1434 result = _invoke(["snapshot", "show", snap_id], env=_env(tmp_path))
1435 assert result.exit_code == 0
1436 data: _ShowOut = json.loads(result.output)
1437 keys = list(data["manifest"].keys())
1438 assert keys == sorted(keys)
1439
1440 def test_show_prefix_resolves_short_id(self, tmp_path: pathlib.Path) -> None:
1441 """A 12-char prefix resolves to the full snapshot."""
1442 _init_repo(tmp_path)
1443 snap_id = _write_snapshot(tmp_path, note="prefix-test")
1444 result = _invoke(["snapshot", "show", snap_id[:12]], env=_env(tmp_path))
1445 assert result.exit_code == 0
1446 data: _ShowOut = json.loads(result.output)
1447 assert data["snapshot_id"] == snap_id
1448
1449 def test_show_not_found_exits_1(self, tmp_path: pathlib.Path) -> None:
1450 """Unknown snapshot ID exits with code 1."""
1451 _init_repo(tmp_path)
1452 result = _invoke(["snapshot", "show", "deadbeef"], env=_env(tmp_path))
1453 assert result.exit_code == 1
1454
1455 def test_show_not_found_error_to_stderr(self, tmp_path: pathlib.Path) -> None:
1456 """Not-found message goes to stderr (captured in output by CliRunner)."""
1457 _init_repo(tmp_path)
1458 result = _invoke(["snapshot", "show", "deadbeef"], env=_env(tmp_path))
1459 assert result.exit_code != 0
1460 assert "not found" in result.output.lower()
1461
1462 def test_show_text_flag_human_readable(self, tmp_path: pathlib.Path) -> None:
1463 """--text emits human-readable output (not JSON)."""
1464 _init_repo(tmp_path)
1465 snap_id = _write_snapshot(tmp_path, note="text-mode")
1466 result = _invoke(["snapshot", "show", snap_id, "--text"], env=_env(tmp_path))
1467 assert result.exit_code == 0
1468 # Text output starts with "snapshot_id:" label, not a JSON brace
1469 assert not result.output.strip().startswith("{")
1470 assert "snapshot_id:" in result.output
1471
1472 def test_show_text_includes_note(self, tmp_path: pathlib.Path) -> None:
1473 """--text output includes the note label."""
1474 _init_repo(tmp_path)
1475 snap_id = _write_snapshot(tmp_path, note="my-note")
1476 result = _invoke(["snapshot", "show", snap_id, "--text"], env=_env(tmp_path))
1477 assert result.exit_code == 0
1478 assert "my-note" in result.output
1479
1480 def test_show_text_lists_files(self, tmp_path: pathlib.Path) -> None:
1481 """--text output lists file names from the manifest."""
1482 _init_repo(tmp_path)
1483 snap_id = _write_snapshot(tmp_path, n_files=3)
1484 result = _invoke(["snapshot", "show", snap_id, "--text"], env=_env(tmp_path))
1485 assert result.exit_code == 0
1486 assert "file_0.txt" in result.output
1487
1488 def test_show_file_count_matches_manifest(self, tmp_path: pathlib.Path) -> None:
1489 """file_count in JSON equals the number of keys in manifest."""
1490 _init_repo(tmp_path)
1491 snap_id = _write_snapshot(tmp_path, n_files=9)
1492 result = _invoke(["snapshot", "show", snap_id], env=_env(tmp_path))
1493 assert result.exit_code == 0
1494 data: _ShowOut = json.loads(result.output)
1495 assert data["file_count"] == len(data["manifest"])
1496
1497 def test_show_created_at_iso8601(self, tmp_path: pathlib.Path) -> None:
1498 """created_at field is ISO-8601 (contains 'T' separator)."""
1499 _init_repo(tmp_path)
1500 snap_id = _write_snapshot(tmp_path)
1501 result = _invoke(["snapshot", "show", snap_id], env=_env(tmp_path))
1502 assert result.exit_code == 0
1503 assert "T" in json.loads(result.output)["created_at"]
1504
1505 def test_show_note_empty_string_when_not_set(self, tmp_path: pathlib.Path) -> None:
1506 """note is empty string in JSON when not supplied at create time."""
1507 _init_repo(tmp_path)
1508 snap_id = _write_snapshot(tmp_path, note="")
1509 result = _invoke(["snapshot", "show", snap_id], env=_env(tmp_path))
1510 assert result.exit_code == 0
1511 assert json.loads(result.output)["note"] == ""
1512
1513 def test_show_text_no_note_line_when_empty(self, tmp_path: pathlib.Path) -> None:
1514 """--text output has no 'note:' line when note is empty."""
1515 _init_repo(tmp_path)
1516 snap_id = _write_snapshot(tmp_path, note="")
1517 result = _invoke(["snapshot", "show", snap_id, "--text"], env=_env(tmp_path))
1518 assert result.exit_code == 0
1519 assert "note:" not in result.output.lower()
1520
1521
1522 class TestSnapshotShowSecurity:
1523 """Security tests for ``muse snapshot show``."""
1524
1525 def test_show_ansi_note_stripped_in_text(self, tmp_path: pathlib.Path) -> None:
1526 """ANSI in note is stripped from --text output."""
1527 _init_repo(tmp_path)
1528 evil = "\x1b[31mDanger\x1b[0m"
1529 snap_id = _write_snapshot(tmp_path, note=evil)
1530 result = _invoke(["snapshot", "show", snap_id, "--text"], env=_env(tmp_path))
1531 assert result.exit_code == 0
1532 assert "\x1b[31m" not in result.output
1533
1534 def test_show_ansi_note_raw_in_json(self, tmp_path: pathlib.Path) -> None:
1535 """ANSI in note is preserved raw in JSON output (agent data)."""
1536 _init_repo(tmp_path)
1537 evil = "\x1b[31mDanger\x1b[0m"
1538 snap_id = _write_snapshot(tmp_path, note=evil)
1539 result = _invoke(["snapshot", "show", snap_id], env=_env(tmp_path))
1540 assert result.exit_code == 0
1541 assert json.loads(result.output)["note"] == evil
1542
1543 def test_show_ansi_rel_path_stripped_in_text(self, tmp_path: pathlib.Path) -> None:
1544 """ANSI in a manifest path is stripped from --text output."""
1545 _init_repo(tmp_path)
1546 evil_path = "\x1b[32msrc/evil.py\x1b[0m"
1547 manifest = {evil_path: _sha(b"evil")}
1548 snap_id = compute_snapshot_id(manifest)
1549 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
1550 result = _invoke(["snapshot", "show", snap_id, "--text"], env=_env(tmp_path))
1551 assert result.exit_code == 0
1552 assert "\x1b[32m" not in result.output
1553
1554 def test_show_control_char_note_stripped_in_text(self, tmp_path: pathlib.Path) -> None:
1555 """CRLF in note is stripped from --text output."""
1556 _init_repo(tmp_path)
1557 snap_id = _write_snapshot(tmp_path, note="good\r\ninjected")
1558 result = _invoke(["snapshot", "show", snap_id, "--text"], env=_env(tmp_path))
1559 assert result.exit_code == 0
1560 assert "\r" not in result.output
1561
1562 def test_show_not_found_id_sanitized_in_error(self, tmp_path: pathlib.Path) -> None:
1563 """ANSI in a not-found snapshot ID is stripped from the error message."""
1564 _init_repo(tmp_path)
1565 evil_id = "\x1b[31mdeadbeef\x1b[0m"
1566 result = _invoke(["snapshot", "show", evil_id], env=_env(tmp_path))
1567 assert result.exit_code != 0
1568 assert "\x1b[31m" not in result.output
1569
1570 def test_show_symlink_in_prefix_scan_skipped(self, tmp_path: pathlib.Path) -> None:
1571 """A symlink in the snapshots dir is skipped during prefix scan."""
1572 _init_repo(tmp_path)
1573 real_id = _write_snapshot(tmp_path, note="real")
1574 snaps_dir = tmp_path / ".muse" / "snapshots"
1575 # Plant a symlink whose name starts with the same prefix as real_id
1576 fake = snaps_dir / (real_id[:4] + "f" * 60 + ".msgpack")
1577 fake.symlink_to("/etc/passwd")
1578 # Full ID lookup should still work
1579 result = _invoke(["snapshot", "show", real_id], env=_env(tmp_path))
1580 assert result.exit_code == 0
1581 assert json.loads(result.output)["note"] == "real"
1582
1583
1584 class TestSnapshotShowStress:
1585 """Stress tests for ``muse snapshot show``."""
1586
1587 def test_show_500_file_manifest_json(self, tmp_path: pathlib.Path) -> None:
1588 """show returns all 500 manifest entries for a large snapshot."""
1589 _init_repo(tmp_path)
1590 manifest = {f"f{i:04d}.dat": _sha(f"data{i}".encode()) for i in range(500)}
1591 snap_id = compute_snapshot_id(manifest)
1592 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
1593 result = _invoke(["snapshot", "show", snap_id], env=_env(tmp_path))
1594 assert result.exit_code == 0
1595 data: _ShowOut = json.loads(result.output)
1596 assert data["file_count"] == 500
1597 assert len(data["manifest"]) == 500
1598
1599 def test_show_50_consecutive_shows(self, tmp_path: pathlib.Path) -> None:
1600 """50 consecutive show calls on different snapshots all succeed."""
1601 _init_repo(tmp_path)
1602 ids = [_write_snapshot(tmp_path, note=f"snap-{i}") for i in range(50)]
1603 for snap_id in ids:
1604 r = _invoke(["snapshot", "show", snap_id], env=_env(tmp_path))
1605 assert r.exit_code == 0, f"Failed for {snap_id[:12]}: {r.output}"
1606 assert json.loads(r.output)["snapshot_id"] == snap_id
1607
1608 def test_show_concurrent_reads_safe(self, tmp_path: pathlib.Path) -> None:
1609 """Concurrent _resolve_snapshot calls on the same snapshot do not crash."""
1610 _init_repo(tmp_path)
1611 snap_id = _write_snapshot(tmp_path, note="concurrent", n_files=5)
1612 errors: list[str] = []
1613
1614 def _do_show() -> None:
1615 try:
1616 rec = _resolve_snapshot(tmp_path, snap_id)
1617 assert rec is not None
1618 assert rec.snapshot_id == snap_id
1619 assert rec.note == "concurrent"
1620 except Exception as exc: # noqa: BLE001
1621 errors.append(str(exc))
1622
1623 threads = [threading.Thread(target=_do_show) for _ in range(20)]
1624 for t in threads:
1625 t.start()
1626 for t in threads:
1627 t.join()
1628 assert not errors, f"Concurrent failures: {errors}"
1629
1630
1631 # ---------------------------------------------------------------------------
1632 # Extended / Security / Stress tests for ``muse snapshot export``
1633 # ---------------------------------------------------------------------------
1634
1635
1636 class TestSnapshotExportExtended:
1637 """Unit, integration, and edge-case tests for ``muse snapshot export``."""
1638
1639 def test_export_help_contains_agent_quickstart(self) -> None:
1640 result = runner.invoke(cli, ["snapshot", "export", "--help"])
1641 assert result.exit_code == 0
1642 assert "quickstart" in result.output.lower() or "muse snapshot export" in result.output
1643
1644 def test_export_help_contains_json_schema(self) -> None:
1645 result = runner.invoke(cli, ["snapshot", "export", "--help"])
1646 assert result.exit_code == 0
1647 assert "size_bytes" in result.output
1648
1649 def test_export_help_contains_exit_codes(self) -> None:
1650 result = runner.invoke(cli, ["snapshot", "export", "--help"])
1651 assert result.exit_code == 0
1652 assert "exit code" in result.output.lower() or "0 —" in result.output
1653
1654 def test_export_j_alias(self, tmp_path: pathlib.Path) -> None:
1655 """-j is an alias for --json."""
1656 _init_repo(tmp_path)
1657 _create_files(tmp_path, 2)
1658 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
1659 snap_id: str = json.loads(create_res.output)["snapshot_id"]
1660 out_file = tmp_path / "alias.tar.gz"
1661 result = _invoke(
1662 ["snapshot", "export", snap_id, "--output", str(out_file), "-j"],
1663 env=_env(tmp_path),
1664 )
1665 assert result.exit_code == 0
1666 data: _ExportOut = json.loads(result.output)
1667 assert data["snapshot_id"] == snap_id
1668
1669 def test_export_tar_gz_default_format(self, tmp_path: pathlib.Path) -> None:
1670 """Default format is tar.gz; JSON reports format correctly."""
1671 _init_repo(tmp_path)
1672 _create_files(tmp_path, 1)
1673 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
1674 snap_id: str = json.loads(create_res.output)["snapshot_id"]
1675 out_file = tmp_path / "out.tar.gz"
1676 result = _invoke(
1677 ["snapshot", "export", snap_id, "--output", str(out_file), "--json"],
1678 env=_env(tmp_path),
1679 )
1680 assert result.exit_code == 0
1681 assert json.loads(result.output)["format"] == "tar.gz"
1682 assert tarfile.is_tarfile(str(out_file))
1683
1684 def test_export_zip_format(self, tmp_path: pathlib.Path) -> None:
1685 """--format zip writes a valid zip archive."""
1686 _init_repo(tmp_path)
1687 _create_files(tmp_path, 2)
1688 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
1689 snap_id: str = json.loads(create_res.output)["snapshot_id"]
1690 out_file = tmp_path / "out.zip"
1691 result = _invoke(
1692 ["snapshot", "export", snap_id, "-f", "zip", "--output", str(out_file), "--json"],
1693 env=_env(tmp_path),
1694 )
1695 assert result.exit_code == 0
1696 assert json.loads(result.output)["format"] == "zip"
1697 assert zipfile.is_zipfile(str(out_file))
1698
1699 def test_export_json_all_fields_present(self, tmp_path: pathlib.Path) -> None:
1700 """JSON output contains all five required fields."""
1701 _init_repo(tmp_path)
1702 _create_files(tmp_path, 2)
1703 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
1704 snap_id: str = json.loads(create_res.output)["snapshot_id"]
1705 out_file = tmp_path / "fields.tar.gz"
1706 result = _invoke(
1707 ["snapshot", "export", snap_id, "--output", str(out_file), "--json"],
1708 env=_env(tmp_path),
1709 )
1710 assert result.exit_code == 0
1711 data: _ExportOut = json.loads(result.output)
1712 assert "snapshot_id" in data
1713 assert "output" in data
1714 assert "format" in data
1715 assert "file_count" in data
1716 assert "size_bytes" in data
1717
1718 def test_export_json_compact_no_indent(self, tmp_path: pathlib.Path) -> None:
1719 """JSON output is compact (no indentation)."""
1720 _init_repo(tmp_path)
1721 _create_files(tmp_path, 1)
1722 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
1723 snap_id: str = json.loads(create_res.output)["snapshot_id"]
1724 out_file = tmp_path / "compact.tar.gz"
1725 result = _invoke(
1726 ["snapshot", "export", snap_id, "--output", str(out_file), "--json"],
1727 env=_env(tmp_path),
1728 )
1729 assert result.exit_code == 0
1730 assert "\n " not in result.output.strip()
1731
1732 def test_export_size_bytes_positive(self, tmp_path: pathlib.Path) -> None:
1733 """size_bytes > 0 for a non-empty archive."""
1734 _init_repo(tmp_path)
1735 _create_files(tmp_path, 3)
1736 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
1737 snap_id: str = json.loads(create_res.output)["snapshot_id"]
1738 out_file = tmp_path / "size.tar.gz"
1739 result = _invoke(
1740 ["snapshot", "export", snap_id, "--output", str(out_file), "--json"],
1741 env=_env(tmp_path),
1742 )
1743 assert result.exit_code == 0
1744 assert json.loads(result.output)["size_bytes"] > 0
1745
1746 def test_export_file_count_matches(self, tmp_path: pathlib.Path) -> None:
1747 """file_count in JSON matches number of files created."""
1748 _init_repo(tmp_path)
1749 _create_files(tmp_path, 5)
1750 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
1751 snap_id: str = json.loads(create_res.output)["snapshot_id"]
1752 out_file = tmp_path / "count.tar.gz"
1753 result = _invoke(
1754 ["snapshot", "export", snap_id, "--output", str(out_file), "--json"],
1755 env=_env(tmp_path),
1756 )
1757 assert result.exit_code == 0
1758 assert json.loads(result.output)["file_count"] >= 5
1759
1760 def test_export_output_path_in_json(self, tmp_path: pathlib.Path) -> None:
1761 """output field in JSON matches the --output argument."""
1762 _init_repo(tmp_path)
1763 _create_files(tmp_path, 1)
1764 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
1765 snap_id: str = json.loads(create_res.output)["snapshot_id"]
1766 out_file = tmp_path / "myarchive.tar.gz"
1767 result = _invoke(
1768 ["snapshot", "export", snap_id, "--output", str(out_file), "--json"],
1769 env=_env(tmp_path),
1770 )
1771 assert result.exit_code == 0
1772 assert json.loads(result.output)["output"] == str(out_file)
1773
1774 def test_export_archive_actually_created(self, tmp_path: pathlib.Path) -> None:
1775 """The archive file is present on disk after export."""
1776 _init_repo(tmp_path)
1777 _create_files(tmp_path, 2)
1778 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
1779 snap_id: str = json.loads(create_res.output)["snapshot_id"]
1780 out_file = tmp_path / "present.tar.gz"
1781 result = _invoke(
1782 ["snapshot", "export", snap_id, "--output", str(out_file)],
1783 env=_env(tmp_path),
1784 )
1785 assert result.exit_code == 0
1786 assert out_file.exists()
1787
1788 def test_export_prefix_nests_files_in_tar(self, tmp_path: pathlib.Path) -> None:
1789 """--prefix nests all files under a directory inside the tar archive."""
1790 _init_repo(tmp_path)
1791 _create_files(tmp_path, 2)
1792 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
1793 snap_id: str = json.loads(create_res.output)["snapshot_id"]
1794 out_file = tmp_path / "prefixed.tar.gz"
1795 result = _invoke(
1796 ["snapshot", "export", snap_id, "--output", str(out_file), "--prefix", "mydir"],
1797 env=_env(tmp_path),
1798 )
1799 assert result.exit_code == 0
1800 with tarfile.open(str(out_file)) as tf:
1801 names = tf.getnames()
1802 assert all(n.startswith("mydir/") for n in names)
1803
1804 def test_export_not_found_exits_1(self, tmp_path: pathlib.Path) -> None:
1805 """Unknown snapshot ID exits with code 1."""
1806 _init_repo(tmp_path)
1807 out_file = tmp_path / "nope.tar.gz"
1808 result = _invoke(
1809 ["snapshot", "export", "deadbeef", "--output", str(out_file)],
1810 env=_env(tmp_path),
1811 )
1812 assert result.exit_code == 1
1813
1814 def test_export_prefix_scan_resolves_short_id(self, tmp_path: pathlib.Path) -> None:
1815 """A 12-char prefix resolves to the correct snapshot for export."""
1816 _init_repo(tmp_path)
1817 _create_files(tmp_path, 1)
1818 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
1819 snap_id: str = json.loads(create_res.output)["snapshot_id"]
1820 out_file = tmp_path / "prefix_resolve.tar.gz"
1821 result = _invoke(
1822 ["snapshot", "export", snap_id[:12], "--output", str(out_file), "--json"],
1823 env=_env(tmp_path),
1824 )
1825 assert result.exit_code == 0
1826 assert json.loads(result.output)["snapshot_id"] == snap_id
1827
1828 def test_export_snapshot_id_in_json_is_full_hex(self, tmp_path: pathlib.Path) -> None:
1829 """snapshot_id in JSON is the full 64-char hex ID."""
1830 _init_repo(tmp_path)
1831 _create_files(tmp_path, 1)
1832 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
1833 snap_id: str = json.loads(create_res.output)["snapshot_id"]
1834 out_file = tmp_path / "id_check.tar.gz"
1835 result = _invoke(
1836 ["snapshot", "export", snap_id, "--output", str(out_file), "--json"],
1837 env=_env(tmp_path),
1838 )
1839 assert result.exit_code == 0
1840 sid = json.loads(result.output)["snapshot_id"]
1841 assert len(sid) == 64
1842 assert all(c in "0123456789abcdef" for c in sid)
1843
1844 def test_export_text_output_mentions_path(self, tmp_path: pathlib.Path) -> None:
1845 """Text output mentions the archive filename."""
1846 _init_repo(tmp_path)
1847 _create_files(tmp_path, 1)
1848 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
1849 snap_id: str = json.loads(create_res.output)["snapshot_id"]
1850 out_file = tmp_path / "mentioned.tar.gz"
1851 result = _invoke(
1852 ["snapshot", "export", snap_id, "--output", str(out_file)],
1853 env=_env(tmp_path),
1854 )
1855 assert result.exit_code == 0
1856 assert "mentioned.tar.gz" in result.output
1857
1858
1859 class TestSnapshotExportSecurity:
1860 """Security tests for ``muse snapshot export``."""
1861
1862 def test_export_not_found_id_sanitized(self, tmp_path: pathlib.Path) -> None:
1863 """ANSI in a not-found snapshot ID is stripped from the error message."""
1864 _init_repo(tmp_path)
1865 evil_id = "\x1b[31mdeadbeef\x1b[0m"
1866 out_file = tmp_path / "nope.tar.gz"
1867 result = _invoke(
1868 ["snapshot", "export", evil_id, "--output", str(out_file)],
1869 env=_env(tmp_path),
1870 )
1871 assert result.exit_code != 0
1872 assert "\x1b[31m" not in result.output
1873
1874 def test_export_text_output_no_ansi(self, tmp_path: pathlib.Path) -> None:
1875 """Normal text output from export contains no ANSI escape sequences."""
1876 _init_repo(tmp_path)
1877 _create_files(tmp_path, 1)
1878 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
1879 snap_id: str = json.loads(create_res.output)["snapshot_id"]
1880 out_file = tmp_path / "clean.tar.gz"
1881 result = _invoke(
1882 ["snapshot", "export", snap_id, "--output", str(out_file)],
1883 env=_env(tmp_path),
1884 )
1885 assert result.exit_code == 0
1886 assert "\x1b[" not in result.output
1887
1888 def test_export_zip_slip_dotdot_skipped(self, tmp_path: pathlib.Path) -> None:
1889 """A manifest entry with '..' segments is skipped (zip-slip guard)."""
1890 _init_repo(tmp_path)
1891 evil_path = "../../../etc/passwd"
1892 obj_data = b"evil content"
1893 obj_id = _sha(obj_data)
1894 write_object(tmp_path, obj_id, obj_data)
1895 manifest = {evil_path: obj_id}
1896 snap_id = compute_snapshot_id(manifest)
1897 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
1898 out_file = tmp_path / "slip.tar.gz"
1899 result = _invoke(
1900 ["snapshot", "export", snap_id, "--output", str(out_file), "--json"],
1901 env=_env(tmp_path),
1902 )
1903 assert result.exit_code == 0
1904 assert json.loads(result.output)["file_count"] == 0
1905
1906 def test_export_zip_slip_absolute_skipped(self, tmp_path: pathlib.Path) -> None:
1907 """A manifest entry with an absolute path is skipped (zip-slip guard)."""
1908 _init_repo(tmp_path)
1909 obj_data = b"absolute evil"
1910 obj_id = _sha(obj_data)
1911 write_object(tmp_path, obj_id, obj_data)
1912 manifest = {"/etc/passwd": obj_id}
1913 snap_id = compute_snapshot_id(manifest)
1914 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
1915 out_file = tmp_path / "abs.tar.gz"
1916 result = _invoke(
1917 ["snapshot", "export", snap_id, "--output", str(out_file), "--json"],
1918 env=_env(tmp_path),
1919 )
1920 assert result.exit_code == 0
1921 assert json.loads(result.output)["file_count"] == 0
1922
1923 def test_export_prefix_dotdot_skipped(self, tmp_path: pathlib.Path) -> None:
1924 """A --prefix containing '..' causes all entries to be skipped."""
1925 _init_repo(tmp_path)
1926 _create_files(tmp_path, 1)
1927 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
1928 snap_id: str = json.loads(create_res.output)["snapshot_id"]
1929 out_file = tmp_path / "dotdot.tar.gz"
1930 result = _invoke(
1931 ["snapshot", "export", snap_id, "--output", str(out_file),
1932 "--prefix", "../escape", "--json"],
1933 env=_env(tmp_path),
1934 )
1935 assert result.exit_code == 0
1936 assert json.loads(result.output)["file_count"] == 0
1937
1938 def test_export_missing_object_skipped(self, tmp_path: pathlib.Path) -> None:
1939 """A manifest entry whose object is missing from the store is skipped."""
1940 _init_repo(tmp_path)
1941 ghost_id = _sha(b"ghost")
1942 manifest = {"ghost.txt": ghost_id}
1943 snap_id = compute_snapshot_id(manifest)
1944 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
1945 out_file = tmp_path / "ghost.tar.gz"
1946 result = _invoke(
1947 ["snapshot", "export", snap_id, "--output", str(out_file), "--json"],
1948 env=_env(tmp_path),
1949 )
1950 assert result.exit_code == 0
1951 assert json.loads(result.output)["file_count"] == 0
1952
1953
1954 class TestSnapshotExportStress:
1955 """Stress tests for ``muse snapshot export``."""
1956
1957 def test_export_500_file_tar_gz(self, tmp_path: pathlib.Path) -> None:
1958 """Export of a 500-file snapshot produces a valid tar.gz with all files."""
1959 _init_repo(tmp_path)
1960 manifest: Manifest = {}
1961 for i in range(500):
1962 data = f"content-{i}".encode()
1963 obj_id = _sha(data)
1964 write_object(tmp_path, obj_id, data)
1965 manifest[f"f{i:04d}.dat"] = obj_id
1966 snap_id = compute_snapshot_id(manifest)
1967 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
1968 out_file = tmp_path / "big.tar.gz"
1969 result = _invoke(
1970 ["snapshot", "export", snap_id, "--output", str(out_file), "--json"],
1971 env=_env(tmp_path),
1972 )
1973 assert result.exit_code == 0
1974 data_out: _ExportOut = json.loads(result.output)
1975 assert data_out["file_count"] == 500
1976 assert data_out["size_bytes"] > 0
1977 assert tarfile.is_tarfile(str(out_file))
1978
1979 def test_export_500_file_zip(self, tmp_path: pathlib.Path) -> None:
1980 """Export of a 500-file snapshot produces a valid zip with all files."""
1981 _init_repo(tmp_path)
1982 manifest: Manifest = {}
1983 for i in range(500):
1984 data = f"zip-content-{i}".encode()
1985 obj_id = _sha(data)
1986 write_object(tmp_path, obj_id, data)
1987 manifest[f"z{i:04d}.dat"] = obj_id
1988 snap_id = compute_snapshot_id(manifest)
1989 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
1990 out_file = tmp_path / "big.zip"
1991 result = _invoke(
1992 ["snapshot", "export", snap_id, "-f", "zip", "--output", str(out_file), "--json"],
1993 env=_env(tmp_path),
1994 )
1995 assert result.exit_code == 0
1996 data_out: _ExportOut = json.loads(result.output)
1997 assert data_out["file_count"] == 500
1998 assert zipfile.is_zipfile(str(out_file))
1999
2000 def test_export_10_consecutive_exports_same_snapshot(self, tmp_path: pathlib.Path) -> None:
2001 """10 consecutive exports of the same snapshot all succeed with consistent results."""
2002 _init_repo(tmp_path)
2003 _create_files(tmp_path, 5)
2004 create_res = _invoke(["snapshot", "create", "--json"], env=_env(tmp_path))
2005 snap_id: str = json.loads(create_res.output)["snapshot_id"]
2006 for i in range(10):
2007 out_file = tmp_path / f"repeat_{i}.tar.gz"
2008 result = _invoke(
2009 ["snapshot", "export", snap_id, "--output", str(out_file), "--json"],
2010 env=_env(tmp_path),
2011 )
2012 assert result.exit_code == 0, f"Iteration {i} failed: {result.output}"
2013 data_out: _ExportOut = json.loads(result.output)
2014 assert data_out["snapshot_id"] == snap_id
2015 assert data_out["file_count"] >= 5
File History 5 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 27 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 27 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e Merge branch 'dev' into main Human 30 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce Merge branch 'dev' into main Human 49 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 101 days ago