"""TDD — Phase 6: caches and indices from binary msgpack to plain JSON. Phase 6 requirements (issue #12): - All cache files under .muse/cache/ use plain JSON (not msgpack) - All index files under .muse/indices/ use plain JSON (not msgpack) - File extensions change from .msgpack to .json - Cache schema versions bumped to invalidate old msgpack files - Old msgpack bytes in any cache/index file → cache miss (no crash) """ from __future__ import annotations import json import pathlib from collections.abc import Mapping import msgpack import pytest # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path: (tmp_path / ".muse").mkdir(parents=True) return tmp_path def _write_msgpack_cache(path: pathlib.Path, data: Mapping[str, object]) -> None: """Write a file in old binary msgpack format.""" path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(msgpack.packb(data, use_bin_type=True)) # --------------------------------------------------------------------------- # stat_cache — .json extension, JSON roundtrip # --------------------------------------------------------------------------- class TestStatCacheJson: def test_cache_file_has_json_extension(self, tmp_path: pathlib.Path) -> None: """StatCache.save() must write a .json file, not .msgpack.""" from muse.core.stat_cache import StatCache from muse.core.paths import cache_dir repo = _make_repo(tmp_path) muse_dir = repo / ".muse" cache = StatCache(muse_dir / "cache", {}) cache._dirty = True cache.save() cache_files = list((muse_dir / "cache").glob("stat.*")) assert len(cache_files) == 1 assert cache_files[0].suffix == ".json", f"Expected .json, got {cache_files[0].suffix}" def test_stat_cache_json_roundtrip(self, tmp_path: pathlib.Path) -> None: """StatCache survives a save → load cycle with all entry fields intact.""" from muse.core.stat_cache import StatCache, FileCacheEntry from muse.core.ids import hash_blob repo = _make_repo(tmp_path) muse_dir = repo / ".muse" cache = StatCache(muse_dir / "cache", {}) cache._entries["src/main.py"] = FileCacheEntry( mtime=1234567890.5, size=1024, ino=99, object_hash=hash_blob(b"main"), dimensions={"symbols": hash_blob(b"syms")}, ) cache._dirty = True cache.save() loaded = StatCache.load(muse_dir) assert "src/main.py" in loaded._entries entry = loaded._entries["src/main.py"] assert entry["size"] == 1024 assert entry["ino"] == 99 assert entry["dimensions"]["symbols"] == hash_blob(b"syms") def test_stat_cache_file_is_json(self, tmp_path: pathlib.Path) -> None: """The stat cache file on disk must be valid UTF-8 JSON.""" from muse.core.stat_cache import StatCache, FileCacheEntry from muse.core.ids import hash_blob repo = _make_repo(tmp_path) muse_dir = repo / ".muse" cache = StatCache(muse_dir / "cache", {}) cache._entries["a.py"] = FileCacheEntry( mtime=1.0, size=10, ino=1, object_hash=hash_blob(b"a"), dimensions={}, ) cache._dirty = True cache.save() cache_file = next((muse_dir / "cache").glob("stat.*")) data = json.loads(cache_file.read_bytes()) assert isinstance(data, dict) assert "entries" in data def test_stale_msgpack_stat_cache_is_miss(self, tmp_path: pathlib.Path) -> None: """Old binary msgpack stat cache → StatCache.load() returns empty cache.""" from muse.core.stat_cache import StatCache, _CACHE_FILENAME repo = _make_repo(tmp_path) muse_dir = repo / ".muse" cache_dir = muse_dir / "cache" cache_dir.mkdir() _write_msgpack_cache(cache_dir / _CACHE_FILENAME, { "version": 3, "entries": {"x.py": {"mtime": 1.0, "size": 1, "ino": 1, "object_hash": "sha256:" + "a" * 64, "dimensions": {}}} }) loaded = StatCache.load(muse_dir) assert loaded._entries == {} # --------------------------------------------------------------------------- # callgraph_cache — JSON via MsgpackCache base # --------------------------------------------------------------------------- class TestCallgraphCacheJson: def test_callgraph_cache_json_roundtrip(self, tmp_path: pathlib.Path) -> None: """CallGraphCache survives save → load with all callee sets intact.""" from muse.core.callgraph_cache import CallGraphCache repo = _make_repo(tmp_path) muse_dir = repo / ".muse" cache = CallGraphCache.empty() cache._cache_dir = muse_dir / "cache" cache.put("sha256:" + "a" * 64, {"callees": ["src/a.py::foo", "src/b.py::bar"]}) cache.save() loaded = CallGraphCache.load(muse_dir) entry = loaded.get("sha256:" + "a" * 64) assert entry is not None assert set(entry["callees"]) == {"src/a.py::foo", "src/b.py::bar"} def test_callgraph_cache_file_is_json(self, tmp_path: pathlib.Path) -> None: """The callgraph cache file must be valid UTF-8 JSON.""" from muse.core.callgraph_cache import CallGraphCache, _CACHE_FILENAME repo = _make_repo(tmp_path) muse_dir = repo / ".muse" cache = CallGraphCache.empty() cache._cache_dir = muse_dir / "cache" cache.put("sha256:" + "b" * 64, {"callees": []}) cache.save() cache_file = (muse_dir / "cache") / _CACHE_FILENAME data = json.loads(cache_file.read_bytes()) assert isinstance(data, dict) def test_callgraph_cache_has_json_extension(self, tmp_path: pathlib.Path) -> None: from muse.core.callgraph_cache import _CACHE_FILENAME assert _CACHE_FILENAME.endswith(".json"), f"Expected .json, got {_CACHE_FILENAME}" def test_stale_msgpack_callgraph_is_miss(self, tmp_path: pathlib.Path) -> None: from muse.core.callgraph_cache import CallGraphCache, _CACHE_FILENAME repo = _make_repo(tmp_path) muse_dir = repo / ".muse" (muse_dir / "cache").mkdir() _write_msgpack_cache((muse_dir / "cache") / _CACHE_FILENAME, {"version": 1, "entries": {"k": {"callees": ["x"]}}}) loaded = CallGraphCache.load(muse_dir) assert loaded.size == 0 # --------------------------------------------------------------------------- # symbol_cache — JSON via MsgpackCache base # --------------------------------------------------------------------------- class TestSymbolCacheJson: def test_symbol_cache_json_roundtrip(self, tmp_path: pathlib.Path) -> None: """SymbolCache survives save → load with symbol tree intact.""" from muse.core.symbol_cache import SymbolCache repo = _make_repo(tmp_path) muse_dir = repo / ".muse" cache = SymbolCache.empty() cache._cache_dir = muse_dir / "cache" obj_id = "sha256:" + "c" * 64 tree = { "src/main.py::foo": { "kind": "function", "name": "foo", "qualified_name": "foo", "content_id": "sha256:" + "a" * 64, "body_hash": "sha256:" + "b" * 64, "signature_id": "sha256:" + "d" * 64, "metadata_id": "sha256:" + "e" * 64, "canonical_key": "src/main.py#foo#function#foo#1", "lineno": 1, "end_lineno": 5, } } cache.put(obj_id, tree) cache.save() loaded = SymbolCache.load(muse_dir) result = loaded.get(obj_id) assert result is not None assert "src/main.py::foo" in result assert result["src/main.py::foo"]["name"] == "foo" def test_symbol_cache_has_json_extension(self, tmp_path: pathlib.Path) -> None: from muse.core.symbol_cache import _CACHE_FILENAME assert _CACHE_FILENAME.endswith(".json"), f"Expected .json, got {_CACHE_FILENAME}" def test_stale_msgpack_symbol_cache_is_miss(self, tmp_path: pathlib.Path) -> None: from muse.core.symbol_cache import SymbolCache, _CACHE_FILENAME repo = _make_repo(tmp_path) muse_dir = repo / ".muse" (muse_dir / "cache").mkdir() _write_msgpack_cache((muse_dir / "cache") / _CACHE_FILENAME, {"version": 1, "entries": {}}) loaded = SymbolCache.load(muse_dir) assert loaded.size == 0 # --------------------------------------------------------------------------- # indices — JSON, .json extension # --------------------------------------------------------------------------- class TestIndicesJson: def test_indices_json_roundtrip(self, tmp_path: pathlib.Path) -> None: """save_symbol_history → load_symbol_history roundtrip preserves entries.""" from muse.core.indices import save_symbol_history, load_symbol_history, SymbolHistoryEntry repo = _make_repo(tmp_path) entry = SymbolHistoryEntry( commit_id="sha256:" + "d" * 64, committed_at="2026-05-21T00:00:00+00:00", op="added", content_id="sha256:" + "f" * 64, body_hash="sha256:" + "e" * 64, signature_id="sha256:" + "a" * 64, ) index = {"src/main.py::foo": [entry]} save_symbol_history(repo, index) loaded = load_symbol_history(repo) assert "src/main.py::foo" in loaded assert loaded["src/main.py::foo"][0].op == "added" def test_symbol_history_file_is_json(self, tmp_path: pathlib.Path) -> None: """The symbol_history index file must be valid UTF-8 JSON.""" from muse.core.indices import save_symbol_history, SymbolHistoryEntry, _index_path repo = _make_repo(tmp_path) entry = SymbolHistoryEntry( commit_id="sha256:" + "f" * 64, committed_at="2026-05-21T00:00:00+00:00", op="added", content_id="sha256:" + "b" * 64, body_hash="sha256:" + "a" * 64, signature_id="sha256:" + "c" * 64, ) save_symbol_history(repo, {"a.py::fn": [entry]}) path = _index_path(repo, "symbol_history") data = json.loads(path.read_bytes()) assert isinstance(data, dict) def test_index_file_has_json_extension(self, tmp_path: pathlib.Path) -> None: """Index files must have .json extension.""" from muse.core.indices import _index_path repo = _make_repo(tmp_path) p = _index_path(repo, "symbol_history") assert p.suffix == ".json", f"Expected .json, got {p.suffix}" def test_hash_occurrence_json_roundtrip(self, tmp_path: pathlib.Path) -> None: """save_hash_occurrence → load_hash_occurrence roundtrip.""" from muse.core.indices import save_hash_occurrence, load_hash_occurrence repo = _make_repo(tmp_path) index = {"sha256:" + "a" * 64: ["src/main.py::foo", "src/util.py::bar"]} save_hash_occurrence(repo, index) loaded = load_hash_occurrence(repo) assert "sha256:" + "a" * 64 in loaded assert "src/main.py::foo" in loaded["sha256:" + "a" * 64] def test_stale_msgpack_index_is_miss(self, tmp_path: pathlib.Path) -> None: """Old msgpack index file → load returns empty dict, no crash.""" from muse.core.indices import load_symbol_history, _index_path repo = _make_repo(tmp_path) path = _index_path(repo, "symbol_history") path.parent.mkdir(parents=True, exist_ok=True) _write_msgpack_cache(path, {"schema_version": "0.1", "entries": {}}) result = load_symbol_history(repo) assert result == {} # --------------------------------------------------------------------------- # test_history — JSON # --------------------------------------------------------------------------- class TestHistoryJson: def test_test_history_json_roundtrip(self, tmp_path: pathlib.Path) -> None: """save_history → load_history roundtrip preserves run records.""" from muse.core.test_history import save_history, load_history, RunRecord, CaseRecord repo = _make_repo(tmp_path) record = RunRecord( run_id="sha256:" + "a" * 64, timestamp="2026-05-21T00:00:00Z", commit_id=None, branch="dev", total=3, passed=2, failed=1, errored=0, skipped=0, results=[ CaseRecord( node_id="tests/test_foo.py::test_bar", outcome="passed", duration_ms=12.5, symbol_addresses=["src/foo.py::bar"], ) ], ) save_history(repo, [record]) loaded = load_history(repo) assert len(loaded) == 1 assert loaded[0]["run_id"] == record["run_id"] assert loaded[0]["passed"] == 2 def test_test_history_file_is_json(self, tmp_path: pathlib.Path) -> None: """The test history file on disk must be valid UTF-8 JSON.""" from muse.core.test_history import save_history, load_history, RunRecord, _history_path repo = _make_repo(tmp_path) save_history(repo, []) path = _history_path(repo) data = json.loads(path.read_bytes()) assert isinstance(data, dict) def test_test_history_file_has_json_extension(self, tmp_path: pathlib.Path) -> None: from muse.core.test_history import _history_path repo = _make_repo(tmp_path) p = _history_path(repo) assert p.suffix == ".json", f"Expected .json, got {p.suffix}" def test_stale_msgpack_history_is_miss(self, tmp_path: pathlib.Path) -> None: """Old binary msgpack test history → load_history returns [].""" from muse.core.test_history import load_history, _history_path repo = _make_repo(tmp_path) path = _history_path(repo) path.parent.mkdir(parents=True, exist_ok=True) _write_msgpack_cache(path, {"version": 1, "runs": []}) result = load_history(repo) assert result == []