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