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