Spaces:
Sleeping
Sleeping
File size: 7,058 Bytes
dc1b199 3c31a2a dc1b199 3c31a2a 7d37f11 3c31a2a dc1b199 3c31a2a dc1b199 3c31a2a b76f199 879e4e0 3c31a2a dc1b199 3c31a2a dc1b199 3c31a2a dc1b199 3c31a2a dc1b199 b76f199 dc1b199 3c31a2a 7d37f11 dc1b199 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 | """Tests for the section cache.
Key acceptance criterion: calling generation twice with the same payload
should NOT trigger a second LLM call (verified via mock call-count assertion).
"""
from pathlib import Path
import pytest
from app.cache.section_cache import clear_all, compute_cache_key, get, invalidate, set
@pytest.fixture(autouse=True)
def isolated_cache(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Redirect cache to a per-test temporary directory."""
monkeypatch.setattr("app.cache.section_cache.settings.cache_dir", tmp_path)
monkeypatch.setattr("app.config.settings.cache_dir", tmp_path)
# ββ compute_cache_key βββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def test_cache_key_deterministic() -> None:
"""Same inputs should always produce the same key."""
k1 = compute_cache_key("E4", ["bullet one", "bullet two"], "tenant_a")
k2 = compute_cache_key("E4", ["bullet one", "bullet two"], "tenant_a")
assert k1 == k2
def test_cache_key_differs_for_different_inputs() -> None:
"""Different bullets should produce different cache keys."""
k1 = compute_cache_key("E4", ["bullet one"], "tenant_a")
k2 = compute_cache_key("E4", ["bullet two"], "tenant_a")
assert k1 != k2
def test_cache_key_differs_for_different_tenants() -> None:
"""Same template+bullets but different tenant must produce different keys."""
k1 = compute_cache_key("E4", ["bullet one"], "tenant_a")
k2 = compute_cache_key("E4", ["bullet one"], "tenant_b")
assert k1 != k2
def test_cache_key_is_64_hex_chars() -> None:
"""SHA-256 cache key should be 64 hexadecimal characters."""
key = compute_cache_key("E4", ["fact"], "tenant_a")
assert len(key) == 64
assert all(c in "0123456789abcdef" for c in key)
def test_cache_key_differs_for_different_ai_levels() -> None:
"""Different ai_level values must produce different cache keys.
This prevents a RAG-only result (level 1) being served to a full-AI
request (level 5) just because the bullets are identical.
"""
k1 = compute_cache_key("E4", ["fact one"], "tenant_a", ai_level=1)
k3 = compute_cache_key("E4", ["fact one"], "tenant_a", ai_level=3)
k5 = compute_cache_key("E4", ["fact one"], "tenant_a", ai_level=5)
assert k1 != k3
assert k3 != k5
assert k1 != k5
def test_cache_key_differs_for_different_ai_percent() -> None:
"""Different ai_percent values must produce different cache keys."""
k0 = compute_cache_key("E4", ["fact one"], "tenant_a", ai_percent=0)
k50 = compute_cache_key("E4", ["fact one"], "tenant_a", ai_percent=50)
k100 = compute_cache_key("E4", ["fact one"], "tenant_a", ai_percent=100)
assert k0 != k50
assert k50 != k100
assert k0 != k100
def test_cache_key_differs_for_different_interference_level() -> None:
"""Same ai_percent but different interference level must not share a cache entry."""
k_medium = compute_cache_key("E4", ["fact"], "tenant_a", ai_percent=52, interference_level="medium")
k_max = compute_cache_key("E4", ["fact"], "tenant_a", ai_percent=52, interference_level="maximum")
k_none = compute_cache_key("E4", ["fact"], "tenant_a", ai_percent=52, interference_level=None)
assert k_medium != k_max
assert k_medium != k_none
assert k_max != k_none
def test_cache_key_default_ai_level_equals_3() -> None:
"""Default ai_level of 3 must match an explicit ai_level=3 call."""
k_default = compute_cache_key("E4", ["fact"], "tenant_a")
k_explicit = compute_cache_key("E4", ["fact"], "tenant_a", ai_level=3)
assert k_default == k_explicit
# ββ get / set / invalidate ββββββββββββββββββββββββββββββββββββββββββββββββββββ
def test_get_returns_none_on_miss() -> None:
"""get should return None for a key that has not been set."""
assert get("nonexistent-key-abc") is None
def test_set_and_get_roundtrip() -> None:
"""set then get should return the same payload."""
key = compute_cache_key("E4", ["test bullet"], "tenant_a")
payload = {"text": "FIXTURE output", "confidence": 0.9, "provenance": []}
set(key, payload)
result = get(key)
assert result is not None
assert result["text"] == "FIXTURE output"
assert result["confidence"] == 0.9
def test_invalidate_removes_entry() -> None:
"""invalidate should delete the cache entry so get returns None."""
key = compute_cache_key("E4", ["bullet"], "tenant_a")
set(key, {"text": "x", "confidence": 0.0, "provenance": []})
assert get(key) is not None
invalidate(key)
assert get(key) is None
def test_invalidate_nonexistent_returns_false() -> None:
"""invalidate on a missing key should return False."""
assert invalidate("totally-missing-key") is False
def test_clear_all_removes_entries(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""clear_all should remove all .json files in the cache directory."""
monkeypatch.setattr("app.cache.section_cache.settings.cache_dir", tmp_path)
key1 = compute_cache_key("E2", ["b1"], "tenant_a")
key2 = compute_cache_key("E4", ["b2"], "tenant_a")
set(key1, {"text": "a", "confidence": 0.5, "provenance": []})
set(key2, {"text": "b", "confidence": 0.5, "provenance": []})
count = clear_all()
assert count == 2
assert get(key1) is None
assert get(key2) is None
# ββ Cache prevents duplicate LLM calls (acceptance test) βββββββββββββββββββββ
def test_cache_hit_prevents_llm_call(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
"""Generating the same section twice must only call the LLM once.
This is the key acceptance criterion for the caching layer.
"""
from app.generator.adapter import MockLLMAdapter
call_count = {"n": 0}
class CountingAdapter(MockLLMAdapter):
def generate_section(self, skeleton: str, bullets: list[str], snippets: list[str], **kwargs) -> str:
call_count["n"] += 1
return "FIXTURE generated text."
adapter = CountingAdapter()
bullets = ["fact one", "fact two"]
template_id = "E4"
tenant_id = "tenant_a"
key = compute_cache_key(template_id, bullets, tenant_id)
def _generate(sk: str, bl: list[str], sn: list[str]) -> str:
cached = get(key)
if cached:
return cached["text"]
result = adapter.generate_section(sk, bl, sn)
set(key, {"text": result, "confidence": 0.8, "provenance": []})
return result
# First call β LLM should be invoked
out1 = _generate("skeleton", bullets, [])
# Second call β should use cache
out2 = _generate("skeleton", bullets, [])
assert out1 == out2
assert call_count["n"] == 1, (
f"LLM was called {call_count['n']} times; expected exactly 1"
)
|