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