gabriel / muse public
test_cmd_symbolic_ref.py python
521 lines 18.8 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 133 days ago
1 """Tests for muse symbolic-ref.
2
3 Coverage tiers
4 --------------
5 Unit — _read_symbolic_ref, _branch_exists, _SymbolicRefResult schema
6 Integration — read mode (branch / detached HEAD), write mode (--set),
7 --create-branch, --short, --format text, --json shorthand
8 Security — ANSI injection in branch names, error output to stderr,
9 unsupported ref rejected, symlink branch rejected, no traceback
10 Stress — 200 sequential reads, 50-branch repo round-trip
11 """
12
13 from __future__ import annotations
14
15 import datetime
16 import json
17 import os
18 import pathlib
19
20 import pytest
21 from tests.cli_test_helper import CliRunner, InvokeResult
22
23 from muse.cli.commands.symbolic_ref import (
24 _SymbolicRefResult,
25 _branch_exists,
26 _read_symbolic_ref,
27 )
28 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
29 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
30 from muse.core._types import Manifest, long_id
31
32 cli = None # argparse-based CLI; CliRunner ignores this arg
33 runner = CliRunner()
34
35
36 # ---------------------------------------------------------------------------
37 # Helpers
38 # ---------------------------------------------------------------------------
39
40
41
42 def _init_repo(path: pathlib.Path, branch: str = "main") -> pathlib.Path:
43 muse = path / ".muse"
44 (muse / "commits").mkdir(parents=True)
45 (muse / "snapshots").mkdir(parents=True)
46 (muse / "objects").mkdir(parents=True)
47 (muse / "refs" / "heads").mkdir(parents=True)
48 (muse / "HEAD").write_text(f"ref: refs/heads/{branch}\n", encoding="utf-8")
49 (muse / "repo.json").write_text(
50 json.dumps({"repo_id": "test-repo", "domain": "midi"}), encoding="utf-8"
51 )
52 return path
53
54
55 def _env(repo: pathlib.Path) -> Manifest:
56 return {"MUSE_REPO_ROOT": str(repo)}
57
58
59 _TS = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
60
61
62 def _snap(repo: pathlib.Path, manifest: Manifest | None = None) -> str:
63 sid = compute_snapshot_id(manifest or {})
64 write_snapshot(
65 repo,
66 SnapshotRecord(
67 snapshot_id=sid,
68 manifest=manifest or {},
69 created_at=_TS,
70 ),
71 )
72 return sid
73
74
75 def _commit(
76 repo: pathlib.Path,
77 snap_id: str,
78 branch: str = "main",
79 parent: str | None = None,
80 message: str = "test",
81 ) -> str:
82 parents = [parent] if parent else []
83 cid = compute_commit_id(
84 repo_id="test-repo",
85 parent_ids=parents,
86 snapshot_id=snap_id,
87 message=message,
88 committed_at_iso=_TS.isoformat(),
89 author="tester",)
90 write_commit(
91 repo,
92 CommitRecord(
93 commit_id=cid,
94 repo_id="test-repo",
95 created_on_branch=branch,
96 snapshot_id=snap_id,
97 message=message,
98 committed_at=_TS,
99 author="tester",
100 parent_commit_id=parent,
101 parent2_commit_id=None,
102 ),
103 )
104 ref_path = repo / ".muse" / "refs" / "heads" / branch
105 ref_path.parent.mkdir(parents=True, exist_ok=True)
106 ref_path.write_text(cid, encoding="utf-8")
107 return cid
108
109
110 def _sr(repo: pathlib.Path, *args: str, **kw: str) -> InvokeResult:
111 return runner.invoke(cli, ["symbolic-ref", *args], env=_env(repo))
112
113
114 # ---------------------------------------------------------------------------
115 # Unit — _SymbolicRefResult schema
116 # ---------------------------------------------------------------------------
117
118
119 class TestSymbolicRefResultSchema:
120 def test_required_fields_present(self) -> None:
121 keys = _SymbolicRefResult.__annotations__
122 assert "ref" in keys
123 assert "symbolic_target" in keys
124 assert "branch" in keys
125 assert "commit_id" in keys
126 assert "detached" in keys
127
128 def test_branch_allows_none(self) -> None:
129 # str | None — detached HEAD support
130 ann = _SymbolicRefResult.__annotations__
131 assert "None" in str(ann["branch"]) or type(None) in getattr(ann["branch"], "__args__", ())
132
133 def test_symbolic_target_allows_none(self) -> None:
134 ann = _SymbolicRefResult.__annotations__
135 assert "None" in str(ann["symbolic_target"]) or type(None) in getattr(
136 ann["symbolic_target"], "__args__", ()
137 )
138
139
140 # ---------------------------------------------------------------------------
141 # Unit — _branch_exists
142 # ---------------------------------------------------------------------------
143
144
145 class TestBranchExists:
146 def test_returns_true_for_real_file(self, tmp_path: pathlib.Path) -> None:
147 _init_repo(tmp_path)
148 sid = _snap(tmp_path)
149 _commit(tmp_path, sid)
150 assert _branch_exists(tmp_path, "main") is True
151
152 def test_returns_false_when_missing(self, tmp_path: pathlib.Path) -> None:
153 _init_repo(tmp_path)
154 assert _branch_exists(tmp_path, "nonexistent") is False
155
156 def test_returns_false_for_symlink(self, tmp_path: pathlib.Path) -> None:
157 _init_repo(tmp_path)
158 sid = _snap(tmp_path)
159 _commit(tmp_path, sid, "main")
160 real = tmp_path / ".muse" / "refs" / "heads" / "main"
161 link = tmp_path / ".muse" / "refs" / "heads" / "sym-branch"
162 link.symlink_to(real)
163 assert _branch_exists(tmp_path, "sym-branch") is False
164
165
166 # ---------------------------------------------------------------------------
167 # Unit — _read_symbolic_ref
168 # ---------------------------------------------------------------------------
169
170
171 class TestReadSymbolicRef:
172 def test_reads_branch_head(self, tmp_path: pathlib.Path) -> None:
173 _init_repo(tmp_path)
174 sid = _snap(tmp_path)
175 cid = _commit(tmp_path, sid)
176 result = _read_symbolic_ref(tmp_path)
177 assert result["ref"] == "HEAD"
178 assert result["branch"] == "main"
179 assert result["symbolic_target"] == "refs/heads/main"
180 assert result["commit_id"] == cid
181 assert result["detached"] is False
182
183 def test_no_commits_returns_null_commit_id(self, tmp_path: pathlib.Path) -> None:
184 _init_repo(tmp_path)
185 result = _read_symbolic_ref(tmp_path)
186 assert result["commit_id"] is None
187 assert result["detached"] is False
188
189 def test_detached_head_is_structured(self, tmp_path: pathlib.Path) -> None:
190 """Detached HEAD must return structured data, not raise."""
191 _init_repo(tmp_path)
192 fake_cid = long_id("a" * 64)
193 (tmp_path / ".muse" / "HEAD").write_text(
194 f"commit: {fake_cid}\n", encoding="utf-8"
195 )
196 result = _read_symbolic_ref(tmp_path)
197 assert result["detached"] is True
198 assert result["branch"] is None
199 assert result["symbolic_target"] is None
200 assert result["commit_id"] == fake_cid
201
202
203 # ---------------------------------------------------------------------------
204 # Integration — read mode (JSON)
205 # ---------------------------------------------------------------------------
206
207
208 class TestReadModeJson:
209 def test_json_flag_outputs_json(self, tmp_path: pathlib.Path) -> None:
210 _init_repo(tmp_path)
211 r = _sr(tmp_path, "--json", "HEAD")
212 assert r.exit_code == 0
213 data = json.loads(r.output)
214 assert data["ref"] == "HEAD"
215 assert data["branch"] == "main"
216 assert data["symbolic_target"] == "refs/heads/main"
217 assert data["detached"] is False
218
219 def test_json_shorthand_alias(self, tmp_path: pathlib.Path) -> None:
220 _init_repo(tmp_path)
221 r = _sr(tmp_path, "--json", "HEAD")
222 assert r.exit_code == 0
223 data = json.loads(r.output)
224 assert "ref" in data
225
226 def test_commit_id_populated_after_commit(self, tmp_path: pathlib.Path) -> None:
227 _init_repo(tmp_path)
228 sid = _snap(tmp_path)
229 cid = _commit(tmp_path, sid)
230 r = _sr(tmp_path, "--json", "HEAD")
231 assert r.exit_code == 0
232 assert json.loads(r.output)["commit_id"] == cid
233
234 def test_no_commits_commit_id_null(self, tmp_path: pathlib.Path) -> None:
235 _init_repo(tmp_path)
236 r = _sr(tmp_path, "--json", "HEAD")
237 assert r.exit_code == 0
238 assert json.loads(r.output)["commit_id"] is None
239
240 def test_detached_head_json(self, tmp_path: pathlib.Path) -> None:
241 """Detached HEAD must return structured JSON, not crash."""
242 _init_repo(tmp_path)
243 fake_cid = long_id("b" * 64)
244 (tmp_path / ".muse" / "HEAD").write_text(
245 f"commit: {fake_cid}\n", encoding="utf-8"
246 )
247 r = _sr(tmp_path, "--json", "HEAD")
248 assert r.exit_code == 0, f"Crashed: {r.output}"
249 data = json.loads(r.output)
250 assert data["detached"] is True
251 assert data["branch"] is None
252 assert data["symbolic_target"] is None
253 assert data["commit_id"] == fake_cid
254
255
256 # ---------------------------------------------------------------------------
257 # Integration — read mode (text)
258 # ---------------------------------------------------------------------------
259
260
261 class TestReadModeText:
262 def test_text_full_path(self, tmp_path: pathlib.Path) -> None:
263 _init_repo(tmp_path)
264 r = _sr(tmp_path, "HEAD")
265 assert r.exit_code == 0
266 assert r.output.strip() == "refs/heads/main"
267
268 def test_text_short_flag(self, tmp_path: pathlib.Path) -> None:
269 _init_repo(tmp_path)
270 r = _sr(tmp_path, "--short", "HEAD")
271 assert r.exit_code == 0
272 assert r.output.strip() == "main"
273
274 def test_text_detached_head_shows_commit(self, tmp_path: pathlib.Path) -> None:
275 _init_repo(tmp_path)
276 fake_cid = long_id("c" * 64)
277 (tmp_path / ".muse" / "HEAD").write_text(
278 f"commit: {fake_cid}\n", encoding="utf-8"
279 )
280 r = _sr(tmp_path, "HEAD")
281 assert r.exit_code == 0
282 # Should show something useful, not crash
283 assert "detached" in r.output.lower() or "cccccccc" in r.output
284
285
286 # ---------------------------------------------------------------------------
287 # Integration — write mode (--set)
288 # ---------------------------------------------------------------------------
289
290
291 class TestWriteMode:
292 def test_set_switches_existing_branch(self, tmp_path: pathlib.Path) -> None:
293 _init_repo(tmp_path)
294 sid = _snap(tmp_path)
295 _commit(tmp_path, sid, "main", message="c1")
296 _commit(tmp_path, sid, "dev", message="c2")
297 r = _sr(tmp_path, "--json", "--set", "dev", "HEAD")
298 assert r.exit_code == 0
299 data = json.loads(r.output)
300 assert data["branch"] == "dev"
301 assert data["symbolic_target"] == "refs/heads/dev"
302 assert data["detached"] is False
303
304 def test_set_updates_head_file(self, tmp_path: pathlib.Path) -> None:
305 _init_repo(tmp_path)
306 sid = _snap(tmp_path)
307 _commit(tmp_path, sid, "main", message="c1")
308 _commit(tmp_path, sid, "feature", message="c2")
309 _sr(tmp_path, "--set", "feature", "HEAD")
310 head_raw = (tmp_path / ".muse" / "HEAD").read_text()
311 assert "feature" in head_raw
312
313 def test_set_nonexistent_branch_errors(self, tmp_path: pathlib.Path) -> None:
314 _init_repo(tmp_path)
315 r = _sr(tmp_path, "--json", "--set", "ghost", "HEAD")
316 assert r.exit_code != 0
317 data = json.loads(r.stdout)
318 assert "error" in data
319
320 def test_set_text_format(self, tmp_path: pathlib.Path) -> None:
321 _init_repo(tmp_path)
322 sid = _snap(tmp_path)
323 _commit(tmp_path, sid, "main", message="c1")
324 _commit(tmp_path, sid, "dev", message="c2")
325 r = _sr(tmp_path, "--set", "dev", "HEAD")
326 assert r.exit_code == 0
327 assert r.output.strip() == "refs/heads/dev"
328
329 def test_set_text_format_short(self, tmp_path: pathlib.Path) -> None:
330 _init_repo(tmp_path)
331 sid = _snap(tmp_path)
332 _commit(tmp_path, sid, "main", message="c1")
333 _commit(tmp_path, sid, "dev", message="c2")
334 r = _sr(tmp_path, "--set", "dev", "--short", "HEAD")
335 assert r.exit_code == 0
336 assert r.output.strip() == "dev"
337
338 def test_set_invalid_branch_name_errors(self, tmp_path: pathlib.Path) -> None:
339 _init_repo(tmp_path)
340 r = _sr(tmp_path, "--json", "--set", "bad\x00branch", "HEAD")
341 assert r.exit_code != 0
342 data = json.loads(r.stdout)
343 assert "error" in data
344
345
346 # ---------------------------------------------------------------------------
347 # Integration — --create-branch (orphan mode)
348 # ---------------------------------------------------------------------------
349
350
351 class TestCreateBranch:
352 def test_create_branch_points_to_empty_branch(self, tmp_path: pathlib.Path) -> None:
353 _init_repo(tmp_path)
354 sid = _snap(tmp_path)
355 _commit(tmp_path, sid, "main")
356 r = _sr(tmp_path, "--json", "--set", "orphan", "--create-branch", "HEAD")
357 assert r.exit_code == 0
358 data = json.loads(r.output)
359 assert data["branch"] == "orphan"
360 # No commits on orphan yet
361 assert data["commit_id"] is None
362
363 def test_create_branch_writes_head_file(self, tmp_path: pathlib.Path) -> None:
364 _init_repo(tmp_path)
365 _sr(tmp_path, "--set", "newbranch", "--create-branch", "HEAD")
366 head_raw = (tmp_path / ".muse" / "HEAD").read_text()
367 assert "newbranch" in head_raw
368
369 def test_without_create_branch_nonexistent_fails(self, tmp_path: pathlib.Path) -> None:
370 _init_repo(tmp_path)
371 r = _sr(tmp_path, "--set", "nonexistent", "HEAD")
372 assert r.exit_code != 0
373
374 def test_create_branch_hint_in_error_message(self, tmp_path: pathlib.Path) -> None:
375 _init_repo(tmp_path)
376 r = _sr(tmp_path, "--set", "nonexistent", "HEAD")
377 # Error message should mention --create-branch so agent knows the fix
378 assert "create-branch" in r.stderr.lower() or "create-branch" in r.output.lower()
379
380
381 # ---------------------------------------------------------------------------
382 # Security
383 # ---------------------------------------------------------------------------
384
385
386 class TestSecurity:
387 def test_ansi_injection_in_branch_name_stripped(self, tmp_path: pathlib.Path) -> None:
388 """Branch names from HEAD file must be sanitized in text output."""
389 _init_repo(tmp_path)
390 # Write a branch name that contains ANSI escape sequence into HEAD directly
391 (tmp_path / ".muse" / "HEAD").write_text(
392 "ref: refs/heads/\x1b[31mred\x1b[0m\n", encoding="utf-8"
393 )
394 # This is an invalid branch name so read_head raises — we just confirm
395 # the command doesn't produce raw ANSI in its output.
396 r = _sr(tmp_path, "HEAD")
397 assert "\x1b" not in r.output
398 assert "\x1b" not in r.stderr
399
400 def test_unsupported_ref_goes_to_stderr_in_text_mode(self, tmp_path: pathlib.Path) -> None:
401 _init_repo(tmp_path)
402 r = _sr(tmp_path, "MERGE_HEAD")
403 assert r.exit_code != 0
404 assert r.stderr != ""
405
406 def test_unsupported_ref_json_error_to_stdout(self, tmp_path: pathlib.Path) -> None:
407 _init_repo(tmp_path)
408 r = _sr(tmp_path, "--json", "MERGE_HEAD")
409 assert r.exit_code != 0
410 data = json.loads(r.stdout)
411 assert "error" in data
412
413 def test_no_traceback_on_unsupported_ref(self, tmp_path: pathlib.Path) -> None:
414 _init_repo(tmp_path)
415 r = _sr(tmp_path, "MERGE_HEAD")
416 assert "Traceback" not in r.output
417 assert "Traceback" not in r.stderr
418
419 def test_no_traceback_on_detached_head(self, tmp_path: pathlib.Path) -> None:
420 """Previously crashed with unhandled ValueError — must not raise."""
421 _init_repo(tmp_path)
422 fake_cid = long_id("d" * 64)
423 (tmp_path / ".muse" / "HEAD").write_text(
424 f"commit: {fake_cid}\n", encoding="utf-8"
425 )
426 r = _sr(tmp_path, "HEAD")
427 assert "Traceback" not in r.output
428 assert "Traceback" not in r.stderr
429
430 def test_symlink_branch_rejected_by_branch_exists(self, tmp_path: pathlib.Path) -> None:
431 """A symlink at refs/heads/<branch> is not treated as a valid branch."""
432 _init_repo(tmp_path)
433 sid = _snap(tmp_path)
434 _commit(tmp_path, sid, "main")
435 # Create a symlink named 'linked' pointing to main's ref file
436 real = tmp_path / ".muse" / "refs" / "heads" / "main"
437 link = tmp_path / ".muse" / "refs" / "heads" / "linked"
438 link.symlink_to(real)
439 r = _sr(tmp_path, "--set", "linked", "HEAD")
440 assert r.exit_code != 0
441
442 def test_no_repo_exits_cleanly(self, tmp_path: pathlib.Path) -> None:
443 r = runner.invoke(
444 cli,
445 ["symbolic-ref", "HEAD"],
446 env={"MUSE_REPO_ROOT": str(tmp_path / "nonexistent")},
447 )
448 assert r.exit_code != 0
449 assert "Traceback" not in r.output
450 assert "Traceback" not in r.stderr
451
452
453 # ---------------------------------------------------------------------------
454 # Stress
455 # ---------------------------------------------------------------------------
456
457
458 class TestStress:
459 def test_200_sequential_reads(self, tmp_path: pathlib.Path) -> None:
460 _init_repo(tmp_path)
461 sid = _snap(tmp_path)
462 _commit(tmp_path, sid)
463 for _ in range(200):
464 r = _sr(tmp_path, "--json", "HEAD")
465 assert r.exit_code == 0
466 data = json.loads(r.output)
467 assert data["branch"] == "main"
468
469 def test_50_branch_round_trip(self, tmp_path: pathlib.Path) -> None:
470 """Create 50 branches, round-trip HEAD to each, verify output."""
471 _init_repo(tmp_path)
472 sid = _snap(tmp_path)
473 branches = [f"branch-{i:03d}" for i in range(50)]
474 for b in branches:
475 _commit(tmp_path, sid, b, message=f"c-{b}")
476
477 for b in branches:
478 r = _sr(tmp_path, "--json", "--set", b, "HEAD")
479 assert r.exit_code == 0
480 data = json.loads(r.output)
481 assert data["branch"] == b
482
483 def test_200_sequential_detached_reads(self, tmp_path: pathlib.Path) -> None:
484 """Detached HEAD must never crash under repeated reads."""
485 _init_repo(tmp_path)
486 fake_cid = long_id("e" * 64)
487 (tmp_path / ".muse" / "HEAD").write_text(
488 f"commit: {fake_cid}\n", encoding="utf-8"
489 )
490 for _ in range(200):
491 r = _sr(tmp_path, "--json", "HEAD")
492 assert r.exit_code == 0
493 data = json.loads(r.output)
494 assert data["detached"] is True
495
496
497 # ---------------------------------------------------------------------------
498 # Flag registration
499 # ---------------------------------------------------------------------------
500
501
502 class TestRegisterFlags:
503 def _parse(self, *args: str):
504 import argparse
505 from muse.cli.commands.symbolic_ref import register
506 p = argparse.ArgumentParser()
507 sub = p.add_subparsers()
508 register(sub)
509 return p.parse_args(["symbolic-ref", *args])
510
511 def test_default_json_out_is_false(self) -> None:
512 ns = self._parse("HEAD")
513 assert ns.json_out is False
514
515 def test_json_flag_sets_json_out(self) -> None:
516 ns = self._parse("--json", "HEAD")
517 assert ns.json_out is True
518
519 def test_j_shorthand_sets_json_out(self) -> None:
520 ns = self._parse("-j", "HEAD")
521 assert ns.json_out is True
File History 3 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 133 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 139 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor 142 days ago