gabriel / muse public
test_core_stat_cache.py python
411 lines 15.0 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 120 days ago
1 """Tests for muse.core.stat_cache.
2
3 Coverage
4 --------
5 - Cache hit: file with unchanged (mtime, size) returns stored hash without I/O.
6 - Cache miss: new or modified file is re-hashed and entry is updated.
7 - Stale-entry pruning: entries for deleted files are removed.
8 - Dimension hash round-trip: set_dimension / get_dimension.
9 - Dimension eviction on object-hash miss: dimensions reset when file changes.
10 - Persistence: save() / load() round-trip via .muse/cache/stat.msgpack.
11 - Atomic write: temp file is cleaned up; no corruption on concurrent use.
12 - empty(): no-op — save() is a no-op without a cache_dir.
13 - load_cache() convenience helper.
14 - walk_workdir() integration: cache is used and persisted automatically.
15 """
16
17 from __future__ import annotations
18
19 import json
20 import pathlib
21 import time
22
23 import pytest
24
25 from muse.core.types import blob_id
26 from muse.core.stat_cache import FileCacheEntry, StatCache, _hash_bytes, load_cache
27 from muse.core.snapshot import walk_workdir
28 from muse.core.paths import muse_dir
29
30
31 # ---------------------------------------------------------------------------
32 # Helpers
33 # ---------------------------------------------------------------------------
34
35
36 def _make_muse_dir(tmp_path: pathlib.Path) -> pathlib.Path:
37 dot_muse = muse_dir(tmp_path)
38 dot_muse.mkdir()
39 (dot_muse / "cache").mkdir()
40 return dot_muse
41
42
43 def _write(path: pathlib.Path, content: str = "hello") -> pathlib.Path:
44 path.parent.mkdir(parents=True, exist_ok=True)
45 path.write_text(content, encoding="utf-8")
46 return path
47
48
49 # ---------------------------------------------------------------------------
50 # _hash_bytes — canonical hash function
51 # ---------------------------------------------------------------------------
52
53
54 class TestHashBytes:
55 def test_matches_hashlib(self, tmp_path: pathlib.Path) -> None:
56 f = _write(tmp_path / "f.txt", "muse")
57 assert _hash_bytes(f) == blob_id(b"muse")
58
59 def test_empty_file(self, tmp_path: pathlib.Path) -> None:
60 f = tmp_path / "empty.txt"
61 f.write_bytes(b"")
62 assert _hash_bytes(f) == blob_id(b"")
63
64 def test_large_file_chunked(self, tmp_path: pathlib.Path) -> None:
65 data = b"x" * (200 * 1024) # 200 KiB — forces multiple 64 KiB chunks
66 f = tmp_path / "big.bin"
67 f.write_bytes(data)
68 assert _hash_bytes(f) == blob_id(data)
69
70
71 # ---------------------------------------------------------------------------
72 # StatCache — construction
73 # ---------------------------------------------------------------------------
74
75
76 class TestStatCacheConstruction:
77 def test_load_missing_file_returns_empty(self, tmp_path: pathlib.Path) -> None:
78 muse_dir = _make_muse_dir(tmp_path)
79 cache = StatCache.load(muse_dir)
80 assert cache._entries == {}
81
82 def test_load_corrupt_msgpack_returns_empty(self, tmp_path: pathlib.Path) -> None:
83 muse_dir = _make_muse_dir(tmp_path)
84 (muse_dir / "cache" / "stat.msgpack").write_bytes(b"not msgpack")
85 cache = StatCache.load(muse_dir)
86 assert cache._entries == {}
87
88 def test_load_wrong_version_returns_empty(self, tmp_path: pathlib.Path) -> None:
89 import msgpack as _mp
90 muse_dir = _make_muse_dir(tmp_path)
91 (muse_dir / "cache" / "stat.msgpack").write_bytes(
92 _mp.packb({"version": 99, "entries": {}}, use_bin_type=True)
93 )
94 cache = StatCache.load(muse_dir)
95 assert cache._entries == {}
96
97 def test_empty_has_no_cache_dir(self, tmp_path: pathlib.Path) -> None:
98 cache = StatCache.empty()
99 assert cache._cache_dir is None
100 assert cache._entries == {}
101
102 def test_load_cache_helper_with_muse_dir(self, tmp_path: pathlib.Path) -> None:
103 _make_muse_dir(tmp_path)
104 cache = load_cache(tmp_path)
105 assert isinstance(cache, StatCache)
106 assert cache._cache_dir == muse_dir(tmp_path) / "cache"
107
108 def test_load_cache_helper_without_muse_dir(self, tmp_path: pathlib.Path) -> None:
109 cache = load_cache(tmp_path)
110 assert cache._cache_dir is None
111
112
113 # ---------------------------------------------------------------------------
114 # StatCache — get_object_hash (hit / miss)
115 # ---------------------------------------------------------------------------
116
117
118 class TestGetObjectHash:
119 def test_first_call_is_cache_miss(self, tmp_path: pathlib.Path) -> None:
120 muse_dir = _make_muse_dir(tmp_path)
121 f = _write(tmp_path / "a.py", "x = 1")
122 cache = StatCache.load(muse_dir)
123
124 h = cache.get_object_hash(tmp_path, f)
125
126 assert h == _hash_bytes(f)
127 assert cache._dirty is True
128 assert "a.py" in cache._entries
129
130 def test_second_call_is_cache_hit_no_dirty(self, tmp_path: pathlib.Path) -> None:
131 muse_dir = _make_muse_dir(tmp_path)
132 f = _write(tmp_path / "a.py", "x = 1")
133 cache = StatCache.load(muse_dir)
134 cache.get_object_hash(tmp_path, f)
135 cache._dirty = False # reset after first miss
136
137 h2 = cache.get_object_hash(tmp_path, f)
138
139 assert h2 == _hash_bytes(f)
140 assert cache._dirty is False # no re-hash, no dirty flag
141
142 def test_modified_file_triggers_miss(self, tmp_path: pathlib.Path) -> None:
143 muse_dir = _make_muse_dir(tmp_path)
144 f = _write(tmp_path / "a.py", "x = 1")
145 cache = StatCache.load(muse_dir)
146 h1 = cache.get_object_hash(tmp_path, f)
147
148 # Modify file content (ensure mtime changes on this filesystem).
149 time.sleep(0.01)
150 f.write_text("x = 2", encoding="utf-8")
151 h2 = cache.get_object_hash(tmp_path, f)
152
153 assert h1 != h2
154 assert h2 == _hash_bytes(f)
155
156 def test_same_content_new_mtime_triggers_miss_but_same_hash(
157 self, tmp_path: pathlib.Path
158 ) -> None:
159 muse_dir = _make_muse_dir(tmp_path)
160 f = _write(tmp_path / "a.py", "identical")
161 cache = StatCache.load(muse_dir)
162 h1 = cache.get_object_hash(tmp_path, f)
163
164 time.sleep(0.01)
165 f.write_text("identical", encoding="utf-8")
166 h2 = cache.get_object_hash(tmp_path, f)
167
168 # Cache miss because mtime changed, but hash is still the same.
169 assert h1 == h2
170
171
172 # ---------------------------------------------------------------------------
173 # StatCache — dimension hashes
174 # ---------------------------------------------------------------------------
175
176
177 class TestDimensionHashes:
178 def test_set_and_get_dimension(self, tmp_path: pathlib.Path) -> None:
179 muse_dir = _make_muse_dir(tmp_path)
180 f = _write(tmp_path / "src.py")
181 cache = StatCache.load(muse_dir)
182 cache.get_object_hash(tmp_path, f) # ensure entry exists
183
184 cache.set_dimension(tmp_path, f, "symbols", "abc123")
185
186 assert cache.get_dimension(tmp_path, f, "symbols") == "abc123"
187
188 def test_get_dimension_missing_key_returns_none(self, tmp_path: pathlib.Path) -> None:
189 muse_dir = _make_muse_dir(tmp_path)
190 f = _write(tmp_path / "src.py")
191 cache = StatCache.load(muse_dir)
192 cache.get_object_hash(tmp_path, f)
193
194 assert cache.get_dimension(tmp_path, f, "nonexistent") is None
195
196 def test_get_dimension_missing_entry_returns_none(self, tmp_path: pathlib.Path) -> None:
197 muse_dir = _make_muse_dir(tmp_path)
198 f = _write(tmp_path / "src.py")
199 cache = StatCache.load(muse_dir)
200 # Never called get_object_hash, so no entry exists.
201 assert cache.get_dimension(tmp_path, f, "symbols") is None
202
203 def test_dimension_evicted_on_object_hash_miss(self, tmp_path: pathlib.Path) -> None:
204 """When a file changes, its dimension hashes must be cleared."""
205 muse_dir = _make_muse_dir(tmp_path)
206 f = _write(tmp_path / "src.py", "v1")
207 cache = StatCache.load(muse_dir)
208 cache.get_object_hash(tmp_path, f)
209 cache.set_dimension(tmp_path, f, "symbols", "stale-hash")
210
211 time.sleep(0.01)
212 f.write_text("v2", encoding="utf-8")
213 cache.get_object_hash(tmp_path, f) # triggers miss → evicts dimensions
214
215 assert cache.get_dimension(tmp_path, f, "symbols") is None
216
217 def test_multiple_dimensions(self, tmp_path: pathlib.Path) -> None:
218 muse_dir = _make_muse_dir(tmp_path)
219 f = _write(tmp_path / "src.py")
220 cache = StatCache.load(muse_dir)
221 cache.get_object_hash(tmp_path, f)
222 cache.set_dimension(tmp_path, f, "symbols", "sym-hash")
223 cache.set_dimension(tmp_path, f, "imports", "imp-hash")
224
225 assert cache.get_dimension(tmp_path, f, "symbols") == "sym-hash"
226 assert cache.get_dimension(tmp_path, f, "imports") == "imp-hash"
227
228 def test_set_dimension_noop_for_unknown_file(self, tmp_path: pathlib.Path) -> None:
229 """set_dimension on a file with no entry must not crash."""
230 muse_dir = _make_muse_dir(tmp_path)
231 f = _write(tmp_path / "ghost.py")
232 cache = StatCache.load(muse_dir)
233 # No get_object_hash call → no entry.
234 cache.set_dimension(tmp_path, f, "symbols", "x") # must not raise
235
236
237 # ---------------------------------------------------------------------------
238 # StatCache — prune
239 # ---------------------------------------------------------------------------
240
241
242 class TestPrune:
243 def test_prune_removes_stale_entries(self, tmp_path: pathlib.Path) -> None:
244 muse_dir = _make_muse_dir(tmp_path)
245 f1 = _write(tmp_path / "keep.py")
246 f2 = _write(tmp_path / "gone.py")
247 cache = StatCache.load(muse_dir)
248 cache.get_object_hash(tmp_path, f1)
249 cache.get_object_hash(tmp_path, f2)
250
251 cache.prune({"keep.py"})
252
253 assert "keep.py" in cache._entries
254 assert "gone.py" not in cache._entries
255
256 def test_prune_noop_when_all_present(self, tmp_path: pathlib.Path) -> None:
257 muse_dir = _make_muse_dir(tmp_path)
258 f = _write(tmp_path / "a.py")
259 cache = StatCache.load(muse_dir)
260 cache.get_object_hash(tmp_path, f)
261 cache._dirty = False
262
263 cache.prune({"a.py"})
264
265 assert cache._dirty is False
266
267 def test_prune_empty_known_set_clears_all(self, tmp_path: pathlib.Path) -> None:
268 muse_dir = _make_muse_dir(tmp_path)
269 f = _write(tmp_path / "a.py")
270 cache = StatCache.load(muse_dir)
271 cache.get_object_hash(tmp_path, f)
272
273 cache.prune(set())
274
275 assert cache._entries == {}
276
277
278 # ---------------------------------------------------------------------------
279 # StatCache — persistence (save / load round-trip)
280 # ---------------------------------------------------------------------------
281
282
283 class TestPersistence:
284 def test_save_and_reload(self, tmp_path: pathlib.Path) -> None:
285 muse_dir = _make_muse_dir(tmp_path)
286 f = _write(tmp_path / "mod.py", "print('hi')")
287 cache = StatCache.load(muse_dir)
288 h = cache.get_object_hash(tmp_path, f)
289 cache.save()
290
291 assert (muse_dir / "cache" / "stat.msgpack").is_file()
292
293 cache2 = StatCache.load(muse_dir)
294 cache2._dirty = False
295 h2 = cache2.get_object_hash(tmp_path, f)
296
297 assert h2 == h
298 assert cache2._dirty is False # served from cache, no re-hash
299
300 def test_save_is_atomic_no_tmp_left(self, tmp_path: pathlib.Path) -> None:
301 muse_dir = _make_muse_dir(tmp_path)
302 f = _write(tmp_path / "x.py")
303 cache = StatCache.load(muse_dir)
304 cache.get_object_hash(tmp_path, f)
305 cache.save()
306
307 assert not any((muse_dir / "cache").glob(".stat_cache_*.tmp"))
308
309 def test_save_noop_when_not_dirty(self, tmp_path: pathlib.Path) -> None:
310 muse_dir = _make_muse_dir(tmp_path)
311 cache = StatCache.load(muse_dir)
312 cache.save() # nothing written
313 assert not (muse_dir / "cache" / "stat.msgpack").exists()
314
315 def test_empty_cache_save_is_noop(self) -> None:
316 cache = StatCache.empty()
317 cache.save() # must not raise
318
319 def test_dimensions_persisted(self, tmp_path: pathlib.Path) -> None:
320 muse_dir = _make_muse_dir(tmp_path)
321 f = _write(tmp_path / "s.py")
322 cache = StatCache.load(muse_dir)
323 cache.get_object_hash(tmp_path, f)
324 cache.set_dimension(tmp_path, f, "symbols", "sym42")
325 cache.save()
326
327 cache2 = StatCache.load(muse_dir)
328 # Validate entry shape — mtime/size unchanged so entry is still valid.
329 assert cache2.get_dimension(tmp_path, f, "symbols") == "sym42"
330
331 def test_json_format_is_versioned(self, tmp_path: pathlib.Path) -> None:
332 muse_dir = _make_muse_dir(tmp_path)
333 f = _write(tmp_path / "v.py")
334 cache = StatCache.load(muse_dir)
335 cache.get_object_hash(tmp_path, f)
336 cache.save()
337
338 import msgpack as _msgpack
339 raw = _msgpack.unpackb((muse_dir / "cache" / "stat.msgpack").read_bytes(), raw=False)
340 assert raw["version"] == 2
341 assert "v.py" in raw["entries"]
342
343
344 # ---------------------------------------------------------------------------
345 # walk_workdir integration
346 # ---------------------------------------------------------------------------
347
348
349 class TestWalkWorkdirCacheIntegration:
350 def test_walk_creates_cache_file(self, tmp_path: pathlib.Path) -> None:
351 dot_muse = muse_dir(tmp_path)
352 dot_muse.mkdir()
353 _write(tmp_path / "a.py", "x = 1")
354 _write(tmp_path / "b.py", "y = 2")
355
356 walk_workdir(tmp_path)
357
358 assert (dot_muse / "cache" / "stat.msgpack").is_file()
359
360 def test_walk_second_call_uses_cache(self, tmp_path: pathlib.Path) -> None:
361 """Second walk should hit cache for both files — no dirty flag set."""
362 dot_muse = muse_dir(tmp_path)
363 dot_muse.mkdir()
364 _write(tmp_path / "a.py", "x = 1")
365
366 walk_workdir(tmp_path) # cold — populates cache
367
368 cache = StatCache.load(dot_muse)
369 cache._dirty = False
370 cache.get_object_hash(tmp_path, tmp_path / "a.py")
371 # Should not set dirty because mtime/size unchanged.
372 assert cache._dirty is False
373
374 def test_walk_excludes_secrets_from_cache(self, tmp_path: pathlib.Path) -> None:
375 """Secrets excluded by built-in blocklist must not appear in the manifest."""
376 dot_muse = muse_dir(tmp_path)
377 dot_muse.mkdir()
378 _write(tmp_path / "visible.py")
379 _write(tmp_path / ".env")
380
381 manifest = walk_workdir(tmp_path)
382
383 assert "visible.py" in manifest
384 assert ".env" not in manifest
385
386 def test_walk_tracks_non_secret_dotfiles(self, tmp_path: pathlib.Path) -> None:
387 """Non-secret dotfiles like .cursorrules are now tracked by default."""
388 dot_muse = muse_dir(tmp_path)
389 dot_muse.mkdir()
390 _write(tmp_path / ".cursorrules")
391 _write(tmp_path / ".editorconfig")
392
393 manifest = walk_workdir(tmp_path)
394
395 assert ".cursorrules" in manifest
396 assert ".editorconfig" in manifest
397
398 def test_walk_without_muse_dir_still_works(self, tmp_path: pathlib.Path) -> None:
399 """walk_workdir must work correctly even with no .muse directory."""
400 _write(tmp_path / "a.py", "ok")
401 manifest = walk_workdir(tmp_path)
402 assert "a.py" in manifest
403
404 def test_walk_hashes_match_direct_hash(self, tmp_path: pathlib.Path) -> None:
405 dot_muse = muse_dir(tmp_path)
406 dot_muse.mkdir()
407 f = _write(tmp_path / "c.py", "content")
408
409 manifest = walk_workdir(tmp_path)
410
411 assert manifest["c.py"] == _hash_bytes(f)
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 120 days ago