gabriel / muse public
test_cmd_rev_parse.py python
540 lines 21.1 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 121 days ago
1 """Comprehensive tests for ``muse rev-parse``.
2
3 Coverage tiers
4 --------------
5 - Integration: branch, HEAD, SHA prefix, full SHA, --abbrev-ref, --format text
6 - Edge cases: empty repo (no commits), empty ref, ambiguous prefix, HEAD→branch
7 - Security: ANSI/control chars in ref → JSON-escaped, empty ref clean error
8 - Stress: 200 rapid resolves
9 """
10 from __future__ import annotations
11 from collections.abc import Mapping
12
13 import datetime
14 import json
15 import pathlib
16
17 import pytest
18 from muse.core.errors import ExitCode
19 from muse.core.object_store import write_object
20 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
21 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
22 from muse.core.types import Manifest, long_id, split_id
23 from muse.core.paths import muse_dir, ref_path
24 from tests.cli_test_helper import CliRunner, InvokeResult
25
26 runner = CliRunner()
27
28 # ---------------------------------------------------------------------------
29 # Helpers
30 # ---------------------------------------------------------------------------
31
32 def _make_repo(tmp_path: pathlib.Path, branch: str = "main") -> pathlib.Path:
33 repo = tmp_path / "repo"
34 dot_muse = muse_dir(repo)
35 for sub in ("objects", "commits", "snapshots", "refs/heads"):
36 (dot_muse / sub).mkdir(parents=True)
37 (dot_muse / "HEAD").write_text(f"ref: refs/heads/{branch}")
38 (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "test", "domain": "code"}))
39 return repo
40
41
42 _TS = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
43
44
45 def _store_snap(repo: pathlib.Path, manifest: Manifest | None = None) -> str:
46 sid = compute_snapshot_id(manifest or {})
47 write_snapshot(repo, SnapshotRecord(
48 snapshot_id=sid,
49 manifest=manifest or {},
50 created_at=_TS,
51 ))
52 return sid
53
54
55 def _make_commit(
56 repo: pathlib.Path,
57 snapshot_id: str,
58 *,
59 branch: str = "main",
60 parent: str | None = None,
61 message: str = "test",
62 ) -> str:
63 parents = [parent] if parent else []
64 cid = compute_commit_id(
65 parent_ids=parents,
66 snapshot_id=snapshot_id,
67 message=message,
68 committed_at_iso=_TS.isoformat(),
69 author="tester",
70 )
71 rec = CommitRecord(
72 repo_id="test",
73 commit_id=cid,
74 branch=branch,
75 snapshot_id=snapshot_id,
76 message=message,
77 committed_at=_TS,
78 author="tester",
79 parent_commit_id=parent,
80 )
81 write_commit(repo, rec)
82 return cid
83
84
85 def _set_head(repo: pathlib.Path, branch: str, commit_id: str) -> None:
86 ref = ref_path(repo, branch)
87 ref.parent.mkdir(parents=True, exist_ok=True)
88 ref.write_text(commit_id)
89
90
91 def _rev(repo: pathlib.Path, *args: str) -> InvokeResult:
92 from muse.cli.app import main as cli
93 return runner.invoke(
94 cli,
95 ["rev-parse", *args],
96 env={"MUSE_REPO_ROOT": str(repo)},
97 )
98
99
100 def _populated_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]:
101 """Return (repo, commit_id) with one commit on main, using real content-addressed IDs."""
102 repo = _make_repo(tmp_path)
103 sid = _store_snap(repo)
104 cid = _make_commit(repo, sid)
105 _set_head(repo, "main", cid)
106 return repo, cid
107
108
109 # ---------------------------------------------------------------------------
110 # Integration — branch resolution
111 # ---------------------------------------------------------------------------
112
113
114 # ---------------------------------------------------------------------------
115 # New: default format is text, --json makes it meaningful
116 # ---------------------------------------------------------------------------
117
118
119 class TestDefaultFormat:
120 def test_default_output_is_text(self, tmp_path: pathlib.Path) -> None:
121 """Without --json the output is a plain commit ID."""
122 repo, cid = _populated_repo(tmp_path)
123 result = _rev(repo, "main")
124 assert result.exit_code == 0
125 assert result.output.strip() == cid
126
127 def test_no_flags_output_is_not_json(self, tmp_path: pathlib.Path) -> None:
128 """Default plain-text output is not parseable as JSON."""
129 repo, cid = _populated_repo(tmp_path)
130 result = _rev(repo, "main")
131 assert result.exit_code == 0
132 with pytest.raises((json.JSONDecodeError, ValueError)):
133 json.loads(result.output)
134
135 def test_json_flag_gives_dict_output(self, tmp_path: pathlib.Path) -> None:
136 """With --json output is a dict."""
137 repo, cid = _populated_repo(tmp_path)
138 result = _rev(repo, "--json", "main")
139 assert result.exit_code == 0
140 data = json.loads(result.output)
141 assert data["commit_id"] == cid
142 assert data["ref"] == "main"
143
144 def test_text_vs_json_differ(self, tmp_path: pathlib.Path) -> None:
145 """Plain text and --json outputs differ in structure."""
146 repo, cid = _populated_repo(tmp_path)
147 text_result = _rev(repo, "main")
148 json_result = _rev(repo, "--json", "main")
149 assert text_result.output.strip() == cid
150 assert json.loads(json_result.output)["commit_id"] == cid
151
152
153 # ---------------------------------------------------------------------------
154 # New: sha256: prefix is required; bare hex is rejected
155 # ---------------------------------------------------------------------------
156
157
158 class TestSha256PrefixRequired:
159 def test_bare_full_hex_rejected(self, tmp_path: pathlib.Path) -> None:
160 """64-char bare hex without sha256: prefix must be rejected."""
161 repo, cid = _populated_repo(tmp_path)
162 result = _rev(repo, split_id(cid)[1])
163 assert result.exit_code == ExitCode.USER_ERROR
164 data = json.loads(result.output)
165 assert "sha256:" in data["error"]
166
167 def test_bare_short_hex_rejected(self, tmp_path: pathlib.Path) -> None:
168 """Short bare hex without sha256: prefix must be rejected."""
169 repo, cid = _populated_repo(tmp_path)
170 result = _rev(repo, split_id(cid)[1][:8]) # 8 bare hex chars, no prefix
171 assert result.exit_code == ExitCode.USER_ERROR
172 data = json.loads(result.output)
173 assert "sha256:" in data["error"]
174
175 def test_canonical_full_id_resolves(self, tmp_path: pathlib.Path) -> None:
176 """sha256:<64hex> must resolve to the commit."""
177 repo, cid = _populated_repo(tmp_path)
178 result = _rev(repo, "--json", cid)
179 assert result.exit_code == 0
180 assert json.loads(result.output)["commit_id"] == cid
181
182 def test_canonical_prefix_resolves(self, tmp_path: pathlib.Path) -> None:
183 """sha256:<8hex> prefix must resolve to the commit."""
184 repo, cid = _populated_repo(tmp_path)
185 prefix = long_id(split_id(cid)[1][:8])# sha256: + 8 hex chars
186 result = _rev(repo, "--json", prefix)
187 assert result.exit_code == 0
188 assert json.loads(result.output)["commit_id"] == cid
189
190
191 # ---------------------------------------------------------------------------
192 # Integration — branch resolution
193 # ---------------------------------------------------------------------------
194
195
196 class TestBranchResolution:
197 def test_resolve_branch_json(self, tmp_path: pathlib.Path) -> None:
198 repo, cid = _populated_repo(tmp_path)
199 result = _rev(repo, "--json", "main")
200 assert result.exit_code == 0
201 data = json.loads(result.output)
202 assert data["commit_id"] == cid
203 assert data["ref"] == "main"
204
205 def test_resolve_branch_text(self, tmp_path: pathlib.Path) -> None:
206 repo, cid = _populated_repo(tmp_path)
207 result = _rev(repo, "main")
208 assert result.exit_code == 0
209 assert result.output.strip() == cid
210
211 def test_json_flag_shorthand(self, tmp_path: pathlib.Path) -> None:
212 repo, cid = _populated_repo(tmp_path)
213 result = _rev(repo, "--json", "main")
214 assert result.exit_code == 0
215 data = json.loads(result.output)
216 assert data["commit_id"] == cid
217
218 def test_unknown_branch_not_found(self, tmp_path: pathlib.Path) -> None:
219 repo = _make_repo(tmp_path)
220 result = _rev(repo, "nonexistent-branch")
221 assert result.exit_code == ExitCode.USER_ERROR
222 data = json.loads(result.output)
223 assert data["commit_id"] is None
224 assert data["error"] == "not found"
225
226
227 # ---------------------------------------------------------------------------
228 # Integration — HEAD resolution
229 # ---------------------------------------------------------------------------
230
231
232 class TestHeadResolution:
233 def test_resolve_head(self, tmp_path: pathlib.Path) -> None:
234 repo, cid = _populated_repo(tmp_path)
235 result = _rev(repo, "--json", "HEAD")
236 assert result.exit_code == 0
237 data = json.loads(result.output)
238 assert data["commit_id"] == cid
239
240 def test_head_lowercase_also_resolves(self, tmp_path: pathlib.Path) -> None:
241 """HEAD resolution is case-insensitive (matches git behaviour)."""
242 repo, cid = _populated_repo(tmp_path)
243 result = _rev(repo, "--json", "head")
244 assert result.exit_code == 0
245 data = json.loads(result.output)
246 assert data["commit_id"] == cid
247
248 def test_head_on_empty_repo_errors(self, tmp_path: pathlib.Path) -> None:
249 """HEAD on a repo with no commits should error cleanly."""
250 repo = _make_repo(tmp_path)
251 result = _rev(repo, "HEAD")
252 assert result.exit_code == ExitCode.USER_ERROR
253 data = json.loads(result.output)
254 assert data["commit_id"] is None
255 assert "no commits" in data["error"]
256
257
258 # ---------------------------------------------------------------------------
259 # Integration — SHA prefix resolution
260 # ---------------------------------------------------------------------------
261
262
263 class TestShaResolution:
264 def test_resolve_full_sha(self, tmp_path: pathlib.Path) -> None:
265 repo, cid = _populated_repo(tmp_path)
266 result = _rev(repo, "--json", cid)
267 assert result.exit_code == 0
268 data = json.loads(result.output)
269 assert data["commit_id"] == cid
270
271 def test_resolve_8char_prefix(self, tmp_path: pathlib.Path) -> None:
272 repo, cid = _populated_repo(tmp_path)
273 prefix = long_id(split_id(cid)[1][:8])# sha256: + first 8 hex chars
274 result = _rev(repo, "--json", prefix)
275 assert result.exit_code == 0
276 data = json.loads(result.output)
277 assert data["commit_id"] == cid
278
279 def test_ambiguous_prefix_returns_candidates(self, tmp_path: pathlib.Path) -> None:
280 """Two commits sharing a prefix → error with candidates list."""
281 # Messages "commit-search-121" and "commit-search-154" produce IDs
282 # sharing the 4-char hex prefix "9d05" (same snapshot, same timestamp).
283 _AMBIG_MSG_1 = "commit-search-121"
284 _AMBIG_MSG_2 = "commit-search-154"
285 _AMBIG_PREFIX = "9d05"
286
287 repo = _make_repo(tmp_path)
288 sid = _store_snap(repo)
289 cid1 = _make_commit(repo, sid, branch="main", message=_AMBIG_MSG_1)
290 cid2 = _make_commit(repo, sid, branch="dev", message=_AMBIG_MSG_2)
291 # cid1/cid2 are sha256:<hex>; compare the hex portion only
292 assert split_id(cid1)[1][:4] == split_id(cid2)[1][:4] == _AMBIG_PREFIX
293 _set_head(repo, "main", cid1)
294 _set_head(repo, "dev", cid2)
295
296 result = _rev(repo, long_id(_AMBIG_PREFIX))
297 assert result.exit_code == ExitCode.USER_ERROR
298 data = json.loads(result.output)
299 assert data["error"] == "ambiguous"
300 assert set(data["candidates"]) == {cid1, cid2}
301
302 def test_nonexistent_full_sha_not_found(self, tmp_path: pathlib.Path) -> None:
303 repo = _make_repo(tmp_path)
304 result = _rev(repo, long_id("f" * 64))
305 assert result.exit_code == ExitCode.USER_ERROR
306 data = json.loads(result.output)
307 assert data["error"] == "not found"
308
309
310 # ---------------------------------------------------------------------------
311 # Integration — --abbrev-ref
312 # ---------------------------------------------------------------------------
313
314
315 class TestAbbrevRef:
316 def test_abbrev_ref_head_returns_branch_name(self, tmp_path: pathlib.Path) -> None:
317 """The canonical agent UX: what branch am I on?"""
318 repo = _make_repo(tmp_path, branch="feat/my-feature")
319 result = _rev(repo, "--abbrev-ref", "--json", "HEAD")
320 assert result.exit_code == 0
321 data = json.loads(result.output)
322 assert data["branch"] == "feat/my-feature"
323 assert data["ref"] == "HEAD"
324
325 def test_abbrev_ref_text_format(self, tmp_path: pathlib.Path) -> None:
326 repo = _make_repo(tmp_path, branch="dev")
327 result = _rev(repo, "--abbrev-ref", "HEAD")
328 assert result.exit_code == 0
329 assert result.output.strip() == "dev"
330
331 def test_abbrev_ref_main(self, tmp_path: pathlib.Path) -> None:
332 repo = _make_repo(tmp_path, branch="main")
333 result = _rev(repo, "--abbrev-ref", "--json", "HEAD")
334 assert result.exit_code == 0
335 assert json.loads(result.output)["branch"] == "main"
336
337
338 # ---------------------------------------------------------------------------
339 # Edge cases
340 # ---------------------------------------------------------------------------
341
342
343 class TestEdgeCases:
344 def test_empty_ref_clean_error(self, tmp_path: pathlib.Path) -> None:
345 """Empty string ref must give a clear 'ref must not be empty' error."""
346 repo = _make_repo(tmp_path)
347 result = _rev(repo, "")
348 assert result.exit_code == ExitCode.USER_ERROR
349 data = json.loads(result.output)
350 assert "empty" in data["error"]
351
352 def test_unrecognized_flag_errors(self, tmp_path: pathlib.Path) -> None:
353 repo, _ = _populated_repo(tmp_path)
354 result = _rev(repo, "--no-such-flag", "main")
355 assert result.exit_code != 0
356
357 def test_branch_with_slash_resolves(self, tmp_path: pathlib.Path) -> None:
358 repo = _make_repo(tmp_path, branch="feat/my-feature")
359 sid = _store_snap(repo)
360 cid = _make_commit(repo, sid, branch="feat/my-feature", message="feat-init")
361 _set_head(repo, "feat/my-feature", cid)
362 result = _rev(repo, "--json", "feat/my-feature")
363 assert result.exit_code == 0
364 assert json.loads(result.output)["commit_id"] == cid
365
366
367 # ---------------------------------------------------------------------------
368 # Security
369 # ---------------------------------------------------------------------------
370
371
372 class TestSecurity:
373 def test_ansi_in_ref_is_json_escaped(self, tmp_path: pathlib.Path) -> None:
374 """ANSI escape in ref is safely JSON-encoded, never echoed raw."""
375 repo = _make_repo(tmp_path)
376 malicious = "\x1b[31mmalicious\x1b[0m"
377 result = _rev(repo, malicious)
378 assert result.exit_code == ExitCode.USER_ERROR
379 # Output is JSON — ANSI must be encoded as \u001b, not emitted raw
380 assert "\x1b" not in result.output
381 data = json.loads(result.output)
382 assert data["error"] == "not found"
383
384 def test_path_traversal_ref_gives_not_found(self, tmp_path: pathlib.Path) -> None:
385 repo = _make_repo(tmp_path)
386 result = _rev(repo, "../../../etc/passwd")
387 assert result.exit_code == ExitCode.USER_ERROR
388
389 def test_null_byte_in_ref(self, tmp_path: pathlib.Path) -> None:
390 repo = _make_repo(tmp_path)
391 result = _rev(repo, "branch\x00null")
392 assert result.exit_code == ExitCode.USER_ERROR
393
394 def test_no_traceback_on_bad_input(self, tmp_path: pathlib.Path) -> None:
395 repo = _make_repo(tmp_path)
396 result = _rev(repo, "")
397 assert "Traceback" not in result.output
398
399
400 # ---------------------------------------------------------------------------
401 # JSON schema — duration_ms + exit_code on every output path
402 # ---------------------------------------------------------------------------
403
404
405 class TestJsonSchema:
406 """Every JSON response must carry duration_ms (float ≥ 0) and exit_code (int)."""
407
408 def _assert_schema(self, d: Mapping[str, object], expected_exit: int = 0) -> None:
409 assert "duration_ms" in d, f"duration_ms missing: {d}"
410 assert isinstance(d["duration_ms"], (int, float))
411 assert d["duration_ms"] >= 0
412 assert "exit_code" in d, f"exit_code missing: {d}"
413 assert d["exit_code"] == expected_exit
414
415 def test_branch_resolution_has_schema(self, tmp_path: pathlib.Path) -> None:
416 repo, cid = _populated_repo(tmp_path)
417 result = _rev(repo, "--json", "main")
418 self._assert_schema(json.loads(result.output))
419
420 def test_head_resolution_has_schema(self, tmp_path: pathlib.Path) -> None:
421 repo, cid = _populated_repo(tmp_path)
422 result = _rev(repo, "--json", "HEAD")
423 self._assert_schema(json.loads(result.output))
424
425 def test_sha_resolution_has_schema(self, tmp_path: pathlib.Path) -> None:
426 repo, cid = _populated_repo(tmp_path)
427 result = _rev(repo, "--json", cid)
428 self._assert_schema(json.loads(result.output))
429
430 def test_abbrev_ref_has_schema(self, tmp_path: pathlib.Path) -> None:
431 repo = _make_repo(tmp_path, branch="feat/x")
432 result = _rev(repo, "--abbrev-ref", "--json", "HEAD")
433 self._assert_schema(json.loads(result.output))
434
435 def test_prefix_resolution_has_schema(self, tmp_path: pathlib.Path) -> None:
436 repo, cid = _populated_repo(tmp_path)
437 prefix = long_id(split_id(cid)[1][:8])
438 result = _rev(repo, "--json", prefix)
439 self._assert_schema(json.loads(result.output))
440
441
442 # ---------------------------------------------------------------------------
443 # Error JSON — all error paths emit structured JSON to stdout
444 # ---------------------------------------------------------------------------
445
446
447 class TestErrorJson:
448 """Every error must emit a parseable JSON dict to stdout (not stderr)."""
449
450 def _assert_error(self, result: InvokeResult) -> Mapping[str, object]:
451 assert result.exit_code != 0, "expected non-zero exit"
452 d = json.loads(result.output) # stdout, not stderr
453 assert "error" in d
454 assert "duration_ms" in d, f"duration_ms missing from error: {d}"
455 assert "exit_code" in d
456 assert d["exit_code"] != 0
457 return d
458
459 def test_empty_ref_emits_json_to_stdout(self, tmp_path: pathlib.Path) -> None:
460 """Empty ref error must land on stdout as JSON, not stderr."""
461 repo = _make_repo(tmp_path)
462 result = _rev(repo, "")
463 self._assert_error(result)
464 assert "empty" in json.loads(result.output)["error"]
465
466 def test_not_found_emits_json_to_stdout(self, tmp_path: pathlib.Path) -> None:
467 """Not-found error must land on stdout as JSON, not stderr."""
468 repo = _make_repo(tmp_path)
469 result = _rev(repo, "no-such-ref")
470 self._assert_error(result)
471 assert json.loads(result.output)["error"] == "not found"
472
473 def test_not_found_has_schema(self, tmp_path: pathlib.Path) -> None:
474 repo = _make_repo(tmp_path)
475 result = _rev(repo, "nonexistent-branch")
476 self._assert_error(result)
477 assert json.loads(result.output)["error"] == "not found"
478
479 def test_ambiguous_prefix_has_schema(self, tmp_path: pathlib.Path) -> None:
480 _AMBIG_MSG_1 = "commit-search-121"
481 _AMBIG_MSG_2 = "commit-search-154"
482 _AMBIG_PREFIX = "9d05"
483 repo = _make_repo(tmp_path)
484 sid = _store_snap(repo)
485 cid1 = _make_commit(repo, sid, branch="main", message=_AMBIG_MSG_1)
486 cid2 = _make_commit(repo, sid, branch="dev", message=_AMBIG_MSG_2)
487 assert split_id(cid1)[1][:4] == split_id(cid2)[1][:4] == _AMBIG_PREFIX
488 _set_head(repo, "main", cid1)
489 _set_head(repo, "dev", cid2)
490 result = _rev(repo, long_id(_AMBIG_PREFIX))
491 d = self._assert_error(result)
492 assert d["error"] == "ambiguous"
493
494 def test_head_no_commits_has_schema(self, tmp_path: pathlib.Path) -> None:
495 repo = _make_repo(tmp_path)
496 result = _rev(repo, "HEAD")
497 self._assert_error(result)
498 assert "no commits" in json.loads(result.output)["error"]
499
500 def test_bare_hex_has_schema(self, tmp_path: pathlib.Path) -> None:
501 repo, cid = _populated_repo(tmp_path)
502 result = _rev(repo, split_id(cid)[1])
503 d = self._assert_error(result)
504 assert "sha256:" in d["error"]
505
506 def test_error_json_has_ref_key(self, tmp_path: pathlib.Path) -> None:
507 """Every error dict must echo back the ref the caller passed."""
508 repo = _make_repo(tmp_path)
509 result = _rev(repo, "missing-branch")
510 d = json.loads(result.output)
511 assert d["ref"] == "missing-branch"
512
513
514 class TestRegisterFlags:
515 def test_default_json_out_is_false(self) -> None:
516 import argparse
517 from muse.cli.commands.rev_parse import register
518 p = argparse.ArgumentParser()
519 subs = p.add_subparsers()
520 register(subs)
521 args = p.parse_args(["rev-parse", "HEAD"])
522 assert args.json_out is False
523
524 def test_json_flag_sets_json_out(self) -> None:
525 import argparse
526 from muse.cli.commands.rev_parse import register
527 p = argparse.ArgumentParser()
528 subs = p.add_subparsers()
529 register(subs)
530 args = p.parse_args(["rev-parse", "HEAD", "--json"])
531 assert args.json_out is True
532
533 def test_j_shorthand_sets_json_out(self) -> None:
534 import argparse
535 from muse.cli.commands.rev_parse import register
536 p = argparse.ArgumentParser()
537 subs = p.add_subparsers()
538 register(subs)
539 args = p.parse_args(["rev-parse", "HEAD", "-j"])
540 assert args.json_out is True
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 121 days ago