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