gabriel / muse public
test_mpack_cmd_pack_unpack.py python
371 lines 13.2 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 121 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 muse.core.paths import heads_dir, muse_dir
24 from tests.cli_test_helper import CliRunner, InvokeResult
25
26 runner = CliRunner()
27
28
29 # ---------------------------------------------------------------------------
30 # Helpers
31 # ---------------------------------------------------------------------------
32
33 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
34 repo = tmp_path / "repo"
35 dot_muse = muse_dir(repo)
36 for sub in ("objects", "commits", "snapshots", "refs/heads"):
37 (dot_muse / sub).mkdir(parents=True)
38 (dot_muse / "HEAD").write_text("ref: refs/heads/main")
39 (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"}))
40 return repo
41
42
43 def _snap(repo: pathlib.Path, manifest: Manifest | None = None) -> str:
44 m = manifest or {}
45 snap_id = compute_snapshot_id(m)
46 write_snapshot(repo, SnapshotRecord(
47 snapshot_id=snap_id,
48 manifest=m,
49 created_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
50 ))
51 return snap_id
52
53
54 def _commit(
55 repo: pathlib.Path,
56 snap_id: str,
57 *,
58 parent: str | None = None,
59 message: str = "test",
60 ) -> str:
61 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
62 parent_ids: list[str] = [parent] if parent else []
63 commit_id = compute_commit_id(
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 repo_id="test-repo",
71 commit_id=commit_id,
72 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 = heads_dir(repo) / "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, f"\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 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 121 days ago