Bhargav
Initial KinChat env: models, personas, scenarios, rubrics, grader, env loop, FastAPI app, client, dashboard, baseline inference (377 tests passing)
2e8387b | """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" | |