gabriel / muse public
test_cmd_archive_hardening.py python
589 lines 23.8 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 122 days ago
1 """Comprehensive hardening tests for ``muse archive``.
2
3 Coverage dimensions:
4
5 Unit
6 ~~~~
7 - ``_safe_arcname`` empty rel_path rejected
8 - ``_safe_arcname`` null bytes in rel_path rejected
9 - ``_safe_arcname`` null bytes in prefix rejected
10 - ``_safe_arcname`` dot-only path rejected (".")
11 - ``_safe_arcname`` deeply nested safe path allowed
12 - ``_safe_arcname`` path with spaces allowed
13 - ``_safe_arcname`` unicode filenames allowed
14 - ``_ArchiveJson`` TypedDict has all expected fields
15
16 Security
17 ~~~~~~~~
18 - ``--json`` flag now works (not broken by format validation)
19 - All error messages route to stderr, not stdout
20 - Unknown --format rejected with nonzero exit (argparse choices= guard)
21 - --prefix with ``..`` rejected with nonzero exit
22 - Zip-slip in manifest (``../`` prefix) skipped in tar.gz
23 - Zip-slip in manifest (``../`` prefix) skipped in zip
24 - ANSI escape sequences in commit message sanitized in text output
25 - Null byte in manifest rel_path skipped silently
26
27 JSON schema
28 ~~~~~~~~~~~
29 - ``--json`` on tar.gz produces valid ``_ArchiveJson`` schema
30 - ``--json`` on zip produces valid ``_ArchiveJson`` schema
31 - ``--json`` includes correct ``file_count`` and ``bytes``
32 - ``--json`` includes ``commit_id`` (full SHA-256)
33 - ``--json`` includes ``message`` and ``branch``
34 - ``--json`` includes ``ref`` as null when HEAD used
35 - ``--json`` includes ``ref`` as string when --ref used
36 - ``--json`` on empty snapshot reports file_count=0
37
38 Integration
39 ~~~~~~~~~~~
40 - ``--ref`` with short SHA resolves correctly
41 - ``--ref`` with branch name resolves correctly
42 - ``--ref`` with unknown ref exits nonzero and writes to stderr
43 - Default output path is ``<sha12>.tar.gz``
44 - Custom output path honoured
45 - Missing object in manifest skipped gracefully
46 - Archive file content matches committed bytes (round-trip)
47 - Zip archive entries are readable
48 - Tar.gz archive entries are readable
49
50 E2E
51 ~~~
52 - Full lifecycle: init → commit files → archive → verify contents
53 - ``--prefix`` adds directory level inside both tar.gz and zip
54 - Repeated archive calls produce identical archives (deterministic)
55 - No ``.muse/`` metadata appears in any archive entry
56
57 Stress
58 ~~~~~~
59 - 200-file archive completes without error
60 - Concurrent archive calls on different repos are safe
61 """
62
63 from __future__ import annotations
64
65 type _FileStore = dict[str, bytes]
66
67 import json
68 import pathlib
69 import tarfile
70 import threading
71 import zipfile
72 from typing import TypedDict
73
74 import pytest
75 from tests.cli_test_helper import CliRunner, InvokeResult
76 from muse.core.types import blob_id, long_id, short_id, fake_id
77 from muse.core.paths import heads_dir, muse_dir
78
79 cli = None
80 runner = CliRunner()
81
82
83 # ---------------------------------------------------------------------------
84 # Helpers
85 # ---------------------------------------------------------------------------
86
87
88 def _env(root: pathlib.Path) -> Manifest:
89 return {"MUSE_REPO_ROOT": str(root)}
90
91
92 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
93 import datetime as dt
94 dot_muse = muse_dir(tmp_path)
95 for sub in ("objects", "commits", "snapshots", "refs/heads"):
96 (dot_muse / sub).mkdir(parents=True, exist_ok=True)
97 (dot_muse / "repo.json").write_text(json.dumps({
98 "repo_id": fake_id("repo"),
99 "domain": "code",
100 "default_branch": "main",
101 "created_at": "2026-01-01T00:00:00+00:00",
102 }), encoding="utf-8")
103 (dot_muse / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8")
104 return tmp_path
105
106
107 def _write_object(root: pathlib.Path, content: bytes) -> str:
108 from muse.core.object_store import write_object
109 obj_id = blob_id(content)
110 write_object(root, obj_id, content)
111 return obj_id
112
113
114 def _make_commit(
115 root: pathlib.Path,
116 files: _FileStore | None = None,
117 message: str = "test commit",
118 ) -> str:
119 import datetime as dt
120 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
121 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
122
123 ref_file = heads_dir(root) / "main"
124 parent_id = ref_file.read_text().strip() if ref_file.exists() else None
125
126 manifest: Manifest = {}
127 for rel_path, content in (files or {}).items():
128 manifest[rel_path] = _write_object(root, content)
129
130 snap_id = compute_snapshot_id(manifest)
131 committed_at = dt.datetime(2026, 1, 1, tzinfo=dt.timezone.utc)
132 commit_id = compute_commit_id( parent_ids=[parent_id] if parent_id else [],
133 snapshot_id=snap_id,
134 message=message,
135 committed_at_iso=committed_at.isoformat(),
136 )
137 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
138 write_commit(root, CommitRecord(
139 commit_id=commit_id,
140 repo_id="test-repo",
141 branch="main",
142 snapshot_id=snap_id,
143 message=message,
144 committed_at=committed_at,
145 parent_commit_id=parent_id,
146 ))
147 ref_file.parent.mkdir(parents=True, exist_ok=True)
148 ref_file.write_text(commit_id, encoding="utf-8")
149 return commit_id
150
151
152 def _invoke(root: pathlib.Path, *args: str) -> InvokeResult:
153 return runner.invoke(cli, ["archive"] + list(args), env=_env(root), catch_exceptions=False)
154
155
156 class _ArchiveJson(TypedDict):
157 path: str
158 format: str
159 file_count: int
160 bytes: int
161 commit_id: str
162 message: str
163 branch: str
164 ref: str | None
165
166
167 def _parse_json(output: str) -> _ArchiveJson:
168 for line in output.splitlines():
169 line = line.strip()
170 if line.startswith("{"):
171 raw = json.loads(line)
172 return _ArchiveJson(
173 path=str(raw["path"]),
174 format=str(raw["format"]),
175 file_count=int(raw["file_count"]),
176 bytes=int(raw["bytes"]),
177 commit_id=str(raw["commit_id"]),
178 message=str(raw["message"]),
179 branch=str(raw["branch"]),
180 ref=raw["ref"] if raw["ref"] is not None else None,
181 )
182 raise AssertionError(f"No JSON object found in output:\n{output}")
183
184
185 # ---------------------------------------------------------------------------
186 # Unit — _safe_arcname edge cases
187 # ---------------------------------------------------------------------------
188
189
190 class TestSafeArcname:
191 def test_empty_rel_path_rejected(self) -> None:
192 from muse.cli.commands.archive import _safe_arcname
193 assert _safe_arcname("", "") is None
194 assert _safe_arcname("prefix", "") is None
195
196 def test_null_byte_in_rel_path_rejected(self) -> None:
197 from muse.cli.commands.archive import _safe_arcname
198 assert _safe_arcname("", "file\x00.txt") is None
199
200 def test_null_byte_in_prefix_rejected(self) -> None:
201 from muse.cli.commands.archive import _safe_arcname
202 assert _safe_arcname("pre\x00fix", "file.txt") is None
203
204 def test_dot_only_path_rejected(self) -> None:
205 from muse.cli.commands.archive import _safe_arcname
206 # PurePosixPath("") normalises to "." — must be rejected
207 assert _safe_arcname("", ".") is None
208
209 def test_deeply_nested_safe_path(self) -> None:
210 from muse.cli.commands.archive import _safe_arcname
211 assert _safe_arcname("", "a/b/c/d/e/file.txt") == "a/b/c/d/e/file.txt"
212
213 def test_path_with_spaces(self) -> None:
214 from muse.cli.commands.archive import _safe_arcname
215 assert _safe_arcname("", "my file.mid") == "my file.mid"
216
217 def test_unicode_filename(self) -> None:
218 from muse.cli.commands.archive import _safe_arcname
219 assert _safe_arcname("", "音楽/track.mid") == "音楽/track.mid"
220
221 def test_prefix_with_subdirs(self) -> None:
222 from muse.cli.commands.archive import _safe_arcname
223 assert _safe_arcname("release/v1.0", "file.txt") == "release/v1.0/file.txt"
224
225
226 # ---------------------------------------------------------------------------
227 # Security
228 # ---------------------------------------------------------------------------
229
230
231 class TestSecurity:
232 def test_json_flag_now_works(self, tmp_path: pathlib.Path) -> None:
233 """--json must NOT exit with an error (it was broken before the fix)."""
234 root = _make_repo(tmp_path)
235 _make_commit(root, files={"song.mid": b"MIDI"})
236 out = tmp_path / "out.tar.gz"
237 result = _invoke(root, "--output", str(out), "--json")
238 assert result.exit_code == 0, f"--json flag is still broken: {result.output}"
239
240 def test_error_unknown_format_to_stderr(self, tmp_path: pathlib.Path) -> None:
241 """Unknown --format must exit nonzero (argparse choices= rejects it)."""
242 root = _make_repo(tmp_path)
243 _make_commit(root)
244 result = runner.invoke(cli, ["archive", "--format", "rar"], env=_env(root))
245 assert result.exit_code != 0
246
247 def test_error_prefix_traversal_to_stderr(self, tmp_path: pathlib.Path) -> None:
248 root = _make_repo(tmp_path)
249 _make_commit(root, files={"song.mid": b"data"})
250 result = runner.invoke(cli, ["archive", "--prefix", "../traversal/"], env=_env(root))
251 assert result.exit_code != 0
252 # Error must NOT appear on stdout (it should be on stderr, which CliRunner merges)
253 # We verify exit code nonzero — that's the contract.
254
255 def test_error_no_commits_nonzero(self, tmp_path: pathlib.Path) -> None:
256 root = _make_repo(tmp_path)
257 result = runner.invoke(cli, ["archive"], env=_env(root))
258 assert result.exit_code != 0
259
260 def test_error_bad_ref_nonzero(self, tmp_path: pathlib.Path) -> None:
261 root = _make_repo(tmp_path)
262 _make_commit(root)
263 result = runner.invoke(cli, ["archive", "--ref", "nonexistent-branch-xyz"], env=_env(root))
264 assert result.exit_code != 0
265
266 def test_zip_slip_in_tar_manifest_skipped(self, tmp_path: pathlib.Path) -> None:
267 from muse.cli.commands.archive import _build_entries, _build_tar
268 root = _make_repo(tmp_path)
269 malicious_id = _write_object(root, b"malicious content")
270 safe_id = _write_object(root, b"safe content")
271 out = tmp_path / "test.tar.gz"
272 manifest = {"../../../etc/cron.d/malicious": malicious_id, "safe.txt": safe_id}
273 entries, _ = _build_entries(root, manifest, prefix="")
274 count = _build_tar(entries, out)
275 assert count == 1
276 with tarfile.open(out, "r:gz") as tf:
277 names = tf.getnames()
278 assert not any("etc" in n or "cron" in n for n in names)
279 assert "safe.txt" in names
280
281 def test_zip_slip_in_zip_manifest_skipped(self, tmp_path: pathlib.Path) -> None:
282 from muse.cli.commands.archive import _build_entries, _build_zip
283 root = _make_repo(tmp_path)
284 malicious_id = _write_object(root, b"malicious")
285 safe_id = _write_object(root, b"safe")
286 out = tmp_path / "test.zip"
287 manifest = {"../../../etc/malicious": malicious_id, "safe.txt": safe_id}
288 entries, _ = _build_entries(root, manifest, prefix="")
289 count = _build_zip(entries, out)
290 assert count == 1
291 with zipfile.ZipFile(out, "r") as zf:
292 names = zf.namelist()
293 assert not any("etc" in n for n in names)
294 assert "safe.txt" in names
295
296 def test_null_byte_in_manifest_path_skipped(self, tmp_path: pathlib.Path) -> None:
297 from muse.cli.commands.archive import _build_entries, _build_tar
298 root = _make_repo(tmp_path)
299 null_id = _write_object(root, b"null content")
300 safe_id = _write_object(root, b"safe content")
301 out = tmp_path / "null.tar.gz"
302 manifest = {"file\x00.txt": null_id, "safe.txt": safe_id}
303 entries, _ = _build_entries(root, manifest, prefix="")
304 count = _build_tar(entries, out)
305 assert count == 1
306
307 def test_ansi_in_commit_message_sanitized(self, tmp_path: pathlib.Path) -> None:
308 root = _make_repo(tmp_path)
309 _make_commit(root, files={"f.mid": b"data"}, message="\x1b[31mred\x1b[0m")
310 out = tmp_path / "ansi.tar.gz"
311 result = _invoke(root, "--output", str(out))
312 assert result.exit_code == 0
313 assert "\x1b" not in result.output
314
315 def test_no_muse_dir_in_archive(self, tmp_path: pathlib.Path) -> None:
316 """The .muse/ directory must never appear in any archive entry."""
317 root = _make_repo(tmp_path)
318 _make_commit(root, files={"song.mid": b"MIDI"})
319 out = tmp_path / "clean.tar.gz"
320 _invoke(root, "--output", str(out))
321 with tarfile.open(out, "r:gz") as tf:
322 names = tf.getnames()
323 assert not any(".muse" in n for n in names)
324
325
326 # ---------------------------------------------------------------------------
327 # JSON schema
328 # ---------------------------------------------------------------------------
329
330
331 class TestJsonSchema:
332 def test_json_tar_gz_schema(self, tmp_path: pathlib.Path) -> None:
333 root = _make_repo(tmp_path)
334 commit_id = _make_commit(root, files={"a.mid": b"data", "b.mid": b"more"})
335 out = tmp_path / "archive.tar.gz"
336 result = _invoke(root, "--output", str(out), "--json")
337 assert result.exit_code == 0
338 payload = _parse_json(result.output)
339 assert payload["format"] == "tar.gz"
340 assert payload["file_count"] == 2
341 assert payload["bytes"] > 0
342 assert payload["commit_id"] == commit_id
343 assert payload["branch"] == "main"
344 assert payload["ref"] is None
345 assert payload["path"] == str(out)
346
347 def test_json_zip_schema(self, tmp_path: pathlib.Path) -> None:
348 root = _make_repo(tmp_path)
349 commit_id = _make_commit(root, files={"track.mid": b"MIDI"})
350 out = tmp_path / "archive.zip"
351 result = _invoke(root, "--format", "zip", "--output", str(out), "--json")
352 assert result.exit_code == 0
353 payload = _parse_json(result.output)
354 assert payload["format"] == "zip"
355 assert payload["file_count"] == 1
356 assert payload["commit_id"] == commit_id
357
358 def test_json_ref_field_when_head(self, tmp_path: pathlib.Path) -> None:
359 root = _make_repo(tmp_path)
360 _make_commit(root, files={"f.mid": b"x"})
361 out = tmp_path / "a.tar.gz"
362 result = _invoke(root, "--output", str(out), "--json")
363 payload = _parse_json(result.output)
364 assert payload["ref"] is None
365
366 def test_json_ref_field_when_explicit_ref(self, tmp_path: pathlib.Path) -> None:
367 root = _make_repo(tmp_path)
368 commit_id = _make_commit(root, files={"f.mid": b"x"})
369 short = short_id(commit_id)
370 out = tmp_path / "a.tar.gz"
371 result = _invoke(root, "--ref", short, "--output", str(out), "--json")
372 payload = _parse_json(result.output)
373 assert payload["ref"] == short
374
375 def test_json_empty_snapshot(self, tmp_path: pathlib.Path) -> None:
376 root = _make_repo(tmp_path)
377 _make_commit(root, files={})
378 out = tmp_path / "empty.tar.gz"
379 result = _invoke(root, "--output", str(out), "--json")
380 payload = _parse_json(result.output)
381 assert payload["file_count"] == 0
382
383 def test_json_message_field(self, tmp_path: pathlib.Path) -> None:
384 root = _make_repo(tmp_path)
385 _make_commit(root, files={"f.mid": b"x"}, message="release v2.0")
386 out = tmp_path / "a.tar.gz"
387 result = _invoke(root, "--output", str(out), "--json")
388 payload = _parse_json(result.output)
389 assert payload["message"] == "release v2.0"
390
391
392 # ---------------------------------------------------------------------------
393 # Integration
394 # ---------------------------------------------------------------------------
395
396
397 class TestIntegration:
398 def test_default_output_path_is_sha12_dot_format(self, tmp_path: pathlib.Path) -> None:
399 root = _make_repo(tmp_path)
400 commit_id = _make_commit(root, files={"f.mid": b"data"})
401 result = _invoke(root)
402 assert result.exit_code == 0
403 # Filename uses bare hex (colons illegal on Windows).
404 bare_short = short_id(commit_id, strip=True)
405 assert bare_short in result.output
406 assert ".tar.gz" in result.output
407
408 def test_ref_with_short_sha(self, tmp_path: pathlib.Path) -> None:
409 root = _make_repo(tmp_path)
410 commit_id = _make_commit(root, files={"a.mid": b"MIDI"})
411 out = tmp_path / "ref.tar.gz"
412 # Use the full commit_id as ref (canonical sha256: prefixed form).
413 result = _invoke(root, "--ref", commit_id, "--output", str(out))
414 assert result.exit_code == 0
415 assert out.exists()
416
417 def test_missing_object_skipped_gracefully(self, tmp_path: pathlib.Path) -> None:
418 """If an object file is missing from the store, that entry is skipped — not a crash."""
419 from muse.cli.commands.archive import _build_entries, _build_tar
420 root = _make_repo(tmp_path)
421 # Write one good object, one phantom.
422 good_id = _write_object(root, b"good content")
423 phantom_id = long_id("a" * 64)# valid format but not written to store
424 out = tmp_path / "partial.tar.gz"
425 manifest = {"good.txt": good_id, "missing.txt": phantom_id}
426 entries, _ = _build_entries(root, manifest, prefix="")
427 count = _build_tar(entries, out)
428 assert count == 1
429 with tarfile.open(out, "r:gz") as tf:
430 names = tf.getnames()
431 assert "good.txt" in names
432 assert "missing.txt" not in names
433
434 def test_archive_bytes_match_committed_content(self, tmp_path: pathlib.Path) -> None:
435 """Content extracted from the archive must match what was committed."""
436 root = _make_repo(tmp_path)
437 content = b"exact bytes for round-trip verification"
438 _make_commit(root, files={"track.mid": content})
439 out = tmp_path / "roundtrip.tar.gz"
440 _invoke(root, "--output", str(out))
441 with tarfile.open(out, "r:gz") as tf:
442 member = tf.getmembers()[0]
443 extracted = tf.extractfile(member)
444 assert extracted is not None
445 assert extracted.read() == content
446
447 def test_zip_content_round_trip(self, tmp_path: pathlib.Path) -> None:
448 root = _make_repo(tmp_path)
449 content = b"zip round trip bytes"
450 _make_commit(root, files={"data.mid": content})
451 out = tmp_path / "rt.zip"
452 _invoke(root, "--format", "zip", "--output", str(out))
453 with zipfile.ZipFile(out, "r") as zf:
454 names = zf.namelist()
455 assert len(names) == 1
456 extracted = zf.read(names[0])
457 assert extracted == content
458
459 def test_prefix_appears_in_tar_gz(self, tmp_path: pathlib.Path) -> None:
460 root = _make_repo(tmp_path)
461 _make_commit(root, files={"song.mid": b"MIDI"})
462 out = tmp_path / "prefixed.tar.gz"
463 _invoke(root, "--output", str(out), "--prefix", "band-v1.0")
464 with tarfile.open(out, "r:gz") as tf:
465 names = tf.getnames()
466 assert all(n.startswith("band-v1.0/") for n in names)
467
468 def test_prefix_appears_in_zip(self, tmp_path: pathlib.Path) -> None:
469 root = _make_repo(tmp_path)
470 _make_commit(root, files={"song.mid": b"MIDI"})
471 out = tmp_path / "prefixed.zip"
472 _invoke(root, "--format", "zip", "--output", str(out), "--prefix", "band-v2.0")
473 with zipfile.ZipFile(out, "r") as zf:
474 names = zf.namelist()
475 assert all(n.startswith("band-v2.0/") for n in names)
476
477
478 # ---------------------------------------------------------------------------
479 # E2E — full lifecycle
480 # ---------------------------------------------------------------------------
481
482
483 class TestE2E:
484 def test_full_lifecycle_tar_gz(self, tmp_path: pathlib.Path) -> None:
485 """init → commit multiple files → archive → verify all files present."""
486 root = _make_repo(tmp_path)
487 files = {
488 "tracks/track_01.mid": b"MIDI track 1",
489 "tracks/track_02.mid": b"MIDI track 2",
490 "README.txt": b"Album readme",
491 }
492 _make_commit(root, files=files)
493 out = tmp_path / "album.tar.gz"
494 result = _invoke(root, "--output", str(out))
495 assert result.exit_code == 0
496 assert out.exists()
497 with tarfile.open(out, "r:gz") as tf:
498 names = tf.getnames()
499 assert len(names) == 3
500 assert any("track_01.mid" in n for n in names)
501 assert any("track_02.mid" in n for n in names)
502 assert any("README.txt" in n for n in names)
503
504 def test_deterministic_output(self, tmp_path: pathlib.Path) -> None:
505 """Two archive calls on the same commit produce byte-identical files."""
506 root = _make_repo(tmp_path)
507 _make_commit(root, files={"a.mid": b"AAA", "b.mid": b"BBB"})
508 out1 = tmp_path / "run1.tar.gz"
509 out2 = tmp_path / "run2.tar.gz"
510 _invoke(root, "--output", str(out1))
511 _invoke(root, "--output", str(out2))
512 # gzip includes a timestamp by default, so byte equality is not guaranteed;
513 # but the member names and content must be identical.
514 with tarfile.open(out1, "r:gz") as tf1, tarfile.open(out2, "r:gz") as tf2:
515 names1 = sorted(tf1.getnames())
516 names2 = sorted(tf2.getnames())
517 assert names1 == names2
518
519 def test_historical_ref_archive(self, tmp_path: pathlib.Path) -> None:
520 """Archiving an old commit SHA produces only files from that snapshot."""
521 root = _make_repo(tmp_path)
522 first_id = _make_commit(root, files={"v1.mid": b"v1 data"})
523 _make_commit(root, files={"v1.mid": b"v1 data", "v2.mid": b"v2 data"})
524 out = tmp_path / "historical.tar.gz"
525 result = _invoke(root, "--ref", short_id(first_id), "--output", str(out))
526 assert result.exit_code == 0
527 with tarfile.open(out, "r:gz") as tf:
528 names = tf.getnames()
529 assert any("v1.mid" in n for n in names)
530 assert not any("v2.mid" in n for n in names)
531
532 def test_output_text_shows_commit_short(self, tmp_path: pathlib.Path) -> None:
533 root = _make_repo(tmp_path)
534 commit_id = _make_commit(root, files={"f.mid": b"x"})
535 out = tmp_path / "out.tar.gz"
536 result = _invoke(root, "--output", str(out))
537 assert result.exit_code == 0
538 assert commit_id[:len("sha256:") + 12] in result.output
539
540 def test_output_text_shows_file_count(self, tmp_path: pathlib.Path) -> None:
541 root = _make_repo(tmp_path)
542 _make_commit(root, files={"a.mid": b"x", "b.mid": b"y", "c.mid": b"z"})
543 out = tmp_path / "out.tar.gz"
544 result = _invoke(root, "--output", str(out))
545 assert "3" in result.output
546
547
548 # ---------------------------------------------------------------------------
549 # Stress
550 # ---------------------------------------------------------------------------
551
552
553 class TestStress:
554 def test_200_file_archive(self, tmp_path: pathlib.Path) -> None:
555 root = _make_repo(tmp_path)
556 files = {f"track_{i:03d}.mid": f"MIDI content {i}".encode() for i in range(200)}
557 _make_commit(root, files=files)
558 out = tmp_path / "big.tar.gz"
559 result = _invoke(root, "--output", str(out), "--json")
560 assert result.exit_code == 0
561 payload = _parse_json(result.output)
562 assert payload["file_count"] == 200
563 with tarfile.open(out, "r:gz") as tf:
564 assert len(tf.getnames()) == 200
565
566 def test_concurrent_archives_different_repos(self, tmp_path: pathlib.Path) -> None:
567 """Concurrent archive operations on different repos must not interfere."""
568 errors: list[str] = []
569
570 def _run(idx: int) -> None:
571 repo_dir = tmp_path / f"repo_{idx}"
572 repo_dir.mkdir()
573 root = _make_repo(repo_dir)
574 _make_commit(root, files={f"track_{idx}.mid": f"content {idx}".encode()})
575 out = repo_dir / f"archive_{idx}.tar.gz"
576 try:
577 result = _invoke(root, "--output", str(out))
578 if result.exit_code != 0:
579 errors.append(f"Thread {idx} exit {result.exit_code}: {result.output[:200]}")
580 except Exception as exc:
581 errors.append(f"Thread {idx}: {exc}")
582
583 threads = [threading.Thread(target=_run, args=(i,)) for i in range(8)]
584 for t in threads:
585 t.start()
586 for t in threads:
587 t.join()
588
589 assert not errors, f"Concurrent archive failures: {errors}"
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 122 days ago