"""Tests for muse symbolic-ref. Coverage tiers -------------- Unit — _read_symbolic_ref, _branch_exists, _SymbolicRefResult schema Integration — read mode (branch / detached HEAD), write mode (--set), --create-branch, --short, --format text, --json shorthand Security — ANSI injection in branch names, error output to stderr, unsupported ref rejected, symlink branch rejected, no traceback Stress — 200 sequential reads, 50-branch repo round-trip """ from __future__ import annotations import datetime import json import os import pathlib import pytest from tests.cli_test_helper import CliRunner, InvokeResult from muse.cli.commands.symbolic_ref import ( _SymbolicRefResult, _branch_exists, _read_symbolic_ref, ) from muse.core.ids import hash_commit, hash_snapshot from muse.core.commits import ( CommitRecord, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, write_snapshot, ) from muse.core.types import Manifest, long_id from muse.core.paths import head_path, heads_dir, muse_dir, ref_path cli = None # argparse-based CLI; CliRunner ignores this arg runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _init_repo(path: pathlib.Path, branch: str = "main") -> pathlib.Path: muse = muse_dir(path) (muse / "commits").mkdir(parents=True) (muse / "snapshots").mkdir(parents=True) (muse / "objects").mkdir(parents=True) (muse / "refs" / "heads").mkdir(parents=True) (muse / "HEAD").write_text(f"ref: refs/heads/{branch}\n", encoding="utf-8") (muse / "repo.json").write_text( json.dumps({"repo_id": "test-repo", "domain": "midi"}), encoding="utf-8" ) return path def _env(repo: pathlib.Path) -> Manifest: return {"MUSE_REPO_ROOT": str(repo)} _TS = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) def _snap(repo: pathlib.Path, manifest: Manifest | None = None) -> str: sid = hash_snapshot(manifest or {}) write_snapshot( repo, SnapshotRecord( snapshot_id=sid, manifest=manifest or {}, created_at=_TS, ), ) return sid def _commit( repo: pathlib.Path, snap_id: str, branch: str = "main", parent: str | None = None, message: str = "test", ) -> str: parents = [parent] if parent else [] cid = hash_commit( parent_ids=parents, snapshot_id=snap_id, message=message, committed_at_iso=_TS.isoformat(), author="tester", ) write_commit( repo, CommitRecord( commit_id=cid, branch=branch, snapshot_id=snap_id, message=message, committed_at=_TS, author="tester", parent_commit_id=parent, parent2_commit_id=None, ), ) branch_ref = ref_path(repo, branch) branch_ref.parent.mkdir(parents=True, exist_ok=True) branch_ref.write_text(cid, encoding="utf-8") return cid def _sr(repo: pathlib.Path, *args: str, **kw: str) -> InvokeResult: return runner.invoke(cli, ["symbolic-ref", *args], env=_env(repo)) # --------------------------------------------------------------------------- # Unit — _SymbolicRefResult schema # --------------------------------------------------------------------------- class TestSymbolicRefResultSchema: def test_required_fields_present(self) -> None: keys = _SymbolicRefResult.__annotations__ assert "ref" in keys assert "symbolic_target" in keys assert "branch" in keys assert "commit_id" in keys assert "detached" in keys def test_branch_allows_none(self) -> None: # str | None — detached HEAD support ann = _SymbolicRefResult.__annotations__ assert "None" in str(ann["branch"]) or type(None) in getattr(ann["branch"], "__args__", ()) def test_symbolic_target_allows_none(self) -> None: ann = _SymbolicRefResult.__annotations__ assert "None" in str(ann["symbolic_target"]) or type(None) in getattr( ann["symbolic_target"], "__args__", () ) # --------------------------------------------------------------------------- # Unit — _branch_exists # --------------------------------------------------------------------------- class TestBranchExists: def test_returns_true_for_real_file(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) sid = _snap(tmp_path) _commit(tmp_path, sid) assert _branch_exists(tmp_path, "main") is True def test_returns_false_when_missing(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) assert _branch_exists(tmp_path, "nonexistent") is False def test_returns_false_for_symlink(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) sid = _snap(tmp_path) _commit(tmp_path, sid, "main") real = heads_dir(tmp_path) / "main" link = heads_dir(tmp_path) / "sym-branch" link.symlink_to(real) assert _branch_exists(tmp_path, "sym-branch") is False # --------------------------------------------------------------------------- # Unit — _read_symbolic_ref # --------------------------------------------------------------------------- class TestReadSymbolicRef: def test_reads_branch_head(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) sid = _snap(tmp_path) cid = _commit(tmp_path, sid) result = _read_symbolic_ref(tmp_path) assert result["ref"] == "HEAD" assert result["branch"] == "main" assert result["symbolic_target"] == "refs/heads/main" assert result["commit_id"] == cid assert result["detached"] is False def test_no_commits_returns_null_commit_id(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _read_symbolic_ref(tmp_path) assert result["commit_id"] is None assert result["detached"] is False def test_detached_head_is_structured(self, tmp_path: pathlib.Path) -> None: """Detached HEAD must return structured data, not raise.""" _init_repo(tmp_path) fake_cid = long_id("a" * 64) (head_path(tmp_path)).write_text( f"commit: {fake_cid}\n", encoding="utf-8" ) result = _read_symbolic_ref(tmp_path) assert result["detached"] is True assert result["branch"] is None assert result["symbolic_target"] is None assert result["commit_id"] == fake_cid # --------------------------------------------------------------------------- # Integration — read mode (JSON) # --------------------------------------------------------------------------- class TestReadModeJson: def test_json_flag_outputs_json(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) r = _sr(tmp_path, "--json", "HEAD") assert r.exit_code == 0 data = json.loads(r.output) assert data["ref"] == "HEAD" assert data["branch"] == "main" assert data["symbolic_target"] == "refs/heads/main" assert data["detached"] is False def test_json_shorthand_alias(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) r = _sr(tmp_path, "--json", "HEAD") assert r.exit_code == 0 data = json.loads(r.output) assert "ref" in data def test_commit_id_populated_after_commit(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) sid = _snap(tmp_path) cid = _commit(tmp_path, sid) r = _sr(tmp_path, "--json", "HEAD") assert r.exit_code == 0 assert json.loads(r.output)["commit_id"] == cid def test_no_commits_commit_id_null(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) r = _sr(tmp_path, "--json", "HEAD") assert r.exit_code == 0 assert json.loads(r.output)["commit_id"] is None def test_detached_head_json(self, tmp_path: pathlib.Path) -> None: """Detached HEAD must return structured JSON, not crash.""" _init_repo(tmp_path) fake_cid = long_id("b" * 64) (head_path(tmp_path)).write_text( f"commit: {fake_cid}\n", encoding="utf-8" ) r = _sr(tmp_path, "--json", "HEAD") assert r.exit_code == 0, f"Crashed: {r.output}" data = json.loads(r.output) assert data["detached"] is True assert data["branch"] is None assert data["symbolic_target"] is None assert data["commit_id"] == fake_cid # --------------------------------------------------------------------------- # Integration — read mode (text) # --------------------------------------------------------------------------- class TestReadModeText: def test_text_full_path(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) r = _sr(tmp_path, "HEAD") assert r.exit_code == 0 assert r.output.strip() == "refs/heads/main" def test_text_short_flag(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) r = _sr(tmp_path, "--short", "HEAD") assert r.exit_code == 0 assert r.output.strip() == "main" def test_text_detached_head_shows_commit(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) fake_cid = long_id("c" * 64) (head_path(tmp_path)).write_text( f"commit: {fake_cid}\n", encoding="utf-8" ) r = _sr(tmp_path, "HEAD") assert r.exit_code == 0 # Should show something useful, not crash assert "detached" in r.output.lower() or "cccccccc" in r.output # --------------------------------------------------------------------------- # Integration — write mode (--set) # --------------------------------------------------------------------------- class TestWriteMode: def test_set_switches_existing_branch(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) sid = _snap(tmp_path) _commit(tmp_path, sid, "main", message="c1") _commit(tmp_path, sid, "dev", message="c2") r = _sr(tmp_path, "--json", "--set", "dev", "HEAD") assert r.exit_code == 0 data = json.loads(r.output) assert data["branch"] == "dev" assert data["symbolic_target"] == "refs/heads/dev" assert data["detached"] is False def test_set_updates_head_file(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) sid = _snap(tmp_path) _commit(tmp_path, sid, "main", message="c1") _commit(tmp_path, sid, "feature", message="c2") _sr(tmp_path, "--set", "feature", "HEAD") head_raw = (head_path(tmp_path)).read_text() assert "feature" in head_raw def test_set_nonexistent_branch_errors(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) r = _sr(tmp_path, "--json", "--set", "ghost", "HEAD") assert r.exit_code != 0 data = json.loads(r.stdout) assert "error" in data def test_set_text_format(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) sid = _snap(tmp_path) _commit(tmp_path, sid, "main", message="c1") _commit(tmp_path, sid, "dev", message="c2") r = _sr(tmp_path, "--set", "dev", "HEAD") assert r.exit_code == 0 assert r.output.strip() == "refs/heads/dev" def test_set_text_format_short(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) sid = _snap(tmp_path) _commit(tmp_path, sid, "main", message="c1") _commit(tmp_path, sid, "dev", message="c2") r = _sr(tmp_path, "--set", "dev", "--short", "HEAD") assert r.exit_code == 0 assert r.output.strip() == "dev" def test_set_invalid_branch_name_errors(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) r = _sr(tmp_path, "--json", "--set", "bad\x00branch", "HEAD") assert r.exit_code != 0 data = json.loads(r.stdout) assert "error" in data # --------------------------------------------------------------------------- # Integration — --create-branch (orphan mode) # --------------------------------------------------------------------------- class TestCreateBranch: def test_create_branch_points_to_empty_branch(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) sid = _snap(tmp_path) _commit(tmp_path, sid, "main") r = _sr(tmp_path, "--json", "--set", "orphan", "--create-branch", "HEAD") assert r.exit_code == 0 data = json.loads(r.output) assert data["branch"] == "orphan" # No commits on orphan yet assert data["commit_id"] is None def test_create_branch_writes_head_file(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _sr(tmp_path, "--set", "newbranch", "--create-branch", "HEAD") head_raw = (head_path(tmp_path)).read_text() assert "newbranch" in head_raw def test_without_create_branch_nonexistent_fails(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) r = _sr(tmp_path, "--set", "nonexistent", "HEAD") assert r.exit_code != 0 def test_create_branch_hint_in_error_message(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) r = _sr(tmp_path, "--set", "nonexistent", "HEAD") # Error message should mention --create-branch so agent knows the fix assert "create-branch" in r.stderr.lower() or "create-branch" in r.output.lower() # --------------------------------------------------------------------------- # Security # --------------------------------------------------------------------------- class TestSecurity: def test_ansi_injection_in_branch_name_stripped(self, tmp_path: pathlib.Path) -> None: """Branch names from HEAD file must be sanitized in text output.""" _init_repo(tmp_path) # Write a branch name that contains ANSI escape sequence into HEAD directly (head_path(tmp_path)).write_text( "ref: refs/heads/\x1b[31mred\x1b[0m\n", encoding="utf-8" ) # This is an invalid branch name so read_head raises — we just confirm # the command doesn't produce raw ANSI in its output. r = _sr(tmp_path, "HEAD") assert "\x1b" not in r.output assert "\x1b" not in r.stderr def test_unsupported_ref_goes_to_stderr_in_text_mode(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) r = _sr(tmp_path, "MERGE_HEAD") assert r.exit_code != 0 assert r.stderr != "" def test_unsupported_ref_json_error_to_stdout(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) r = _sr(tmp_path, "--json", "MERGE_HEAD") assert r.exit_code != 0 data = json.loads(r.stdout) assert "error" in data def test_no_traceback_on_unsupported_ref(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) r = _sr(tmp_path, "MERGE_HEAD") assert "Traceback" not in r.output assert "Traceback" not in r.stderr def test_no_traceback_on_detached_head(self, tmp_path: pathlib.Path) -> None: """Previously crashed with unhandled ValueError — must not raise.""" _init_repo(tmp_path) fake_cid = long_id("d" * 64) (head_path(tmp_path)).write_text( f"commit: {fake_cid}\n", encoding="utf-8" ) r = _sr(tmp_path, "HEAD") assert "Traceback" not in r.output assert "Traceback" not in r.stderr def test_symlink_branch_rejected_by_branch_exists(self, tmp_path: pathlib.Path) -> None: """A symlink at refs/heads/ is not treated as a valid branch.""" _init_repo(tmp_path) sid = _snap(tmp_path) _commit(tmp_path, sid, "main") # Create a symlink named 'linked' pointing to main's ref file real = heads_dir(tmp_path) / "main" link = heads_dir(tmp_path) / "linked" link.symlink_to(real) r = _sr(tmp_path, "--set", "linked", "HEAD") assert r.exit_code != 0 def test_no_repo_exits_cleanly(self, tmp_path: pathlib.Path) -> None: r = runner.invoke( cli, ["symbolic-ref", "HEAD"], env={"MUSE_REPO_ROOT": str(tmp_path / "nonexistent")}, ) assert r.exit_code != 0 assert "Traceback" not in r.output assert "Traceback" not in r.stderr # --------------------------------------------------------------------------- # Stress # --------------------------------------------------------------------------- class TestStress: def test_200_sequential_reads(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) sid = _snap(tmp_path) _commit(tmp_path, sid) for _ in range(200): r = _sr(tmp_path, "--json", "HEAD") assert r.exit_code == 0 data = json.loads(r.output) assert data["branch"] == "main" def test_50_branch_round_trip(self, tmp_path: pathlib.Path) -> None: """Create 50 branches, round-trip HEAD to each, verify output.""" _init_repo(tmp_path) sid = _snap(tmp_path) branches = [f"branch-{i:03d}" for i in range(50)] for b in branches: _commit(tmp_path, sid, b, message=f"c-{b}") for b in branches: r = _sr(tmp_path, "--json", "--set", b, "HEAD") assert r.exit_code == 0 data = json.loads(r.output) assert data["branch"] == b def test_200_sequential_detached_reads(self, tmp_path: pathlib.Path) -> None: """Detached HEAD must never crash under repeated reads.""" _init_repo(tmp_path) fake_cid = long_id("e" * 64) (head_path(tmp_path)).write_text( f"commit: {fake_cid}\n", encoding="utf-8" ) for _ in range(200): r = _sr(tmp_path, "--json", "HEAD") assert r.exit_code == 0 data = json.loads(r.output) assert data["detached"] is True # --------------------------------------------------------------------------- # Flag registration # --------------------------------------------------------------------------- class TestRegisterFlags: def _parse(self, *args: str) -> "argparse.Namespace": import argparse from muse.cli.commands.symbolic_ref import register p = argparse.ArgumentParser() sub = p.add_subparsers() register(sub) return p.parse_args(["symbolic-ref", *args]) def test_default_json_out_is_false(self) -> None: ns = self._parse("HEAD") assert ns.json_out is False def test_json_flag_sets_json_out(self) -> None: ns = self._parse("--json", "HEAD") assert ns.json_out is True def test_j_shorthand_sets_json_out(self) -> None: ns = self._parse("-j", "HEAD") assert ns.json_out is True