Spaces:
Runtime error
Runtime error
| """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 | |
| 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" | |
| ) | |