Coverage for agentos/tests/test_serial_cache.py: 0%

1"""Tests for agentos.tools.serial_cache — Serializer, TTLCache, SmartCache.""" 

2 

3import json 

4import pickle 

5import pytest 

6import threading 

7import time 

8from agentos.tools.serial_cache import ( 

9 EvictionPolicy, 

10 SerialFormat, 

11 Serializer, 

12 SmartCache, 

13 TTLCache, 

14) 

15 

16 

17# ============================================================================ 

18# SerialFormat 

19# ============================================================================ 

20 

21class TestSerialFormat: 

22 def test_enum_values(self): 

23 assert SerialFormat.JSON.value == "json" 

24 assert SerialFormat.PICKLE.value == "pickle" 

25 assert SerialFormat.MSGPACK.value == "msgpack" 

26 assert SerialFormat.AUTO.value == "auto" 

27 

28 def test_detect_explicit_json(self): 

29 fmt = SerialFormat.JSON 

30 assert fmt.detect(b'{"a":1}') == SerialFormat.JSON 

31 

32 def test_detect_explicit_pickle(self): 

33 fmt = SerialFormat.PICKLE 

34 assert fmt.detect(b"\x80\x04") == SerialFormat.PICKLE 

35 

36 def test_detect_auto_json(self): 

37 fmt = SerialFormat.AUTO 

38 assert fmt.detect(b'{"a":1}') == SerialFormat.JSON 

39 assert fmt.detect(b"[1,2,3]") == SerialFormat.JSON 

40 

41 def test_detect_auto_pickle(self): 

42 fmt = SerialFormat.AUTO 

43 data = pickle.dumps({"x": 1}) 

44 assert fmt.detect(data) == SerialFormat.PICKLE 

45 

46 def test_detect_auto_unknown(self): 

47 fmt = SerialFormat.AUTO 

48 with pytest.raises(ValueError, match="Cannot auto-detect"): 

49 fmt.detect(b"\xff\xfe") 

50 

51 

52# ============================================================================ 

53# Serializer 

54# ============================================================================ 

55 

56class TestSerializer: 

57 def test_default_format(self): 

58 s = Serializer() 

59 assert s._fmt == SerialFormat.JSON 

60 

61 def test_custom_format(self): 

62 s = Serializer(fmt=SerialFormat.PICKLE) 

63 assert s._fmt == SerialFormat.PICKLE 

64 

65 def test_auto_format_falls_back_to_json(self): 

66 s = Serializer(fmt=SerialFormat.AUTO) 

67 data = s.dumps({"a": 1}) 

68 result = s.loads(data) 

69 assert result == {"a": 1} 

70 

71 def test_dumps_json(self): 

72 s = Serializer(fmt=SerialFormat.JSON) 

73 data = s.dumps({"hello": "world"}) 

74 assert json.loads(data) == {"hello": "world"} 

75 

76 def test_dumps_pickle(self): 

77 s = Serializer(fmt=SerialFormat.PICKLE) 

78 data = s.dumps({"x": 42}) 

79 assert pickle.loads(data) == {"x": 42} 

80 

81 def test_loads_json(self): 

82 s = Serializer() 

83 data = json.dumps({"a": 1, "b": 2}).encode("utf-8") 

84 result = s.loads(data) 

85 assert result == {"a": 1, "b": 2} 

86 

87 def test_loads_pickle(self): 

88 s = Serializer() 

89 data = pickle.dumps({"z": 99}) 

90 result = s.loads(data) 

91 assert result == {"z": 99} 

92 

93 def test_loads_explicit_format(self): 

94 s = Serializer() 

95 data = json.dumps([1, 2, 3]).encode("utf-8") 

96 result = s.loads(data, fmt=SerialFormat.JSON) 

97 assert result == [1, 2, 3] 

98 

99 def test_roundtrip_json(self): 

100 s = Serializer(fmt=SerialFormat.JSON) 

101 obj = {"nested": {"key": [1, 2, 3]}} 

102 assert s.loads(s.dumps(obj)) == obj 

103 

104 def test_roundtrip_pickle(self): 

105 s = Serializer(fmt=SerialFormat.PICKLE) 

106 obj = {"tuple": (1, 2), "set": {3, 4}} 

107 assert s.loads(s.dumps(obj)) == obj 

108 

109 def test_stats(self): 

110 s = Serializer() 

111 s.dumps({"a": 1}) 

112 s.loads(json.dumps({"b": 2}).encode()) 

113 st = s.stats 

114 assert st["total_serialized"] == 1 

115 assert st["total_deserialized"] == 1 

116 assert st["format"] == "json" 

117 

118 def test_unsupported_format(self): 

119 s = Serializer(fmt=SerialFormat.PICKLE) 

120 # Create a serializer with a bad internal state and test dumps 

121 with pytest.raises(ValueError, match="Unsupported"): 

122 s.dumps({}, use_msgpack=False) 

123 s._fmt = "bad_fmt" 

124 s.dumps({}) 

125 

126 def test_use_msgpack_flag(self): 

127 pytest.importorskip("msgpack") 

128 s = Serializer(fmt=SerialFormat.JSON) 

129 data = s.dumps({"a": 1}, use_msgpack=True) 

130 import msgpack 

131 assert msgpack.unpackb(data) == {"a": 1} 

132 

133 def test_loads_msgpack(self): 

134 pytest.importorskip("msgpack") 

135 import msgpack 

136 s = Serializer() 

137 data = msgpack.packb({"m": "p"}) 

138 result = s.loads(data) 

139 assert result == {"m": "p"} 

140 

141 def test_stats_pickle_format(self): 

142 s = Serializer(fmt=SerialFormat.PICKLE) 

143 s.dumps({"x": 1}) 

144 assert s.stats["format"] == "pickle" 

145 

146 

147# ============================================================================ 

148# EvictionPolicy 

149# ============================================================================ 

150 

151class TestEvictionPolicy: 

152 def test_enum_values(self): 

153 assert EvictionPolicy.LRU.value == "lru" 

154 assert EvictionPolicy.LFU.value == "lfu" 

155 assert EvictionPolicy.TTL_ONLY.value == "ttl_only" 

156 

157 

158# ============================================================================ 

159# TTLCache — Basic Operations 

160# ============================================================================ 

161 

162class TestTTLCacheBasic: 

163 def test_defaults(self): 

164 cache = TTLCache[int]() 

165 assert cache._max_size == 1000 

166 assert cache._ttl == 300.0 

167 assert cache._policy == EvictionPolicy.LRU 

168 assert cache.size == 0 

169 

170 def test_custom_params(self): 

171 cache = TTLCache[str](max_size=10, ttl=60.0, policy=EvictionPolicy.LFU) 

172 assert cache._max_size == 10 

173 assert cache._ttl == 60.0 

174 assert cache._policy == EvictionPolicy.LFU 

175 

176 def test_set_get(self): 

177 cache = TTLCache[int]() 

178 cache.set("a", 1) 

179 assert cache.get("a") == 1 

180 

181 def test_get_missing(self): 

182 cache = TTLCache[str]() 

183 assert cache.get("no-key") is None 

184 

185 def test_set_overwrite(self): 

186 cache = TTLCache[int]() 

187 cache.set("a", 1) 

188 cache.set("a", 99) 

189 assert cache.get("a") == 99 

190 assert cache.size == 1 

191 

192 def test_delete(self): 

193 cache = TTLCache[int]() 

194 cache.set("x", 42) 

195 assert cache.delete("x") is True 

196 assert cache.get("x") is None 

197 assert cache.size == 0 

198 

199 def test_delete_missing(self): 

200 cache = TTLCache[int]() 

201 assert cache.delete("nope") is False 

202 

203 def test_clear(self): 

204 cache = TTLCache[int]() 

205 cache.set("a", 1) 

206 cache.set("b", 2) 

207 cache.clear() 

208 assert cache.size == 0 

209 assert cache.get("a") is None 

210 

211 def test_size(self): 

212 cache = TTLCache[int]() 

213 assert cache.size == 0 

214 cache.set("a", 1) 

215 assert cache.size == 1 

216 cache.set("b", 2) 

217 assert cache.size == 2 

218 

219 

220# ============================================================================ 

221# TTLCache — TTL Expiration 

222# ============================================================================ 

223 

224class TestTTLCacheTTL: 

225 def test_expired_entry(self): 

226 cache = TTLCache[int](ttl=0.05) 

227 cache.set("a", 1) 

228 assert cache.get("a") == 1 

229 time.sleep(0.1) 

230 assert cache.get("a") is None 

231 

232 def test_custom_ttl_per_entry(self): 

233 cache = TTLCache[int](ttl=10.0) 

234 cache.set("a", 1, ttl=0.05) 

235 assert cache.get("a") == 1 

236 time.sleep(0.1) 

237 assert cache.get("a") is None 

238 

239 def test_cleanup(self): 

240 cache = TTLCache[int](ttl=0.05) 

241 cache.set("a", 1) 

242 cache.set("b", 2) 

243 time.sleep(0.1) 

244 removed = cache.cleanup() 

245 assert removed == 2 

246 assert cache.size == 0 

247 

248 def test_cleanup_partial(self): 

249 cache = TTLCache[int](ttl=0.05) 

250 cache.set("a", 1) 

251 time.sleep(0.1) 

252 cache.set("b", 2, ttl=10.0) 

253 removed = cache.cleanup() 

254 assert removed == 1 

255 assert cache.size == 1 

256 assert cache.get("b") == 2 

257 

258 

259# ============================================================================ 

260# TTLCache — Eviction 

261# ============================================================================ 

262 

263class TestTTLCacheEviction: 

264 def test_lru_eviction(self): 

265 cache = TTLCache[int](max_size=2, ttl=300, policy=EvictionPolicy.LRU) 

266 cache.set("a", 1) 

267 cache.set("b", 2) 

268 cache.get("a") # access "a" — now "b" is least recently used 

269 cache.set("c", 3) 

270 assert cache.get("b") is None 

271 assert cache.get("a") == 1 

272 assert cache.get("c") == 3 

273 

274 def test_lfu_eviction(self): 

275 cache = TTLCache[int](max_size=2, ttl=300, policy=EvictionPolicy.LFU) 

276 cache.set("a", 1) 

277 cache.set("b", 2) 

278 cache.get("b") # b access_count = 1 

279 cache.get("b") # b access_count = 2 

280 cache.get("a") # a access_count = 1 

281 cache.set("c", 3) 

282 # a has lowest access_count (1), should be evicted 

283 assert cache.get("a") is None 

284 assert cache.get("b") == 2 

285 assert cache.get("c") == 3 

286 

287 def test_ttl_only_eviction(self): 

288 cache = TTLCache[int](max_size=2, ttl=300, policy=EvictionPolicy.TTL_ONLY) 

289 cache.set("a", 1) 

290 cache.set("b", 2) 

291 cache.get("b") # access b 

292 cache.set("c", 3) 

293 # TTL_ONLY evicts oldest (first inserted) regardless of access 

294 assert cache.get("a") is None 

295 assert cache.get("b") == 2 

296 

297 

298# ============================================================================ 

299# TTLCache — Stats 

300# ============================================================================ 

301 

302class TestTTLCacheStats: 

303 def test_default_stats(self): 

304 cache = TTLCache[int](max_size=50, ttl=10, policy=EvictionPolicy.LFU) 

305 s = cache.stats 

306 assert s["size"] == 0 

307 assert s["max_size"] == 50 

308 assert s["ttl"] == 10 

309 assert s["policy"] == "lfu" 

310 assert s["hits"] == 0 

311 assert s["misses"] == 0 

312 assert s["evictions"] == 0 

313 

314 def test_hits_misses(self): 

315 cache = TTLCache[int]() 

316 cache.set("a", 1) 

317 cache.get("a") # hit 

318 cache.get("b") # miss 

319 s = cache.stats 

320 assert s["hits"] == 1 

321 assert s["misses"] == 1 

322 

323 def test_hit_rate(self): 

324 cache = TTLCache[int]() 

325 cache.set("a", 1) 

326 cache.get("a") # hit 

327 cache.get("a") # hit 

328 cache.get("b") # miss 

329 s = cache.stats 

330 assert s["hit_rate"] == round(2 / 3, 3) 

331 

332 def test_hit_rate_no_ops(self): 

333 cache = TTLCache[int]() 

334 s = cache.stats 

335 assert s["hit_rate"] == 0.0 

336 

337 def test_evictions_count(self): 

338 cache = TTLCache[int](max_size=2, ttl=300, policy=EvictionPolicy.LRU) 

339 cache.set("a", 1) 

340 cache.set("b", 2) 

341 cache.set("c", 3) # evicts "a" 

342 assert cache.stats["evictions"] == 1 

343 

344 

345# ============================================================================ 

346# TTLCache — Thread Safety 

347# ============================================================================ 

348 

349class TestTTLCacheThreadSafety: 

350 def test_concurrent_access(self): 

351 cache = TTLCache[int]() 

352 errors = [] 

353 

354 def worker(start_idx): 

355 try: 

356 for i in range(start_idx, start_idx + 50): 

357 cache.set(f"k{i}", i) 

358 cache.get(f"k{i}") 

359 cache.delete(f"k{i}") 

360 except Exception as e: 

361 errors.append(e) 

362 

363 threads = [threading.Thread(target=worker, args=(i * 100,)) for i in range(4)] 

364 for t in threads: 

365 t.start() 

366 for t in threads: 

367 t.join() 

368 

369 assert len(errors) == 0 

370 

371 

372# ============================================================================ 

373# SmartCache 

374# ============================================================================ 

375 

376class TestSmartCache: 

377 def test_get_missing(self): 

378 sc = SmartCache[int]() 

379 assert sc.get("x") is None 

380 

381 def test_get_or_compute(self): 

382 sc = SmartCache[int]() 

383 result = sc.get_or_compute("key", lambda: 42) 

384 assert result == 42 

385 # Second call should use cache 

386 result = sc.get_or_compute("key", lambda: 99) 

387 assert result == 42 

388 

389 def test_get_or_compute_complex(self): 

390 sc = SmartCache[str](ttl=10.0) 

391 calls = [] 

392 

393 def factory(): 

394 calls.append(1) 

395 return "computed" 

396 

397 r1 = sc.get_or_compute("k", factory) 

398 assert r1 == "computed" 

399 assert len(calls) == 1 

400 r2 = sc.get_or_compute("k", factory) 

401 assert r2 == "computed" 

402 assert len(calls) == 1 

403 

404 def test_get_or_compute_expired(self): 

405 sc = SmartCache[int](ttl=0.05) 

406 sc.get_or_compute("k", lambda: 42) 

407 time.sleep(0.1) 

408 result = sc.get_or_compute("k", lambda: 99) 

409 assert result == 99 # recomputed after expiry 

410 

411 def test_get_or_compute_custom_ttl(self): 

412 sc = SmartCache[int]() 

413 r1 = sc.get_or_compute("k", lambda: 42, ttl=0.05) 

414 assert r1 == 42 

415 time.sleep(0.1) 

416 r2 = sc.get_or_compute("k", lambda: 99) 

417 assert r2 == 99 

418 

419 def test_set_and_get(self): 

420 sc = SmartCache[int]() 

421 sc.set("x", 100) 

422 assert sc.get("x") == 100 

423 

424 def test_delete(self): 

425 sc = SmartCache[int]() 

426 sc.set("x", 1) 

427 assert sc.delete("x") is True 

428 assert sc.get("x") is None 

429 

430 def test_clear(self): 

431 sc = SmartCache[int]() 

432 sc.set("a", 1) 

433 sc.set("b", 2) 

434 sc.clear() 

435 assert sc.size == 0 

436 

437 def test_size(self): 

438 sc = SmartCache[int]() 

439 assert sc.size == 0 

440 sc.set("a", 1) 

441 assert sc.size == 1 

442 sc.set("b", 2) 

443 assert sc.size == 2 

444 

445 def test_stats(self): 

446 sc = SmartCache[int]() 

447 sc.get_or_compute("k", lambda: 42) 

448 s = sc.stats 

449 assert s["size"] == 1 

450 assert s["hits"] >= 0 

451 assert s["max_size"] == 1000 

452 

453 def test_dump_load(self): 

454 sc = SmartCache[str]() 

455 sc.set("a", "hello") 

456 sc.set("b", "world") 

457 

458 data = sc.dump() 

459 assert len(data) > 0 

460 

461 sc2 = SmartCache[str]() 

462 loaded = sc2.load(data) 

463 assert loaded == 2 

464 assert sc2.get("a") == "hello" 

465 assert sc2.get("b") == "world" 

466 

467 def test_dump_load_expired_filters(self): 

468 sc = SmartCache[int]() 

469 sc.set("live", 42, ttl=10.0) 

470 sc.get_or_compute("dead", lambda: 1, ttl=0.01) 

471 time.sleep(0.05) 

472 

473 data = sc.dump() 

474 sc2 = SmartCache[int]() 

475 loaded = sc2.load(data) 

476 assert loaded == 1 

477 assert sc2.get("live") == 42 

478 assert sc2.get("dead") is None 

479 

480 def test_custom_params_passthrough(self): 

481 sc = SmartCache[int](max_size=5, ttl=30.0, policy=EvictionPolicy.LFU) 

482 assert sc.size == 0 

483 assert sc._cache._max_size == 5