File size: 3,597 Bytes
7af280e | 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 | """Tests for the in-memory TTL cache in data/cache.py — the Cache class and
the @cached decorator, tested in isolation with no live API calls and no
dependency on wmata_realtime.py/weather.py.
"""
import time
from data.cache import Cache, cached
def test_get_returns_value_before_ttl_expires():
cache = Cache()
cache.set("station", "Metro Center", ttl_seconds=5)
assert cache.get("station") == "Metro Center"
def test_get_returns_none_after_ttl_expires():
cache = Cache()
cache.set("station", "Metro Center", ttl_seconds=0.1)
time.sleep(0.15)
assert cache.get("station") is None
def test_expired_entry_is_evicted_from_store():
cache = Cache()
cache.set("station", "Metro Center", ttl_seconds=0.1)
time.sleep(0.15)
cache.get("station") # triggers lazy eviction
assert cache.stats()["size"] == 0
def test_get_missing_key_returns_none():
cache = Cache()
assert cache.get("does-not-exist") is None
def test_clear_removes_all_entries():
cache = Cache()
cache.set("a", 1, ttl_seconds=5)
cache.set("b", 2, ttl_seconds=5)
cache.clear()
assert cache.get("a") is None
assert cache.get("b") is None
assert cache.stats()["size"] == 0
def test_stats_tracks_hits_misses_and_size():
cache = Cache()
cache.set("a", 1, ttl_seconds=5)
cache.get("a") # hit
cache.get("a") # hit
cache.get("missing") # miss
stats = cache.stats()
assert stats["hits"] == 2
assert stats["misses"] == 1
assert stats["size"] == 1
def test_stats_counts_expiry_as_a_miss():
cache = Cache()
cache.set("a", 1, ttl_seconds=0.1)
time.sleep(0.15)
cache.get("a")
stats = cache.stats()
assert stats["hits"] == 0
assert stats["misses"] == 1
def test_cached_decorator_returns_cached_value_before_ttl_expires():
calls = []
@cached(ttl_seconds=5)
def slow_double(n):
calls.append(n)
return n * 2
assert slow_double(3) == 6
assert slow_double(3) == 6
assert calls == [3] # second call was served from cache, not recomputed
def test_cached_decorator_recomputes_after_ttl_expires():
calls = []
@cached(ttl_seconds=0.1)
def slow_double(n):
calls.append(n)
return n * 2
assert slow_double(3) == 6
time.sleep(0.15)
assert slow_double(3) == 6
assert calls == [3, 3] # recomputed after expiry
def test_cached_decorator_uses_different_keys_for_different_args():
calls = []
@cached(ttl_seconds=5)
def slow_double(n):
calls.append(n)
return n * 2
assert slow_double(3) == 6
assert slow_double(4) == 8
assert slow_double(3) == 6 # still cached from the first call
assert slow_double(4) == 8 # still cached from the second call
assert calls == [3, 4]
def test_cached_decorator_uses_different_keys_for_different_kwargs():
calls = []
@cached(ttl_seconds=5)
def get_thing(station=None):
calls.append(station)
return f"info for {station}"
assert get_thing(station="Metro Center") == "info for Metro Center"
assert get_thing(station="Union Station") == "info for Union Station"
assert get_thing(station="Metro Center") == "info for Metro Center"
assert calls == ["Metro Center", "Union Station"]
def test_cached_decorator_exposes_underlying_cache_for_clearing():
@cached(ttl_seconds=5)
def slow_double(n):
return n * 2
slow_double(3)
assert slow_double.cache.stats()["size"] == 1
slow_double.cache.clear()
assert slow_double.cache.stats()["size"] == 0
|