gabriel / muse public
test_mv_supercharge.py python
409 lines 15.7 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 127 days ago
1 """Supercharge tests for ``muse mv``.
2
3 Coverage tiers
4 --------------
5 - JSON envelope: exit_code and duration_ms present on success and dry_run
6 - Error payload: errors go to stdout as JSON in --json mode, no dual stderr prose
7 - Data integrity: sha256: OID prefix preserved; source/dest echoed verbatim
8 - TypedDicts: _MvResultJson and _MvErrorJson with required annotations
9 - Docstring: module docstring covers exit_code and duration_ms
10 - No-prose pollution: JSON stdout is valid on all non-error paths
11 - Stress: 50-file move-all with --json, correct stage state
12 """
13 from __future__ import annotations
14 from collections.abc import Mapping
15
16 import datetime
17 import argparse
18 import json
19 import pathlib
20 from typing import get_type_hints
21
22 from muse.core.object_store import write_object
23 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
24 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
25 from muse.core.types import Manifest, blob_id
26 from muse.plugins.code.stage import make_entry, read_stage, write_stage
27 from muse.core.paths import heads_dir, muse_dir
28 from tests.cli_test_helper import CliRunner, InvokeResult
29
30 runner = CliRunner()
31
32 _REPO_ID = "mv-supercharge"
33 _DT = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
34
35
36 # ---------------------------------------------------------------------------
37 # Helpers
38 # ---------------------------------------------------------------------------
39
40
41
42
43 def _init_repo(tmp_path: pathlib.Path) -> pathlib.Path:
44 dot_muse = muse_dir(tmp_path)
45 for d in ("commits", "snapshots", "objects", "refs/heads", "code"):
46 (dot_muse / d).mkdir(parents=True, exist_ok=True)
47 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
48 (dot_muse / "repo.json").write_text(
49 json.dumps({"repo_id": _REPO_ID, "domain": "code"}), encoding="utf-8"
50 )
51 return tmp_path
52
53
54 def _env(root: pathlib.Path) -> Mapping[str, str]:
55 return {"MUSE_REPO_ROOT": str(root)}
56
57
58 def _commit(root: pathlib.Path, files: Mapping[str, bytes]) -> str:
59 manifest: Manifest = {}
60 for rel, content in files.items():
61 oid = blob_id(content)
62 write_object(root, oid, content)
63 manifest[rel] = oid
64 abs_p = root / rel
65 abs_p.parent.mkdir(parents=True, exist_ok=True)
66 abs_p.write_bytes(content)
67 snap_id = compute_snapshot_id(manifest)
68 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest, created_at=_DT))
69 cid = compute_commit_id( parent_ids=[],
70 snapshot_id=snap_id,
71 message="test",
72 committed_at_iso=_DT.isoformat(),
73 )
74 write_commit(root, CommitRecord(
75 commit_id=cid, repo_id=_REPO_ID, branch="main",
76 snapshot_id=snap_id, message="test", committed_at=_DT,
77 ))
78 (heads_dir(root) / "main").write_text(cid, encoding="utf-8")
79 return cid
80
81
82 def _mv(root: pathlib.Path, *args: str) -> InvokeResult:
83 from muse.cli.app import main as cli
84 return runner.invoke(cli, ["mv", *args], env=_env(root))
85
86
87 # ---------------------------------------------------------------------------
88 # JSON envelope — exit_code
89 # ---------------------------------------------------------------------------
90
91
92 class TestJsonEnvelopeExitCode:
93 def test_success_has_exit_code_zero(self, tmp_path: pathlib.Path) -> None:
94 root = _init_repo(tmp_path)
95 _commit(root, {"a.py": b"# a\n"})
96 r = _mv(root, "--json", "a.py", "b.py")
97 assert r.exit_code == 0
98 d = json.loads(r.output)
99 assert "exit_code" in d, "exit_code missing from success envelope"
100 assert d["exit_code"] == 0
101
102 def test_dry_run_has_exit_code_zero(self, tmp_path: pathlib.Path) -> None:
103 root = _init_repo(tmp_path)
104 _commit(root, {"a.py": b"# a\n"})
105 r = _mv(root, "--json", "--dry-run", "a.py", "b.py")
106 assert r.exit_code == 0
107 d = json.loads(r.output)
108 assert "exit_code" in d, "exit_code missing from dry_run envelope"
109 assert d["exit_code"] == 0
110
111 def test_error_payload_has_exit_code_nonzero(self, tmp_path: pathlib.Path) -> None:
112 root = _init_repo(tmp_path)
113 _commit(root, {"anchor.py": b"# a\n"})
114 r = _mv(root, "--json", "ghost.py", "dest.py")
115 d = json.loads(r.output)
116 assert "exit_code" in d
117 assert d["exit_code"] != 0
118
119
120 # ---------------------------------------------------------------------------
121 # JSON envelope — duration_ms
122 # ---------------------------------------------------------------------------
123
124
125 class TestJsonEnvelopeDurationMs:
126 def test_success_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
127 root = _init_repo(tmp_path)
128 _commit(root, {"a.py": b"# a\n"})
129 r = _mv(root, "--json", "a.py", "b.py")
130 d = json.loads(r.output)
131 assert "duration_ms" in d, "duration_ms missing from success envelope"
132 assert isinstance(d["duration_ms"], float)
133 assert d["duration_ms"] >= 0.0
134
135 def test_dry_run_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
136 root = _init_repo(tmp_path)
137 _commit(root, {"a.py": b"# a\n"})
138 r = _mv(root, "--json", "--dry-run", "a.py", "b.py")
139 d = json.loads(r.output)
140 assert "duration_ms" in d, "duration_ms missing from dry_run envelope"
141 assert isinstance(d["duration_ms"], float)
142
143 def test_force_move_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
144 root = _init_repo(tmp_path)
145 _commit(root, {"src.py": b"# s\n", "dst.py": b"# d\n"})
146 r = _mv(root, "--json", "--force", "src.py", "dst.py")
147 d = json.loads(r.output)
148 assert "duration_ms" in d
149
150
151 # ---------------------------------------------------------------------------
152 # Error payload — errors route to stdout as JSON in --json mode
153 # ---------------------------------------------------------------------------
154
155
156 class TestErrorPayload:
157 def test_untracked_source_error_goes_to_stdout(self, tmp_path: pathlib.Path) -> None:
158 root = _init_repo(tmp_path)
159 _commit(root, {"anchor.py": b"# a\n"})
160 (root / "ghost.py").write_text("# untracked\n")
161 r = _mv(root, "--json", "ghost.py", "dest.py")
162 assert r.exit_code != 0
163 assert not r.stderr.strip(), f"unexpected stderr: {r.stderr!r}"
164 d = json.loads(r.output)
165 assert d["status"] == "error"
166
167 def test_missing_disk_source_error_goes_to_stdout(self, tmp_path: pathlib.Path) -> None:
168 root = _init_repo(tmp_path)
169 _commit(root, {"a.py": b"# a\n"})
170 (root / "a.py").unlink()
171 r = _mv(root, "--json", "a.py", "b.py")
172 assert r.exit_code != 0
173 assert not r.stderr.strip(), f"unexpected stderr: {r.stderr!r}"
174 d = json.loads(r.output)
175 assert d["status"] == "error"
176
177 def test_dest_already_tracked_error_goes_to_stdout(self, tmp_path: pathlib.Path) -> None:
178 root = _init_repo(tmp_path)
179 _commit(root, {"src.py": b"# s\n", "dst.py": b"# d\n"})
180 r = _mv(root, "--json", "src.py", "dst.py")
181 assert r.exit_code != 0
182 assert not r.stderr.strip(), f"unexpected stderr: {r.stderr!r}"
183 d = json.loads(r.output)
184 assert d["status"] == "error"
185
186 def test_dest_exists_on_disk_error_goes_to_stdout(self, tmp_path: pathlib.Path) -> None:
187 root = _init_repo(tmp_path)
188 _commit(root, {"src.py": b"# s\n"})
189 (root / "dst.py").write_text("# existing\n")
190 r = _mv(root, "--json", "src.py", "dst.py")
191 assert r.exit_code != 0
192 assert not r.stderr.strip(), f"unexpected stderr: {r.stderr!r}"
193 d = json.loads(r.output)
194 assert d["status"] == "error"
195
196 def test_path_traversal_error_goes_to_stderr_in_text_mode(self, tmp_path: pathlib.Path) -> None:
197 """Path traversal in text mode is caught before --json is parsed — stderr is OK."""
198 root = _init_repo(tmp_path)
199 _commit(root, {"anchor.py": b"# a\n"})
200 r = _mv(root, "../../../etc/passwd", "dest.py")
201 assert r.exit_code != 0
202
203 def test_error_payload_has_status_error(self, tmp_path: pathlib.Path) -> None:
204 root = _init_repo(tmp_path)
205 _commit(root, {"anchor.py": b"# a\n"})
206 r = _mv(root, "--json", "ghost.py", "dest.py")
207 d = json.loads(r.output)
208 assert d["status"] == "error"
209
210 def test_error_payload_has_error_field(self, tmp_path: pathlib.Path) -> None:
211 root = _init_repo(tmp_path)
212 _commit(root, {"anchor.py": b"# a\n"})
213 r = _mv(root, "--json", "ghost.py", "dest.py")
214 d = json.loads(r.output)
215 assert "error" in d
216 assert d["error"]
217
218 def test_no_emoji_on_stderr_in_json_mode(self, tmp_path: pathlib.Path) -> None:
219 root = _init_repo(tmp_path)
220 _commit(root, {"anchor.py": b"# a\n"})
221 r = _mv(root, "--json", "ghost.py", "dest.py")
222 assert "❌" not in r.stderr
223
224
225 # ---------------------------------------------------------------------------
226 # Data integrity
227 # ---------------------------------------------------------------------------
228
229
230 class TestDataIntegrity:
231 def test_object_id_is_sha256_prefixed(self, tmp_path: pathlib.Path) -> None:
232 """object_id in JSON response must carry sha256: prefix."""
233 root = _init_repo(tmp_path)
234 content = b"# content\n"
235 _commit(root, {"a.py": content})
236 r = _mv(root, "--json", "a.py", "b.py")
237 d = json.loads(r.output)
238 assert d["object_id"].startswith("sha256:"), (
239 f"object_id missing sha256: prefix: {d['object_id']!r}"
240 )
241
242 def test_source_dest_echoed_verbatim(self, tmp_path: pathlib.Path) -> None:
243 root = _init_repo(tmp_path)
244 _commit(root, {"src/module.py": b"# m\n"})
245 r = _mv(root, "--json", "src/module.py", "lib/module.py")
246 d = json.loads(r.output)
247 assert d["source"] == "src/module.py"
248 assert d["dest"] == "lib/module.py"
249
250 def test_object_id_matches_staged_entry(self, tmp_path: pathlib.Path) -> None:
251 """object_id in JSON must match what was written to the stage."""
252 root = _init_repo(tmp_path)
253 content = b"# exact\n"
254 _commit(root, {"x.py": content})
255 r = _mv(root, "--json", "x.py", "y.py")
256 d = json.loads(r.output)
257 stage = read_stage(root)
258 assert stage["y.py"]["object_id"] == d["object_id"]
259
260 def test_staged_only_move_object_id_preserved(self, tmp_path: pathlib.Path) -> None:
261 """Staged-only file move must echo the correct object_id."""
262 root = _init_repo(tmp_path)
263 _commit(root, {"anchor.py": b"# anchor\n"})
264 content = b"# staged only\n"
265 oid = blob_id(content)
266 write_object(root, oid, content)
267 (root / "new.py").write_bytes(content)
268 stage = read_stage(root)
269 stage["new.py"] = make_entry(oid, "A")
270 write_stage(root, stage)
271 r = _mv(root, "--json", "new.py", "renamed.py")
272 d = json.loads(r.output)
273 assert d["object_id"] == oid
274
275
276 # ---------------------------------------------------------------------------
277 # No-prose pollution
278 # ---------------------------------------------------------------------------
279
280
281 class TestNoProsePollution:
282 def test_success_stdout_is_valid_json(self, tmp_path: pathlib.Path) -> None:
283 root = _init_repo(tmp_path)
284 _commit(root, {"a.py": b"# a\n"})
285 r = _mv(root, "--json", "a.py", "b.py")
286 json.loads(r.output) # must not raise
287
288 def test_dry_run_stdout_is_valid_json(self, tmp_path: pathlib.Path) -> None:
289 root = _init_repo(tmp_path)
290 _commit(root, {"a.py": b"# a\n"})
291 r = _mv(root, "--json", "--dry-run", "a.py", "b.py")
292 json.loads(r.output) # must not raise
293
294 def test_no_emoji_in_success_json(self, tmp_path: pathlib.Path) -> None:
295 root = _init_repo(tmp_path)
296 _commit(root, {"a.py": b"# a\n"})
297 r = _mv(root, "--json", "a.py", "b.py")
298 assert "✅" not in r.output
299 assert "❌" not in r.output
300
301
302 # ---------------------------------------------------------------------------
303 # TypedDicts
304 # ---------------------------------------------------------------------------
305
306
307 class TestTypedDicts:
308 def test_mv_result_json_typeddict_exists(self) -> None:
309 from muse.cli.commands.mv import _MvResultJson
310 assert _MvResultJson is not None
311
312 def test_mv_error_json_typeddict_exists(self) -> None:
313 from muse.cli.commands.mv import _MvErrorJson
314 assert _MvErrorJson is not None
315
316 def test_mv_result_json_has_exit_code_annotation(self) -> None:
317 from muse.cli.commands.mv import _MvResultJson
318 hints = get_type_hints(_MvResultJson)
319 assert "exit_code" in hints
320
321 def test_mv_result_json_has_duration_ms_annotation(self) -> None:
322 from muse.cli.commands.mv import _MvResultJson
323 hints = get_type_hints(_MvResultJson)
324 assert "duration_ms" in hints
325
326 def test_mv_error_json_has_required_fields(self) -> None:
327 from muse.cli.commands.mv import _MvErrorJson
328 hints = get_type_hints(_MvErrorJson)
329 for field in ("status", "error", "exit_code"):
330 assert field in hints, f"Missing annotation: {field!r}"
331
332
333 # ---------------------------------------------------------------------------
334 # Docstring coverage
335 # ---------------------------------------------------------------------------
336
337
338 class TestDocstring:
339 def _doc(self) -> str:
340 import muse.cli.commands.mv as mod
341 return mod.__doc__ or ""
342
343 def test_docstring_documents_exit_code(self) -> None:
344 assert "exit_code" in self._doc()
345
346 def test_docstring_documents_duration_ms(self) -> None:
347 assert "duration_ms" in self._doc()
348
349
350 # ---------------------------------------------------------------------------
351 # Stress
352 # ---------------------------------------------------------------------------
353
354
355 class TestStress:
356 def test_50_files_move_all_json(self, tmp_path: pathlib.Path) -> None:
357 """Move all 50 tracked files with --json; all envelopes must be valid."""
358 root = _init_repo(tmp_path)
359 n = 50
360 files = {f"f{i:03d}.py": f"# {i}\n".encode() for i in range(n)}
361 _commit(root, files)
362
363 for i in range(n):
364 r = _mv(root, "--json", f"f{i:03d}.py", f"moved/f{i:03d}.py")
365 assert r.exit_code == 0, f"move {i} failed: {r.output}"
366 d = json.loads(r.output)
367 assert d["exit_code"] == 0
368 assert "duration_ms" in d
369 assert d["source"] == f"f{i:03d}.py"
370 assert d["dest"] == f"moved/f{i:03d}.py"
371
372 stage = read_stage(root)
373 for i in range(n):
374 assert stage[f"f{i:03d}.py"]["mode"] == "D"
375 assert stage[f"moved/f{i:03d}.py"]["mode"] == "A"
376
377
378 # ---------------------------------------------------------------------------
379 # TestRegisterFlags — argparse-level verification
380 # ---------------------------------------------------------------------------
381
382
383 class TestRegisterFlags:
384 """Verify that register() wires --json / -j correctly."""
385
386 def _make_parser(self) -> "argparse.ArgumentParser":
387 import argparse
388 from muse.cli.commands.mv import register
389 ap = argparse.ArgumentParser()
390 subs = ap.add_subparsers()
391 register(subs)
392 return ap
393
394 def test_json_flag_long(self) -> None:
395 ns = self._make_parser().parse_args(["mv", "src.py", "dst.py", "--json"])
396 assert ns.json_out is True
397
398 def test_j_alias(self) -> None:
399 ns = self._make_parser().parse_args(["mv", "src.py", "dst.py", "-j"])
400 assert ns.json_out is True
401
402 def test_default_is_text(self) -> None:
403 ns = self._make_parser().parse_args(["mv", "src.py", "dst.py"])
404 assert ns.json_out is False
405
406 def test_dest_is_json_out(self) -> None:
407 ns = self._make_parser().parse_args(["mv", "src.py", "dst.py", "-j"])
408 assert hasattr(ns, "json_out")
409 assert not hasattr(ns, "fmt")
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 127 days ago