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