feat(notion-search): add SHA256-keyed JSON file cache

Cache layer for rerank results. Key is SHA256 of query + sorted
candidate IDs, so changing the candidate set automatically invalidates.
1-day TTL by default; corrupted cache files are silently dropped.

5 tests cover hit, miss, different-candidates, TTL expiry, corruption.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-04-28 07:42:48 +09:00
parent 89b20aef16
commit 32a1c9d538
2 changed files with 138 additions and 0 deletions

View File

@@ -58,11 +58,80 @@ def test_call_claude_raises_when_neither_available():
"error mentions setup options")
def test_cache_miss_returns_none():
import tempfile
from pathlib import Path
import _search_cache
with tempfile.TemporaryDirectory() as tmpdir:
result = _search_cache.cache_get("query", ["id1", "id2"], cache_dir=Path(tmpdir))
_assert(result is None, "cache miss returns None")
def test_cache_hit_returns_cached_results():
import tempfile
from pathlib import Path
import _search_cache
with tempfile.TemporaryDirectory() as tmpdir:
cache_dir = Path(tmpdir)
results = [{"id": "abc", "title": "Test"}]
_search_cache.cache_put("query", ["id1", "id2"], results, cache_dir=cache_dir)
hit = _search_cache.cache_get("query", ["id1", "id2"], cache_dir=cache_dir)
_assert(hit == results, "cache hit returns the stored results")
def test_cache_miss_on_different_candidates():
import tempfile
from pathlib import Path
import _search_cache
with tempfile.TemporaryDirectory() as tmpdir:
cache_dir = Path(tmpdir)
_search_cache.cache_put("query", ["id1", "id2"], [{"x": 1}], cache_dir=cache_dir)
miss = _search_cache.cache_get("query", ["id1", "id3"], cache_dir=cache_dir)
_assert(miss is None, "different candidate set hashes to different key")
def test_cache_miss_after_ttl_expires():
import tempfile
from pathlib import Path
import _search_cache
with tempfile.TemporaryDirectory() as tmpdir:
cache_dir = Path(tmpdir)
_search_cache.cache_put("query", ["id1"], [{"x": 1}], cache_dir=cache_dir)
# TTL=0 means immediately expired
miss = _search_cache.cache_get("query", ["id1"], cache_dir=cache_dir, ttl_seconds=0)
_assert(miss is None, "expired cache returns None")
def test_cache_handles_corrupted_file():
import tempfile
import hashlib
from pathlib import Path
import _search_cache
with tempfile.TemporaryDirectory() as tmpdir:
cache_dir = Path(tmpdir)
cache_dir.mkdir(exist_ok=True)
# Write garbage to the expected key path
key = hashlib.sha256("query|id1".encode()).hexdigest()
(cache_dir / f"{key}.json").write_text("not json {{{")
result = _search_cache.cache_get("query", ["id1"], cache_dir=cache_dir)
_assert(result is None, "corrupted cache file returns None")
def run_all():
tests = [
test_call_claude_dispatches_to_sdk_when_available,
test_call_claude_falls_back_to_cli_when_no_sdk,
test_call_claude_raises_when_neither_available,
test_cache_miss_returns_none,
test_cache_hit_returns_cached_results,
test_cache_miss_on_different_candidates,
test_cache_miss_after_ttl_expires,
test_cache_handles_corrupted_file,
]
for t in tests:
print(f"\n{t.__name__}")