gabriel / muse public
test_cmd_archive.py python
280 lines 10.7 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 125 days ago
1 """Comprehensive tests for ``muse archive``.
2
3 Covers:
4 - Unit: _safe_arcname zip-slip guard
5 - Integration: archive a commit to tar.gz and zip
6 - E2E: full CLI via CliRunner with output path
7 - Security: --prefix validation, zip-slip prevention in manifest paths
8 - Stress: archive with many tracked files
9 """
10
11 from __future__ import annotations
12
13 type _FileStore = dict[str, bytes]
14
15 import datetime
16 import json
17 import pathlib
18 import tarfile
19 import zipfile
20
21 import pytest
22 from tests.cli_test_helper import CliRunner
23 from muse.core.types import blob_id, fake_id
24 from muse.core.paths import heads_dir, muse_dir
25
26 cli = None # argparse migration — CliRunner ignores this arg
27
28 runner = CliRunner()
29
30
31 # ---------------------------------------------------------------------------
32 # Helpers
33 # ---------------------------------------------------------------------------
34
35 def _env(root: pathlib.Path) -> Manifest:
36 return {"MUSE_REPO_ROOT": str(root)}
37
38
39 def _init_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]:
40 dot_muse = muse_dir(tmp_path)
41 dot_muse.mkdir()
42 repo_id = fake_id("repo")
43 (dot_muse / "repo.json").write_text(json.dumps({
44 "repo_id": repo_id,
45 "domain": "midi",
46 "default_branch": "main",
47 "created_at": "2025-01-01T00:00:00+00:00",
48 }), encoding="utf-8")
49 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
50 (dot_muse / "refs" / "heads").mkdir(parents=True)
51 (dot_muse / "snapshots").mkdir()
52 (dot_muse / "commits").mkdir()
53 (dot_muse / "objects").mkdir()
54 return tmp_path, repo_id
55
56
57 def _make_commit_with_files(
58 root: pathlib.Path, repo_id: str, files: _FileStore | None = None
59 ) -> str:
60 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
61 from muse.core.snapshot import compute_snapshot_id, compute_commit_id
62
63 ref_file = heads_dir(root) / "main"
64 parent_id = ref_file.read_text().strip() if ref_file.exists() else None
65
66 from muse.core.object_store import write_object
67 manifest: Manifest = {}
68 if files:
69 for rel_path, content in files.items():
70 obj_id = blob_id(content)
71 write_object(root, obj_id, content)
72 manifest[rel_path] = obj_id
73
74 snap_id = compute_snapshot_id(manifest)
75 committed_at = datetime.datetime.now(datetime.timezone.utc)
76 commit_id = compute_commit_id( parent_ids=[parent_id] if parent_id else [],
77 snapshot_id=snap_id, message="archive test",
78 committed_at_iso=committed_at.isoformat(),
79 )
80 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
81 write_commit(root, CommitRecord(
82 commit_id=commit_id, repo_id=repo_id, branch="main",
83 snapshot_id=snap_id, message="archive test",
84 committed_at=committed_at, parent_commit_id=parent_id,
85 ))
86 ref_file.parent.mkdir(parents=True, exist_ok=True)
87 ref_file.write_text(commit_id, encoding="utf-8")
88 return commit_id
89
90
91 # ---------------------------------------------------------------------------
92 # Unit tests
93 # ---------------------------------------------------------------------------
94
95 class TestArchiveUnit:
96 def test_safe_arcname_normal_path(self) -> None:
97 from muse.cli.commands.archive import _safe_arcname
98 assert _safe_arcname("myproject", "state/song.mid") == "myproject/state/song.mid"
99
100 def test_safe_arcname_no_prefix(self) -> None:
101 from muse.cli.commands.archive import _safe_arcname
102 assert _safe_arcname("", "state/song.mid") == "state/song.mid"
103
104 def test_safe_arcname_traversal_in_rel_path_rejected(self) -> None:
105 from muse.cli.commands.archive import _safe_arcname
106 assert _safe_arcname("prefix", "../../../etc/passwd") is None
107
108 def test_safe_arcname_absolute_rel_path_rejected(self) -> None:
109 from muse.cli.commands.archive import _safe_arcname
110 assert _safe_arcname("prefix", "/etc/passwd") is None
111
112 def test_safe_arcname_traversal_in_prefix_rejected(self) -> None:
113 from muse.cli.commands.archive import _safe_arcname
114 assert _safe_arcname("../traversal", "file.txt") is None
115
116 def test_safe_arcname_trailing_slash_normalised(self) -> None:
117 from muse.cli.commands.archive import _safe_arcname
118 assert _safe_arcname("myproject/", "file.txt") == "myproject/file.txt"
119
120
121 # ---------------------------------------------------------------------------
122 # Integration tests
123 # ---------------------------------------------------------------------------
124
125 class TestArchiveIntegration:
126 def test_archive_empty_commit(self, tmp_path: pathlib.Path) -> None:
127 root, repo_id = _init_repo(tmp_path)
128 _make_commit_with_files(root, repo_id, files={})
129 out = tmp_path / "out.tar.gz"
130 result = runner.invoke(cli, ["archive", "--output", str(out)], env=_env(root), catch_exceptions=False)
131 assert result.exit_code == 0
132 assert out.exists()
133
134 def test_archive_tar_gz_contains_files(self, tmp_path: pathlib.Path) -> None:
135 root, repo_id = _init_repo(tmp_path)
136 _make_commit_with_files(root, repo_id, files={"state/song.mid": b"\x00\x00MIDI"})
137 out = tmp_path / "archive.tar.gz"
138 result = runner.invoke(cli, ["archive", "--output", str(out)], env=_env(root), catch_exceptions=False)
139 assert result.exit_code == 0
140 with tarfile.open(out, "r:gz") as tf:
141 names = tf.getnames()
142 assert any("song.mid" in n for n in names)
143
144 def test_archive_zip_contains_files(self, tmp_path: pathlib.Path) -> None:
145 root, repo_id = _init_repo(tmp_path)
146 _make_commit_with_files(root, repo_id, files={"track.mid": b"MIDIdata"})
147 out = tmp_path / "archive.zip"
148 result = runner.invoke(
149 cli, ["archive", "--format", "zip", "--output", str(out)],
150 env=_env(root), catch_exceptions=False,
151 )
152 assert result.exit_code == 0
153 with zipfile.ZipFile(out, "r") as zf:
154 names = zf.namelist()
155 assert any("track.mid" in n for n in names)
156
157 def test_archive_with_prefix(self, tmp_path: pathlib.Path) -> None:
158 root, repo_id = _init_repo(tmp_path)
159 _make_commit_with_files(root, repo_id, files={"song.mid": b"data"})
160 out = tmp_path / "prefixed.tar.gz"
161 result = runner.invoke(
162 cli, ["archive", "--output", str(out), "--prefix", "myband-v1.0/"],
163 env=_env(root), catch_exceptions=False,
164 )
165 assert result.exit_code == 0
166 with tarfile.open(out, "r:gz") as tf:
167 names = tf.getnames()
168 assert any("myband-v1.0" in n for n in names)
169
170 def test_archive_unknown_format_fails(self, tmp_path: pathlib.Path) -> None:
171 root, repo_id = _init_repo(tmp_path)
172 _make_commit_with_files(root, repo_id)
173 result = runner.invoke(cli, ["archive", "--format", "rar"], env=_env(root))
174 assert result.exit_code != 0
175
176 def test_archive_no_commits_fails(self, tmp_path: pathlib.Path) -> None:
177 root, repo_id = _init_repo(tmp_path)
178 result = runner.invoke(cli, ["archive"], env=_env(root))
179 assert result.exit_code != 0
180
181 def test_archive_short_flags(self, tmp_path: pathlib.Path) -> None:
182 root, repo_id = _init_repo(tmp_path)
183 _make_commit_with_files(root, repo_id, files={"test.mid": b"data"})
184 out = tmp_path / "short.tar.gz"
185 result = runner.invoke(
186 cli, ["archive", "-f", "tar.gz", "-o", str(out)],
187 env=_env(root), catch_exceptions=False,
188 )
189 assert result.exit_code == 0
190
191
192 # ---------------------------------------------------------------------------
193 # Security tests
194 # ---------------------------------------------------------------------------
195
196 class TestArchiveSecurity:
197 def test_prefix_traversal_rejected(self, tmp_path: pathlib.Path) -> None:
198 root, repo_id = _init_repo(tmp_path)
199 _make_commit_with_files(root, repo_id, files={"song.mid": b"data"})
200 out = tmp_path / "malicious.tar.gz"
201 result = runner.invoke(
202 cli, ["archive", "--output", str(out), "--prefix", "../traversal/"],
203 env=_env(root),
204 )
205 assert result.exit_code != 0
206
207 def test_zip_slip_manifest_path_skipped(self, tmp_path: pathlib.Path) -> None:
208 """A manifest entry with '../' is skipped, not written to archive."""
209 root, repo_id = _init_repo(tmp_path)
210 from muse.core.object_store import write_object
211 from muse.cli.commands.archive import _build_entries, _build_tar
212 content = b"malicious content"
213 obj_id = blob_id(content)
214 write_object(root, obj_id, content)
215
216 out = tmp_path / "safe.tar.gz"
217 manifest = {"../../../etc/passwd": obj_id, "safe.txt": obj_id}
218 entries, _ = _build_entries(root, manifest, prefix="")
219 count = _build_tar(entries, out)
220 assert count == 1 # only safe.txt
221 with tarfile.open(out, "r:gz") as tf:
222 names = tf.getnames()
223 assert all("etc" not in n for n in names)
224
225
226 # ---------------------------------------------------------------------------
227 # Stress tests
228 # ---------------------------------------------------------------------------
229
230 class TestArchiveStress:
231 def test_archive_many_files(self, tmp_path: pathlib.Path) -> None:
232 root, repo_id = _init_repo(tmp_path)
233 files = {f"track_{i:03d}.mid": f"MIDI{i}".encode() for i in range(50)}
234 _make_commit_with_files(root, repo_id, files=files)
235 out = tmp_path / "many.tar.gz"
236 result = runner.invoke(cli, ["archive", "--output", str(out)], env=_env(root), catch_exceptions=False)
237 assert result.exit_code == 0
238 with tarfile.open(out, "r:gz") as tf:
239 names = tf.getnames()
240 assert len(names) == 50
241
242
243 import argparse as _argparse
244
245
246 class TestRegisterFlags:
247 def _parse(self, *args: str) -> _argparse.Namespace:
248 from muse.cli.commands.archive import register
249 p = _argparse.ArgumentParser()
250 sub = p.add_subparsers()
251 register(sub)
252 return p.parse_args(["archive", *args])
253
254 def test_default_json_out_is_false(self) -> None:
255 ns = self._parse()
256 assert ns.json_out is False
257
258 def test_json_flag_sets_json_out(self) -> None:
259 ns = self._parse("--json")
260 assert ns.json_out is True
261
262 def test_j_shorthand_sets_json_out(self) -> None:
263 ns = self._parse("-j")
264 assert ns.json_out is True
265
266 def test_format_default(self) -> None:
267 ns = self._parse()
268 assert ns.fmt == "tar.gz"
269
270 def test_format_flag(self) -> None:
271 ns = self._parse("--format", "zip")
272 assert ns.fmt == "zip"
273
274 def test_list_default(self) -> None:
275 ns = self._parse()
276 assert ns.list_mode is False
277
278 def test_output_default(self) -> None:
279 ns = self._parse()
280 assert ns.output is None
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 125 days ago