"""TDD tests for kinchat.server.cache — PersonaResponseCache.""" import pytest from kinchat.models import ChatMsg, PersonaState from kinchat.server.cache import PersonaResponseCache # ---- Fixtures ---- def _state(persona_id: str = "mom", trust: float = 0.5, **kw) -> PersonaState: return PersonaState(persona_id=persona_id, trust=trust, **kw) def _msg(sender: str, text: str, turn: int = 0) -> ChatMsg: return ChatMsg(sender=sender, text=text, turn=turn) # ---- Basic put/get ---- def test_put_then_get_returns_value(): cache = PersonaResponseCache() key = cache.make_key("mom", _state(), [_msg("user", "hi")]) cache.put(key, "Hello dear!") assert cache.get(key) == "Hello dear!" def test_get_missing_returns_none(): cache = PersonaResponseCache() assert cache.get("nonexistent-key") is None # ---- Determinism ---- def test_same_inputs_same_key(): cache = PersonaResponseCache() msgs = [_msg("user", "hi", turn=1), _msg("mom", "hello", turn=2)] k1 = cache.make_key("mom", _state(), msgs) k2 = cache.make_key("mom", _state(), msgs) assert k1 == k2 def test_different_persona_id_different_key(): cache = PersonaResponseCache() msgs = [_msg("user", "hi")] k1 = cache.make_key("mom", _state("mom"), msgs) k2 = cache.make_key("dad", _state("dad"), msgs) assert k1 != k2 def test_state_delta_changes_key(): cache = PersonaResponseCache() msgs = [_msg("user", "hi")] k1 = cache.make_key("mom", _state(trust=0.50), msgs) k2 = cache.make_key("mom", _state(trust=0.51), msgs) assert k1 != k2 def test_different_last_message_changes_key(): cache = PersonaResponseCache() k1 = cache.make_key("mom", _state(), [_msg("user", "hi")]) k2 = cache.make_key("mom", _state(), [_msg("user", "bye")]) assert k1 != k2 def test_only_last_n_messages_affect_key(): """Prepending older messages beyond last_n should NOT change the key.""" cache = PersonaResponseCache(last_n=2) tail = [_msg("user", "a", turn=4), _msg("mom", "b", turn=5)] short = tail long = [_msg("user", "old1", turn=1), _msg("user", "old2", turn=2), _msg("user", "old3", turn=3)] + tail k_short = cache.make_key("mom", _state(), short) k_long = cache.make_key("mom", _state(), long) assert k_short == k_long def test_known_facts_set_hashes_consistently_across_insertion_orders(): """Set ordering is non-deterministic in Python; ensure key is stable.""" cache = PersonaResponseCache() # Build two PersonaStates with the same facts in different insertion orders. s1 = PersonaState(persona_id="mom") for f in ["loves_chai", "knows_birthday", "saw_letter"]: s1.known_facts.add(f) s2 = PersonaState(persona_id="mom") for f in ["saw_letter", "loves_chai", "knows_birthday"]: s2.known_facts.add(f) s3 = PersonaState(persona_id="mom") for f in ["knows_birthday", "saw_letter", "loves_chai"]: s3.known_facts.add(f) msgs = [_msg("user", "hi")] k1 = cache.make_key("mom", s1, msgs) k2 = cache.make_key("mom", s2, msgs) k3 = cache.make_key("mom", s3, msgs) assert k1 == k2 == k3 def test_known_facts_change_changes_key(): cache = PersonaResponseCache() s1 = PersonaState(persona_id="mom", known_facts={"a", "b"}) s2 = PersonaState(persona_id="mom", known_facts={"a", "b", "c"}) msgs = [_msg("user", "hi")] assert cache.make_key("mom", s1, msgs) != cache.make_key("mom", s2, msgs) # ---- LRU behavior ---- def test_lru_eviction_when_capacity_exceeded(): cache = PersonaResponseCache(capacity=2) cache.put("k1", "v1") cache.put("k2", "v2") cache.put("k3", "v3") # evicts k1 assert cache.get("k1") is None assert cache.get("k2") == "v2" assert cache.get("k3") == "v3" assert cache.stats()["evictions"] == 1 def test_lru_recency_reorders_on_get(): """Access first key, then put new — second-inserted (now LRU) should evict.""" cache = PersonaResponseCache(capacity=2) cache.put("k1", "v1") cache.put("k2", "v2") # Access k1 -> it becomes most recent; k2 is now LRU. assert cache.get("k1") == "v1" cache.put("k3", "v3") # should evict k2, not k1 assert cache.get("k1") == "v1" assert cache.get("k2") is None assert cache.get("k3") == "v3" def test_put_updates_existing_key_without_eviction(): cache = PersonaResponseCache(capacity=2) cache.put("k1", "v1") cache.put("k1", "v1b") cache.put("k2", "v2") assert cache.get("k1") == "v1b" assert cache.get("k2") == "v2" assert cache.stats()["evictions"] == 0 # ---- Hit rate / stats ---- def test_hit_rate_zero_at_start(): cache = PersonaResponseCache() assert cache.hit_rate == 0.0 def test_hit_rate_after_mixed_lookups(): cache = PersonaResponseCache() cache.put("k1", "v1") cache.get("k1") # hit cache.get("k1") # hit cache.get("missing") # miss assert cache.hit_rate == pytest.approx(2 / 3) def test_stats_returns_all_six_keys(): cache = PersonaResponseCache(capacity=4) cache.put("k1", "v1") cache.get("k1") cache.get("missing") s = cache.stats() for field in ("hits", "misses", "evictions", "size", "capacity", "hit_rate"): assert field in s assert s["hits"] == 1 assert s["misses"] == 1 assert s["evictions"] == 0 assert s["size"] == 1 assert s["capacity"] == 4 assert s["hit_rate"] == pytest.approx(0.5) def test_get_or_none_hit_path(): cache = PersonaResponseCache() state = _state() msgs = [_msg("user", "hi")] key = cache.make_key("mom", state, msgs) cache.put(key, "cached!") assert cache.get_or_none("mom", state, msgs) == "cached!" assert cache.stats()["hits"] == 1 def test_get_or_none_miss_path(): cache = PersonaResponseCache() assert cache.get_or_none("mom", _state(), [_msg("user", "hi")]) is None assert cache.stats()["misses"] == 1 def test_size_reflects_entries(): cache = PersonaResponseCache(capacity=10) assert cache.stats()["size"] == 0 cache.put("k1", "v1") cache.put("k2", "v2") assert cache.stats()["size"] == 2 def test_empty_recent_msgs_is_valid(): cache = PersonaResponseCache() k1 = cache.make_key("mom", _state(), []) k2 = cache.make_key("mom", _state(), []) assert k1 == k2 cache.put(k1, "silent") assert cache.get(k1) == "silent"