gabriel / muse public
test_cmd_show_ref.py python
472 lines 18.4 KB
Raw
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2 fix: remove commit_exists filter from have anchors — server… Sonnet 4.6 patch 21 days ago
1 """Comprehensive tests for ``muse show-ref``.
2
3 Audit findings addressed here
4 ------------------------------
5 Security
6 - Format error now goes to stderr (was stdout) — verified below.
7 - ANSI injection in branch names and commit IDs stripped in text mode.
8 - Symlink refs in .muse/refs/heads/ are silently skipped.
9 - Ref files with non-hex content are silently skipped.
10
11 Agent UX
12 - ``--verify`` now emits JSON when combined with ``--json`` (was silent).
13 - ``--count`` added for branch inventory without reading all commit IDs.
14 - ``--pattern`` default changed from ``""`` to ``None`` (cleaner guard).
15
16 Performance
17 - Symlink check and commit-ID validation happen in ``_list_branch_refs``
18 before the output path, so corrupt refs never surface to callers.
19
20 Coverage tiers
21 --------------
22 - Unit: _list_branch_refs, _head_info, _ShowRefResult schema
23 - Integration: JSON/text output, --head, --verify (json + exit), --count,
24 --pattern, empty repo, no HEAD commit, multi-branch sorting
25 - Security: ANSI stripped in text mode, symlinks skipped, invalid commit IDs
26 skipped, format error to stderr, no traceback on errors
27 - Stress: 200 sequential full-list calls, 100-branch repo listing
28 """
29 from __future__ import annotations
30
31 import json
32 import os
33 import pathlib
34
35 import pytest
36
37 from muse.core.types import long_id
38 from muse.core.errors import ExitCode
39 from muse.core.paths import heads_dir, ref_path, muse_dir
40 from tests.cli_test_helper import CliRunner, InvokeResult
41
42 runner = CliRunner()
43
44 # ---------------------------------------------------------------------------
45 # Helpers
46 # ---------------------------------------------------------------------------
47
48 _FAKE_OID = long_id("a" * 64)
49
50
51 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
52 repo = tmp_path / "repo"
53 muse = muse_dir(repo)
54 (muse / "objects").mkdir(parents=True)
55 (muse / "commits").mkdir(parents=True)
56 (muse / "snapshots").mkdir(parents=True)
57 (muse / "refs" / "heads").mkdir(parents=True)
58 (muse / "HEAD").write_text("ref: refs/heads/main")
59 (muse / "repo.json").write_text(json.dumps({"repo_id": "r1", "domain": "code"}))
60 return repo
61
62
63 def _write_ref(repo: pathlib.Path, branch: str, commit_id: str = _FAKE_OID) -> None:
64 branch_ref = ref_path(repo, branch)
65 branch_ref.write_text(commit_id)
66
67
68 def _sr(repo: pathlib.Path, *args: str) -> InvokeResult:
69 from muse.cli.app import main as cli
70 return runner.invoke(cli, ["show-ref", *args],
71 env={"MUSE_REPO_ROOT": str(repo)})
72
73
74 # ---------------------------------------------------------------------------
75 # Unit — private helpers
76 # ---------------------------------------------------------------------------
77
78
79 class TestListBranchRefs:
80 def test_empty_heads_dir(self, tmp_path: pathlib.Path) -> None:
81 from muse.cli.commands.show_ref import _list_branch_refs
82 repo = _make_repo(tmp_path)
83 assert _list_branch_refs(repo) == []
84
85 def test_single_ref(self, tmp_path: pathlib.Path) -> None:
86 from muse.cli.commands.show_ref import _list_branch_refs
87 repo = _make_repo(tmp_path)
88 _write_ref(repo, "main")
89 refs = _list_branch_refs(repo)
90 assert len(refs) == 1
91 assert refs[0]["ref"] == "refs/heads/main"
92 assert refs[0]["commit_id"] == _FAKE_OID
93
94 def test_multiple_refs_sorted(self, tmp_path: pathlib.Path) -> None:
95 from muse.cli.commands.show_ref import _list_branch_refs
96 repo = _make_repo(tmp_path)
97 _write_ref(repo, "zeta", long_id("b" * 64))
98 _write_ref(repo, "alpha", long_id("c" * 64))
99 _write_ref(repo, "main", long_id("d" * 64))
100 refs = _list_branch_refs(repo)
101 names = [r["ref"] for r in refs]
102 assert names == ["refs/heads/alpha", "refs/heads/main", "refs/heads/zeta"]
103
104 def test_invalid_commit_id_skipped(self, tmp_path: pathlib.Path) -> None:
105 from muse.cli.commands.show_ref import _list_branch_refs
106 repo = _make_repo(tmp_path)
107 _write_ref(repo, "good", long_id("a" * 64))
108 _write_ref(repo, "bad", "not-a-sha256")
109 refs = _list_branch_refs(repo)
110 assert len(refs) == 1
111 assert refs[0]["ref"] == "refs/heads/good"
112
113 def test_empty_ref_file_skipped(self, tmp_path: pathlib.Path) -> None:
114 from muse.cli.commands.show_ref import _list_branch_refs
115 repo = _make_repo(tmp_path)
116 _write_ref(repo, "empty", "")
117 assert _list_branch_refs(repo) == []
118
119 def test_symlink_ref_skipped(self, tmp_path: pathlib.Path) -> None:
120 from muse.cli.commands.show_ref import _list_branch_refs
121 repo = _make_repo(tmp_path)
122 _write_ref(repo, "real", long_id("a" * 64))
123 sym = heads_dir(repo) / "sym-branch"
124 sym.symlink_to(heads_dir(repo) / "real")
125 refs = _list_branch_refs(repo)
126 # Only the real branch should appear; the symlink is skipped.
127 assert len(refs) == 1
128 assert refs[0]["ref"] == "refs/heads/real"
129
130 def test_nonexistent_heads_dir(self, tmp_path: pathlib.Path) -> None:
131 from muse.cli.commands.show_ref import _list_branch_refs
132 repo = _make_repo(tmp_path)
133 import shutil
134 shutil.rmtree(heads_dir(repo))
135 assert _list_branch_refs(repo) == []
136
137
138 class TestHeadInfo:
139 def test_returns_none_on_empty_branch(self, tmp_path: pathlib.Path) -> None:
140 from muse.cli.commands.show_ref import _head_info
141 repo = _make_repo(tmp_path)
142 # HEAD points to main but no ref file written → commit_id is None
143 assert _head_info(repo) is None
144
145 def test_returns_info_when_commit_present(self, tmp_path: pathlib.Path) -> None:
146 from muse.cli.commands.show_ref import _head_info
147 repo = _make_repo(tmp_path)
148 _write_ref(repo, "main")
149 info = _head_info(repo)
150 assert info is not None
151 assert info["branch"] == "main"
152 assert info["commit_id"] == _FAKE_OID
153 assert info["ref"] == "refs/heads/main"
154
155 def test_schema_fields_present(self, tmp_path: pathlib.Path) -> None:
156 from muse.cli.commands.show_ref import _HeadInfo
157 fields = set(_HeadInfo.__annotations__)
158 assert fields == {"ref", "branch", "commit_id"}
159
160
161 class TestShowRefResultSchema:
162 def test_schema_fields(self) -> None:
163 from muse.cli.commands.show_ref import _ShowRefResult
164 fields = set(_ShowRefResult.__annotations__)
165 assert "refs" in fields
166 assert "head" in fields
167 assert "count" in fields
168 assert "duration_ms" in fields
169 assert "exit_code" in fields
170
171
172 # ---------------------------------------------------------------------------
173 # Integration — JSON output
174 # ---------------------------------------------------------------------------
175
176
177 class TestJsonOutput:
178 def test_empty_repo_zero_refs(self, tmp_path: pathlib.Path) -> None:
179 repo = _make_repo(tmp_path)
180 result = _sr(repo, "--json")
181 assert result.exit_code == 0
182 data = json.loads(result.output)
183 assert data["refs"] == []
184 assert data["count"] == 0
185
186 def test_single_branch_present(self, tmp_path: pathlib.Path) -> None:
187 repo = _make_repo(tmp_path)
188 _write_ref(repo, "main")
189 data = json.loads(_sr(repo, "--json").output)
190 assert data["count"] == 1
191 assert data["refs"][0]["ref"] == "refs/heads/main"
192 assert data["refs"][0]["commit_id"] == _FAKE_OID
193
194 def test_head_present_when_commit_exists(self, tmp_path: pathlib.Path) -> None:
195 repo = _make_repo(tmp_path)
196 _write_ref(repo, "main")
197 data = json.loads(_sr(repo, "--json").output)
198 assert data["head"] is not None
199 assert data["head"]["branch"] == "main"
200
201 def test_head_null_when_no_commit(self, tmp_path: pathlib.Path) -> None:
202 repo = _make_repo(tmp_path)
203 data = json.loads(_sr(repo, "--json").output)
204 assert data["head"] is None
205
206 def test_json_shorthand_flag(self, tmp_path: pathlib.Path) -> None:
207 repo = _make_repo(tmp_path)
208 result = _sr(repo, "--json")
209 assert result.exit_code == 0
210 assert "refs" in json.loads(result.output)
211
212 def test_multi_branch_sorted(self, tmp_path: pathlib.Path) -> None:
213 repo = _make_repo(tmp_path)
214 _write_ref(repo, "zeta", "b" * 64)
215 _write_ref(repo, "alpha", "c" * 64)
216 data = json.loads(_sr(repo, "--json").output)
217 refs = [r["ref"] for r in data["refs"]]
218 assert refs == sorted(refs)
219
220
221 # ---------------------------------------------------------------------------
222 # Integration — text output
223 # ---------------------------------------------------------------------------
224
225
226 class TestTextOutput:
227 def test_commit_id_in_output(self, tmp_path: pathlib.Path) -> None:
228 repo = _make_repo(tmp_path)
229 _write_ref(repo, "main")
230 result = _sr(repo)
231 assert result.exit_code == 0
232 assert _FAKE_OID in result.output
233
234 def test_head_marker_present(self, tmp_path: pathlib.Path) -> None:
235 repo = _make_repo(tmp_path)
236 _write_ref(repo, "main")
237 result = _sr(repo)
238 assert "* " in result.output
239 assert "(HEAD)" in result.output
240
241 def test_empty_repo_no_output(self, tmp_path: pathlib.Path) -> None:
242 repo = _make_repo(tmp_path)
243 result = _sr(repo)
244 assert result.exit_code == 0
245 assert result.output.strip() == ""
246
247
248 # ---------------------------------------------------------------------------
249 # Integration — --head mode
250 # ---------------------------------------------------------------------------
251
252
253 class TestHeadMode:
254 def test_json_head_present(self, tmp_path: pathlib.Path) -> None:
255 repo = _make_repo(tmp_path)
256 _write_ref(repo, "main")
257 data = json.loads(_sr(repo, "--head", "--json").output)
258 assert data["head"]["branch"] == "main"
259
260 def test_json_head_null(self, tmp_path: pathlib.Path) -> None:
261 repo = _make_repo(tmp_path)
262 data = json.loads(_sr(repo, "--head", "--json").output)
263 assert data["head"] is None
264
265 def test_text_head_present(self, tmp_path: pathlib.Path) -> None:
266 repo = _make_repo(tmp_path)
267 _write_ref(repo, "main")
268 result = _sr(repo, "--head")
269 assert "(HEAD)" in result.output
270 assert _FAKE_OID in result.output
271
272 def test_text_no_head(self, tmp_path: pathlib.Path) -> None:
273 repo = _make_repo(tmp_path)
274 result = _sr(repo, "--head")
275 assert "no HEAD commit" in result.output
276
277
278 # ---------------------------------------------------------------------------
279 # Integration — --verify mode (agent UX supercharge)
280 # ---------------------------------------------------------------------------
281
282
283 class TestVerifyMode:
284 def test_existing_ref_exits_0(self, tmp_path: pathlib.Path) -> None:
285 repo = _make_repo(tmp_path)
286 _write_ref(repo, "main")
287 result = _sr(repo, "--verify", "refs/heads/main")
288 assert result.exit_code == 0
289
290 def test_missing_ref_exits_1(self, tmp_path: pathlib.Path) -> None:
291 repo = _make_repo(tmp_path)
292 result = _sr(repo, "--verify", "refs/heads/nonexistent")
293 assert result.exit_code == ExitCode.USER_ERROR
294
295 def test_json_verify_exists_true(self, tmp_path: pathlib.Path) -> None:
296 """JSON output is now emitted — critical agent UX improvement."""
297 repo = _make_repo(tmp_path)
298 _write_ref(repo, "main")
299 result = _sr(repo, "--verify", "refs/heads/main", "--json")
300 assert result.exit_code == 0
301 data = json.loads(result.output)
302 assert data["exists"] is True
303 assert data["ref"] == "refs/heads/main"
304
305 def test_json_verify_exists_false(self, tmp_path: pathlib.Path) -> None:
306 repo = _make_repo(tmp_path)
307 result = _sr(repo, "--verify", "refs/heads/ghost", "--json")
308 assert result.exit_code == ExitCode.USER_ERROR
309 data = json.loads(result.output)
310 assert data["exists"] is False
311 assert data["ref"] == "refs/heads/ghost"
312
313
314 # ---------------------------------------------------------------------------
315 # Integration — --count mode (new)
316 # ---------------------------------------------------------------------------
317
318
319 class TestCountMode:
320 def test_json_count_zero(self, tmp_path: pathlib.Path) -> None:
321 repo = _make_repo(tmp_path)
322 data = json.loads(_sr(repo, "--count", "--json").output)
323 assert data["count"] == 0
324
325 def test_json_count_with_branches(self, tmp_path: pathlib.Path) -> None:
326 repo = _make_repo(tmp_path)
327 _write_ref(repo, "main")
328 _write_ref(repo, "dev", long_id("b" * 64))
329 data = json.loads(_sr(repo, "--count", "--json").output)
330 assert data["count"] == 2
331
332 def test_text_count(self, tmp_path: pathlib.Path) -> None:
333 repo = _make_repo(tmp_path)
334 _write_ref(repo, "main")
335 result = _sr(repo, "--count")
336 assert result.exit_code == 0
337 assert result.output.strip() == "1"
338
339 def test_count_with_pattern(self, tmp_path: pathlib.Path) -> None:
340 """--count respects --pattern filter."""
341 repo = _make_repo(tmp_path)
342 _write_ref(repo, "main")
343 _write_ref(repo, "feat-x", long_id("b" * 64))
344 _write_ref(repo, "feat-y", long_id("c" * 64))
345 data = json.loads(_sr(repo, "--count", "--pattern", "refs/heads/feat*", "--json").output)
346 assert data["count"] == 2
347
348
349 # ---------------------------------------------------------------------------
350 # Integration — --pattern filter
351 # ---------------------------------------------------------------------------
352
353
354 class TestPatternFilter:
355 def test_pattern_matches(self, tmp_path: pathlib.Path) -> None:
356 repo = _make_repo(tmp_path)
357 _write_ref(repo, "feat-a", long_id("b" * 64))
358 _write_ref(repo, "feat-b", long_id("c" * 64))
359 _write_ref(repo, "main")
360 data = json.loads(_sr(repo, "--pattern", "refs/heads/feat*", "--json").output)
361 assert data["count"] == 2
362 for r in data["refs"]:
363 assert r["ref"].startswith("refs/heads/feat")
364
365 def test_pattern_no_match(self, tmp_path: pathlib.Path) -> None:
366 repo = _make_repo(tmp_path)
367 _write_ref(repo, "main")
368 data = json.loads(_sr(repo, "--pattern", "refs/heads/release/*", "--json").output)
369 assert data["count"] == 0
370 assert data["refs"] == []
371
372
373 # ---------------------------------------------------------------------------
374 # Security
375 # ---------------------------------------------------------------------------
376
377
378 class TestSecurity:
379 def test_ansi_in_branch_name_stripped_text(self, tmp_path: pathlib.Path) -> None:
380 """Branch name with ANSI escape in ref path is sanitized in text output."""
381 repo = _make_repo(tmp_path)
382 ansi_branch = "\x1b[31mmalicious\x1b[0m"
383 branch_ref = ref_path(repo, ansi_branch)
384 branch_ref.write_text(long_id("a" * 64))
385 result = _sr(repo)
386 assert "\x1b" not in result.output
387
388 def test_ansi_in_commit_id_stripped_text(self, tmp_path: pathlib.Path) -> None:
389 """Commit ID with ANSI in ref file content is sanitized.
390 (The ref is also skipped by validate_object_id — no ANSI ever reaches output.)
391 """
392 repo = _make_repo(tmp_path)
393 _write_ref(repo, "main", f"\x1b[31m{'a' * 60}")
394 result = _sr(repo)
395 assert "\x1b" not in result.output
396
397 def test_format_error_to_stderr(self, tmp_path: pathlib.Path) -> None:
398 repo = _make_repo(tmp_path)
399 result = _sr(repo, "--format", "yaml")
400 assert result.exit_code != 0
401 assert "error" in result.stderr.lower()
402 assert result.stdout_bytes == b""
403
404 def test_no_traceback_on_bad_format(self, tmp_path: pathlib.Path) -> None:
405 repo = _make_repo(tmp_path)
406 result = _sr(repo, "--format", "xml")
407 assert "Traceback" not in result.output
408
409 def test_symlink_ref_not_included(self, tmp_path: pathlib.Path) -> None:
410 repo = _make_repo(tmp_path)
411 _write_ref(repo, "real", long_id("a" * 64))
412 sym = heads_dir(repo) / "sym"
413 sym.symlink_to(heads_dir(repo) / "real")
414 data = json.loads(_sr(repo, "--json").output)
415 ref_names = [r["ref"] for r in data["refs"]]
416 assert "refs/heads/sym" not in ref_names
417 assert "refs/heads/real" in ref_names
418
419 def test_invalid_commit_id_ref_not_included(self, tmp_path: pathlib.Path) -> None:
420 repo = _make_repo(tmp_path)
421 _write_ref(repo, "good", long_id("a" * 64))
422 _write_ref(repo, "corrupt", "not-a-sha256-at-all")
423 data = json.loads(_sr(repo, "--json").output)
424 ref_names = [r["ref"] for r in data["refs"]]
425 assert "refs/heads/good" in ref_names
426 assert "refs/heads/corrupt" not in ref_names
427
428 def test_path_traversal_in_pattern_safe(self, tmp_path: pathlib.Path) -> None:
429 """A crafted pattern cannot escape the ref listing via fnmatch."""
430 repo = _make_repo(tmp_path)
431 _write_ref(repo, "main")
432 result = _sr(repo, "--pattern", "../../../../etc/*", "--json")
433 assert result.exit_code == 0
434 data = json.loads(result.output)
435 assert data["count"] == 0
436
437
438 # ---------------------------------------------------------------------------
439 # Stress
440 # ---------------------------------------------------------------------------
441
442
443 class TestStress:
444 def test_200_sequential_calls(self, tmp_path: pathlib.Path) -> None:
445 repo = _make_repo(tmp_path)
446 _write_ref(repo, "main")
447 for i in range(200):
448 result = _sr(repo, "--json")
449 assert result.exit_code == 0, f"failed at iteration {i}"
450 assert json.loads(result.output)["count"] == 1
451
452 def test_100_branch_repo(self, tmp_path: pathlib.Path) -> None:
453 """Listing 100 branches must complete and return the correct count."""
454 repo = _make_repo(tmp_path)
455 hex_chars = "0123456789abcdef"
456 for i in range(100):
457 # Build a deterministic valid sha256:-prefixed commit ID.
458 oid = long_id((hex_chars[i % 16]) * 64)
459 _write_ref(repo, f"branch-{i:04d}", oid)
460 data = json.loads(_sr(repo, "--json").output)
461 assert data["count"] == 100
462 # All refs must be sorted lexicographically.
463 names = [r["ref"] for r in data["refs"]]
464 assert names == sorted(names)
465
466 def test_100_verify_calls(self, tmp_path: pathlib.Path) -> None:
467 repo = _make_repo(tmp_path)
468 _write_ref(repo, "main")
469 for i in range(100):
470 result = _sr(repo, "--verify", "refs/heads/main", "--json")
471 assert result.exit_code == 0, f"failed at iteration {i}"
472 assert json.loads(result.output)["exists"] is True
File History 4 commits
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2 fix: remove commit_exists filter from have anchors — server… Sonnet 4.6 patch 21 days ago
sha256:36c3cb3e76619d4c30a6d9bf81b5ec4ff148e30dcfed913e3114ca7b43b81c7e fix: rename objects→blobs in push client and all stale test… Sonnet 4.6 patch 22 days ago
sha256:c06a9b9b9fee26c68ea725b44d54b2c0a171301ce9de746d5b656617b4463a9a fix: repair four test failures from post-migration audit Sonnet 4.6 patch 28 days ago
sha256:1900655993c83c4107067375548a7be823e471d2515830842f1a12cba4bd3cdf fix: unified object store migration — idempotent writes, JS… Sonnet 4.6 minor 29 days ago