gabriel / muse public
test_cmd_archive.py python
287 lines 10.7 KB
Raw
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2 fix: remove commit_exists filter from have anchors — server… Sonnet 4.6 patch 21 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.commits import (
61 CommitRecord,
62 write_commit,
63 )
64 from muse.core.snapshots import (
65 SnapshotRecord,
66 write_snapshot,
67 )
68 from muse.core.ids import hash_commit, hash_snapshot
69
70 ref_file = heads_dir(root) / "main"
71 parent_id = ref_file.read_text().strip() if ref_file.exists() else None
72
73 from muse.core.object_store import write_object
74 manifest: Manifest = {}
75 if files:
76 for rel_path, content in files.items():
77 obj_id = blob_id(content)
78 write_object(root, obj_id, content)
79 manifest[rel_path] = obj_id
80
81 snap_id = hash_snapshot(manifest)
82 committed_at = datetime.datetime.now(datetime.timezone.utc)
83 commit_id = hash_commit( parent_ids=[parent_id] if parent_id else [],
84 snapshot_id=snap_id, message="archive test",
85 committed_at_iso=committed_at.isoformat(),
86 )
87 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
88 write_commit(root, CommitRecord(
89 commit_id=commit_id, branch="main",
90 snapshot_id=snap_id, message="archive test",
91 committed_at=committed_at, parent_commit_id=parent_id,
92 ))
93 ref_file.parent.mkdir(parents=True, exist_ok=True)
94 ref_file.write_text(commit_id, encoding="utf-8")
95 return commit_id
96
97
98 # ---------------------------------------------------------------------------
99 # Unit tests
100 # ---------------------------------------------------------------------------
101
102 class TestArchiveUnit:
103 def test_safe_arcname_normal_path(self) -> None:
104 from muse.cli.commands.archive import _safe_arcname
105 assert _safe_arcname("myproject", "state/song.mid") == "myproject/state/song.mid"
106
107 def test_safe_arcname_no_prefix(self) -> None:
108 from muse.cli.commands.archive import _safe_arcname
109 assert _safe_arcname("", "state/song.mid") == "state/song.mid"
110
111 def test_safe_arcname_traversal_in_rel_path_rejected(self) -> None:
112 from muse.cli.commands.archive import _safe_arcname
113 assert _safe_arcname("prefix", "../../../etc/passwd") is None
114
115 def test_safe_arcname_absolute_rel_path_rejected(self) -> None:
116 from muse.cli.commands.archive import _safe_arcname
117 assert _safe_arcname("prefix", "/etc/passwd") is None
118
119 def test_safe_arcname_traversal_in_prefix_rejected(self) -> None:
120 from muse.cli.commands.archive import _safe_arcname
121 assert _safe_arcname("../traversal", "file.txt") is None
122
123 def test_safe_arcname_trailing_slash_normalised(self) -> None:
124 from muse.cli.commands.archive import _safe_arcname
125 assert _safe_arcname("myproject/", "file.txt") == "myproject/file.txt"
126
127
128 # ---------------------------------------------------------------------------
129 # Integration tests
130 # ---------------------------------------------------------------------------
131
132 class TestArchiveIntegration:
133 def test_archive_empty_commit(self, tmp_path: pathlib.Path) -> None:
134 root, repo_id = _init_repo(tmp_path)
135 _make_commit_with_files(root, repo_id, files={})
136 out = tmp_path / "out.tar.gz"
137 result = runner.invoke(cli, ["archive", "--output", str(out)], env=_env(root), catch_exceptions=False)
138 assert result.exit_code == 0
139 assert out.exists()
140
141 def test_archive_tar_gz_contains_files(self, tmp_path: pathlib.Path) -> None:
142 root, repo_id = _init_repo(tmp_path)
143 _make_commit_with_files(root, repo_id, files={"state/song.mid": b"\x00\x00MIDI"})
144 out = tmp_path / "archive.tar.gz"
145 result = runner.invoke(cli, ["archive", "--output", str(out)], env=_env(root), catch_exceptions=False)
146 assert result.exit_code == 0
147 with tarfile.open(out, "r:gz") as tf:
148 names = tf.getnames()
149 assert any("song.mid" in n for n in names)
150
151 def test_archive_zip_contains_files(self, tmp_path: pathlib.Path) -> None:
152 root, repo_id = _init_repo(tmp_path)
153 _make_commit_with_files(root, repo_id, files={"track.mid": b"MIDIdata"})
154 out = tmp_path / "archive.zip"
155 result = runner.invoke(
156 cli, ["archive", "--format", "zip", "--output", str(out)],
157 env=_env(root), catch_exceptions=False,
158 )
159 assert result.exit_code == 0
160 with zipfile.ZipFile(out, "r") as zf:
161 names = zf.namelist()
162 assert any("track.mid" in n for n in names)
163
164 def test_archive_with_prefix(self, tmp_path: pathlib.Path) -> None:
165 root, repo_id = _init_repo(tmp_path)
166 _make_commit_with_files(root, repo_id, files={"song.mid": b"data"})
167 out = tmp_path / "prefixed.tar.gz"
168 result = runner.invoke(
169 cli, ["archive", "--output", str(out), "--prefix", "myband-v1.0/"],
170 env=_env(root), catch_exceptions=False,
171 )
172 assert result.exit_code == 0
173 with tarfile.open(out, "r:gz") as tf:
174 names = tf.getnames()
175 assert any("myband-v1.0" in n for n in names)
176
177 def test_archive_unknown_format_fails(self, tmp_path: pathlib.Path) -> None:
178 root, repo_id = _init_repo(tmp_path)
179 _make_commit_with_files(root, repo_id)
180 result = runner.invoke(cli, ["archive", "--format", "rar"], env=_env(root))
181 assert result.exit_code != 0
182
183 def test_archive_no_commits_fails(self, tmp_path: pathlib.Path) -> None:
184 root, repo_id = _init_repo(tmp_path)
185 result = runner.invoke(cli, ["archive"], env=_env(root))
186 assert result.exit_code != 0
187
188 def test_archive_short_flags(self, tmp_path: pathlib.Path) -> None:
189 root, repo_id = _init_repo(tmp_path)
190 _make_commit_with_files(root, repo_id, files={"test.mid": b"data"})
191 out = tmp_path / "short.tar.gz"
192 result = runner.invoke(
193 cli, ["archive", "-f", "tar.gz", "-o", str(out)],
194 env=_env(root), catch_exceptions=False,
195 )
196 assert result.exit_code == 0
197
198
199 # ---------------------------------------------------------------------------
200 # Security tests
201 # ---------------------------------------------------------------------------
202
203 class TestArchiveSecurity:
204 def test_prefix_traversal_rejected(self, tmp_path: pathlib.Path) -> None:
205 root, repo_id = _init_repo(tmp_path)
206 _make_commit_with_files(root, repo_id, files={"song.mid": b"data"})
207 out = tmp_path / "malicious.tar.gz"
208 result = runner.invoke(
209 cli, ["archive", "--output", str(out), "--prefix", "../traversal/"],
210 env=_env(root),
211 )
212 assert result.exit_code != 0
213
214 def test_zip_slip_manifest_path_skipped(self, tmp_path: pathlib.Path) -> None:
215 """A manifest entry with '../' is skipped, not written to archive."""
216 root, repo_id = _init_repo(tmp_path)
217 from muse.core.object_store import write_object
218 from muse.cli.commands.archive import _build_entries, _build_tar
219 content = b"malicious content"
220 obj_id = blob_id(content)
221 write_object(root, obj_id, content)
222
223 out = tmp_path / "safe.tar.gz"
224 manifest = {"../../../etc/passwd": obj_id, "safe.txt": obj_id}
225 entries, _ = _build_entries(root, manifest, prefix="")
226 count = _build_tar(entries, out)
227 assert count == 1 # only safe.txt
228 with tarfile.open(out, "r:gz") as tf:
229 names = tf.getnames()
230 assert all("etc" not in n for n in names)
231
232
233 # ---------------------------------------------------------------------------
234 # Stress tests
235 # ---------------------------------------------------------------------------
236
237 class TestArchiveStress:
238 def test_archive_many_files(self, tmp_path: pathlib.Path) -> None:
239 root, repo_id = _init_repo(tmp_path)
240 files = {f"track_{i:03d}.mid": f"MIDI{i}".encode() for i in range(50)}
241 _make_commit_with_files(root, repo_id, files=files)
242 out = tmp_path / "many.tar.gz"
243 result = runner.invoke(cli, ["archive", "--output", str(out)], env=_env(root), catch_exceptions=False)
244 assert result.exit_code == 0
245 with tarfile.open(out, "r:gz") as tf:
246 names = tf.getnames()
247 assert len(names) == 50
248
249
250 import argparse as _argparse
251
252
253 class TestRegisterFlags:
254 def _parse(self, *args: str) -> _argparse.Namespace:
255 from muse.cli.commands.archive import register
256 p = _argparse.ArgumentParser()
257 sub = p.add_subparsers()
258 register(sub)
259 return p.parse_args(["archive", *args])
260
261 def test_default_json_out_is_false(self) -> None:
262 ns = self._parse()
263 assert ns.json_out is False
264
265 def test_json_flag_sets_json_out(self) -> None:
266 ns = self._parse("--json")
267 assert ns.json_out is True
268
269 def test_j_shorthand_sets_json_out(self) -> None:
270 ns = self._parse("-j")
271 assert ns.json_out is True
272
273 def test_format_default(self) -> None:
274 ns = self._parse()
275 assert ns.fmt == "tar.gz"
276
277 def test_format_flag(self) -> None:
278 ns = self._parse("--format", "zip")
279 assert ns.fmt == "zip"
280
281 def test_list_default(self) -> None:
282 ns = self._parse()
283 assert ns.list_mode is False
284
285 def test_output_default(self) -> None:
286 ns = self._parse()
287 assert ns.output is None
File History 4 commits
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2 fix: remove commit_exists filter from have anchors — server… Sonnet 4.6 patch 21 days ago
sha256:36c3cb3e76619d4c30a6d9bf81b5ec4ff148e30dcfed913e3114ca7b43b81c7e fix: rename objects→blobs in push client and all stale test… Sonnet 4.6 patch 22 days ago
sha256:c06a9b9b9fee26c68ea725b44d54b2c0a171301ce9de746d5b656617b4463a9a fix: repair four test failures from post-migration audit Sonnet 4.6 patch 28 days ago
sha256:1900655993c83c4107067375548a7be823e471d2515830842f1a12cba4bd3cdf fix: unified object store migration — idempotent writes, JS… Sonnet 4.6 minor ⚠ 29 days ago