gabriel / muse public
test_indices.py python
290 lines 10.8 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/indices.py — optional local index layer.
2
3 Coverage
4 --------
5 SymbolHistoryEntry
6 - to_dict / from_dict round-trip.
7 - All six fields preserved.
8
9 symbol_history index
10 - save_symbol_history writes a valid JSON file.
11 - load_symbol_history reads it back correctly.
12 - load returns empty dict when file absent.
13 - load returns empty dict on corrupt JSON.
14 - Sorting: entries dict is sorted by address.
15 - Multiple addresses, multiple events per address.
16
17 hash_occurrence index
18 - save_hash_occurrence writes a valid JSON file.
19 - load_hash_occurrence reads it back correctly.
20 - load returns empty dict when file absent.
21 - load returns empty dict on corrupt JSON.
22 - Addresses within each hash entry are sorted.
23
24 index_info
25 - Reports "absent" for missing indexes.
26 - Reports "present" + correct entry count for existing indexes.
27 - Reports "corrupt" for malformed JSON.
28 - Reports both indexes.
29
30 Schema compliance
31 - schema_version == __version__.
32 - updated_at is present and is a non-empty string.
33 - index field matches the index name.
34 """
35
36 import pathlib
37
38 import msgpack
39 import pytest
40
41 from muse._version import __version__
42 from muse.core.indices import (
43 HashOccurrenceIndex,
44 SymbolHistoryEntry,
45 SymbolHistoryIndex,
46 index_info,
47 load_hash_occurrence,
48 load_symbol_history,
49 save_hash_occurrence,
50 save_symbol_history,
51 )
52 from muse.core.paths import indices_dir
53
54
55 # ---------------------------------------------------------------------------
56 # SymbolHistoryEntry
57 # ---------------------------------------------------------------------------
58
59
60 class TestSymbolHistoryEntry:
61 def test_to_dict_from_dict_round_trip(self) -> None:
62 entry = SymbolHistoryEntry(
63 commit_id="abc123",
64 committed_at="2026-01-01T00:00:00+00:00",
65 op="insert",
66 content_id="content_abc",
67 body_hash="body_hash_xyz",
68 signature_id="sig_id_pqr",
69 )
70 d = entry.to_dict()
71 entry2 = SymbolHistoryEntry.from_dict(d)
72 assert entry2.commit_id == "abc123"
73 assert entry2.committed_at == "2026-01-01T00:00:00+00:00"
74 assert entry2.op == "insert"
75 assert entry2.content_id == "content_abc"
76 assert entry2.body_hash == "body_hash_xyz"
77 assert entry2.signature_id == "sig_id_pqr"
78
79 def test_all_ops_preserved(self) -> None:
80 for op in ("insert", "delete", "replace", "patch"):
81 e = SymbolHistoryEntry("c", "t", op, "cid", "bh", "sig")
82 assert SymbolHistoryEntry.from_dict(e.to_dict()).op == op
83
84
85 # ---------------------------------------------------------------------------
86 # symbol_history index — save / load
87 # ---------------------------------------------------------------------------
88
89
90 class TestSymbolHistoryIndex:
91 def _make_entry(self, op: str = "insert") -> SymbolHistoryEntry:
92 return SymbolHistoryEntry(
93 commit_id="commit1",
94 committed_at="2026-01-01T00:00:00+00:00",
95 op=op,
96 content_id="cid1",
97 body_hash="bh1",
98 signature_id="sig1",
99 )
100
101 def test_save_creates_file(self, tmp_path: pathlib.Path) -> None:
102 index: SymbolHistoryIndex = {
103 "src/a.py::f": [self._make_entry()],
104 }
105 save_symbol_history(tmp_path, index)
106 path = indices_dir(tmp_path) / "symbol_history.msgpack"
107 assert path.exists()
108
109 def test_round_trip(self, tmp_path: pathlib.Path) -> None:
110 entry = self._make_entry("replace")
111 index: SymbolHistoryIndex = {
112 "src/billing.py::compute_total": [entry],
113 }
114 save_symbol_history(tmp_path, index)
115 loaded = load_symbol_history(tmp_path)
116 assert "src/billing.py::compute_total" in loaded
117 entries = loaded["src/billing.py::compute_total"]
118 assert len(entries) == 1
119 assert entries[0].op == "replace"
120 assert entries[0].commit_id == "commit1"
121
122 def test_multiple_addresses(self, tmp_path: pathlib.Path) -> None:
123 index: SymbolHistoryIndex = {
124 "src/a.py::alpha": [self._make_entry("insert")],
125 "src/b.py::beta": [self._make_entry("insert"), self._make_entry("replace")],
126 }
127 save_symbol_history(tmp_path, index)
128 loaded = load_symbol_history(tmp_path)
129 assert len(loaded["src/a.py::alpha"]) == 1
130 assert len(loaded["src/b.py::beta"]) == 2
131
132 def test_load_absent_returns_empty(self, tmp_path: pathlib.Path) -> None:
133 result = load_symbol_history(tmp_path)
134 assert result == {}
135
136 def test_load_corrupt_returns_empty(self, tmp_path: pathlib.Path) -> None:
137 idx_dir = indices_dir(tmp_path)
138 idx_dir.mkdir(parents=True, exist_ok=True)
139 (idx_dir / "symbol_history.msgpack").write_bytes(b"\xff\xfe not valid msgpack")
140 result = load_symbol_history(tmp_path)
141 assert result == {}
142
143 def test_schema_compliance(self, tmp_path: pathlib.Path) -> None:
144 index: SymbolHistoryIndex = {"x.py::f": [self._make_entry()]}
145 save_symbol_history(tmp_path, index)
146 raw = msgpack.unpackb(
147 (indices_dir(tmp_path) / "symbol_history.msgpack").read_bytes(),
148 raw=False,
149 )
150 assert raw["schema_version"] == __version__
151 assert raw["index"] == "symbol_history"
152 assert raw["updated_at"] # non-empty string
153 assert "x.py::f" in raw["entries"]
154
155 def test_empty_index_saved(self, tmp_path: pathlib.Path) -> None:
156 save_symbol_history(tmp_path, {})
157 loaded = load_symbol_history(tmp_path)
158 assert loaded == {}
159
160 def test_entries_sorted_by_address(self, tmp_path: pathlib.Path) -> None:
161 index: SymbolHistoryIndex = {
162 "z.py::z": [self._make_entry()],
163 "a.py::a": [self._make_entry()],
164 "m.py::m": [self._make_entry()],
165 }
166 save_symbol_history(tmp_path, index)
167 raw = msgpack.unpackb(
168 (indices_dir(tmp_path) / "symbol_history.msgpack").read_bytes(),
169 raw=False,
170 )
171 keys = list(raw["entries"].keys())
172 assert keys == sorted(keys)
173
174
175 # ---------------------------------------------------------------------------
176 # hash_occurrence index — save / load
177 # ---------------------------------------------------------------------------
178
179
180 class TestHashOccurrenceIndex:
181 def test_save_creates_file(self, tmp_path: pathlib.Path) -> None:
182 index: HashOccurrenceIndex = {
183 "deadbeef": ["src/a.py::f", "src/b.py::g"],
184 }
185 save_hash_occurrence(tmp_path, index)
186 path = indices_dir(tmp_path) / "hash_occurrence.msgpack"
187 assert path.exists()
188
189 def test_round_trip(self, tmp_path: pathlib.Path) -> None:
190 index: HashOccurrenceIndex = {
191 "abc123": ["src/a.py::f", "src/b.py::g"],
192 "def456": ["src/c.py::h"],
193 }
194 save_hash_occurrence(tmp_path, index)
195 loaded = load_hash_occurrence(tmp_path)
196 assert "abc123" in loaded
197 assert set(loaded["abc123"]) == {"src/a.py::f", "src/b.py::g"}
198 assert loaded["def456"] == ["src/c.py::h"]
199
200 def test_addresses_sorted_within_hash(self, tmp_path: pathlib.Path) -> None:
201 index: HashOccurrenceIndex = {
202 "hash1": ["z.py::z", "a.py::a", "m.py::m"],
203 }
204 save_hash_occurrence(tmp_path, index)
205 raw = msgpack.unpackb(
206 (indices_dir(tmp_path) / "hash_occurrence.msgpack").read_bytes(),
207 raw=False,
208 )
209 addrs = raw["entries"]["hash1"]
210 assert addrs == sorted(addrs)
211
212 def test_hashes_sorted(self, tmp_path: pathlib.Path) -> None:
213 index: HashOccurrenceIndex = {
214 "zzz": ["a.py::f"],
215 "aaa": ["b.py::g"],
216 }
217 save_hash_occurrence(tmp_path, index)
218 raw = msgpack.unpackb(
219 (indices_dir(tmp_path) / "hash_occurrence.msgpack").read_bytes(),
220 raw=False,
221 )
222 keys = list(raw["entries"].keys())
223 assert keys == sorted(keys)
224
225 def test_load_absent_returns_empty(self, tmp_path: pathlib.Path) -> None:
226 assert load_hash_occurrence(tmp_path) == {}
227
228 def test_load_corrupt_returns_empty(self, tmp_path: pathlib.Path) -> None:
229 idx_dir = indices_dir(tmp_path)
230 idx_dir.mkdir(parents=True, exist_ok=True)
231 (idx_dir / "hash_occurrence.msgpack").write_bytes(b"\xff\xfe garbage bytes")
232 assert load_hash_occurrence(tmp_path) == {}
233
234 def test_schema_compliance(self, tmp_path: pathlib.Path) -> None:
235 save_hash_occurrence(tmp_path, {"h": ["a.py::f"]})
236 raw = msgpack.unpackb(
237 (indices_dir(tmp_path) / "hash_occurrence.msgpack").read_bytes(),
238 raw=False,
239 )
240 assert raw["schema_version"] == __version__
241 assert raw["index"] == "hash_occurrence"
242 assert raw["updated_at"]
243
244 def test_empty_index(self, tmp_path: pathlib.Path) -> None:
245 save_hash_occurrence(tmp_path, {})
246 assert load_hash_occurrence(tmp_path) == {}
247
248
249 # ---------------------------------------------------------------------------
250 # index_info
251 # ---------------------------------------------------------------------------
252
253
254 class TestIndexInfo:
255 def test_both_absent(self, tmp_path: pathlib.Path) -> None:
256 info = index_info(tmp_path)
257 assert len(info) == 2
258 names = {i["name"] for i in info}
259 assert names == {"symbol_history", "hash_occurrence"}
260 for item in info:
261 assert item["status"] == "absent"
262
263 def test_symbol_history_present(self, tmp_path: pathlib.Path) -> None:
264 entry = SymbolHistoryEntry("c", "t", "insert", "cid", "bh", "sig")
265 save_symbol_history(tmp_path, {"a.py::f": [entry], "b.py::g": [entry]})
266 info = index_info(tmp_path)
267 sh = next(i for i in info if i["name"] == "symbol_history")
268 assert sh["status"] == "present"
269 assert sh["entries"] == 2
270
271 def test_hash_occurrence_present(self, tmp_path: pathlib.Path) -> None:
272 save_hash_occurrence(tmp_path, {"h1": ["a.py::f"], "h2": ["b.py::g"]})
273 info = index_info(tmp_path)
274 ho = next(i for i in info if i["name"] == "hash_occurrence")
275 assert ho["status"] == "present"
276 assert ho["entries"] == 2
277
278 def test_corrupt_index_reported(self, tmp_path: pathlib.Path) -> None:
279 idx_dir = indices_dir(tmp_path)
280 idx_dir.mkdir(parents=True, exist_ok=True)
281 (idx_dir / "symbol_history.msgpack").write_bytes(b"\xff\xfe garbage")
282 info = index_info(tmp_path)
283 sh = next(i for i in info if i["name"] == "symbol_history")
284 assert sh["status"] == "corrupt"
285
286 def test_updated_at_present_when_index_exists(self, tmp_path: pathlib.Path) -> None:
287 save_hash_occurrence(tmp_path, {"h": ["f.py::x"]})
288 info = index_info(tmp_path)
289 ho = next(i for i in info if i["name"] == "hash_occurrence")
290 assert ho["updated_at"] # non-empty string
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 121 days ago