File size: 5,969 Bytes
d4d0bc7 | 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 | """Tests for backend/cache.py — TTLCache and module-level singletons."""
import time
import threading
import pytest
from backend.cache import TTLCache, plans_cache, stats_cache
# ── Basic get/set ─────────────────────────────────────────────────────────────
def test_set_and_get_basic():
cache = TTLCache(ttl=60)
cache.set("key1", "value1")
assert cache.get("key1") == "value1"
def test_get_missing_key_returns_none():
cache = TTLCache(ttl=60)
assert cache.get("nonexistent") is None
def test_set_various_value_types():
cache = TTLCache(ttl=60)
cache.set("int_key", 42)
cache.set("dict_key", {"a": 1, "b": [1, 2, 3]})
cache.set("list_key", [1, 2, 3])
assert cache.get("int_key") == 42
assert cache.get("dict_key") == {"a": 1, "b": [1, 2, 3]}
assert cache.get("list_key") == [1, 2, 3]
# ── TTL expiry ────────────────────────────────────────────────────────────────
def test_expired_entry_returns_none():
cache = TTLCache(ttl=1)
cache.set("expiring", "will_expire", ttl=1)
time.sleep(1.1)
assert cache.get("expiring") is None
def test_ttl_not_yet_expired():
cache = TTLCache(ttl=60)
cache.set("fresh", "still_alive", ttl=60)
assert cache.get("fresh") == "still_alive"
def test_per_entry_ttl_overrides_default():
cache = TTLCache(ttl=60)
# Use a very short TTL for this specific entry
cache.set("short_lived", "gone_soon", ttl=1)
assert cache.get("short_lived") == "gone_soon"
time.sleep(1.1)
assert cache.get("short_lived") is None
def test_default_ttl_is_used_when_not_specified():
"""Setting an entry without explicit TTL uses the cache's default TTL."""
cache = TTLCache(ttl=300)
cache.set("default_ttl_key", "hello")
# Should be present because 300s haven't passed
assert cache.get("default_ttl_key") == "hello"
# ── delete() ──────────────────────────────────────────────────────────────────
def test_delete_removes_entry():
cache = TTLCache(ttl=60)
cache.set("to_delete", "present")
assert cache.get("to_delete") == "present"
cache.delete("to_delete")
assert cache.get("to_delete") is None
def test_delete_nonexistent_key_does_not_raise():
cache = TTLCache(ttl=60)
cache.delete("ghost_key") # should not raise
# ── clear() ───────────────────────────────────────────────────────────────────
def test_clear_empties_cache():
cache = TTLCache(ttl=60)
cache.set("a", 1)
cache.set("b", 2)
cache.set("c", 3)
cache.clear()
assert cache.get("a") is None
assert cache.get("b") is None
assert cache.get("c") is None
def test_clear_on_empty_cache_does_not_raise():
cache = TTLCache(ttl=60)
cache.clear() # should not raise
# ── size() ────────────────────────────────────────────────────────────────────
def test_size_returns_correct_count():
cache = TTLCache(ttl=60)
assert cache.size() == 0
cache.set("x", 1)
assert cache.size() == 1
cache.set("y", 2)
assert cache.size() == 2
cache.delete("x")
assert cache.size() == 1
cache.clear()
assert cache.size() == 0
def test_size_does_not_count_expired_entries():
"""Expired entries that have been accessed are removed, shrinking size."""
cache = TTLCache(ttl=1)
cache.set("expires", "soon", ttl=1)
assert cache.size() == 1
time.sleep(1.1)
# Access triggers eviction
cache.get("expires")
assert cache.size() == 0
# ── Thread safety ─────────────────────────────────────────────────────────────
def test_concurrent_reads_writes_do_not_crash():
"""Multiple threads reading and writing should not raise or corrupt data."""
cache = TTLCache(ttl=60)
errors = []
def writer(n):
try:
for i in range(50):
cache.set(f"key_{n}_{i}", f"val_{n}_{i}")
except Exception as e:
errors.append(e)
def reader(n):
try:
for i in range(50):
cache.get(f"key_{n}_{i}")
except Exception as e:
errors.append(e)
threads = []
for t in range(5):
threads.append(threading.Thread(target=writer, args=(t,)))
threads.append(threading.Thread(target=reader, args=(t,)))
for th in threads:
th.start()
for th in threads:
th.join()
assert errors == [], f"Thread errors: {errors}"
# ── Module-level singletons ───────────────────────────────────────────────────
def test_plans_cache_is_ttlcache_instance():
assert isinstance(plans_cache, TTLCache)
def test_stats_cache_is_ttlcache_instance():
assert isinstance(stats_cache, TTLCache)
def test_plans_cache_ttl_is_600():
assert plans_cache.ttl == 600
def test_stats_cache_ttl_is_120():
assert stats_cache.ttl == 120
def test_plans_cache_and_stats_cache_are_different_objects():
assert plans_cache is not stats_cache
def test_singletons_are_same_object_on_reimport():
from backend.cache import plans_cache as pc2, stats_cache as sc2
assert plans_cache is pc2
assert stats_cache is sc2
|