Coverage for agentos/tests/test_serial_cache.py: 0%
361 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-09 10:28 +0800
1"""Tests for agentos.tools.serial_cache — Serializer, TTLCache, SmartCache."""
3import json
4import pickle
5import threading
6import time
8import pytest
10from agentos.tools.serial_cache import (
11 EvictionPolicy,
12 SerialFormat,
13 Serializer,
14 SmartCache,
15 TTLCache,
16)
18# ============================================================================
19# SerialFormat
20# ============================================================================
22class TestSerialFormat:
23 def test_enum_values(self):
24 assert SerialFormat.JSON.value == "json"
25 assert SerialFormat.PICKLE.value == "pickle"
26 assert SerialFormat.MSGPACK.value == "msgpack"
27 assert SerialFormat.AUTO.value == "auto"
29 def test_detect_explicit_json(self):
30 fmt = SerialFormat.JSON
31 assert fmt.detect(b'{"a":1}') == SerialFormat.JSON
33 def test_detect_explicit_pickle(self):
34 fmt = SerialFormat.PICKLE
35 assert fmt.detect(b"\x80\x04") == SerialFormat.PICKLE
37 def test_detect_auto_json(self):
38 fmt = SerialFormat.AUTO
39 assert fmt.detect(b'{"a":1}') == SerialFormat.JSON
40 assert fmt.detect(b"[1,2,3]") == SerialFormat.JSON
42 def test_detect_auto_pickle(self):
43 fmt = SerialFormat.AUTO
44 data = pickle.dumps({"x": 1})
45 assert fmt.detect(data) == SerialFormat.PICKLE
47 def test_detect_auto_unknown(self):
48 fmt = SerialFormat.AUTO
49 with pytest.raises(ValueError, match="Cannot auto-detect"):
50 fmt.detect(b"\xff\xfe")
53# ============================================================================
54# Serializer
55# ============================================================================
57class TestSerializer:
58 def test_default_format(self):
59 s = Serializer()
60 assert s._fmt == SerialFormat.JSON
62 def test_custom_format(self):
63 s = Serializer(fmt=SerialFormat.PICKLE)
64 assert s._fmt == SerialFormat.PICKLE
66 def test_auto_format_falls_back_to_json(self):
67 s = Serializer(fmt=SerialFormat.AUTO)
68 data = s.dumps({"a": 1})
69 result = s.loads(data)
70 assert result == {"a": 1}
72 def test_dumps_json(self):
73 s = Serializer(fmt=SerialFormat.JSON)
74 data = s.dumps({"hello": "world"})
75 assert json.loads(data) == {"hello": "world"}
77 def test_dumps_pickle(self):
78 s = Serializer(fmt=SerialFormat.PICKLE)
79 data = s.dumps({"x": 42})
80 assert pickle.loads(data) == {"x": 42}
82 def test_loads_json(self):
83 s = Serializer()
84 data = json.dumps({"a": 1, "b": 2}).encode("utf-8")
85 result = s.loads(data)
86 assert result == {"a": 1, "b": 2}
88 def test_loads_pickle(self):
89 s = Serializer()
90 data = pickle.dumps({"z": 99})
91 result = s.loads(data)
92 assert result == {"z": 99}
94 def test_loads_explicit_format(self):
95 s = Serializer()
96 data = json.dumps([1, 2, 3]).encode("utf-8")
97 result = s.loads(data, fmt=SerialFormat.JSON)
98 assert result == [1, 2, 3]
100 def test_roundtrip_json(self):
101 s = Serializer(fmt=SerialFormat.JSON)
102 obj = {"nested": {"key": [1, 2, 3]}}
103 assert s.loads(s.dumps(obj)) == obj
105 def test_roundtrip_pickle(self):
106 s = Serializer(fmt=SerialFormat.PICKLE)
107 obj = {"tuple": (1, 2), "set": {3, 4}}
108 assert s.loads(s.dumps(obj)) == obj
110 def test_stats(self):
111 s = Serializer()
112 s.dumps({"a": 1})
113 s.loads(json.dumps({"b": 2}).encode())
114 st = s.stats
115 assert st["total_serialized"] == 1
116 assert st["total_deserialized"] == 1
117 assert st["format"] == "json"
119 def test_unsupported_format(self):
120 s = Serializer(fmt=SerialFormat.PICKLE)
121 # Create a serializer with a bad internal state and test dumps
122 with pytest.raises(ValueError, match="Unsupported"):
123 s.dumps({}, use_msgpack=False)
124 s._fmt = "bad_fmt"
125 s.dumps({})
127 def test_use_msgpack_flag(self):
128 pytest.importorskip("msgpack")
129 s = Serializer(fmt=SerialFormat.JSON)
130 data = s.dumps({"a": 1}, use_msgpack=True)
131 import msgpack
132 assert msgpack.unpackb(data) == {"a": 1}
134 def test_loads_msgpack(self):
135 pytest.importorskip("msgpack")
136 import msgpack
137 s = Serializer()
138 data = msgpack.packb({"m": "p"})
139 result = s.loads(data)
140 assert result == {"m": "p"}
142 def test_stats_pickle_format(self):
143 s = Serializer(fmt=SerialFormat.PICKLE)
144 s.dumps({"x": 1})
145 assert s.stats["format"] == "pickle"
148# ============================================================================
149# EvictionPolicy
150# ============================================================================
152class TestEvictionPolicy:
153 def test_enum_values(self):
154 assert EvictionPolicy.LRU.value == "lru"
155 assert EvictionPolicy.LFU.value == "lfu"
156 assert EvictionPolicy.TTL_ONLY.value == "ttl_only"
159# ============================================================================
160# TTLCache — Basic Operations
161# ============================================================================
163class TestTTLCacheBasic:
164 def test_defaults(self):
165 cache = TTLCache[int]()
166 assert cache._max_size == 1000
167 assert cache._ttl == 300.0
168 assert cache._policy == EvictionPolicy.LRU
169 assert cache.size == 0
171 def test_custom_params(self):
172 cache = TTLCache[str](max_size=10, ttl=60.0, policy=EvictionPolicy.LFU)
173 assert cache._max_size == 10
174 assert cache._ttl == 60.0
175 assert cache._policy == EvictionPolicy.LFU
177 def test_set_get(self):
178 cache = TTLCache[int]()
179 cache.set("a", 1)
180 assert cache.get("a") == 1
182 def test_get_missing(self):
183 cache = TTLCache[str]()
184 assert cache.get("no-key") is None
186 def test_set_overwrite(self):
187 cache = TTLCache[int]()
188 cache.set("a", 1)
189 cache.set("a", 99)
190 assert cache.get("a") == 99
191 assert cache.size == 1
193 def test_delete(self):
194 cache = TTLCache[int]()
195 cache.set("x", 42)
196 assert cache.delete("x") is True
197 assert cache.get("x") is None
198 assert cache.size == 0
200 def test_delete_missing(self):
201 cache = TTLCache[int]()
202 assert cache.delete("nope") is False
204 def test_clear(self):
205 cache = TTLCache[int]()
206 cache.set("a", 1)
207 cache.set("b", 2)
208 cache.clear()
209 assert cache.size == 0
210 assert cache.get("a") is None
212 def test_size(self):
213 cache = TTLCache[int]()
214 assert cache.size == 0
215 cache.set("a", 1)
216 assert cache.size == 1
217 cache.set("b", 2)
218 assert cache.size == 2
221# ============================================================================
222# TTLCache — TTL Expiration
223# ============================================================================
225class TestTTLCacheTTL:
226 def test_expired_entry(self):
227 cache = TTLCache[int](ttl=0.05)
228 cache.set("a", 1)
229 assert cache.get("a") == 1
230 time.sleep(0.1)
231 assert cache.get("a") is None
233 def test_custom_ttl_per_entry(self):
234 cache = TTLCache[int](ttl=10.0)
235 cache.set("a", 1, ttl=0.05)
236 assert cache.get("a") == 1
237 time.sleep(0.1)
238 assert cache.get("a") is None
240 def test_cleanup(self):
241 cache = TTLCache[int](ttl=0.05)
242 cache.set("a", 1)
243 cache.set("b", 2)
244 time.sleep(0.1)
245 removed = cache.cleanup()
246 assert removed == 2
247 assert cache.size == 0
249 def test_cleanup_partial(self):
250 cache = TTLCache[int](ttl=0.05)
251 cache.set("a", 1)
252 time.sleep(0.1)
253 cache.set("b", 2, ttl=10.0)
254 removed = cache.cleanup()
255 assert removed == 1
256 assert cache.size == 1
257 assert cache.get("b") == 2
260# ============================================================================
261# TTLCache — Eviction
262# ============================================================================
264class TestTTLCacheEviction:
265 def test_lru_eviction(self):
266 cache = TTLCache[int](max_size=2, ttl=300, policy=EvictionPolicy.LRU)
267 cache.set("a", 1)
268 cache.set("b", 2)
269 cache.get("a") # access "a" — now "b" is least recently used
270 cache.set("c", 3)
271 assert cache.get("b") is None
272 assert cache.get("a") == 1
273 assert cache.get("c") == 3
275 def test_lfu_eviction(self):
276 cache = TTLCache[int](max_size=2, ttl=300, policy=EvictionPolicy.LFU)
277 cache.set("a", 1)
278 cache.set("b", 2)
279 cache.get("b") # b access_count = 1
280 cache.get("b") # b access_count = 2
281 cache.get("a") # a access_count = 1
282 cache.set("c", 3)
283 # a has lowest access_count (1), should be evicted
284 assert cache.get("a") is None
285 assert cache.get("b") == 2
286 assert cache.get("c") == 3
288 def test_ttl_only_eviction(self):
289 cache = TTLCache[int](max_size=2, ttl=300, policy=EvictionPolicy.TTL_ONLY)
290 cache.set("a", 1)
291 cache.set("b", 2)
292 cache.get("b") # access b
293 cache.set("c", 3)
294 # TTL_ONLY evicts oldest (first inserted) regardless of access
295 assert cache.get("a") is None
296 assert cache.get("b") == 2
299# ============================================================================
300# TTLCache — Stats
301# ============================================================================
303class TestTTLCacheStats:
304 def test_default_stats(self):
305 cache = TTLCache[int](max_size=50, ttl=10, policy=EvictionPolicy.LFU)
306 s = cache.stats
307 assert s["size"] == 0
308 assert s["max_size"] == 50
309 assert s["ttl"] == 10
310 assert s["policy"] == "lfu"
311 assert s["hits"] == 0
312 assert s["misses"] == 0
313 assert s["evictions"] == 0
315 def test_hits_misses(self):
316 cache = TTLCache[int]()
317 cache.set("a", 1)
318 cache.get("a") # hit
319 cache.get("b") # miss
320 s = cache.stats
321 assert s["hits"] == 1
322 assert s["misses"] == 1
324 def test_hit_rate(self):
325 cache = TTLCache[int]()
326 cache.set("a", 1)
327 cache.get("a") # hit
328 cache.get("a") # hit
329 cache.get("b") # miss
330 s = cache.stats
331 assert s["hit_rate"] == round(2 / 3, 3)
333 def test_hit_rate_no_ops(self):
334 cache = TTLCache[int]()
335 s = cache.stats
336 assert s["hit_rate"] == 0.0
338 def test_evictions_count(self):
339 cache = TTLCache[int](max_size=2, ttl=300, policy=EvictionPolicy.LRU)
340 cache.set("a", 1)
341 cache.set("b", 2)
342 cache.set("c", 3) # evicts "a"
343 assert cache.stats["evictions"] == 1
346# ============================================================================
347# TTLCache — Thread Safety
348# ============================================================================
350class TestTTLCacheThreadSafety:
351 def test_concurrent_access(self):
352 cache = TTLCache[int]()
353 errors = []
355 def worker(start_idx):
356 try:
357 for i in range(start_idx, start_idx + 50):
358 cache.set(f"k{i}", i)
359 cache.get(f"k{i}")
360 cache.delete(f"k{i}")
361 except Exception as e:
362 errors.append(e)
364 threads = [threading.Thread(target=worker, args=(i * 100,)) for i in range(4)]
365 for t in threads:
366 t.start()
367 for t in threads:
368 t.join()
370 assert len(errors) == 0
373# ============================================================================
374# SmartCache
375# ============================================================================
377class TestSmartCache:
378 def test_get_missing(self):
379 sc = SmartCache[int]()
380 assert sc.get("x") is None
382 def test_get_or_compute(self):
383 sc = SmartCache[int]()
384 result = sc.get_or_compute("key", lambda: 42)
385 assert result == 42
386 # Second call should use cache
387 result = sc.get_or_compute("key", lambda: 99)
388 assert result == 42
390 def test_get_or_compute_complex(self):
391 sc = SmartCache[str](ttl=10.0)
392 calls = []
394 def factory():
395 calls.append(1)
396 return "computed"
398 r1 = sc.get_or_compute("k", factory)
399 assert r1 == "computed"
400 assert len(calls) == 1
401 r2 = sc.get_or_compute("k", factory)
402 assert r2 == "computed"
403 assert len(calls) == 1
405 def test_get_or_compute_expired(self):
406 sc = SmartCache[int](ttl=0.05)
407 sc.get_or_compute("k", lambda: 42)
408 time.sleep(0.1)
409 result = sc.get_or_compute("k", lambda: 99)
410 assert result == 99 # recomputed after expiry
412 def test_get_or_compute_custom_ttl(self):
413 sc = SmartCache[int]()
414 r1 = sc.get_or_compute("k", lambda: 42, ttl=0.05)
415 assert r1 == 42
416 time.sleep(0.1)
417 r2 = sc.get_or_compute("k", lambda: 99)
418 assert r2 == 99
420 def test_set_and_get(self):
421 sc = SmartCache[int]()
422 sc.set("x", 100)
423 assert sc.get("x") == 100
425 def test_delete(self):
426 sc = SmartCache[int]()
427 sc.set("x", 1)
428 assert sc.delete("x") is True
429 assert sc.get("x") is None
431 def test_clear(self):
432 sc = SmartCache[int]()
433 sc.set("a", 1)
434 sc.set("b", 2)
435 sc.clear()
436 assert sc.size == 0
438 def test_size(self):
439 sc = SmartCache[int]()
440 assert sc.size == 0
441 sc.set("a", 1)
442 assert sc.size == 1
443 sc.set("b", 2)
444 assert sc.size == 2
446 def test_stats(self):
447 sc = SmartCache[int]()
448 sc.get_or_compute("k", lambda: 42)
449 s = sc.stats
450 assert s["size"] == 1
451 assert s["hits"] >= 0
452 assert s["max_size"] == 1000
454 def test_dump_load(self):
455 sc = SmartCache[str]()
456 sc.set("a", "hello")
457 sc.set("b", "world")
459 data = sc.dump()
460 assert len(data) > 0
462 sc2 = SmartCache[str]()
463 loaded = sc2.load(data)
464 assert loaded == 2
465 assert sc2.get("a") == "hello"
466 assert sc2.get("b") == "world"
468 def test_dump_load_expired_filters(self):
469 sc = SmartCache[int]()
470 sc.set("live", 42, ttl=10.0)
471 sc.get_or_compute("dead", lambda: 1, ttl=0.01)
472 time.sleep(0.05)
474 data = sc.dump()
475 sc2 = SmartCache[int]()
476 loaded = sc2.load(data)
477 assert loaded == 1
478 assert sc2.get("live") == 42
479 assert sc2.get("dead") is None
481 def test_custom_params_passthrough(self):
482 sc = SmartCache[int](max_size=5, ttl=30.0, policy=EvictionPolicy.LFU)
483 assert sc.size == 0
484 assert sc._cache._max_size == 5