gabriel / muse public
test_archive_command.py python
832 lines 33.2 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 124 days ago
1 """Tests for ``muse archive`` — snapshot export command.
2
3 Tiers
4 -----
5 1. Unit — ``_safe_arcname`` and ``_build_entries`` in isolation.
6 2. Integration — store round-trip: write commit/snapshot, build archive, verify contents.
7 3. End-to-End — full CLI invocations via CliRunner.
8 4. Security — zip-slip, tar-slip, null bytes, ``..`` traversal, unsafe prefixes.
9 5. Stress — large manifests, many files, names at path limits.
10 6. Performance — timing assertions on archive creation and list mode.
11 7. Data Integrity — archive contents match snapshot manifest exactly; JSON schema complete.
12 """
13
14 from __future__ import annotations
15
16 import datetime
17 import json
18 import pathlib
19 import tarfile
20 import time
21 import zipfile
22
23 import pytest
24 from tests.cli_test_helper import CliRunner
25
26 cli = None # argparse migration — CliRunner ignores this arg
27
28 from muse.cli.commands.archive import (
29 _FORMAT_CHOICES,
30 _build_entries,
31 _build_tar,
32 _build_zip,
33 _safe_arcname,
34 )
35 from muse.core.object_store import write_object
36 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
37 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
38 from muse.core.types import blob_id, long_id, short_id, split_id, fake_id
39 from muse.core.paths import heads_dir, muse_dir
40
41 runner = CliRunner()
42
43
44 # ---------------------------------------------------------------------------
45 # Fixtures
46 # ---------------------------------------------------------------------------
47
48
49 @pytest.fixture
50 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
51 """Minimal Muse repo chdir'd into tmp_path."""
52 monkeypatch.chdir(tmp_path)
53 dot_muse = muse_dir(tmp_path)
54 dot_muse.mkdir()
55 (dot_muse / "repo.json").write_text('{"repo_id":"test-repo"}')
56 (dot_muse / "HEAD").write_text("ref: refs/heads/main")
57 (dot_muse / "commits").mkdir()
58 (dot_muse / "snapshots").mkdir()
59 (dot_muse / "refs" / "heads").mkdir(parents=True)
60 (dot_muse / "objects").mkdir()
61 return tmp_path
62
63
64 def _make_commit(
65 root: pathlib.Path,
66 files: dict[str, bytes],
67 message: str = "test commit",
68 ) -> CommitRecord:
69 """Write objects, a snapshot, and a commit; update the branch ref.
70
71 Args:
72 root: Repository root.
73 files: Mapping of relative path → raw file bytes.
74 message: Commit message.
75
76 Returns:
77 The written ``CommitRecord``.
78 """
79 manifest: dict[str, str] = {}
80 for rel_path, content in files.items():
81 oid = blob_id(content)
82 write_object(root, oid, content)
83 manifest[rel_path] = oid
84
85 snap_id = compute_snapshot_id(manifest)
86 snap = SnapshotRecord(
87 snapshot_id=snap_id,
88 manifest=manifest,
89 directories=[],
90 created_at=datetime.datetime(2026, 3, 1, tzinfo=datetime.timezone.utc),
91 note="",
92 )
93 write_snapshot(root, snap)
94
95 committed_at = datetime.datetime(2026, 3, 1, tzinfo=datetime.timezone.utc)
96 cid = compute_commit_id( parent_ids=[],
97 snapshot_id=snap_id,
98 message=message,
99 committed_at_iso=committed_at.isoformat(),
100 author="test-author",
101 )
102 record = CommitRecord(
103 commit_id=cid,
104 repo_id="test-repo",
105 branch="main",
106 snapshot_id=snap_id,
107 message=message,
108 committed_at=committed_at,
109 author="test-author",
110 agent_id="test-agent",
111 model_id="test-model",
112 )
113 write_commit(root, record)
114 (heads_dir(root) / "main").write_text(cid)
115 return record
116
117
118 # ===========================================================================
119 # 1. Unit tests — _safe_arcname and _build_entries
120 # ===========================================================================
121
122
123 class TestSafeArcname:
124 def test_simple_path_no_prefix(self) -> None:
125 assert _safe_arcname("", "src/main.py") == "src/main.py"
126
127 def test_simple_path_with_prefix(self) -> None:
128 assert _safe_arcname("myproject", "src/main.py") == "myproject/src/main.py"
129
130 def test_prefix_trailing_slash_stripped(self) -> None:
131 assert _safe_arcname("myproject/", "a.py") == "myproject/a.py"
132
133 def test_empty_rel_path_returns_none(self) -> None:
134 assert _safe_arcname("", "") is None
135
136 def test_dot_rel_path_returns_none(self) -> None:
137 # PurePosixPath("") → "." — should be rejected
138 assert _safe_arcname("", ".") is None
139
140 def test_absolute_rel_path_returns_none(self) -> None:
141 assert _safe_arcname("", "/etc/passwd") is None
142
143 def test_dotdot_in_rel_path_returns_none(self) -> None:
144 assert _safe_arcname("", "../../etc/passwd") is None
145
146 def test_dotdot_component_in_rel_path_returns_none(self) -> None:
147 assert _safe_arcname("", "src/../../../etc/passwd") is None
148
149 def test_dotdot_in_prefix_returns_none(self) -> None:
150 assert _safe_arcname("../traversal", "a.py") is None
151
152 def test_null_byte_in_rel_path_returns_none(self) -> None:
153 assert _safe_arcname("", "a\x00b.py") is None
154
155 def test_null_byte_in_prefix_returns_none(self) -> None:
156 assert _safe_arcname("pre\x00fix", "a.py") is None
157
158 def test_nested_path(self) -> None:
159 assert _safe_arcname("", "a/b/c/d.txt") == "a/b/c/d.txt"
160
161 def test_single_filename(self) -> None:
162 assert _safe_arcname("", "README.md") == "README.md"
163
164 def test_prefix_with_subdirs(self) -> None:
165 assert _safe_arcname("proj/v2", "src/app.py") == "proj/v2/src/app.py"
166
167
168 class TestBuildEntries:
169 def test_returns_entries_for_valid_manifest(self, repo: pathlib.Path) -> None:
170 c = _make_commit(repo, {"a.py": b"hello"})
171 from muse.core.store import read_snapshot, read_commit
172 commit = read_commit(repo, c.commit_id)
173 assert commit is not None
174 snap = read_snapshot(repo, commit.snapshot_id)
175 assert snap is not None
176 entries, skipped = _build_entries(repo, snap.manifest, "")
177 assert len(entries) == 1
178 assert skipped == []
179 arcname, oid, path = entries[0]
180 assert arcname == "a.py"
181 assert path.exists()
182
183 def test_skips_missing_objects(self, repo: pathlib.Path) -> None:
184 # Fake a manifest entry pointing at a nonexistent object.
185 fake_manifest = {"ghost.py": fake_id("ghost-obj")}
186 entries, skipped = _build_entries(repo, fake_manifest, "")
187 assert entries == []
188 assert len(skipped) == 1
189 assert "missing" in skipped[0]
190
191 def test_entries_sorted_by_arcname(self, repo: pathlib.Path) -> None:
192 c = _make_commit(repo, {"z.py": b"z", "a.py": b"a", "m.py": b"m"})
193 from muse.core.store import read_snapshot, read_commit
194 commit = read_commit(repo, c.commit_id)
195 snap = read_snapshot(repo, commit.snapshot_id)
196 entries, _ = _build_entries(repo, snap.manifest, "")
197 names = [e[0] for e in entries]
198 assert names == sorted(names)
199
200 def test_prefix_applied_to_arcnames(self, repo: pathlib.Path) -> None:
201 c = _make_commit(repo, {"src/app.py": b"app"})
202 from muse.core.store import read_snapshot, read_commit
203 commit = read_commit(repo, c.commit_id)
204 snap = read_snapshot(repo, commit.snapshot_id)
205 entries, _ = _build_entries(repo, snap.manifest, "myproject")
206 assert entries[0][0] == "myproject/src/app.py"
207
208
209 # ===========================================================================
210 # 2. Integration tests — store round-trip + archive contents
211 # ===========================================================================
212
213
214 class TestTarContents:
215 def test_tar_contains_all_files(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
216 _make_commit(repo, {"a.py": b"aaa", "b.py": b"bbb"})
217 out = tmp_path / "out.tar.gz"
218 runner.invoke(cli, ["archive", "--output", str(out)], catch_exceptions=False)
219 with tarfile.open(out, "r:gz") as tar:
220 names = tar.getnames()
221 assert "a.py" in names
222 assert "b.py" in names
223
224 def test_tar_file_contents_match_source(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
225 _make_commit(repo, {"hello.py": b"print('hello')"})
226 out = tmp_path / "out.tar.gz"
227 runner.invoke(cli, ["archive", "--output", str(out)], catch_exceptions=False)
228 with tarfile.open(out, "r:gz") as tar:
229 member = tar.getmember("hello.py")
230 f = tar.extractfile(member)
231 assert f is not None
232 assert f.read() == b"print('hello')"
233
234 def test_tar_prefix_wraps_files(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
235 _make_commit(repo, {"a.py": b"a"})
236 out = tmp_path / "out.tar.gz"
237 runner.invoke(
238 cli, ["archive", "--prefix", "proj", "--output", str(out)],
239 catch_exceptions=False,
240 )
241 with tarfile.open(out, "r:gz") as tar:
242 names = tar.getnames()
243 assert "proj/a.py" in names
244 assert "a.py" not in names
245
246 def test_no_muse_metadata_in_tar(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
247 _make_commit(repo, {"src/app.py": b"app"})
248 out = tmp_path / "out.tar.gz"
249 runner.invoke(cli, ["archive", "--output", str(out)], catch_exceptions=False)
250 with tarfile.open(out, "r:gz") as tar:
251 names = tar.getnames()
252 assert not any(".muse" in n for n in names)
253
254
255 class TestZipContents:
256 def test_zip_contains_all_files(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
257 _make_commit(repo, {"x.py": b"x", "y.py": b"y"})
258 out = tmp_path / "out.zip"
259 runner.invoke(cli, ["archive", "--format", "zip", "--output", str(out)], catch_exceptions=False)
260 with zipfile.ZipFile(out) as zf:
261 names = zf.namelist()
262 assert "x.py" in names
263 assert "y.py" in names
264
265 def test_zip_file_contents_match_source(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
266 _make_commit(repo, {"data.txt": b"hello world"})
267 out = tmp_path / "out.zip"
268 runner.invoke(cli, ["archive", "--format", "zip", "--output", str(out)], catch_exceptions=False)
269 with zipfile.ZipFile(out) as zf:
270 assert zf.read("data.txt") == b"hello world"
271
272 def test_zip_prefix_wraps_files(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
273 _make_commit(repo, {"b.py": b"b"})
274 out = tmp_path / "out.zip"
275 runner.invoke(
276 cli, ["archive", "--format", "zip", "--prefix", "release", "--output", str(out)],
277 catch_exceptions=False,
278 )
279 with zipfile.ZipFile(out) as zf:
280 names = zf.namelist()
281 assert "release/b.py" in names
282 assert "b.py" not in names
283
284
285 # ===========================================================================
286 # 3. End-to-End tests — full CLI
287 # ===========================================================================
288
289
290 class TestDefaultBehavior:
291 def test_exits_0_with_commit(self, repo: pathlib.Path) -> None:
292 _make_commit(repo, {"a.py": b"a"})
293 result = runner.invoke(cli, ["archive"], catch_exceptions=False)
294 assert result.exit_code == 0
295
296 def test_default_filename_no_sha256_prefix(self, repo: pathlib.Path) -> None:
297 c = _make_commit(repo, {"a.py": b"a"})
298 runner.invoke(cli, ["archive"], catch_exceptions=False)
299 _, hex_full = split_id(c.commit_id)
300 assert pathlib.Path(f"{hex_full}.tar.gz").exists()
301
302 def test_default_filename_has_no_colon(self, repo: pathlib.Path) -> None:
303 _make_commit(repo, {"a.py": b"a"})
304 runner.invoke(cli, ["archive"], catch_exceptions=False)
305 created = list(pathlib.Path(".").glob("*.tar.gz"))
306 assert created, "no tar.gz file created"
307 assert ":" not in created[0].name
308
309 def test_no_commits_exits_1(self, repo: pathlib.Path) -> None:
310 result = runner.invoke(cli, ["archive"])
311 assert result.exit_code != 0
312
313 def test_output_includes_file_count(self, repo: pathlib.Path) -> None:
314 _make_commit(repo, {"a.py": b"a", "b.py": b"b"})
315 result = runner.invoke(cli, ["archive"], catch_exceptions=False)
316 assert "2 file(s)" in result.output
317
318 def test_output_includes_commit_short(self, repo: pathlib.Path) -> None:
319 c = _make_commit(repo, {"a.py": b"a"})
320 result = runner.invoke(cli, ["archive"], catch_exceptions=False)
321 short = short_id(c.commit_id, strip=True)
322 assert short in result.output
323
324
325 class TestFormatFlag:
326 def test_zip_format_flag(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
327 _make_commit(repo, {"a.py": b"a"})
328 out = tmp_path / "out.zip"
329 result = runner.invoke(
330 cli, ["archive", "--format", "zip", "--output", str(out)],
331 catch_exceptions=False,
332 )
333 assert result.exit_code == 0
334 assert zipfile.is_zipfile(out)
335
336 def test_tgz_short_flag(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
337 _make_commit(repo, {"a.py": b"a"})
338 out = tmp_path / "out.tar.gz"
339 result = runner.invoke(
340 cli, ["archive", "-f", "tar.gz", "--output", str(out)],
341 catch_exceptions=False,
342 )
343 assert result.exit_code == 0
344 assert tarfile.is_tarfile(out)
345
346 def test_invalid_format_exits_nonzero(self, repo: pathlib.Path) -> None:
347 _make_commit(repo, {"a.py": b"a"})
348 result = runner.invoke(cli, ["archive", "--format", "rar"])
349 assert result.exit_code != 0
350
351
352 class TestRefFlag:
353 def test_ref_to_branch(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
354 _make_commit(repo, {"a.py": b"a"}, message="on main")
355 out = tmp_path / "out.tar.gz"
356 result = runner.invoke(
357 cli, ["archive", "--ref", "main", "--output", str(out)],
358 catch_exceptions=False,
359 )
360 assert result.exit_code == 0
361
362 def test_ref_to_commit_id(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
363 c = _make_commit(repo, {"a.py": b"a"})
364 out = tmp_path / "out.tar.gz"
365 short = c.commit_id[len("sha256:"):len("sha256:") + 8]
366 result = runner.invoke(
367 cli, ["archive", "--ref", short, "--output", str(out)],
368 catch_exceptions=False,
369 )
370 assert result.exit_code == 0
371
372 def test_unknown_ref_exits_1(self, repo: pathlib.Path) -> None:
373 _make_commit(repo, {"a.py": b"a"})
374 result = runner.invoke(cli, ["archive", "--ref", "no-such-branch"])
375 assert result.exit_code != 0
376
377
378 class TestOutputFlag:
379 def test_custom_output_path(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
380 _make_commit(repo, {"a.py": b"a"})
381 out = tmp_path / "release.tar.gz"
382 runner.invoke(cli, ["archive", "--output", str(out)], catch_exceptions=False)
383 assert out.exists()
384
385 def test_output_short_flag(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
386 _make_commit(repo, {"a.py": b"a"})
387 out = tmp_path / "r.tar.gz"
388 result = runner.invoke(
389 cli, ["archive", "-o", str(out)], catch_exceptions=False
390 )
391 assert result.exit_code == 0
392 assert out.exists()
393
394 def test_missing_output_dir_exits_1(self, repo: pathlib.Path) -> None:
395 _make_commit(repo, {"a.py": b"a"})
396 result = runner.invoke(cli, ["archive", "--output", "/nonexistent/dir/out.tar.gz"])
397 assert result.exit_code != 0
398
399
400 class TestListMode:
401 def test_list_exits_0(self, repo: pathlib.Path) -> None:
402 _make_commit(repo, {"a.py": b"a"})
403 result = runner.invoke(cli, ["archive", "--list"], catch_exceptions=False)
404 assert result.exit_code == 0
405
406 def test_list_does_not_create_file(self, repo: pathlib.Path) -> None:
407 _make_commit(repo, {"a.py": b"a"})
408 before = set(pathlib.Path(".").glob("*.tar.gz"))
409 runner.invoke(cli, ["archive", "--list"], catch_exceptions=False)
410 after = set(pathlib.Path(".").glob("*.tar.gz"))
411 assert before == after
412
413 def test_list_shows_file_paths(self, repo: pathlib.Path) -> None:
414 _make_commit(repo, {"src/app.py": b"app", "README.md": b"readme"})
415 result = runner.invoke(cli, ["archive", "--list"], catch_exceptions=False)
416 assert "src/app.py" in result.output
417 assert "README.md" in result.output
418
419 def test_list_shows_file_count(self, repo: pathlib.Path) -> None:
420 _make_commit(repo, {"a.py": b"a", "b.py": b"b", "c.py": b"c"})
421 result = runner.invoke(cli, ["archive", "--list"], catch_exceptions=False)
422 assert "3 file(s)" in result.output
423
424 def test_list_with_prefix_shows_prefixed_paths(self, repo: pathlib.Path) -> None:
425 _make_commit(repo, {"a.py": b"a"})
426 result = runner.invoke(
427 cli, ["archive", "--list", "--prefix", "proj"],
428 catch_exceptions=False,
429 )
430 assert "proj/a.py" in result.output
431
432 def test_list_json_schema(self, repo: pathlib.Path) -> None:
433 _make_commit(repo, {"a.py": b"a"})
434 result = runner.invoke(
435 cli, ["archive", "--list", "--json"], catch_exceptions=False
436 )
437 data = json.loads(result.output)
438 required = {
439 "commit_id", "snapshot_id", "message", "branch", "author",
440 "committed_at", "ref", "prefix", "file_count", "entries",
441 }
442 assert required <= data.keys()
443 assert isinstance(data["entries"], list)
444 assert data["entries"][0].keys() >= {"path", "object_id"}
445
446 def test_list_json_entry_count_matches(self, repo: pathlib.Path) -> None:
447 _make_commit(repo, {"a.py": b"a", "b.py": b"b"})
448 result = runner.invoke(
449 cli, ["archive", "--list", "--json"], catch_exceptions=False
450 )
451 data = json.loads(result.output)
452 assert data["file_count"] == 2
453 assert len(data["entries"]) == 2
454
455
456 class TestJsonOutput:
457 def test_json_exits_0(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
458 _make_commit(repo, {"a.py": b"a"})
459 out = tmp_path / "out.tar.gz"
460 result = runner.invoke(
461 cli, ["archive", "--json", "--output", str(out)],
462 catch_exceptions=False,
463 )
464 assert result.exit_code == 0
465
466 def test_json_is_valid(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
467 _make_commit(repo, {"a.py": b"a"})
468 out = tmp_path / "out.tar.gz"
469 result = runner.invoke(
470 cli, ["archive", "--json", "--output", str(out)],
471 catch_exceptions=False,
472 )
473 data = json.loads(result.output)
474 assert isinstance(data, dict)
475
476 def test_json_has_all_keys(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
477 _make_commit(repo, {"a.py": b"a"})
478 out = tmp_path / "out.tar.gz"
479 result = runner.invoke(
480 cli, ["archive", "--json", "--output", str(out)],
481 catch_exceptions=False,
482 )
483 data = json.loads(result.output)
484 required = {
485 "path", "format", "file_count", "bytes",
486 "commit_id", "snapshot_id", "message", "branch",
487 "author", "agent_id", "model_id", "committed_at",
488 "ref", "prefix",
489 }
490 assert required <= data.keys()
491
492 def test_json_file_count_correct(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
493 _make_commit(repo, {"a.py": b"a", "b.py": b"b"})
494 out = tmp_path / "out.tar.gz"
495 result = runner.invoke(
496 cli, ["archive", "--json", "--output", str(out)],
497 catch_exceptions=False,
498 )
499 data = json.loads(result.output)
500 assert data["file_count"] == 2
501
502 def test_json_bytes_positive(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
503 _make_commit(repo, {"a.py": b"some content here"})
504 out = tmp_path / "out.tar.gz"
505 result = runner.invoke(
506 cli, ["archive", "--json", "--output", str(out)],
507 catch_exceptions=False,
508 )
509 data = json.loads(result.output)
510 assert data["bytes"] > 0
511
512 def test_json_commit_id_matches(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
513 c = _make_commit(repo, {"a.py": b"a"})
514 out = tmp_path / "out.tar.gz"
515 result = runner.invoke(
516 cli, ["archive", "--json", "--output", str(out)],
517 catch_exceptions=False,
518 )
519 data = json.loads(result.output)
520 assert data["commit_id"] == c.commit_id
521
522 def test_json_snapshot_id_present(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
523 c = _make_commit(repo, {"a.py": b"a"})
524 out = tmp_path / "out.tar.gz"
525 result = runner.invoke(
526 cli, ["archive", "--json", "--output", str(out)],
527 catch_exceptions=False,
528 )
529 data = json.loads(result.output)
530 assert data["snapshot_id"] == c.snapshot_id
531
532 def test_json_agent_id_and_model_id(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
533 _make_commit(repo, {"a.py": b"a"})
534 out = tmp_path / "out.tar.gz"
535 result = runner.invoke(
536 cli, ["archive", "--json", "--output", str(out)],
537 catch_exceptions=False,
538 )
539 data = json.loads(result.output)
540 assert data["agent_id"] == "test-agent"
541 assert data["model_id"] == "test-model"
542
543 def test_json_ref_null_for_head(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
544 _make_commit(repo, {"a.py": b"a"})
545 out = tmp_path / "out.tar.gz"
546 result = runner.invoke(
547 cli, ["archive", "--json", "--output", str(out)],
548 catch_exceptions=False,
549 )
550 data = json.loads(result.output)
551 assert data["ref"] is None
552
553 def test_json_ref_set_when_given(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
554 _make_commit(repo, {"a.py": b"a"})
555 out = tmp_path / "out.tar.gz"
556 result = runner.invoke(
557 cli, ["archive", "--json", "--ref", "main", "--output", str(out)],
558 catch_exceptions=False,
559 )
560 data = json.loads(result.output)
561 assert data["ref"] == "main"
562
563 def test_json_prefix_field(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
564 _make_commit(repo, {"a.py": b"a"})
565 out = tmp_path / "out.tar.gz"
566 result = runner.invoke(
567 cli, ["archive", "--json", "--prefix", "myproj", "--output", str(out)],
568 catch_exceptions=False,
569 )
570 data = json.loads(result.output)
571 assert data["prefix"] == "myproj"
572
573 def test_json_format_field(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
574 _make_commit(repo, {"a.py": b"a"})
575 out = tmp_path / "out.zip"
576 result = runner.invoke(
577 cli, ["archive", "--json", "--format", "zip", "--output", str(out)],
578 catch_exceptions=False,
579 )
580 data = json.loads(result.output)
581 assert data["format"] == "zip"
582
583
584 # ===========================================================================
585 # 4. Security tests
586 # ===========================================================================
587
588
589 class TestSecurity:
590 def test_safe_arcname_blocks_traversal(self) -> None:
591 assert _safe_arcname("", "../../etc/passwd") is None
592
593 def test_safe_arcname_blocks_absolute(self) -> None:
594 assert _safe_arcname("", "/etc/passwd") is None
595
596 def test_safe_arcname_blocks_null_byte_path(self) -> None:
597 assert _safe_arcname("", "a\x00b") is None
598
599 def test_safe_arcname_blocks_null_byte_prefix(self) -> None:
600 assert _safe_arcname("pre\x00fix", "a.py") is None
601
602 def test_safe_arcname_blocks_dotdot_prefix(self) -> None:
603 assert _safe_arcname("../../malicious", "a.py") is None
604
605 def test_prefix_dotdot_rejected_by_cli(self, repo: pathlib.Path) -> None:
606 _make_commit(repo, {"a.py": b"a"})
607 result = runner.invoke(cli, ["archive", "--prefix", "../../etc"])
608 assert result.exit_code != 0
609
610 def test_prefix_dotdot_error_on_stderr(self, repo: pathlib.Path) -> None:
611 _make_commit(repo, {"a.py": b"a"})
612 result = runner.invoke(cli, ["archive", "--prefix", "../../etc"])
613 assert "❌" in result.stderr
614
615 def test_unknown_ref_does_not_glob(self, repo: pathlib.Path) -> None:
616 """A glob metacharacter in --ref must not trigger directory scanning."""
617 _make_commit(repo, {"a.py": b"a"})
618 result = runner.invoke(cli, ["archive", "--ref", "../../*"])
619 assert result.exit_code != 0
620
621 def test_tar_archive_has_no_traversal_paths(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
622 _make_commit(repo, {"safe/file.py": b"ok"})
623 out = tmp_path / "out.tar.gz"
624 runner.invoke(cli, ["archive", "--output", str(out)], catch_exceptions=False)
625 with tarfile.open(out, "r:gz") as tar:
626 for name in tar.getnames():
627 assert not name.startswith("/")
628 assert ".." not in name.split("/")
629
630 def test_zip_archive_has_no_traversal_paths(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
631 _make_commit(repo, {"safe/file.py": b"ok"})
632 out = tmp_path / "out.zip"
633 runner.invoke(
634 cli, ["archive", "--format", "zip", "--output", str(out)],
635 catch_exceptions=False,
636 )
637 with zipfile.ZipFile(out) as zf:
638 for name in zf.namelist():
639 assert not name.startswith("/")
640 assert ".." not in name.split("/")
641
642
643 # ===========================================================================
644 # 5. Stress tests
645 # ===========================================================================
646
647
648 class TestStress:
649 def test_100_file_manifest_tar(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
650 files = {f"src/module_{i:03d}.py": f"# module {i}".encode() for i in range(100)}
651 _make_commit(repo, files)
652 out = tmp_path / "out.tar.gz"
653 result = runner.invoke(cli, ["archive", "--output", str(out)], catch_exceptions=False)
654 assert result.exit_code == 0
655 with tarfile.open(out, "r:gz") as tar:
656 assert len(tar.getnames()) == 100
657
658 def test_100_file_manifest_zip(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
659 files = {f"src/module_{i:03d}.py": f"# module {i}".encode() for i in range(100)}
660 _make_commit(repo, files)
661 out = tmp_path / "out.zip"
662 result = runner.invoke(
663 cli, ["archive", "--format", "zip", "--output", str(out)],
664 catch_exceptions=False,
665 )
666 assert result.exit_code == 0
667 with zipfile.ZipFile(out) as zf:
668 assert len(zf.namelist()) == 100
669
670 def test_list_mode_100_files(self, repo: pathlib.Path) -> None:
671 files = {f"f_{i:03d}.txt": b"x" for i in range(100)}
672 _make_commit(repo, files)
673 result = runner.invoke(cli, ["archive", "--list", "--json"], catch_exceptions=False)
674 data = json.loads(result.output)
675 assert data["file_count"] == 100
676 assert len(data["entries"]) == 100
677
678 def test_deeply_nested_paths(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
679 files = {"a/b/c/d/e/f/deep.py": b"deep"}
680 _make_commit(repo, files)
681 out = tmp_path / "out.tar.gz"
682 runner.invoke(cli, ["archive", "--output", str(out)], catch_exceptions=False)
683 with tarfile.open(out, "r:gz") as tar:
684 assert "a/b/c/d/e/f/deep.py" in tar.getnames()
685
686 def test_large_file_content(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
687 big = b"x" * (1024 * 512) # 512 KiB
688 _make_commit(repo, {"big.bin": big})
689 out = tmp_path / "out.tar.gz"
690 result = runner.invoke(cli, ["archive", "--output", str(out)], catch_exceptions=False)
691 assert result.exit_code == 0
692 with tarfile.open(out, "r:gz") as tar:
693 f = tar.extractfile(tar.getmember("big.bin"))
694 assert f is not None
695 assert f.read() == big
696
697
698 # ===========================================================================
699 # 6. Performance tests
700 # ===========================================================================
701
702
703 class TestPerformance:
704 def test_single_file_archive_under_500ms(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
705 _make_commit(repo, {"a.py": b"a"})
706 out = tmp_path / "out.tar.gz"
707 start = time.monotonic()
708 runner.invoke(cli, ["archive", "--output", str(out)], catch_exceptions=False)
709 elapsed = time.monotonic() - start
710 assert elapsed < 0.5, f"single-file archive took {elapsed:.3f}s"
711
712 def test_list_mode_under_300ms(self, repo: pathlib.Path) -> None:
713 files = {f"f_{i}.py": b"x" for i in range(20)}
714 _make_commit(repo, files)
715 start = time.monotonic()
716 runner.invoke(cli, ["archive", "--list", "--json"], catch_exceptions=False)
717 elapsed = time.monotonic() - start
718 assert elapsed < 0.3, f"list mode took {elapsed:.3f}s"
719
720 def test_json_output_under_500ms(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
721 files = {f"f_{i}.py": b"x" for i in range(10)}
722 _make_commit(repo, files)
723 out = tmp_path / "out.tar.gz"
724 start = time.monotonic()
725 runner.invoke(cli, ["archive", "--json", "--output", str(out)], catch_exceptions=False)
726 elapsed = time.monotonic() - start
727 assert elapsed < 0.5, f"json archive took {elapsed:.3f}s"
728
729
730 # ===========================================================================
731 # 7. Data Integrity tests
732 # ===========================================================================
733
734
735 class TestDataIntegrity:
736 def test_archive_contains_exactly_manifest_files(
737 self, repo: pathlib.Path, tmp_path: pathlib.Path
738 ) -> None:
739 """Every file in the snapshot manifest appears in the archive, no more."""
740 files = {"a.py": b"a", "b/c.py": b"bc", "d.txt": b"d"}
741 _make_commit(repo, files)
742 out = tmp_path / "out.tar.gz"
743 runner.invoke(cli, ["archive", "--output", str(out)], catch_exceptions=False)
744 with tarfile.open(out, "r:gz") as tar:
745 names = set(tar.getnames())
746 assert names == set(files.keys())
747
748 def test_file_bytes_match_original(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
749 content = b"\x00\x01\x02binary\xff\xfe"
750 _make_commit(repo, {"binary.bin": content})
751 out = tmp_path / "out.tar.gz"
752 runner.invoke(cli, ["archive", "--output", str(out)], catch_exceptions=False)
753 with tarfile.open(out, "r:gz") as tar:
754 f = tar.extractfile(tar.getmember("binary.bin"))
755 assert f is not None
756 assert f.read() == content
757
758 def test_zip_bytes_match_original(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
759 content = b"exact content"
760 _make_commit(repo, {"f.txt": content})
761 out = tmp_path / "out.zip"
762 runner.invoke(
763 cli, ["archive", "--format", "zip", "--output", str(out)],
764 catch_exceptions=False,
765 )
766 with zipfile.ZipFile(out) as zf:
767 assert zf.read("f.txt") == content
768
769 def test_list_entries_match_archive_entries(
770 self, repo: pathlib.Path, tmp_path: pathlib.Path
771 ) -> None:
772 """Files listed by --list match files written to the archive."""
773 files = {"x.py": b"x", "y/z.py": b"yz"}
774 _make_commit(repo, files)
775 list_result = runner.invoke(
776 cli, ["archive", "--list", "--json"], catch_exceptions=False
777 )
778 list_data = json.loads(list_result.output)
779 listed_paths = {e["path"] for e in list_data["entries"]}
780
781 out = tmp_path / "out.tar.gz"
782 runner.invoke(cli, ["archive", "--output", str(out)], catch_exceptions=False)
783 with tarfile.open(out, "r:gz") as tar:
784 archive_paths = set(tar.getnames())
785
786 assert listed_paths == archive_paths
787
788 def test_list_entries_sorted(self, repo: pathlib.Path) -> None:
789 files = {"z.py": b"z", "a.py": b"a", "m.py": b"m"}
790 _make_commit(repo, files)
791 result = runner.invoke(
792 cli, ["archive", "--list", "--json"], catch_exceptions=False
793 )
794 data = json.loads(result.output)
795 paths = [e["path"] for e in data["entries"]]
796 assert paths == sorted(paths)
797
798 def test_committed_at_iso8601(self, repo: pathlib.Path, tmp_path: pathlib.Path) -> None:
799 _make_commit(repo, {"a.py": b"a"})
800 out = tmp_path / "out.tar.gz"
801 result = runner.invoke(
802 cli, ["archive", "--json", "--output", str(out)], catch_exceptions=False
803 )
804 data = json.loads(result.output)
805 # Must parse without error
806 dt = datetime.datetime.fromisoformat(data["committed_at"])
807 assert dt.tzinfo is not None
808
809 def test_json_path_field_matches_written_file(
810 self, repo: pathlib.Path, tmp_path: pathlib.Path
811 ) -> None:
812 _make_commit(repo, {"a.py": b"a"})
813 out = tmp_path / "exact-name.tar.gz"
814 result = runner.invoke(
815 cli, ["archive", "--json", "--output", str(out)], catch_exceptions=False
816 )
817 data = json.loads(result.output)
818 assert pathlib.Path(data["path"]) == out
819
820 def test_json_bytes_matches_file_size(
821 self, repo: pathlib.Path, tmp_path: pathlib.Path
822 ) -> None:
823 _make_commit(repo, {"a.py": b"content here"})
824 out = tmp_path / "out.tar.gz"
825 result = runner.invoke(
826 cli, ["archive", "--json", "--output", str(out)], catch_exceptions=False
827 )
828 data = json.loads(result.output)
829 assert data["bytes"] == out.stat().st_size
830
831 def test_format_choices_complete(self) -> None:
832 assert _FORMAT_CHOICES == {"tar.gz", "zip"}
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 124 days ago