Spaces:
Sleeping
Sleeping
File size: 2,934 Bytes
44436be | 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 | import time
import fakeredis
from utils import redis_client
from utils.cache import cached
def setup_function():
redis_client.set_redis_client(fakeredis.FakeRedis())
def test_caches_result_for_identical_args():
calls = []
@cached(ttl_seconds=60)
def fn(conn, x):
calls.append(x)
return x * 2
assert fn("conn-a", 5) == 10
assert fn("conn-b", 5) == 10 # different conn, same x -> cache hit
assert calls == [5]
def test_different_args_are_different_cache_entries():
calls = []
@cached(ttl_seconds=60)
def fn(conn, x):
calls.append(x)
return x * 2
fn("conn-a", 5)
fn("conn-a", 7)
assert calls == [5, 7]
def test_expires_after_ttl():
calls = []
@cached(ttl_seconds=1)
def fn(conn, x):
calls.append(x)
return x
fn("conn-a", 1)
fn("conn-a", 1) # still within TTL
assert calls == [1]
time.sleep(1.1)
fn("conn-a", 1) # expired -> recompute
assert calls == [1, 1]
def test_cache_clear_forces_recompute():
calls = []
@cached(ttl_seconds=60)
def fn(conn, x):
calls.append(x)
return x
fn("conn-a", 1)
fn.cache_clear()
fn("conn-a", 1)
assert calls == [1, 1]
def test_cache_clear_only_clears_own_function():
calls_a, calls_b = [], []
@cached(ttl_seconds=60)
def fn_a(conn, x):
calls_a.append(x)
return x
@cached(ttl_seconds=60)
def fn_b(conn, x):
calls_b.append(x)
return x
fn_a("conn-a", 1)
fn_b("conn-a", 1)
fn_a.cache_clear()
fn_a("conn-a", 1) # recomputed
fn_b("conn-a", 1) # still cached
assert calls_a == [1, 1]
assert calls_b == [1]
def test_is_cached_reflects_state():
@cached(ttl_seconds=10)
def fn(conn, x):
return x
assert fn.is_cached(1) is False
fn("conn-a", 1)
assert fn.is_cached(1) is True
assert fn.is_cached(2) is False
def test_is_cached_false_after_ttl_expiry():
@cached(ttl_seconds=1)
def fn(conn, x):
return x
fn("conn-a", 1)
assert fn.is_cached(1) is True
time.sleep(1.1)
assert fn.is_cached(1) is False
def test_kwargs_are_part_of_the_key():
calls = []
@cached(ttl_seconds=60)
def fn(conn, category=None):
calls.append(category)
return category
fn("conn-a", category="built")
fn("conn-a", category="natural")
fn("conn-a", category="built")
assert calls == ["built", "natural"]
def test_cached_value_round_trips_complex_objects():
# Regression guard: cache must survive pickling non-trivial return types
# (services return Pydantic models / lists of dicts, not just scalars).
@cached(ttl_seconds=60)
def fn(conn):
return {"a": [1, 2, {"nested": True}], "b": None}
first = fn("conn-a")
second = fn("conn-a")
assert first == second == {"a": [1, 2, {"nested": True}], "b": None}
|