tests/test_llm_cache.py

"""Tests for the LLM response cache (in-memory + disk layers).

The disk layer is what lets a benchmark that dies midway be re-run for
free: identical prompts replay from `LEXSI_LLM_CACHE_DIR` across process
restarts instead of re-hitting the API.
"""

from __future__ import annotations

import importlib


def _fresh_client(monkeypatch, cache_dir=None):
    """Reimport the client module with a clean cache + chosen env."""
    if cache_dir is not None:
        monkeypatch.setenv("LEXSI_LLM_CACHE_DIR", str(cache_dir))
    else:
        monkeypatch.delenv("LEXSI_LLM_CACHE_DIR", raising=False)
    import lexsi_ds.llm.client as c
    importlib.reload(c)
    c.reset_llm_cache_stats()
    return c


def test_memory_cache_hit(monkeypatch):
    c = _fresh_client(monkeypatch)
    k = c._llm_cache_key("m", "p", 100, "sys", "user")
    c._llm_cache_put(k, c.LLMResult(text="A"))
    assert c._llm_cache_get(k).text == "A"
    assert c.llm_cache_stats() == {"hits": 1, "misses": 1}


def test_disk_cache_survives_restart(monkeypatch, tmp_path):
    c = _fresh_client(monkeypatch, cache_dir=tmp_path)
    k = c._llm_cache_key("m", "p", 100, "sys", "user-X")
    c._llm_cache_put(k, c.LLMResult(text="ANSWER", raw={"usage": 1}))
    assert (tmp_path / f"{k}.json").exists()

    # Simulate a process restart: wipe the in-memory cache, keep disk.
    c._LLM_CACHE.clear()
    c.reset_llm_cache_stats()
    hit = c._llm_cache_get(k)
    assert hit is not None and hit.text == "ANSWER"
    assert hit.raw == {"usage": 1}
    assert c.llm_cache_stats()["hits"] == 1


def test_disk_cache_miss_returns_none(monkeypatch, tmp_path):
    c = _fresh_client(monkeypatch, cache_dir=tmp_path)
    k = c._llm_cache_key("m", "p", 100, "sys", "never-seen")
    assert c._llm_cache_get(k) is None


def test_corrupt_disk_entry_is_a_miss_not_a_crash(monkeypatch, tmp_path):
    c = _fresh_client(monkeypatch, cache_dir=tmp_path)
    k = c._llm_cache_key("m", "p", 100, "sys", "corrupt")
    (tmp_path / f"{k}.json").write_text("{not valid json")
    assert c._llm_cache_get(k) is None


def test_cache_disabled_env(monkeypatch, tmp_path):
    monkeypatch.setenv("LEXSI_LLM_CACHE", "0")
    c = _fresh_client(monkeypatch, cache_dir=tmp_path)
    k = c._llm_cache_key("m", "p", 100, "sys", "user")
    c._llm_cache_put(k, c.LLMResult(text="A"))
    assert c._llm_cache_get(k) is None
    assert not list(tmp_path.glob("*.json"))
    monkeypatch.delenv("LEXSI_LLM_CACHE", raising=False)
    importlib.reload(c)  # restore module default for other tests