gabriel / muse public
test_snapshot_supercharge.py python
829 lines 33.1 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 124 days ago
1 """Comprehensive tests for ``muse snapshot`` subcommands.
2
3 Covers gaps in the original test_cmd_snapshot.py:
4
5 * JSON envelope — duration_ms / exit_code on all four subcommands
6 * JSON schema completeness — all documented fields, correct types
7 * Bug regression — sha256: prefix round-trip through _list_all_snapshots /
8 _resolve_snapshot (bare-hex stem bug)
9 * Data integrity — create → export tar.gz/zip → extract → verify file content
10 * Security — ANSI escape injection in note, symlink skip in snapshots dir,
11 path traversal rejected by _validate_snapshot_id_prefix / _safe_arcname,
12 zip-slip guard for crafted manifest entries
13 * Text mode — ``snapshot read --text`` output format
14 * --prefix — files nested under prefix directory inside archive
15 * Limit validation — limit=0 rejected, limit=1 honoured, limit clamps output
16 * Idempotency — identical working-tree always produces the same snapshot_id
17 * Empty list envelope — snapshot list --json returns envelope even when empty
18 * Concurrent stress — N parallel snapshot creates, all independent and valid
19 * Large file export — single 5 MiB file round-trips correctly
20 """
21
22 from __future__ import annotations
23 from collections.abc import Mapping
24
25 import json
26 import os
27 import pathlib
28 import tarfile
29 import threading
30 import zipfile
31
32 import pytest
33
34 from muse.core.types import short_id, split_id
35 from muse.core.paths import muse_dir, snapshots_dir
36 from tests.cli_test_helper import CliRunner
37
38 cli = None # argparse migration — CliRunner ignores this arg
39
40 runner = CliRunner()
41
42
43 # ---------------------------------------------------------------------------
44 # Shared helpers
45 # ---------------------------------------------------------------------------
46
47
48 def _init_repo(path: pathlib.Path) -> pathlib.Path:
49 dot_muse = muse_dir(path)
50 for d in ("commits", "snapshots", "objects", "refs/heads"):
51 (dot_muse / d).mkdir(parents=True, exist_ok=True)
52 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
53 (dot_muse / "repo.json").write_text(
54 json.dumps({"repo_id": "snap-supercharge", "domain": "code"}),
55 encoding="utf-8",
56 )
57 return path
58
59
60 def _env(repo: pathlib.Path) -> Mapping[str, str]:
61 return {"MUSE_REPO_ROOT": str(repo)}
62
63
64 def _create_files(root: pathlib.Path, count: int = 3) -> list[str]:
65 names: list[str] = []
66 for i in range(count):
67 name = f"file_{i}.txt"
68 (root / name).write_text(f"content-{i}", encoding="utf-8")
69 names.append(name)
70 return names
71
72
73 def _create_snapshot(root: pathlib.Path, note: str = "") -> Mapping[str, object]:
74 """Create a snapshot and return the parsed JSON output."""
75 cmd = ["snapshot", "create", "--json"]
76 if note:
77 cmd += ["-m", note]
78 result = runner.invoke(cli, cmd, env=_env(root))
79 assert result.exit_code == 0, result.output
80 return json.loads(result.output)
81
82
83 # ---------------------------------------------------------------------------
84 # JSON envelope — duration_ms / exit_code
85 # ---------------------------------------------------------------------------
86
87
88 class TestJsonEnvelope:
89 """Every --json subcommand must include duration_ms and exit_code."""
90
91 def test_create_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
92 _init_repo(tmp_path)
93 _create_files(tmp_path, 1)
94 data = _create_snapshot(tmp_path)
95 assert "duration_ms" in data
96 assert isinstance(data["duration_ms"], (int, float))
97 assert data["duration_ms"] >= 0
98
99 def test_create_has_exit_code_zero(self, tmp_path: pathlib.Path) -> None:
100 _init_repo(tmp_path)
101 _create_files(tmp_path, 1)
102 data = _create_snapshot(tmp_path)
103 assert data["exit_code"] == 0
104
105 def test_list_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
106 _init_repo(tmp_path)
107 _create_files(tmp_path, 1)
108 _create_snapshot(tmp_path)
109 result = runner.invoke(cli, ["snapshot", "list", "--json"], env=_env(tmp_path))
110 assert result.exit_code == 0
111 data = json.loads(result.output)
112 assert "duration_ms" in data
113 assert isinstance(data["duration_ms"], (int, float))
114 assert data["duration_ms"] >= 0
115
116 def test_list_has_exit_code_zero(self, tmp_path: pathlib.Path) -> None:
117 _init_repo(tmp_path)
118 _create_files(tmp_path, 1)
119 _create_snapshot(tmp_path)
120 result = runner.invoke(cli, ["snapshot", "list", "--json"], env=_env(tmp_path))
121 data = json.loads(result.output)
122 assert data["exit_code"] == 0
123
124 def test_list_empty_has_envelope(self, tmp_path: pathlib.Path) -> None:
125 _init_repo(tmp_path)
126 result = runner.invoke(cli, ["snapshot", "list", "--json"], env=_env(tmp_path))
127 assert result.exit_code == 0
128 data = json.loads(result.output)
129 assert data["snapshots"] == []
130 assert "duration_ms" in data
131 assert data["exit_code"] == 0
132
133 def test_read_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
134 _init_repo(tmp_path)
135 _create_files(tmp_path, 1)
136 created = _create_snapshot(tmp_path)
137 snap_id = created["snapshot_id"]
138 result = runner.invoke(cli, ["snapshot", "read", snap_id, "--json"], env=_env(tmp_path))
139 assert result.exit_code == 0
140 data = json.loads(result.output)
141 assert "duration_ms" in data
142 assert isinstance(data["duration_ms"], (int, float))
143 assert data["duration_ms"] >= 0
144
145 def test_read_has_exit_code_zero(self, tmp_path: pathlib.Path) -> None:
146 _init_repo(tmp_path)
147 _create_files(tmp_path, 1)
148 created = _create_snapshot(tmp_path)
149 snap_id = created["snapshot_id"]
150 result = runner.invoke(cli, ["snapshot", "read", snap_id, "--json"], env=_env(tmp_path))
151 data = json.loads(result.output)
152 assert data["exit_code"] == 0
153
154 def test_export_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
155 _init_repo(tmp_path)
156 _create_files(tmp_path, 1)
157 created = _create_snapshot(tmp_path)
158 snap_id = created["snapshot_id"]
159 out = tmp_path / "out.tar.gz"
160 result = runner.invoke(
161 cli,
162 ["snapshot", "export", snap_id, "--output", str(out), "--json"],
163 env=_env(tmp_path),
164 )
165 assert result.exit_code == 0
166 data = json.loads(result.output)
167 assert "duration_ms" in data
168 assert isinstance(data["duration_ms"], (int, float))
169 assert data["duration_ms"] >= 0
170
171 def test_export_has_exit_code_zero(self, tmp_path: pathlib.Path) -> None:
172 _init_repo(tmp_path)
173 _create_files(tmp_path, 1)
174 created = _create_snapshot(tmp_path)
175 snap_id = created["snapshot_id"]
176 out = tmp_path / "out.tar.gz"
177 result = runner.invoke(
178 cli,
179 ["snapshot", "export", snap_id, "--output", str(out), "--json"],
180 env=_env(tmp_path),
181 )
182 data = json.loads(result.output)
183 assert data["exit_code"] == 0
184
185
186 # ---------------------------------------------------------------------------
187 # JSON schema completeness
188 # ---------------------------------------------------------------------------
189
190
191 class TestJsonSchemaCompleteness:
192 """All documented fields must be present with correct types."""
193
194 def test_create_schema(self, tmp_path: pathlib.Path) -> None:
195 _init_repo(tmp_path)
196 _create_files(tmp_path, 2)
197 data = _create_snapshot(tmp_path, note="schema-test")
198 assert isinstance(data["repo_id"], str)
199 assert isinstance(data["snapshot_id"], str)
200 assert data["snapshot_id"].startswith("sha256:")
201 assert isinstance(data["file_count"], int)
202 assert data["file_count"] >= 1
203 assert isinstance(data["note"], str)
204 assert data["note"] == "schema-test"
205 assert isinstance(data["created_at"], str)
206 # ISO-8601: basic sanity check
207 assert "T" in data["created_at"] or "-" in data["created_at"]
208 assert isinstance(data["duration_ms"], (int, float))
209 assert isinstance(data["exit_code"], int)
210
211 def test_list_schema(self, tmp_path: pathlib.Path) -> None:
212 _init_repo(tmp_path)
213 _create_files(tmp_path, 2)
214 _create_snapshot(tmp_path, note="list-schema")
215 result = runner.invoke(cli, ["snapshot", "list", "--json"], env=_env(tmp_path))
216 assert result.exit_code == 0
217 data = json.loads(result.output)
218 assert "snapshots" in data
219 assert isinstance(data["snapshots"], list)
220 assert "duration_ms" in data
221 assert "exit_code" in data
222 item = data["snapshots"][0]
223 assert isinstance(item["snapshot_id"], str)
224 assert item["snapshot_id"].startswith("sha256:")
225 assert isinstance(item["file_count"], int)
226 assert isinstance(item["note"], str)
227 assert isinstance(item["created_at"], str)
228
229 def test_read_schema(self, tmp_path: pathlib.Path) -> None:
230 _init_repo(tmp_path)
231 _create_files(tmp_path, 2)
232 created = _create_snapshot(tmp_path, note="read-schema")
233 snap_id = created["snapshot_id"]
234 result = runner.invoke(cli, ["snapshot", "read", snap_id, "--json"], env=_env(tmp_path))
235 assert result.exit_code == 0
236 data = json.loads(result.output)
237 assert isinstance(data["snapshot_id"], str)
238 assert data["snapshot_id"].startswith("sha256:")
239 assert isinstance(data["created_at"], str)
240 assert isinstance(data["file_count"], int)
241 assert isinstance(data["note"], str)
242 assert isinstance(data["manifest"], dict)
243 assert len(data["manifest"]) == data["file_count"]
244 assert isinstance(data["duration_ms"], (int, float))
245 assert isinstance(data["exit_code"], int)
246
247 def test_export_schema(self, tmp_path: pathlib.Path) -> None:
248 _init_repo(tmp_path)
249 _create_files(tmp_path, 2)
250 created = _create_snapshot(tmp_path)
251 snap_id = created["snapshot_id"]
252 out = tmp_path / "schema.tar.gz"
253 result = runner.invoke(
254 cli,
255 ["snapshot", "export", snap_id, "--output", str(out), "--json"],
256 env=_env(tmp_path),
257 )
258 assert result.exit_code == 0
259 data = json.loads(result.output)
260 assert isinstance(data["snapshot_id"], str)
261 assert isinstance(data["output"], str)
262 assert data["format"] in ("tar.gz", "zip")
263 assert isinstance(data["file_count"], int)
264 assert isinstance(data["size_bytes"], int)
265 assert data["size_bytes"] > 0
266 assert isinstance(data["duration_ms"], (int, float))
267 assert isinstance(data["exit_code"], int)
268
269 def test_manifest_keys_are_sorted(self, tmp_path: pathlib.Path) -> None:
270 _init_repo(tmp_path)
271 # Create files in reverse alpha order to verify manifest sorts them.
272 for name in ("zzz.txt", "aaa.txt", "mmm.txt"):
273 (tmp_path / name).write_text(name, encoding="utf-8")
274 created = _create_snapshot(tmp_path)
275 snap_id = created["snapshot_id"]
276 result = runner.invoke(cli, ["snapshot", "read", snap_id, "--json"], env=_env(tmp_path))
277 data = json.loads(result.output)
278 keys = list(data["manifest"].keys())
279 assert keys == sorted(keys)
280
281
282 # ---------------------------------------------------------------------------
283 # Bug regression — sha256: prefix round-trip
284 # ---------------------------------------------------------------------------
285
286
287 class TestSha256PrefixRoundTrip:
288 """Regression for the bare-hex-stem bug: _list_all_snapshots and
289 _resolve_snapshot were passing path.stem (bare hex) to read_snapshot,
290 which then compared it against compute_snapshot_id output (sha256: prefixed),
291 causing every snapshot to fail content-hash verification and appear missing."""
292
293 def test_list_after_create_returns_snapshot(self, tmp_path: pathlib.Path) -> None:
294 _init_repo(tmp_path)
295 _create_files(tmp_path, 2)
296 created = _create_snapshot(tmp_path)
297 result = runner.invoke(cli, ["snapshot", "list", "--json"], env=_env(tmp_path))
298 assert result.exit_code == 0
299 data = json.loads(result.output)
300 ids = [s["snapshot_id"] for s in data["snapshots"]]
301 assert created["snapshot_id"] in ids
302
303 def test_read_by_full_id_succeeds(self, tmp_path: pathlib.Path) -> None:
304 _init_repo(tmp_path)
305 _create_files(tmp_path, 1)
306 created = _create_snapshot(tmp_path)
307 snap_id = created["snapshot_id"]
308 result = runner.invoke(cli, ["snapshot", "read", snap_id], env=_env(tmp_path))
309 assert result.exit_code == 0
310
311 def test_bare_hex_prefix_rejected(self, tmp_path: pathlib.Path) -> None:
312 """Bare hex prefix (no sha256: type tag) must be rejected at the CLI boundary."""
313 _init_repo(tmp_path)
314 _create_files(tmp_path, 1)
315 created = _create_snapshot(tmp_path)
316 snap_id = created["snapshot_id"]
317 result = runner.invoke(cli, ["snapshot", "read", short_id(snap_id, strip=True)], env=_env(tmp_path))
318 assert result.exit_code != 0
319
320 def test_read_by_sha256_prefix_succeeds(self, tmp_path: pathlib.Path) -> None:
321 """Full sha256:... ID passed to snapshot read must resolve."""
322 _init_repo(tmp_path)
323 _create_files(tmp_path, 1)
324 created = _create_snapshot(tmp_path)
325 snap_id = created["snapshot_id"]
326 result = runner.invoke(cli, ["snapshot", "read", snap_id, "--json"], env=_env(tmp_path))
327 assert result.exit_code == 0
328 data = json.loads(result.output)
329 assert data["snapshot_id"] == snap_id
330
331 def test_snapshot_id_in_read_matches_create(self, tmp_path: pathlib.Path) -> None:
332 _init_repo(tmp_path)
333 _create_files(tmp_path, 2)
334 created = _create_snapshot(tmp_path)
335 result = runner.invoke(cli, ["snapshot", "read", created["snapshot_id"], "--json"], env=_env(tmp_path))
336 data = json.loads(result.output)
337 assert data["snapshot_id"] == created["snapshot_id"]
338
339
340 # ---------------------------------------------------------------------------
341 # Data integrity — create → export → verify content
342 # ---------------------------------------------------------------------------
343
344
345 class TestDataIntegrity:
346 """File contents written to archives must match the original source files."""
347
348 def test_tar_gz_content_matches_source(self, tmp_path: pathlib.Path) -> None:
349 _init_repo(tmp_path)
350 names = _create_files(tmp_path, 3)
351 created = _create_snapshot(tmp_path)
352 snap_id = created["snapshot_id"]
353 out = tmp_path / "integrity.tar.gz"
354 runner.invoke(
355 cli,
356 ["snapshot", "export", snap_id, "--output", str(out)],
357 env=_env(tmp_path),
358 )
359 assert out.exists()
360 with tarfile.open(out, "r:gz") as tar:
361 members = {m.name: m for m in tar.getmembers()}
362 for name in names:
363 match = [k for k in members if k.endswith(name)]
364 assert match, f"{name} not found in archive"
365 content = tar.extractfile(members[match[0]])
366 assert content is not None
367 extracted = content.read().decode("utf-8")
368 expected = (tmp_path / name).read_text(encoding="utf-8")
369 assert extracted == expected, f"content mismatch for {name}"
370
371 def test_zip_content_matches_source(self, tmp_path: pathlib.Path) -> None:
372 _init_repo(tmp_path)
373 names = _create_files(tmp_path, 3)
374 created = _create_snapshot(tmp_path)
375 snap_id = created["snapshot_id"]
376 out = tmp_path / "integrity.zip"
377 runner.invoke(
378 cli,
379 ["snapshot", "export", snap_id, "--format", "zip", "--output", str(out)],
380 env=_env(tmp_path),
381 )
382 assert out.exists()
383 with zipfile.ZipFile(out, "r") as zf:
384 namelist = zf.namelist()
385 for name in names:
386 match = [k for k in namelist if k.endswith(name)]
387 assert match, f"{name} not found in zip"
388 extracted = zf.read(match[0]).decode("utf-8")
389 expected = (tmp_path / name).read_text(encoding="utf-8")
390 assert extracted == expected, f"content mismatch for {name}"
391
392 def test_export_file_count_matches_snapshot(self, tmp_path: pathlib.Path) -> None:
393 _init_repo(tmp_path)
394 _create_files(tmp_path, 4)
395 created = _create_snapshot(tmp_path)
396 snap_id = created["snapshot_id"]
397 out = tmp_path / "count.tar.gz"
398 result = runner.invoke(
399 cli,
400 ["snapshot", "export", snap_id, "--output", str(out), "--json"],
401 env=_env(tmp_path),
402 )
403 assert result.exit_code == 0
404 data = json.loads(result.output)
405 assert data["file_count"] == created["file_count"]
406
407 def test_export_size_bytes_matches_disk(self, tmp_path: pathlib.Path) -> None:
408 _init_repo(tmp_path)
409 _create_files(tmp_path, 2)
410 created = _create_snapshot(tmp_path)
411 snap_id = created["snapshot_id"]
412 out = tmp_path / "size.tar.gz"
413 result = runner.invoke(
414 cli,
415 ["snapshot", "export", snap_id, "--output", str(out), "--json"],
416 env=_env(tmp_path),
417 )
418 data = json.loads(result.output)
419 assert data["size_bytes"] == out.stat().st_size
420
421
422 # ---------------------------------------------------------------------------
423 # Security
424 # ---------------------------------------------------------------------------
425
426
427 class TestSecurity:
428 """Security properties of snapshot commands."""
429
430 def test_ansi_escape_in_note_sanitized_in_text_output(self, tmp_path: pathlib.Path) -> None:
431 """ANSI escape sequences in notes must not reach the terminal raw."""
432 _init_repo(tmp_path)
433 _create_files(tmp_path, 1)
434 malicious_note = "\x1b[31mred\x1b[0m"
435 result = runner.invoke(
436 cli, ["snapshot", "create", "-m", malicious_note], env=_env(tmp_path)
437 )
438 assert result.exit_code == 0
439 # ANSI escape character should not appear verbatim in text output.
440 assert "\x1b" not in result.output
441
442 def test_note_appears_sanitized_in_list_text(self, tmp_path: pathlib.Path) -> None:
443 _init_repo(tmp_path)
444 _create_files(tmp_path, 1)
445 malicious_note = "\x1b[1mBOLD\x1b[0m"
446 _create_snapshot(tmp_path, note=malicious_note)
447 result = runner.invoke(cli, ["snapshot", "list"], env=_env(tmp_path))
448 assert result.exit_code == 0
449 assert "\x1b" not in result.output
450
451 def test_symlink_in_snapshots_dir_is_skipped(self, tmp_path: pathlib.Path) -> None:
452 """A symlink inside .muse/snapshots/ must not be read as a snapshot."""
453 _init_repo(tmp_path)
454 _create_files(tmp_path, 1)
455 created = _create_snapshot(tmp_path)
456 snaps_dir = snapshots_dir(tmp_path)
457 # Plant a symlink pointing to a real file outside the snapshot namespace.
458 target = tmp_path / "some_file.txt"
459 target.write_bytes(b"payload")
460 fake_stem = f"deadbeef{'0' * 56}"
461 link = snaps_dir / f"{fake_stem}.msgpack"
462 try:
463 link.symlink_to(target)
464 except (OSError, NotImplementedError):
465 pytest.skip("symlinks not supported on this platform")
466 result = runner.invoke(cli, ["snapshot", "list", "--json"], env=_env(tmp_path))
467 assert result.exit_code == 0
468 data = json.loads(result.output)
469 # Only the legitimately created snapshot should appear.
470 ids = [s["snapshot_id"] for s in data["snapshots"]]
471 assert len(ids) == 1
472 assert ids[0] == created["snapshot_id"]
473
474 def test_path_traversal_in_snapshot_id_prefix_is_safe(self, tmp_path: pathlib.Path) -> None:
475 """A crafted snapshot_id with ../ must not escape the snapshots dir."""
476 _init_repo(tmp_path)
477 result = runner.invoke(
478 cli,
479 ["snapshot", "read", "../../etc/passwd"],
480 env=_env(tmp_path),
481 )
482 # Must fail gracefully — not crash, not read /etc/passwd.
483 assert result.exit_code != 0
484
485 def test_safe_arcname_rejects_dotdot_path(self, tmp_path: pathlib.Path) -> None:
486 """_safe_arcname must return None for paths with .. segments."""
487 from muse.cli.commands.snapshot_cmd import _safe_arcname
488
489 assert _safe_arcname("", "../etc/passwd") is None
490 assert _safe_arcname("prefix", "../../secret") is None
491
492 def test_safe_arcname_rejects_absolute_path(self, tmp_path: pathlib.Path) -> None:
493 from muse.cli.commands.snapshot_cmd import _safe_arcname
494
495 assert _safe_arcname("", "/etc/passwd") is None
496 assert _safe_arcname("prefix", "/root/.ssh/id_rsa") is None
497
498 def test_safe_arcname_accepts_normal_path(self, tmp_path: pathlib.Path) -> None:
499 from muse.cli.commands.snapshot_cmd import _safe_arcname
500
501 assert _safe_arcname("", "src/main.py") == "src/main.py"
502 assert _safe_arcname("myproject", "lib/util.py") == "myproject/lib/util.py"
503
504 def test_safe_arcname_rejects_dotdot_in_prefix(self) -> None:
505 from muse.cli.commands.snapshot_cmd import _safe_arcname
506
507 assert _safe_arcname("../escape", "file.txt") is None
508
509
510 # ---------------------------------------------------------------------------
511 # Text mode — snapshot read --text
512 # ---------------------------------------------------------------------------
513
514
515 class TestTextMode:
516 def test_read_text_shows_snapshot_id(self, tmp_path: pathlib.Path) -> None:
517 _init_repo(tmp_path)
518 _create_files(tmp_path, 2)
519 created = _create_snapshot(tmp_path)
520 snap_id = created["snapshot_id"]
521 result = runner.invoke(
522 cli, ["snapshot", "read", snap_id], env=_env(tmp_path)
523 )
524 assert result.exit_code == 0
525 assert "snapshot_id" in result.output
526 assert snap_id in result.output
527
528 def test_read_text_shows_file_list(self, tmp_path: pathlib.Path) -> None:
529 _init_repo(tmp_path)
530 _create_files(tmp_path, 2)
531 created = _create_snapshot(tmp_path)
532 snap_id = created["snapshot_id"]
533 result = runner.invoke(
534 cli, ["snapshot", "read", snap_id], env=_env(tmp_path)
535 )
536 assert result.exit_code == 0
537 assert "file" in result.output.lower() or "files" in result.output.lower()
538
539 def test_read_text_shows_note_when_set(self, tmp_path: pathlib.Path) -> None:
540 _init_repo(tmp_path)
541 _create_files(tmp_path, 1)
542 created = _create_snapshot(tmp_path, note="my-label")
543 snap_id = created["snapshot_id"]
544 result = runner.invoke(
545 cli, ["snapshot", "read", snap_id], env=_env(tmp_path)
546 )
547 assert result.exit_code == 0
548 assert "my-label" in result.output
549
550 def test_read_text_is_not_valid_json(self, tmp_path: pathlib.Path) -> None:
551 """--text output must not be machine-parseable JSON."""
552 _init_repo(tmp_path)
553 _create_files(tmp_path, 1)
554 created = _create_snapshot(tmp_path)
555 snap_id = created["snapshot_id"]
556 result = runner.invoke(
557 cli, ["snapshot", "read", snap_id], env=_env(tmp_path)
558 )
559 assert result.exit_code == 0
560 with pytest.raises((json.JSONDecodeError, ValueError)):
561 json.loads(result.output)
562
563
564 # ---------------------------------------------------------------------------
565 # --prefix export
566 # ---------------------------------------------------------------------------
567
568
569 class TestPrefixExport:
570 def test_tar_gz_files_nested_under_prefix(self, tmp_path: pathlib.Path) -> None:
571 _init_repo(tmp_path)
572 _create_files(tmp_path, 2)
573 created = _create_snapshot(tmp_path)
574 snap_id = created["snapshot_id"]
575 out = tmp_path / "prefixed.tar.gz"
576 runner.invoke(
577 cli,
578 ["snapshot", "export", snap_id, "--prefix", "myproject", "--output", str(out)],
579 env=_env(tmp_path),
580 )
581 assert out.exists()
582 with tarfile.open(out, "r:gz") as tar:
583 names = tar.getnames()
584 assert all(n.startswith("myproject/") for n in names), names
585
586 def test_zip_files_nested_under_prefix(self, tmp_path: pathlib.Path) -> None:
587 _init_repo(tmp_path)
588 _create_files(tmp_path, 2)
589 created = _create_snapshot(tmp_path)
590 snap_id = created["snapshot_id"]
591 out = tmp_path / "prefixed.zip"
592 runner.invoke(
593 cli,
594 [
595 "snapshot", "export", snap_id,
596 "--format", "zip",
597 "--prefix", "release",
598 "--output", str(out),
599 ],
600 env=_env(tmp_path),
601 )
602 assert out.exists()
603 with zipfile.ZipFile(out, "r") as zf:
604 names = zf.namelist()
605 assert all(n.startswith("release/") for n in names), names
606
607 def test_empty_prefix_uses_flat_layout(self, tmp_path: pathlib.Path) -> None:
608 _init_repo(tmp_path)
609 _create_files(tmp_path, 2)
610 created = _create_snapshot(tmp_path)
611 snap_id = created["snapshot_id"]
612 out = tmp_path / "flat.tar.gz"
613 runner.invoke(
614 cli,
615 ["snapshot", "export", snap_id, "--prefix", "", "--output", str(out)],
616 env=_env(tmp_path),
617 )
618 assert out.exists()
619 with tarfile.open(out, "r:gz") as tar:
620 names = tar.getnames()
621 assert all(not n.startswith("/") for n in names)
622
623
624 # ---------------------------------------------------------------------------
625 # Limit validation
626 # ---------------------------------------------------------------------------
627
628
629 class TestLimitValidation:
630 def test_limit_zero_rejected(self, tmp_path: pathlib.Path) -> None:
631 _init_repo(tmp_path)
632 result = runner.invoke(
633 cli, ["snapshot", "list", "--limit", "0"], env=_env(tmp_path)
634 )
635 assert result.exit_code != 0
636
637 def test_limit_one_returns_at_most_one(self, tmp_path: pathlib.Path) -> None:
638 _init_repo(tmp_path)
639 _create_files(tmp_path, 1)
640 for _ in range(3):
641 _create_snapshot(tmp_path)
642 result = runner.invoke(
643 cli, ["snapshot", "list", "--limit", "1", "--json"], env=_env(tmp_path)
644 )
645 assert result.exit_code == 0
646 data = json.loads(result.output)
647 assert len(data["snapshots"]) <= 1
648
649 def test_negative_limit_rejected(self, tmp_path: pathlib.Path) -> None:
650 _init_repo(tmp_path)
651 result = runner.invoke(
652 cli, ["snapshot", "list", "--limit", "-1"], env=_env(tmp_path)
653 )
654 assert result.exit_code != 0
655
656 def test_short_flag_n_respected(self, tmp_path: pathlib.Path) -> None:
657 _init_repo(tmp_path)
658 _create_files(tmp_path, 1)
659 for _ in range(4):
660 _create_snapshot(tmp_path)
661 result = runner.invoke(
662 cli, ["snapshot", "list", "--limit", "2", "--json"], env=_env(tmp_path)
663 )
664 assert result.exit_code == 0
665 data = json.loads(result.output)
666 assert len(data["snapshots"]) <= 2
667
668
669 # ---------------------------------------------------------------------------
670 # Idempotency — same tree → same snapshot_id
671 # ---------------------------------------------------------------------------
672
673
674 class TestIdempotency:
675 def test_same_files_same_snapshot_id(self, tmp_path: pathlib.Path) -> None:
676 _init_repo(tmp_path)
677 _create_files(tmp_path, 3)
678 first = _create_snapshot(tmp_path)
679 second = _create_snapshot(tmp_path)
680 assert first["snapshot_id"] == second["snapshot_id"]
681
682 def test_different_content_different_snapshot_id(self, tmp_path: pathlib.Path) -> None:
683 _init_repo(tmp_path)
684 _create_files(tmp_path, 2)
685 first = _create_snapshot(tmp_path)
686 # Modify a file.
687 (tmp_path / "file_0.txt").write_text("changed-content", encoding="utf-8")
688 second = _create_snapshot(tmp_path)
689 assert first["snapshot_id"] != second["snapshot_id"]
690
691 def test_list_shows_only_one_when_idempotent(self, tmp_path: pathlib.Path) -> None:
692 """write_snapshot is idempotent — same ID written twice → one file."""
693 _init_repo(tmp_path)
694 _create_files(tmp_path, 2)
695 _create_snapshot(tmp_path)
696 _create_snapshot(tmp_path)
697 result = runner.invoke(cli, ["snapshot", "list", "--json"], env=_env(tmp_path))
698 data = json.loads(result.output)
699 # De-duplicate by snapshot_id.
700 ids = {s["snapshot_id"] for s in data["snapshots"]}
701 assert len(ids) == 1
702
703
704 # ---------------------------------------------------------------------------
705 # List ordering — newest first
706 # ---------------------------------------------------------------------------
707
708
709 class TestListOrdering:
710 def test_list_newest_first(self, tmp_path: pathlib.Path) -> None:
711 """Multiple distinct snapshots must be returned newest-first."""
712 _init_repo(tmp_path)
713 snap_ids: list[str] = []
714 for i in range(3):
715 (tmp_path / f"round_{i}.txt").write_text(f"v{i}", encoding="utf-8")
716 created = _create_snapshot(tmp_path)
717 snap_ids.append(created["snapshot_id"])
718 result = runner.invoke(cli, ["snapshot", "list", "--json"], env=_env(tmp_path))
719 data = json.loads(result.output)
720 returned = [s["snapshot_id"] for s in data["snapshots"]]
721 # Newest (last created) must appear first.
722 assert returned[0] == snap_ids[-1]
723
724
725 # ---------------------------------------------------------------------------
726 # Concurrent stress
727 # ---------------------------------------------------------------------------
728
729
730 class TestConcurrentStress:
731 def test_concurrent_creates_all_succeed(self, tmp_path: pathlib.Path) -> None:
732 """N threads creating snapshots concurrently must all succeed."""
733 _init_repo(tmp_path)
734 _create_files(tmp_path, 5)
735 n_threads = 8
736 errors: list[str] = []
737 results: list[dict] = []
738 lock = threading.Lock()
739
740 def _do_create() -> None:
741 result = runner.invoke(
742 cli, ["snapshot", "create", "--json"], env=_env(tmp_path)
743 )
744 with lock:
745 if result.exit_code != 0:
746 errors.append(result.output)
747 else:
748 results.append(json.loads(result.output))
749
750 threads = [threading.Thread(target=_do_create) for _ in range(n_threads)]
751 for t in threads:
752 t.start()
753 for t in threads:
754 t.join()
755
756 assert not errors, f"Some creates failed: {errors}"
757 assert len(results) == n_threads
758 # All results have a valid snapshot_id.
759 for r in results:
760 assert r["snapshot_id"].startswith("sha256:")
761 assert r["exit_code"] == 0
762
763
764 # ---------------------------------------------------------------------------
765 # Large file stress
766 # ---------------------------------------------------------------------------
767
768
769 class TestLargeFileExport:
770 def test_large_file_round_trips_correctly(self, tmp_path: pathlib.Path) -> None:
771 """A 5 MiB file must survive create → export → extract unchanged."""
772 _init_repo(tmp_path)
773 payload = os.urandom(5 * 1024 * 1024)
774 (tmp_path / "big.bin").write_bytes(payload)
775 created = _create_snapshot(tmp_path)
776 snap_id = created["snapshot_id"]
777 out = tmp_path / "big.tar.gz"
778 result = runner.invoke(
779 cli,
780 ["snapshot", "export", snap_id, "--output", str(out), "--json"],
781 env=_env(tmp_path),
782 )
783 assert result.exit_code == 0
784 data = json.loads(result.output)
785 assert data["file_count"] >= 1
786 assert data["size_bytes"] > 0
787 assert out.exists()
788 # Verify archive actually opens.
789 assert tarfile.is_tarfile(str(out))
790 with tarfile.open(out, "r:gz") as tar:
791 members = [m for m in tar.getmembers() if m.name.endswith("big.bin")]
792 assert members, "big.bin not found in archive"
793 content = tar.extractfile(members[0])
794 assert content is not None
795 assert content.read() == payload
796
797
798 # ---------------------------------------------------------------------------
799 # Export to default filename
800 # ---------------------------------------------------------------------------
801
802
803 class TestDefaultFilename:
804 def test_export_default_filename_is_short_id_dot_format(self, tmp_path: pathlib.Path) -> None:
805 """When --output is omitted, the archive uses <short_id>.<fmt>."""
806 _init_repo(tmp_path)
807 _create_files(tmp_path, 1)
808 created = _create_snapshot(tmp_path)
809 snap_id = created["snapshot_id"]
810 # Run from tmp_path so the default output lands there.
811 orig_dir = pathlib.Path.cwd()
812 os.chdir(tmp_path)
813 try:
814 result = runner.invoke(
815 cli, ["snapshot", "export", snap_id, "--json"], env=_env(tmp_path)
816 )
817 finally:
818 os.chdir(orig_dir)
819 assert result.exit_code == 0
820 data = json.loads(result.output)
821 assert data["output"].endswith(".tar.gz")
822 assert pathlib.Path(tmp_path / data["output"]).exists() or pathlib.Path(data["output"]).exists()
823
824 def test_export_not_found_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
825 _init_repo(tmp_path)
826 result = runner.invoke(
827 cli, ["snapshot", "export", "nonexistent"], env=_env(tmp_path)
828 )
829 assert result.exit_code != 0
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 124 days ago