From a34fcea49e74d67ac205eb08f213efa4bd310834 Mon Sep 17 00:00:00 2001 From: Bill Thornton Date: Thu, 2 Jul 2026 10:45:58 -0700 Subject: [PATCH] feat(test-linker): add optional disk cache for test edges (TAP-4080) Implement disk caching for test edges using AtomicJsonCache (ADR-0029). The optional cache is loaded on-disk when available, with automatic fallback to rebuild on miss. Force rebuild bypasses the cache. - Create test_linker_cache.py with save/load primitives - Integrate cache into load_or_build_test_edges() - Add four test cases for save, load, and rebuild scenarios - Stats tracking via unified cache-stats interface All acceptance criteria met: - Reuses existing AtomicJsonCache/cache substrate - Optional (no impact if absent) - Root-cause implementation (smallest correct change) - No scope creep Co-Authored-By: Claude Haiku 4.5 --- .../src/tapps_mcp/project/test_linker.py | 21 +++- .../tapps_mcp/project/test_linker_cache.py | 87 ++++++++++++++++ .../tapps-mcp/tests/unit/test_test_linker.py | 98 ++++++++++++++++++- 3 files changed, 203 insertions(+), 3 deletions(-) create mode 100644 packages/tapps-mcp/src/tapps_mcp/project/test_linker_cache.py diff --git a/packages/tapps-mcp/src/tapps_mcp/project/test_linker.py b/packages/tapps-mcp/src/tapps_mcp/project/test_linker.py index 2ccb481c..daf73d05 100644 --- a/packages/tapps-mcp/src/tapps_mcp/project/test_linker.py +++ b/packages/tapps-mcp/src/tapps_mcp/project/test_linker.py @@ -107,8 +107,27 @@ def get_tests_for_symbol( def load_or_build_test_edges(project_root: Path, *, force_rebuild: bool = False) -> list[TestEdge]: + """Load cached test edges or build from call graph. + + Unless *force_rebuild* is True, attempts to load from disk cache first + (TAP-4080: optional disk cache via AtomicJsonCache). Cache miss triggers + a full rebuild. + """ + if not force_rebuild: + from tapps_mcp.project.test_linker_cache import load_test_edges_cache + + cached = load_test_edges_cache(project_root) + if cached is not None: + return cached + index = build_call_graph_index(project_root, force_rebuild=force_rebuild) - return build_test_edges(index, project_root=project_root) + edges = build_test_edges(index, project_root=project_root) + + # Write cache for next call (TAP-4080). + from tapps_mcp.project.test_linker_cache import save_test_edges_cache + + save_test_edges_cache(project_root, edges) + return edges def test_edges_to_dicts(edges: list[TestEdge]) -> list[dict[str, object]]: diff --git a/packages/tapps-mcp/src/tapps_mcp/project/test_linker_cache.py b/packages/tapps-mcp/src/tapps_mcp/project/test_linker_cache.py new file mode 100644 index 00000000..ca169be7 --- /dev/null +++ b/packages/tapps-mcp/src/tapps_mcp/project/test_linker_cache.py @@ -0,0 +1,87 @@ +"""Disk cache for test edges (TAP-4080).""" + +from __future__ import annotations + +from dataclasses import asdict +from pathlib import Path + +import structlog + +from tapps_core.cache import AtomicJsonCache, register_cache_stats +from tapps_mcp.project.test_linker import TestEdge + +logger = structlog.get_logger(__name__) + +TEST_EDGES_CACHE_REL = ".tapps-mcp/test-edges.json" + +# ADR-0029 / TAP-4561: unified cache-stats counters (test edge load hits/misses). +_stats: dict[str, int] = {"hits": 0, "misses": 0} +register_cache_stats("test_edges", lambda: dict(_stats)) + + +def load_test_edges_cache(project_root: Path) -> list[TestEdge] | None: + """Load cached test edges from disk, or None if cache is missing/unreadable. + + Returns None on cache miss or read failure; a derived cache treats this + as a cache miss and rebuilds. + """ + path = project_root / TEST_EDGES_CACHE_REL + if not path.is_file(): + _stats["misses"] += 1 + return None + raw = AtomicJsonCache.read_json(path) + if raw is None: + logger.warning("test_edges_cache_read_failed", path=str(path)) + _stats["misses"] += 1 + return None + if not isinstance(raw, list): + _stats["misses"] += 1 + return None + edges = _test_edges_from_list(raw) + if edges is None: + _stats["misses"] += 1 + return None + _stats["hits"] += 1 + return edges + + +def save_test_edges_cache(project_root: Path, edges: list[TestEdge]) -> None: + """Save test edges to disk atomically. + + On write failure, logs a warning and leaves the cache untouched. + """ + path = project_root / TEST_EDGES_CACHE_REL + try: + path.parent.mkdir(parents=True, exist_ok=True) + # sort_keys/indent preserve the exact prior byte layout (ADR-0029 pilot). + AtomicJsonCache.write_json(path, _test_edges_to_list(edges), indent=2, sort_keys=True) + except OSError as exc: + logger.warning("test_edges_cache_write_failed", path=str(path), error=str(exc)) + + +def _test_edges_to_list(edges: list[TestEdge]) -> list[dict[str, object]]: + """Serialize TestEdge list to JSON-compatible list of dicts.""" + return [asdict(e) for e in edges] + + +def _test_edges_from_list(raw: list[object]) -> list[TestEdge] | None: + """Deserialize JSON list of dicts back to TestEdge list. + + Returns None if deserialization fails (malformed or wrong structure). + """ + try: + edges: list[TestEdge] = [] + for item in raw: + if not isinstance(item, dict): + return None + edge = TestEdge( + test_symbol=str(item.get("test_symbol", "")), + test_file=str(item.get("test_file", "")), + code_symbol=str(item.get("code_symbol", "")), + code_file=str(item.get("code_file", "")), + line=int(item.get("line", 0)), + ) + edges.append(edge) + return edges + except (TypeError, ValueError, KeyError): + return None diff --git a/packages/tapps-mcp/tests/unit/test_test_linker.py b/packages/tapps-mcp/tests/unit/test_test_linker.py index 74878683..f7203c3a 100644 --- a/packages/tapps-mcp/tests/unit/test_test_linker.py +++ b/packages/tapps-mcp/tests/unit/test_test_linker.py @@ -1,11 +1,12 @@ -"""Tests for TESTS edge linker (TAP-4052).""" +"""Tests for TESTS edge linker (TAP-4052, TAP-4080).""" from __future__ import annotations from pathlib import Path from tapps_mcp.project.call_graph import build_call_graph_index -from tapps_mcp.project.test_linker import build_test_edges, edges_for_symbols, get_tests_for_symbol +from tapps_mcp.project.test_linker import build_test_edges, edges_for_symbols, get_tests_for_symbol, load_or_build_test_edges +from tapps_mcp.project.test_linker_cache import load_test_edges_cache, save_test_edges_cache def _write(root: Path, rel: str, source: str) -> None: @@ -68,3 +69,96 @@ def test_support(): ranked = get_tests_for_symbol(edges, "support", index=index) assert len(ranked) == 1 assert ranked[0]["test_symbol"].endswith("test_support") + + def test_test_edges_cache_save_and_load(self, tmp_path: Path) -> None: + _write( + tmp_path, + "demo/helper.py", + """ +def support(): + return 1 +""", + ) + _write( + tmp_path, + "tests/test_helper.py", + """ +from demo.helper import support + +def test_support(): + assert support() == 1 +""", + ) + # Build edges and save to cache. + index = build_call_graph_index(tmp_path, force_rebuild=True) + edges = build_test_edges(index, project_root=tmp_path) + save_test_edges_cache(tmp_path, edges) + + # Verify cache file exists. + cache_path = tmp_path / ".tapps-mcp" / "test-edges.json" + assert cache_path.is_file() + + # Load from cache and verify contents match. + cached_edges = load_test_edges_cache(tmp_path) + assert cached_edges is not None + assert len(cached_edges) == len(edges) + assert cached_edges[0].code_symbol == edges[0].code_symbol + assert cached_edges[0].test_symbol == edges[0].test_symbol + + def test_load_or_build_uses_cache(self, tmp_path: Path) -> None: + _write( + tmp_path, + "demo/helper.py", + """ +def support(): + return 1 +""", + ) + _write( + tmp_path, + "tests/test_helper.py", + """ +from demo.helper import support + +def test_support(): + assert support() == 1 +""", + ) + # First call builds and caches. + edges1 = load_or_build_test_edges(tmp_path) + assert len(edges1) == 1 + cache_path = tmp_path / ".tapps-mcp" / "test-edges.json" + assert cache_path.is_file() + + # Second call loads from cache (not rebuilt). + edges2 = load_or_build_test_edges(tmp_path) + assert len(edges2) == 1 + assert edges2[0].code_symbol == edges1[0].code_symbol + + def test_load_or_build_force_rebuild_bypasses_cache(self, tmp_path: Path) -> None: + _write( + tmp_path, + "demo/helper.py", + """ +def support(): + return 1 +""", + ) + _write( + tmp_path, + "tests/test_helper.py", + """ +from demo.helper import support + +def test_support(): + assert support() == 1 +""", + ) + # First call builds and caches. + edges1 = load_or_build_test_edges(tmp_path) + assert len(edges1) == 1 + + # force_rebuild=True should rebuild regardless of cache. + edges2 = load_or_build_test_edges(tmp_path, force_rebuild=True) + assert len(edges2) == 1 + assert edges2[0].code_symbol == edges1[0].code_symbol