"""Tests for muse/core/refs.py — canonical ref-file reading primitives. Coverage -------- read_ref - returns commit ID for a well-formed ref file - returns None for a missing file - returns None for an empty file - returns None for a whitespace-only file - strips leading/trailing whitespace and newlines - returns None on PermissionError (graceful degradation) iter_branch_refs - empty heads dir → yields nothing - missing heads dir → yields nothing - single branch → yields (branch_name, commit_id) - multiple branches → yields all (branch_name, commit_id) pairs - skips symlinks - skips non-file entries (subdirectories) - skips empty ref files - each yielded commit_id is non-empty string - lazy — yields incrementally (returns an iterator, not a list) - branch name is the filename, not the full path """ from __future__ import annotations import json import pathlib import pytest from muse.core.refs import iter_branch_refs, read_ref, write_branch_ref as _store_write_branch_ref from muse.core.types import long_id from muse.core.paths import heads_dir, muse_dir, ref_path # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path: muse = muse_dir(tmp_path) for d in ("objects", "commits", "snapshots", "refs/heads"): (muse / d).mkdir(parents=True, exist_ok=True) (muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo"})) (muse / "HEAD").write_text("ref: refs/heads/main\n") return tmp_path def _write_branch_ref(repo: pathlib.Path, branch: str, commit_id: str) -> pathlib.Path: """Write a branch ref file via the canonical store function; return the path.""" _store_write_branch_ref(repo, branch, commit_id) from muse.core.paths import ref_path as _ref_path return _ref_path(repo, branch) def _write_corrupt_ref(repo: pathlib.Path, branch: str, bare_hex: str) -> pathlib.Path: """Write a ref file with bare hex (no prefix) to simulate a corrupt/legacy file.""" branch_ref = ref_path(repo, branch) branch_ref.parent.mkdir(parents=True, exist_ok=True) branch_ref.write_text(f"{bare_hex}\n", encoding="utf-8") return branch_ref _FAKE_CID = long_id("ab" * 32) _FAKE_CID2 = long_id("cd" * 32) _FAKE_CID3 = long_id("ef" * 32) # --------------------------------------------------------------------------- # read_ref # --------------------------------------------------------------------------- class TestReadRef: def test_returns_commit_id_for_well_formed_ref(self, tmp_path: pathlib.Path) -> None: ref = tmp_path / "myref" ref.write_text(f"{_FAKE_CID}\n", encoding="utf-8") assert read_ref(ref) == _FAKE_CID def test_returns_none_for_missing_file(self, tmp_path: pathlib.Path) -> None: assert read_ref(tmp_path / "nonexistent") is None def test_returns_none_for_empty_file(self, tmp_path: pathlib.Path) -> None: ref = tmp_path / "empty" ref.write_text("", encoding="utf-8") assert read_ref(ref) is None def test_returns_none_for_whitespace_only(self, tmp_path: pathlib.Path) -> None: ref = tmp_path / "ws" ref.write_text(" \n \n", encoding="utf-8") assert read_ref(ref) is None def test_strips_whitespace_and_newlines(self, tmp_path: pathlib.Path) -> None: ref = tmp_path / "ref" ref.write_text(f" {_FAKE_CID} \n", encoding="utf-8") assert read_ref(ref) == _FAKE_CID def test_strips_trailing_newline_only(self, tmp_path: pathlib.Path) -> None: ref = tmp_path / "ref" ref.write_text(f"{_FAKE_CID}\n", encoding="utf-8") assert read_ref(ref) == _FAKE_CID def test_returns_none_on_permission_error(self, tmp_path: pathlib.Path) -> None: ref = tmp_path / "locked" ref.write_text(_FAKE_CID, encoding="utf-8") ref.chmod(0o000) try: result = read_ref(ref) assert result is None finally: ref.chmod(0o644) def test_bare_hex_returns_none(self, tmp_path: pathlib.Path) -> None: """Bare hex without sha256: prefix is invalid — read_ref must return None.""" bare = "ab" * 32 ref = _write_corrupt_ref(tmp_path, "main", bare) assert read_ref(ref) is None def test_prefixed_id_returned_unchanged(self, tmp_path: pathlib.Path) -> None: """Already-prefixed IDs must pass through read_ref without modification.""" ref = tmp_path / "main" ref.write_text(f"{_FAKE_CID}\n", encoding="utf-8") assert read_ref(ref) == _FAKE_CID # --------------------------------------------------------------------------- # iter_branch_refs # --------------------------------------------------------------------------- class TestIterBranchRefs: def test_missing_heads_dir_yields_nothing(self, tmp_path: pathlib.Path) -> None: repo = tmp_path # no .muse directory at all result = list(iter_branch_refs(repo)) assert result == [] def test_empty_heads_dir_yields_nothing(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = list(iter_branch_refs(repo)) assert result == [] def test_single_branch_yields_name_and_commit_id(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _write_branch_ref(repo, "main", _FAKE_CID) result = list(iter_branch_refs(repo)) assert len(result) == 1 name, cid = result[0] assert name == "main" assert cid == _FAKE_CID def test_multiple_branches_yields_all(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _write_branch_ref(repo, "main", _FAKE_CID) _write_branch_ref(repo, "dev", _FAKE_CID2) _write_branch_ref(repo, "feat/x", _FAKE_CID3) result = dict(iter_branch_refs(repo)) assert result == {"main": _FAKE_CID, "dev": _FAKE_CID2, "feat/x": _FAKE_CID3} def test_skips_symlinks(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _write_branch_ref(repo, "main", _FAKE_CID) h_dir = heads_dir(repo) symlink = h_dir / "alias" symlink.symlink_to(h_dir / "main") names = [name for name, _ in iter_branch_refs(repo)] assert "alias" not in names assert "main" in names def test_skips_empty_subdirectories(self, tmp_path: pathlib.Path) -> None: """Empty subdirectories (no ref files) yield nothing for that subtree.""" repo = _make_repo(tmp_path) _write_branch_ref(repo, "main", _FAKE_CID) subdir = heads_dir(repo) / "namespace" subdir.mkdir() names = [name for name, _ in iter_branch_refs(repo)] assert "namespace" not in names assert "main" in names def test_hierarchical_branch_name_uses_posix_slash(self, tmp_path: pathlib.Path) -> None: """Branch names with slashes (task/foo) are yielded as relative POSIX paths.""" repo = _make_repo(tmp_path) _write_branch_ref(repo, "task/my-feature", _FAKE_CID) result = list(iter_branch_refs(repo)) assert len(result) == 1 name, cid = result[0] assert name == "task/my-feature" assert cid == _FAKE_CID def test_skips_empty_ref_files(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _write_branch_ref(repo, "main", _FAKE_CID) (heads_dir(repo) / "empty-branch").write_text("") result = list(iter_branch_refs(repo)) names = [name for name, _ in result] assert "empty-branch" not in names assert "main" in names def test_yields_non_empty_commit_ids(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _write_branch_ref(repo, "main", _FAKE_CID) for name, cid in iter_branch_refs(repo): assert cid # non-empty assert isinstance(cid, str) def test_returns_iterator_not_list(self, tmp_path: pathlib.Path) -> None: """iter_branch_refs must return an iterator (lazy), not a pre-built list.""" import collections.abc repo = _make_repo(tmp_path) result = iter_branch_refs(repo) assert isinstance(result, collections.abc.Iterator) def test_branch_name_is_relative_posix_path(self, tmp_path: pathlib.Path) -> None: """Branch name is relative to h_dir — not an absolute filesystem path.""" repo = _make_repo(tmp_path) _write_branch_ref(repo, "simple", _FAKE_CID) result = list(iter_branch_refs(repo)) assert len(result) == 1 name, _ = result[0] assert name == "simple" assert str(tmp_path) not in name def test_bare_hex_in_file_is_skipped(self, tmp_path: pathlib.Path) -> None: """iter_branch_refs must skip ref files containing bare hex (no prefix).""" repo = _make_repo(tmp_path) _write_corrupt_ref(repo, "main", "cd" * 32) result = list(iter_branch_refs(repo)) assert result == [] def test_uses_generator_not_list_comprehension(self) -> None: """iter_branch_refs must use a generator (structural check).""" import inspect from muse.core import refs as refs_module source = inspect.getsource(refs_module.iter_branch_refs) assert "yield" in source, "iter_branch_refs must use yield (generator)"