gabriel / muse public
test_name_rev_supercharge.py python
359 lines 13.3 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 121 days ago
1 """Supercharge tests for ``muse name-rev``.
2
3 Coverage tiers
4 --------------
5 - JSON envelope: exit_code and duration_ms present on all resolution outcomes
6 - Error payload: errors go to stdout as JSON in --json mode, no dual stderr prose
7 - Prefix resolution: bare hex short-prefix matches sha256:-prefixed keys in name_map
8 - TypedDicts: _NameRevJson and _NameRevErrorJson 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: 100-commit chain, all entries have exit_code and duration_ms
12 """
13 from __future__ import annotations
14 from collections.abc import Mapping
15
16 import datetime
17 import json
18 import pathlib
19 from typing import get_type_hints
20
21 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
22 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
23 from muse.core.paths import ref_path, muse_dir
24 from tests.cli_test_helper import CliRunner, InvokeResult
25
26 runner = CliRunner()
27
28 _DT = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
29
30
31 # ---------------------------------------------------------------------------
32 # Helpers
33 # ---------------------------------------------------------------------------
34
35
36 def _init_repo(tmp_path: pathlib.Path) -> pathlib.Path:
37 dot_muse = muse_dir(tmp_path)
38 for d in ("commits", "snapshots", "objects", "refs/heads"):
39 (dot_muse / d).mkdir(parents=True, exist_ok=True)
40 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
41 (dot_muse / "repo.json").write_text(
42 json.dumps({"repo_id": "nr-supercharge", "domain": "midi"}), encoding="utf-8"
43 )
44 return tmp_path
45
46
47 def _env(root: pathlib.Path) -> Mapping[str, str]:
48 return {"MUSE_REPO_ROOT": str(root)}
49
50
51 def _commit(
52 root: pathlib.Path,
53 msg: str,
54 branch: str = "main",
55 parent: str | None = None,
56 ) -> str:
57 sid = compute_snapshot_id({})
58 write_snapshot(root, SnapshotRecord(snapshot_id=sid, manifest={}, created_at=_DT))
59 parent_ids = [parent] if parent else []
60 cid = compute_commit_id( parent_ids=parent_ids,
61 snapshot_id=sid,
62 message=msg,
63 committed_at_iso=_DT.isoformat(),
64 )
65 write_commit(root, CommitRecord(
66 commit_id=cid, repo_id="nr-supercharge", branch=branch,
67 snapshot_id=sid, message=msg, committed_at=_DT,
68 parent_commit_id=parent,
69 ))
70 ref = ref_path(root, branch)
71 ref.parent.mkdir(parents=True, exist_ok=True)
72 ref.write_text(cid, encoding="utf-8")
73 return cid
74
75
76 def _nr(root: pathlib.Path, *args: str, stdin: str | None = None) -> InvokeResult:
77 from muse.cli.app import main as cli
78 extra = [] if "--json" in args or "-j" in args else ["--json"]
79 return runner.invoke(cli, ["name-rev", *extra, *args], env=_env(root), input=stdin)
80
81
82 def _hex_prefix(cid: str, n: int = 8) -> str:
83 """Extract n hex chars from a sha256:-prefixed commit ID."""
84 return cid[len("sha256:"):len("sha256:") + n]
85
86
87 # ---------------------------------------------------------------------------
88 # JSON envelope — exit_code
89 # ---------------------------------------------------------------------------
90
91
92 class TestJsonEnvelopeExitCode:
93 def test_found_has_exit_code_zero(self, tmp_path: pathlib.Path) -> None:
94 root = _init_repo(tmp_path)
95 cid = _commit(root, "c1")
96 r = _nr(root, cid)
97 assert r.exit_code == 0
98 d = json.loads(r.output)
99 assert "exit_code" in d, "exit_code missing from found envelope"
100 assert d["exit_code"] == 0
101
102 def test_undefined_result_still_exit_code_zero(self, tmp_path: pathlib.Path) -> None:
103 """undefined entries are a valid outcome — exit_code must still be 0."""
104 root = _init_repo(tmp_path)
105 _commit(root, "c1")
106 fake = "a" * 64
107 r = _nr(root, fake)
108 assert r.exit_code == 0
109 d = json.loads(r.output)
110 assert "exit_code" in d
111 assert d["exit_code"] == 0
112
113 def test_mixed_found_and_undefined_exit_code_zero(self, tmp_path: pathlib.Path) -> None:
114 root = _init_repo(tmp_path)
115 cid = _commit(root, "c1")
116 fake = "b" * 64
117 r = _nr(root, cid, fake)
118 assert r.exit_code == 0
119 d = json.loads(r.output)
120 assert d["exit_code"] == 0
121
122
123 # ---------------------------------------------------------------------------
124 # JSON envelope — duration_ms
125 # ---------------------------------------------------------------------------
126
127
128 class TestJsonEnvelopeDurationMs:
129 def test_found_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
130 root = _init_repo(tmp_path)
131 cid = _commit(root, "c1")
132 r = _nr(root, cid)
133 d = json.loads(r.output)
134 assert "duration_ms" in d, "duration_ms missing from found envelope"
135 assert isinstance(d["duration_ms"], float)
136 assert d["duration_ms"] >= 0.0
137
138 def test_undefined_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
139 root = _init_repo(tmp_path)
140 _commit(root, "c1")
141 r = _nr(root, "a" * 64)
142 d = json.loads(r.output)
143 assert "duration_ms" in d
144
145 def test_branches_filter_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
146 root = _init_repo(tmp_path)
147 cid = _commit(root, "c1", "main")
148 r = _nr(root, cid, "--branches", "main")
149 d = json.loads(r.output)
150 assert "duration_ms" in d
151 assert isinstance(d["duration_ms"], float)
152
153
154 # ---------------------------------------------------------------------------
155 # Error payload — errors route to stdout as JSON in --json mode
156 # ---------------------------------------------------------------------------
157
158
159 class TestErrorPayload:
160 def test_no_inputs_error_goes_to_stdout(self, tmp_path: pathlib.Path) -> None:
161 root = _init_repo(tmp_path)
162 r = _nr(root, "--json")
163 assert r.exit_code != 0
164 assert not r.stderr.strip(), f"unexpected stderr: {r.stderr!r}"
165 d = json.loads(r.output)
166 assert d["status"] == "error"
167
168 def test_invalid_hex_error_goes_to_stdout(self, tmp_path: pathlib.Path) -> None:
169 root = _init_repo(tmp_path)
170 r = _nr(root, "--json", "not-hex!")
171 assert r.exit_code != 0
172 assert not r.stderr.strip(), f"unexpected stderr: {r.stderr!r}"
173 d = json.loads(r.output)
174 assert d["status"] == "error"
175
176 def test_bad_max_walk_error_goes_to_stdout(self, tmp_path: pathlib.Path) -> None:
177 root = _init_repo(tmp_path)
178 cid = _commit(root, "c1")
179 r = _nr(root, "--json", cid, "--max-walk", "0")
180 assert r.exit_code != 0
181 assert not r.stderr.strip(), f"unexpected stderr: {r.stderr!r}"
182 d = json.loads(r.output)
183 assert d["status"] == "error"
184
185 def test_error_payload_has_status_error(self, tmp_path: pathlib.Path) -> None:
186 root = _init_repo(tmp_path)
187 r = _nr(root, "--json", "not-valid!")
188 d = json.loads(r.output)
189 assert d["status"] == "error"
190
191 def test_error_payload_has_exit_code(self, tmp_path: pathlib.Path) -> None:
192 root = _init_repo(tmp_path)
193 r = _nr(root, "--json", "not-valid!")
194 d = json.loads(r.output)
195 assert "exit_code" in d
196 assert d["exit_code"] != 0
197
198 def test_error_payload_has_error_field(self, tmp_path: pathlib.Path) -> None:
199 root = _init_repo(tmp_path)
200 r = _nr(root, "--json")
201 d = json.loads(r.output)
202 assert "error" in d
203 assert d["error"]
204
205 def test_no_emoji_on_stderr_in_json_mode(self, tmp_path: pathlib.Path) -> None:
206 root = _init_repo(tmp_path)
207 r = _nr(root, "--json", "not-valid!")
208 assert "❌" not in r.stderr
209
210
211 # ---------------------------------------------------------------------------
212 # Prefix resolution — bare hex prefix against sha256:-prefixed keys
213 # ---------------------------------------------------------------------------
214
215
216 class TestPrefixResolutionSha256:
217 def test_bare_hex_8char_prefix_resolves(self, tmp_path: pathlib.Path) -> None:
218 """Bare 8-char hex prefix must resolve against sha256:-prefixed name_map keys."""
219 root = _init_repo(tmp_path)
220 cid = _commit(root, "c1")
221 prefix = _hex_prefix(cid, 8) # first 8 hex chars, no sha256: prefix
222 r = _nr(root, prefix)
223 assert r.exit_code == 0
224 entry = json.loads(r.output)["results"][0]
225 assert entry["commit_id"] == cid
226 assert entry["undefined"] is False
227
228 def test_bare_hex_4char_prefix_resolves(self, tmp_path: pathlib.Path) -> None:
229 """4-char bare hex prefix must resolve when unambiguous."""
230 root = _init_repo(tmp_path)
231 cid = _commit(root, "unique-c1-msg")
232 prefix = _hex_prefix(cid, 4)
233 r = _nr(root, prefix)
234 assert r.exit_code == 0
235 entry = json.loads(r.output)["results"][0]
236 # Either resolves to cid or is ambiguous — must not crash or error
237 assert entry["input"] == prefix
238 assert r.exit_code == 0
239
240 def test_sha256_prefixed_short_id_resolves(self, tmp_path: pathlib.Path) -> None:
241 """sha256:-prefixed short IDs (e.g. sha256:abcd1234) must also resolve."""
242 root = _init_repo(tmp_path)
243 cid = _commit(root, "c1")
244 short = cid[:len("sha256:") + 8] # keep sha256: + 8 hex chars
245 r = _nr(root, short)
246 assert r.exit_code == 0
247 entry = json.loads(r.output)["results"][0]
248 assert entry["commit_id"] == cid
249
250 def test_input_field_preserves_bare_hex_prefix(self, tmp_path: pathlib.Path) -> None:
251 """input field echoes the caller's original value, not the resolved full ID."""
252 root = _init_repo(tmp_path)
253 cid = _commit(root, "c1")
254 prefix = _hex_prefix(cid, 8)
255 r = _nr(root, prefix)
256 entry = json.loads(r.output)["results"][0]
257 assert entry["input"] == prefix
258
259
260 # ---------------------------------------------------------------------------
261 # No-prose pollution
262 # ---------------------------------------------------------------------------
263
264
265 class TestNoProsePollution:
266 def test_found_stdout_is_valid_json(self, tmp_path: pathlib.Path) -> None:
267 root = _init_repo(tmp_path)
268 cid = _commit(root, "c1")
269 r = _nr(root, cid)
270 json.loads(r.output) # must not raise
271
272 def test_undefined_stdout_is_valid_json(self, tmp_path: pathlib.Path) -> None:
273 root = _init_repo(tmp_path)
274 _commit(root, "c1")
275 json.loads(_nr(root, "a" * 64).output)
276
277 def test_no_emoji_in_success_json(self, tmp_path: pathlib.Path) -> None:
278 root = _init_repo(tmp_path)
279 cid = _commit(root, "c1")
280 r = _nr(root, cid)
281 assert "✅" not in r.output
282 assert "❌" not in r.output
283
284
285 # ---------------------------------------------------------------------------
286 # TypedDicts
287 # ---------------------------------------------------------------------------
288
289
290 class TestTypedDicts:
291 def test_name_rev_json_typeddict_exists(self) -> None:
292 from muse.cli.commands.name_rev import _NameRevJson
293 assert _NameRevJson is not None
294
295 def test_name_rev_error_json_typeddict_exists(self) -> None:
296 from muse.cli.commands.name_rev import _NameRevErrorJson
297 assert _NameRevErrorJson is not None
298
299 def test_name_rev_json_has_exit_code_annotation(self) -> None:
300 from muse.cli.commands.name_rev import _NameRevJson
301 hints = get_type_hints(_NameRevJson)
302 assert "exit_code" in hints
303
304 def test_name_rev_json_has_duration_ms_annotation(self) -> None:
305 from muse.cli.commands.name_rev import _NameRevJson
306 hints = get_type_hints(_NameRevJson)
307 assert "duration_ms" in hints
308
309 def test_name_rev_error_json_has_required_fields(self) -> None:
310 from muse.cli.commands.name_rev import _NameRevErrorJson
311 hints = get_type_hints(_NameRevErrorJson)
312 for field in ("status", "error", "exit_code"):
313 assert field in hints, f"Missing annotation: {field!r}"
314
315
316 # ---------------------------------------------------------------------------
317 # Docstring coverage
318 # ---------------------------------------------------------------------------
319
320
321 class TestDocstring:
322 def _doc(self) -> str:
323 import muse.cli.commands.name_rev as mod
324 return mod.__doc__ or ""
325
326 def test_docstring_documents_exit_code(self) -> None:
327 assert "exit_code" in self._doc()
328
329 def test_docstring_documents_duration_ms(self) -> None:
330 assert "duration_ms" in self._doc()
331
332
333 # ---------------------------------------------------------------------------
334 # Stress
335 # ---------------------------------------------------------------------------
336
337
338 class TestStress:
339 def test_100_commit_chain_all_have_envelope_fields(self, tmp_path: pathlib.Path) -> None:
340 root = _init_repo(tmp_path)
341 parent: str | None = None
342 commits: list[str] = []
343 for i in range(100):
344 cid = _commit(root, f"c{i:03d}", parent=parent)
345 commits.append(cid)
346 parent = cid
347
348 r = _nr(root, commits[0], commits[49], commits[-1])
349 assert r.exit_code == 0
350 d = json.loads(r.output)
351 assert "exit_code" in d
352 assert "duration_ms" in d
353 assert d["exit_code"] == 0
354 assert isinstance(d["duration_ms"], float)
355 # Tip at distance 0, midpoint at 50, root at 99
356 by_id = {e["commit_id"]: e for e in d["results"]}
357 assert by_id[commits[-1]]["distance"] == 0
358 assert by_id[commits[49]]["distance"] == 50
359 assert by_id[commits[0]]["distance"] == 99
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 121 days ago