File size: 6,416 Bytes
2e8387b | 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 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 | """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"
|