test_core_refs.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Tests for muse/core/refs.py — canonical ref-file reading primitives. |
| 2 | |
| 3 | Coverage |
| 4 | -------- |
| 5 | read_ref |
| 6 | - returns commit ID for a well-formed ref file |
| 7 | - returns None for a missing file |
| 8 | - returns None for an empty file |
| 9 | - returns None for a whitespace-only file |
| 10 | - strips leading/trailing whitespace and newlines |
| 11 | - returns None on PermissionError (graceful degradation) |
| 12 | |
| 13 | iter_branch_refs |
| 14 | - empty heads dir → yields nothing |
| 15 | - missing heads dir → yields nothing |
| 16 | - single branch → yields (branch_name, commit_id) |
| 17 | - multiple branches → yields all (branch_name, commit_id) pairs |
| 18 | - skips symlinks |
| 19 | - skips non-file entries (subdirectories) |
| 20 | - skips empty ref files |
| 21 | - each yielded commit_id is non-empty string |
| 22 | - lazy — yields incrementally (returns an iterator, not a list) |
| 23 | - branch name is the filename, not the full path |
| 24 | """ |
| 25 | |
| 26 | from __future__ import annotations |
| 27 | |
| 28 | import json |
| 29 | import pathlib |
| 30 | |
| 31 | import pytest |
| 32 | |
| 33 | from muse.core.refs import iter_branch_refs, read_ref, write_branch_ref as _store_write_branch_ref |
| 34 | from muse.core.types import long_id |
| 35 | from muse.core.paths import heads_dir, muse_dir, ref_path |
| 36 | |
| 37 | |
| 38 | # --------------------------------------------------------------------------- |
| 39 | # Helpers |
| 40 | # --------------------------------------------------------------------------- |
| 41 | |
| 42 | |
| 43 | def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path: |
| 44 | muse = muse_dir(tmp_path) |
| 45 | for d in ("objects", "commits", "snapshots", "refs/heads"): |
| 46 | (muse / d).mkdir(parents=True, exist_ok=True) |
| 47 | (muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo"})) |
| 48 | (muse / "HEAD").write_text("ref: refs/heads/main\n") |
| 49 | return tmp_path |
| 50 | |
| 51 | |
| 52 | def _write_branch_ref(repo: pathlib.Path, branch: str, commit_id: str) -> pathlib.Path: |
| 53 | """Write a branch ref file via the canonical store function; return the path.""" |
| 54 | _store_write_branch_ref(repo, branch, commit_id) |
| 55 | from muse.core.paths import ref_path as _ref_path |
| 56 | return _ref_path(repo, branch) |
| 57 | |
| 58 | |
| 59 | def _write_corrupt_ref(repo: pathlib.Path, branch: str, bare_hex: str) -> pathlib.Path: |
| 60 | """Write a ref file with bare hex (no prefix) to simulate a corrupt/legacy file.""" |
| 61 | branch_ref = ref_path(repo, branch) |
| 62 | branch_ref.parent.mkdir(parents=True, exist_ok=True) |
| 63 | branch_ref.write_text(f"{bare_hex}\n", encoding="utf-8") |
| 64 | return branch_ref |
| 65 | |
| 66 | |
| 67 | _FAKE_CID = long_id("ab" * 32) |
| 68 | _FAKE_CID2 = long_id("cd" * 32) |
| 69 | _FAKE_CID3 = long_id("ef" * 32) |
| 70 | |
| 71 | |
| 72 | # --------------------------------------------------------------------------- |
| 73 | # read_ref |
| 74 | # --------------------------------------------------------------------------- |
| 75 | |
| 76 | |
| 77 | class TestReadRef: |
| 78 | def test_returns_commit_id_for_well_formed_ref(self, tmp_path: pathlib.Path) -> None: |
| 79 | ref = tmp_path / "myref" |
| 80 | ref.write_text(f"{_FAKE_CID}\n", encoding="utf-8") |
| 81 | assert read_ref(ref) == _FAKE_CID |
| 82 | |
| 83 | def test_returns_none_for_missing_file(self, tmp_path: pathlib.Path) -> None: |
| 84 | assert read_ref(tmp_path / "nonexistent") is None |
| 85 | |
| 86 | def test_returns_none_for_empty_file(self, tmp_path: pathlib.Path) -> None: |
| 87 | ref = tmp_path / "empty" |
| 88 | ref.write_text("", encoding="utf-8") |
| 89 | assert read_ref(ref) is None |
| 90 | |
| 91 | def test_returns_none_for_whitespace_only(self, tmp_path: pathlib.Path) -> None: |
| 92 | ref = tmp_path / "ws" |
| 93 | ref.write_text(" \n \n", encoding="utf-8") |
| 94 | assert read_ref(ref) is None |
| 95 | |
| 96 | def test_strips_whitespace_and_newlines(self, tmp_path: pathlib.Path) -> None: |
| 97 | ref = tmp_path / "ref" |
| 98 | ref.write_text(f" {_FAKE_CID} \n", encoding="utf-8") |
| 99 | assert read_ref(ref) == _FAKE_CID |
| 100 | |
| 101 | def test_strips_trailing_newline_only(self, tmp_path: pathlib.Path) -> None: |
| 102 | ref = tmp_path / "ref" |
| 103 | ref.write_text(f"{_FAKE_CID}\n", encoding="utf-8") |
| 104 | assert read_ref(ref) == _FAKE_CID |
| 105 | |
| 106 | def test_returns_none_on_permission_error(self, tmp_path: pathlib.Path) -> None: |
| 107 | ref = tmp_path / "locked" |
| 108 | ref.write_text(_FAKE_CID, encoding="utf-8") |
| 109 | ref.chmod(0o000) |
| 110 | try: |
| 111 | result = read_ref(ref) |
| 112 | assert result is None |
| 113 | finally: |
| 114 | ref.chmod(0o644) |
| 115 | |
| 116 | def test_bare_hex_returns_none(self, tmp_path: pathlib.Path) -> None: |
| 117 | """Bare hex without sha256: prefix is invalid — read_ref must return None.""" |
| 118 | bare = "ab" * 32 |
| 119 | ref = _write_corrupt_ref(tmp_path, "main", bare) |
| 120 | assert read_ref(ref) is None |
| 121 | |
| 122 | def test_prefixed_id_returned_unchanged(self, tmp_path: pathlib.Path) -> None: |
| 123 | """Already-prefixed IDs must pass through read_ref without modification.""" |
| 124 | ref = tmp_path / "main" |
| 125 | ref.write_text(f"{_FAKE_CID}\n", encoding="utf-8") |
| 126 | assert read_ref(ref) == _FAKE_CID |
| 127 | |
| 128 | |
| 129 | # --------------------------------------------------------------------------- |
| 130 | # iter_branch_refs |
| 131 | # --------------------------------------------------------------------------- |
| 132 | |
| 133 | |
| 134 | class TestIterBranchRefs: |
| 135 | def test_missing_heads_dir_yields_nothing(self, tmp_path: pathlib.Path) -> None: |
| 136 | repo = tmp_path # no .muse directory at all |
| 137 | result = list(iter_branch_refs(repo)) |
| 138 | assert result == [] |
| 139 | |
| 140 | def test_empty_heads_dir_yields_nothing(self, tmp_path: pathlib.Path) -> None: |
| 141 | repo = _make_repo(tmp_path) |
| 142 | result = list(iter_branch_refs(repo)) |
| 143 | assert result == [] |
| 144 | |
| 145 | def test_single_branch_yields_name_and_commit_id(self, tmp_path: pathlib.Path) -> None: |
| 146 | repo = _make_repo(tmp_path) |
| 147 | _write_branch_ref(repo, "main", _FAKE_CID) |
| 148 | result = list(iter_branch_refs(repo)) |
| 149 | assert len(result) == 1 |
| 150 | name, cid = result[0] |
| 151 | assert name == "main" |
| 152 | assert cid == _FAKE_CID |
| 153 | |
| 154 | def test_multiple_branches_yields_all(self, tmp_path: pathlib.Path) -> None: |
| 155 | repo = _make_repo(tmp_path) |
| 156 | _write_branch_ref(repo, "main", _FAKE_CID) |
| 157 | _write_branch_ref(repo, "dev", _FAKE_CID2) |
| 158 | _write_branch_ref(repo, "feat/x", _FAKE_CID3) |
| 159 | result = dict(iter_branch_refs(repo)) |
| 160 | assert result == {"main": _FAKE_CID, "dev": _FAKE_CID2, "feat/x": _FAKE_CID3} |
| 161 | |
| 162 | def test_skips_symlinks(self, tmp_path: pathlib.Path) -> None: |
| 163 | repo = _make_repo(tmp_path) |
| 164 | _write_branch_ref(repo, "main", _FAKE_CID) |
| 165 | h_dir = heads_dir(repo) |
| 166 | symlink = h_dir / "alias" |
| 167 | symlink.symlink_to(h_dir / "main") |
| 168 | names = [name for name, _ in iter_branch_refs(repo)] |
| 169 | assert "alias" not in names |
| 170 | assert "main" in names |
| 171 | |
| 172 | def test_skips_empty_subdirectories(self, tmp_path: pathlib.Path) -> None: |
| 173 | """Empty subdirectories (no ref files) yield nothing for that subtree.""" |
| 174 | repo = _make_repo(tmp_path) |
| 175 | _write_branch_ref(repo, "main", _FAKE_CID) |
| 176 | subdir = heads_dir(repo) / "namespace" |
| 177 | subdir.mkdir() |
| 178 | names = [name for name, _ in iter_branch_refs(repo)] |
| 179 | assert "namespace" not in names |
| 180 | assert "main" in names |
| 181 | |
| 182 | def test_hierarchical_branch_name_uses_posix_slash(self, tmp_path: pathlib.Path) -> None: |
| 183 | """Branch names with slashes (task/foo) are yielded as relative POSIX paths.""" |
| 184 | repo = _make_repo(tmp_path) |
| 185 | _write_branch_ref(repo, "task/my-feature", _FAKE_CID) |
| 186 | result = list(iter_branch_refs(repo)) |
| 187 | assert len(result) == 1 |
| 188 | name, cid = result[0] |
| 189 | assert name == "task/my-feature" |
| 190 | assert cid == _FAKE_CID |
| 191 | |
| 192 | def test_skips_empty_ref_files(self, tmp_path: pathlib.Path) -> None: |
| 193 | repo = _make_repo(tmp_path) |
| 194 | _write_branch_ref(repo, "main", _FAKE_CID) |
| 195 | (heads_dir(repo) / "empty-branch").write_text("") |
| 196 | result = list(iter_branch_refs(repo)) |
| 197 | names = [name for name, _ in result] |
| 198 | assert "empty-branch" not in names |
| 199 | assert "main" in names |
| 200 | |
| 201 | def test_yields_non_empty_commit_ids(self, tmp_path: pathlib.Path) -> None: |
| 202 | repo = _make_repo(tmp_path) |
| 203 | _write_branch_ref(repo, "main", _FAKE_CID) |
| 204 | for name, cid in iter_branch_refs(repo): |
| 205 | assert cid # non-empty |
| 206 | assert isinstance(cid, str) |
| 207 | |
| 208 | def test_returns_iterator_not_list(self, tmp_path: pathlib.Path) -> None: |
| 209 | """iter_branch_refs must return an iterator (lazy), not a pre-built list.""" |
| 210 | import collections.abc |
| 211 | repo = _make_repo(tmp_path) |
| 212 | result = iter_branch_refs(repo) |
| 213 | assert isinstance(result, collections.abc.Iterator) |
| 214 | |
| 215 | def test_branch_name_is_relative_posix_path(self, tmp_path: pathlib.Path) -> None: |
| 216 | """Branch name is relative to h_dir — not an absolute filesystem path.""" |
| 217 | repo = _make_repo(tmp_path) |
| 218 | _write_branch_ref(repo, "simple", _FAKE_CID) |
| 219 | result = list(iter_branch_refs(repo)) |
| 220 | assert len(result) == 1 |
| 221 | name, _ = result[0] |
| 222 | assert name == "simple" |
| 223 | assert str(tmp_path) not in name |
| 224 | |
| 225 | def test_bare_hex_in_file_is_skipped(self, tmp_path: pathlib.Path) -> None: |
| 226 | """iter_branch_refs must skip ref files containing bare hex (no prefix).""" |
| 227 | repo = _make_repo(tmp_path) |
| 228 | _write_corrupt_ref(repo, "main", "cd" * 32) |
| 229 | result = list(iter_branch_refs(repo)) |
| 230 | assert result == [] |
| 231 | |
| 232 | def test_uses_generator_not_list_comprehension(self) -> None: |
| 233 | """iter_branch_refs must use a generator (structural check).""" |
| 234 | import inspect |
| 235 | from muse.core import refs as refs_module |
| 236 | source = inspect.getsource(refs_module.iter_branch_refs) |
| 237 | assert "yield" in source, "iter_branch_refs must use yield (generator)" |