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