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