Repository navigation
Expand file tree
/
Copy pathtest_cache.py
More file actions
182 lines (114 loc) · 4.4 KB
/
Copy pathtest_cache.py
File metadata and controls
182 lines (114 loc) · 4.4 KB
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
import time
import pytest
from auth0_api_python._internal.cache_expiry import is_fresh
from auth0_api_python.cache import InMemoryCache
# ===== InMemoryCache Basic Operations =====
def test_in_memory_cache_get_set():
"""Test basic get and set operations."""
cache = InMemoryCache()
cache.set("key1", {"data": "value1"})
cache.set("key2", "string_value")
cache.set("key3", 123)
assert cache.get("key1") == {"data": "value1"}
assert cache.get("key2") == "string_value"
assert cache.get("key3") == 123
def test_in_memory_cache_ttl_expiry():
"""Test that entries expire after TTL."""
cache = InMemoryCache()
cache.set("key1", "value1", ttl_seconds=1)
assert cache.get("key1") == "value1"
time.sleep(1.1)
assert cache.get("key1") is None
def test_in_memory_cache_delete():
"""Test delete operation."""
cache = InMemoryCache()
cache.set("key1", "value1")
assert cache.get("key1") == "value1"
cache.delete("key1")
assert cache.get("key1") is None
cache.delete("nonexistent")
def test_in_memory_cache_clear():
"""Test clear all operation."""
cache = InMemoryCache()
cache.set("key1", "value1")
cache.set("key2", "value2")
cache.set("key3", "value3")
assert cache.get("key1") == "value1"
assert cache.get("key2") == "value2"
cache.clear()
assert cache.get("key1") is None
assert cache.get("key2") is None
assert cache.get("key3") is None
def test_in_memory_cache_get_expired():
"""Test that expired entries return None and are removed."""
cache = InMemoryCache()
cache.set("key1", "value1", ttl_seconds=1)
time.sleep(1.1)
result = cache.get("key1")
assert result is None
result_again = cache.get("key1")
assert result_again is None
def test_in_memory_cache_get_nonexistent():
"""Test that nonexistent keys return None."""
cache = InMemoryCache()
assert cache.get("nonexistent") is None
assert cache.get("another_missing_key") is None
def test_in_memory_cache_overwrite():
"""Test overwriting existing keys."""
cache = InMemoryCache()
cache.set("key1", "original_value")
assert cache.get("key1") == "original_value"
cache.set("key1", "new_value")
assert cache.get("key1") == "new_value"
cache.set("key1", {"complex": "value"}, ttl_seconds=10)
assert cache.get("key1") == {"complex": "value"}
def test_in_memory_cache_no_ttl():
"""Test that entries without TTL never expire."""
cache = InMemoryCache()
cache.set("key1", "value1")
time.sleep(0.5)
assert cache.get("key1") == "value1"
time.sleep(0.5)
assert cache.get("key1") == "value1"
def test_in_memory_cache_ttl_zero_expires_immediately():
"""Test that ttl_seconds=0 means entries expire immediately (always refetch)."""
cache = InMemoryCache()
cache.set("key1", "value1", ttl_seconds=0)
# Entry should be immediately expired — forces refetch on next access
assert cache.get("key1") is None
# ===== LRU Eviction =====
def test_in_memory_cache_max_entries_eviction():
"""Test LRU eviction when max_entries is reached."""
cache = InMemoryCache(max_entries=3)
cache.set("key1", "value1")
cache.set("key2", "value2")
cache.set("key3", "value3")
assert cache.get("key1") == "value1"
assert cache.get("key2") == "value2"
assert cache.get("key3") == "value3"
cache.set("key4", "value4")
assert cache.get("key1") is None
assert cache.get("key2") == "value2"
assert cache.get("key3") == "value3"
assert cache.get("key4") == "value4"
def test_in_memory_cache_lru_access_order():
"""Test that least recently used entry is evicted first."""
cache = InMemoryCache(max_entries=3)
cache.set("key1", "value1")
cache.set("key2", "value2")
cache.set("key3", "value3")
cache.get("key1")
time.sleep(0.01)
cache.get("key2")
time.sleep(0.01)
cache.set("key4", "value4")
assert cache.get("key3") is None
assert cache.get("key1") == "value1"
assert cache.get("key2") == "value2"
assert cache.get("key4") == "value4"
# ===== Expiry Buffer =====
@pytest.mark.parametrize(("seconds_left", "expected"), [(29, False), (30, False), (31, True)])
def test_is_fresh_requires_more_than_the_buffer(seconds_left, expected):
"""An entry is fresh only with more than 30 seconds left."""
now = 1_700_000_000
assert is_fresh(now + seconds_left, now) is expected