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