gabriel / muse public
test_mpack_cmd_pack_unpack.py python
371 lines 13.2 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 133 days ago
1 """Comprehensive tests for ``muse pack-objects`` and ``unpack-objects``.
2
3 Coverage tiers
4 --------------
5 - Integration: pack HEAD, explicit commit, --have pruning, --dry-run,
6 round-trip pack→unpack, text+json format for unpack
7 - Security: invalid want/have IDs rejected, empty stdin, corrupted msgpack
8 - Stress: 5-commit chain pack, 200 unpack rounds (idempotency)
9 """
10 from __future__ import annotations
11
12 import datetime
13 import json
14 import pathlib
15
16 import msgpack
17
18 from muse.core.errors import ExitCode
19 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
20 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
21 from muse.core.object_store import object_path, write_object
22 from muse.core._types import Manifest, blob_id
23 from tests.cli_test_helper import CliRunner, InvokeResult
24
25 runner = CliRunner()
26
27
28 # ---------------------------------------------------------------------------
29 # Helpers
30 # ---------------------------------------------------------------------------
31
32 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
33 repo = tmp_path / "repo"
34 muse = repo / ".muse"
35 for sub in ("objects", "commits", "snapshots", "refs/heads"):
36 (muse / sub).mkdir(parents=True)
37 (muse / "HEAD").write_text("ref: refs/heads/main")
38 (muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"}))
39 return repo
40
41
42 def _snap(repo: pathlib.Path, manifest: Manifest | None = None) -> str:
43 m = manifest or {}
44 snap_id = compute_snapshot_id(m)
45 write_snapshot(repo, SnapshotRecord(
46 snapshot_id=snap_id,
47 manifest=m,
48 created_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
49 ))
50 return snap_id
51
52
53 def _commit(
54 repo: pathlib.Path,
55 snap_id: str,
56 *,
57 parent: str | None = None,
58 message: str = "test",
59 ) -> str:
60 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
61 parent_ids: list[str] = [parent] if parent else []
62 commit_id = compute_commit_id(
63 repo_id="test-repo",
64 parent_ids=parent_ids,
65 snapshot_id=snap_id,
66 message=message,
67 committed_at_iso=committed_at.isoformat(),
68 )
69 write_commit(repo, CommitRecord(
70 commit_id=commit_id,
71 repo_id="test-repo",
72 created_on_branch="main",
73 snapshot_id=snap_id,
74 message=message,
75 committed_at=committed_at,
76 parent_commit_id=parent,
77 ))
78 return commit_id
79
80
81 def _set_head(repo: pathlib.Path, commit_id: str) -> None:
82 ref = repo / ".muse" / "refs" / "heads" / "main"
83 ref.write_text(commit_id)
84
85
86 def _po(repo: pathlib.Path, *args: str) -> InvokeResult:
87 from muse.cli.app import main as cli
88 return runner.invoke(
89 cli,
90 ["pack-objects", *args],
91 env={"MUSE_REPO_ROOT": str(repo)},
92 )
93
94
95 def _uo(repo: pathlib.Path, input_bytes: bytes, *args: str) -> InvokeResult:
96 from muse.cli.app import main as cli
97 return runner.invoke(
98 cli,
99 ["unpack-objects", *args],
100 input=input_bytes,
101 env={"MUSE_REPO_ROOT": str(repo)},
102 )
103
104
105 # ---------------------------------------------------------------------------
106 # pack-objects
107 # ---------------------------------------------------------------------------
108
109
110 class TestPackObjects:
111 def test_pack_head(self, tmp_path: pathlib.Path) -> None:
112 repo = _make_repo(tmp_path)
113 sid = _snap(repo)
114 cid = _commit(repo, sid)
115 _set_head(repo, cid)
116 result = _po(repo, "HEAD")
117 assert result.exit_code == 0
118 bundle = msgpack.unpackb(result.stdout_bytes, raw=False)
119 assert "commits" in bundle
120 assert len(bundle["commits"]) >= 1
121
122 def test_pack_explicit_commit(self, tmp_path: pathlib.Path) -> None:
123 repo = _make_repo(tmp_path)
124 sid = _snap(repo)
125 cid = _commit(repo, sid)
126 result = _po(repo, cid)
127 assert result.exit_code == 0
128 bundle = msgpack.unpackb(result.stdout_bytes, raw=False)
129 ids = [c["commit_id"] for c in bundle["commits"]]
130 assert cid in ids
131
132 def test_dry_run_returns_json(self, tmp_path: pathlib.Path) -> None:
133 repo = _make_repo(tmp_path)
134 sid = _snap(repo)
135 cid = _commit(repo, sid)
136 result = _po(repo, cid, "--dry-run", "--json")
137 assert result.exit_code == 0
138 data = json.loads(result.output)
139 assert data["commits"] >= 1
140 assert "snapshots" in data
141 assert "objects" in data
142 assert cid in data["want"]
143
144 def test_dry_run_head(self, tmp_path: pathlib.Path) -> None:
145 repo = _make_repo(tmp_path)
146 sid = _snap(repo)
147 cid = _commit(repo, sid)
148 _set_head(repo, cid)
149 result = _po(repo, "HEAD", "--dry-run", "--json")
150 assert result.exit_code == 0
151 data = json.loads(result.output)
152 assert cid in data["want"]
153
154 def test_have_prunes_old_commits(self, tmp_path: pathlib.Path) -> None:
155 repo = _make_repo(tmp_path)
156 sid = _snap(repo)
157 c1 = _commit(repo, sid, message="c1")
158 c2 = _commit(repo, sid, parent=c1)
159 result = _po(repo, c2, "--have", c1, "--dry-run", "--json")
160 assert result.exit_code == 0
161 data = json.loads(result.output)
162 # c1 is already in "have" — should not be in the pack
163 assert data["commits"] == 1
164
165 def test_invalid_want_id_rejected(self, tmp_path: pathlib.Path) -> None:
166 repo = _make_repo(tmp_path)
167 result = _po(repo, "not-a-hex-id")
168 assert result.exit_code == ExitCode.USER_ERROR
169
170 def test_invalid_have_id_rejected(self, tmp_path: pathlib.Path) -> None:
171 repo = _make_repo(tmp_path)
172 sid = _snap(repo)
173 cid = _commit(repo, sid)
174 result = _po(repo, cid, "--have", "bad-hex")
175 assert result.exit_code == ExitCode.USER_ERROR
176
177 def test_head_no_commits_errors(self, tmp_path: pathlib.Path) -> None:
178 repo = _make_repo(tmp_path)
179 result = _po(repo, "HEAD")
180 assert result.exit_code == ExitCode.USER_ERROR
181
182 def test_no_traceback_on_bad_want(self, tmp_path: pathlib.Path) -> None:
183 repo = _make_repo(tmp_path)
184 result = _po(repo, "bad")
185 assert "Traceback" not in result.output
186
187
188 # ---------------------------------------------------------------------------
189 # unpack-objects
190 # ---------------------------------------------------------------------------
191
192
193 class TestUnpackObjects:
194 def test_round_trip(self, tmp_path: pathlib.Path) -> None:
195 src = _make_repo(tmp_path / "src")
196 dst = _make_repo(tmp_path / "dst")
197 sid = _snap(src)
198 cid = _commit(src, sid)
199
200 pack_result = _po(src, cid)
201 assert pack_result.exit_code == 0
202
203 unpack_result = _uo(dst, pack_result.stdout_bytes, "--json")
204 assert unpack_result.exit_code == 0
205 data = json.loads(unpack_result.output)
206 assert data["commits_written"] == 1
207
208 def test_idempotent_double_unpack(self, tmp_path: pathlib.Path) -> None:
209 src = _make_repo(tmp_path / "src")
210 dst = _make_repo(tmp_path / "dst")
211 sid = _snap(src)
212 cid = _commit(src, sid)
213
214 pack_bytes = _po(src, cid).stdout_bytes
215 _uo(dst, pack_bytes, "--json")
216 result2 = _uo(dst, pack_bytes, "--json")
217 assert result2.exit_code == 0
218 data = json.loads(result2.output)
219 assert data["commits_written"] == 0 # already present
220
221 def test_json_shorthand(self, tmp_path: pathlib.Path) -> None:
222 src = _make_repo(tmp_path / "src")
223 dst = _make_repo(tmp_path / "dst")
224 sid = _snap(src)
225 cid = _commit(src, sid)
226 pack_bytes = _po(src, cid).stdout_bytes
227 result = _uo(dst, pack_bytes, "--json")
228 assert result.exit_code == 0
229 assert "commits_written" in json.loads(result.output)
230
231 def test_text_format(self, tmp_path: pathlib.Path) -> None:
232 src = _make_repo(tmp_path / "src")
233 dst = _make_repo(tmp_path / "dst")
234 sid = _snap(src)
235 cid = _commit(src, sid)
236 pack_bytes = _po(src, cid).stdout_bytes
237 result = _uo(dst, pack_bytes)
238 assert result.exit_code == 0
239 assert "commits" in result.output
240
241 def test_corrupted_msgpack_errors(self, tmp_path: pathlib.Path) -> None:
242 repo = _make_repo(tmp_path)
243 result = _uo(repo, b"\xff\xfe corrupted bytes")
244 assert result.exit_code == ExitCode.USER_ERROR
245
246 def test_empty_stdin_treated_as_empty_pack(self, tmp_path: pathlib.Path) -> None:
247 """An empty msgpack map {} is a valid empty pack; raw empty bytes are not."""
248 repo = _make_repo(tmp_path)
249 result = _uo(repo, b"")
250 assert result.exit_code == ExitCode.USER_ERROR
251
252 def test_no_traceback_on_corrupt_input(self, tmp_path: pathlib.Path) -> None:
253 repo = _make_repo(tmp_path)
254 result = _uo(repo, b"this is not msgpack at all!")
255 assert "Traceback" not in result.output
256
257
258 # ---------------------------------------------------------------------------
259 # Stress
260 # ---------------------------------------------------------------------------
261
262
263 class TestStress:
264 def test_5_commit_chain_pack(self, tmp_path: pathlib.Path) -> None:
265 repo = _make_repo(tmp_path)
266 sid = _snap(repo)
267 prev: str | None = None
268 for i in range(5):
269 prev = _commit(repo, sid, parent=prev, message=f"commit-{i}")
270 assert prev is not None
271 result = _po(repo, prev, "--dry-run", "--json")
272 assert result.exit_code == 0
273 data = json.loads(result.output)
274 assert data["commits"] == 5
275
276 def test_200_unpack_idempotency_rounds(self, tmp_path: pathlib.Path) -> None:
277 src = _make_repo(tmp_path / "src")
278 dst = _make_repo(tmp_path / "dst")
279 sid = _snap(src)
280 cid = _commit(src, sid)
281 pack_bytes = _po(src, cid).stdout_bytes
282 for i in range(200):
283 result = _uo(dst, pack_bytes)
284 assert result.exit_code == 0, f"failed at iteration {i}"
285
286
287 # ---------------------------------------------------------------------------
288 # Additional security, format, and unit gap-fill tests
289 # ---------------------------------------------------------------------------
290
291
292 class TestPackObjectsSecurity:
293 def test_dry_run_json_has_expected_keys(self, tmp_path: pathlib.Path) -> None:
294 repo = _make_repo(tmp_path)
295 sid = _snap(repo)
296 cid = _commit(repo, sid)
297 _set_head(repo, cid)
298 r = _po(repo, "HEAD", "--dry-run", "--json")
299 assert r.exit_code == 0
300 d = json.loads(r.output)
301 assert "want" in d
302 assert "have" in d
303 assert "commits" in d
304 assert "snapshots" in d
305 assert "objects" in d
306
307 def test_ansi_in_want_rejected(self, tmp_path: pathlib.Path) -> None:
308 repo = _make_repo(tmp_path)
309 r = _po(repo, "\x1b[31m" + "a" * 58 + "\x1b[0m")
310 assert r.exit_code != 0
311 assert "Traceback" not in r.output
312
313 def test_empty_want_list_errors(self, tmp_path: pathlib.Path) -> None:
314 """pack-objects with no want IDs and no HEAD should error gracefully."""
315 repo = _make_repo(tmp_path)
316 r = _po(repo)
317 assert r.exit_code != 0
318
319 def test_200_sequential_dry_run(self, tmp_path: pathlib.Path) -> None:
320 repo = _make_repo(tmp_path)
321 sid = _snap(repo)
322 cid = _commit(repo, sid)
323 _set_head(repo, cid)
324 for i in range(200):
325 r = _po(repo, "HEAD", "--dry-run")
326 assert r.exit_code == 0, f"failed at {i}"
327
328
329 class TestUnpackObjectsSecurity:
330 def test_format_error_to_stderr(self, tmp_path: pathlib.Path) -> None:
331 repo = _make_repo(tmp_path)
332 r = _uo(repo, b"", "--format", "xml")
333 assert r.exit_code != 0
334 assert r.stdout_bytes == b""
335 assert r.stderr.strip() # any error text on stderr
336
337 def test_no_traceback_on_bad_format(self, tmp_path: pathlib.Path) -> None:
338 repo = _make_repo(tmp_path)
339 r = _uo(repo, b"", "--format", "bad")
340 assert "Traceback" not in r.output
341
342 def test_full_round_trip_with_objects(self, tmp_path: pathlib.Path) -> None:
343 """Pack objects included in a snapshot manifest survive the round trip."""
344 src = _make_repo(tmp_path / "src")
345 dst = _make_repo(tmp_path / "dst")
346 content = b"hello round trip"
347 oid = blob_id(content)
348 write_object(src, oid, content)
349 sid = _snap(src, {"hello.txt": oid})
350 cid = _commit(src, sid)
351 pack_bytes = _po(src, cid).stdout_bytes
352 assert len(pack_bytes) > 0
353 r = _uo(dst, pack_bytes)
354 assert r.exit_code == 0
355 # Object should now exist in dst
356 assert object_path(dst, oid).exists()
357
358 def test_unpack_text_output_format(self, tmp_path: pathlib.Path) -> None:
359 src = _make_repo(tmp_path / "src")
360 dst = _make_repo(tmp_path / "dst")
361 sid = _snap(src)
362 cid = _commit(src, sid)
363 pack_bytes = _po(src, cid).stdout_bytes
364 r = _uo(dst, pack_bytes)
365 assert r.exit_code == 0
366 assert "commit" in r.output.lower() or "object" in r.output.lower() or "ok" in r.output.lower()
367
368 def test_no_traceback_on_invalid_msgpack(self, tmp_path: pathlib.Path) -> None:
369 repo = _make_repo(tmp_path)
370 r = _uo(repo, b"\xff\xfe invalid msgpack")
371 assert "Traceback" not in r.output
File History 2 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 133 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 139 days ago