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