PlanOpticon

planopticon / tests / test_api_cache.py

Source Blame History 80 lines

ccf32cc…	leo	1	"""Tests for API response cache."""
ccf32cc…	leo	2
ccf32cc…	leo	3	import time
ccf32cc…	leo	4
ccf32cc…	leo	5	from video_processor.utils.api_cache import ApiCache
ccf32cc…	leo	6
ccf32cc…	leo	7
ccf32cc…	leo	8	class TestApiCache:
ccf32cc…	leo	9	def test_set_and_get(self, tmp_path):
ccf32cc…	leo	10	cache = ApiCache(tmp_path, namespace="test")
ccf32cc…	leo	11	cache.set("key1", {"data": "value"})
ccf32cc…	leo	12	result = cache.get("key1")
ccf32cc…	leo	13	assert result == {"data": "value"}
ccf32cc…	leo	14
ccf32cc…	leo	15	def test_get_missing_key(self, tmp_path):
ccf32cc…	leo	16	cache = ApiCache(tmp_path, namespace="test")
ccf32cc…	leo	17	assert cache.get("nonexistent") is None
ccf32cc…	leo	18
ccf32cc…	leo	19	def test_ttl_expiry(self, tmp_path):
ccf32cc…	leo	20	cache = ApiCache(tmp_path, namespace="test", ttl=0)
ccf32cc…	leo	21	cache.set("key1", "value")
ccf32cc…	leo	22	# With TTL=0, any subsequent access should be expired
ccf32cc…	leo	23	time.sleep(0.01)
ccf32cc…	leo	24	assert cache.get("key1") is None
ccf32cc…	leo	25
ccf32cc…	leo	26	def test_invalidate(self, tmp_path):
ccf32cc…	leo	27	cache = ApiCache(tmp_path, namespace="test")
ccf32cc…	leo	28	cache.set("key1", "value")
ccf32cc…	leo	29	assert cache.get("key1") == "value"
ccf32cc…	leo	30	result = cache.invalidate("key1")
ccf32cc…	leo	31	assert result is True
ccf32cc…	leo	32	assert cache.get("key1") is None
ccf32cc…	leo	33
ccf32cc…	leo	34	def test_invalidate_missing(self, tmp_path):
ccf32cc…	leo	35	cache = ApiCache(tmp_path, namespace="test")
ccf32cc…	leo	36	result = cache.invalidate("nonexistent")
ccf32cc…	leo	37	assert result is False
ccf32cc…	leo	38
ccf32cc…	leo	39	def test_clear_all(self, tmp_path):
ccf32cc…	leo	40	cache = ApiCache(tmp_path, namespace="test")
ccf32cc…	leo	41	cache.set("a", 1)
ccf32cc…	leo	42	cache.set("b", 2)
ccf32cc…	leo	43	cache.set("c", 3)
ccf32cc…	leo	44	count = cache.clear()
ccf32cc…	leo	45	assert count == 3
ccf32cc…	leo	46	assert cache.get("a") is None
ccf32cc…	leo	47
ccf32cc…	leo	48	def test_clear_older_than(self, tmp_path):
ccf32cc…	leo	49	cache = ApiCache(tmp_path, namespace="test")
ccf32cc…	leo	50	cache.set("old", "value")
ccf32cc…	leo	51	# Setting older_than to a very large number means nothing gets cleared
ccf32cc…	leo	52	count = cache.clear(older_than=99999)
ccf32cc…	leo	53	assert count == 0
ccf32cc…	leo	54
ccf32cc…	leo	55	def test_get_stats(self, tmp_path):
ccf32cc…	leo	56	cache = ApiCache(tmp_path, namespace="test")
ccf32cc…	leo	57	cache.set("x", {"hello": "world"})
ccf32cc…	leo	58	cache.set("y", [1, 2, 3])
ccf32cc…	leo	59	stats = cache.get_stats()
ccf32cc…	leo	60	assert stats["namespace"] == "test"
ccf32cc…	leo	61	assert stats["entry_count"] == 2
ccf32cc…	leo	62	assert stats["total_size_bytes"] > 0
ccf32cc…	leo	63
ccf32cc…	leo	64	def test_namespace_isolation(self, tmp_path):
ccf32cc…	leo	65	cache_a = ApiCache(tmp_path, namespace="ns_a")
ccf32cc…	leo	66	cache_b = ApiCache(tmp_path, namespace="ns_b")
ccf32cc…	leo	67	cache_a.set("key", "value_a")
ccf32cc…	leo	68	cache_b.set("key", "value_b")
ccf32cc…	leo	69	assert cache_a.get("key") == "value_a"
ccf32cc…	leo	70	assert cache_b.get("key") == "value_b"
ccf32cc…	leo	71
ccf32cc…	leo	72	def test_creates_namespace_dir(self, tmp_path):
829e24a…	leo	73	ApiCache(tmp_path / "sub", namespace="deep")
ccf32cc…	leo	74	assert (tmp_path / "sub" / "deep").exists()
ccf32cc…	leo	75
ccf32cc…	leo	76	def test_cache_path_uses_hash(self, tmp_path):
ccf32cc…	leo	77	cache = ApiCache(tmp_path, namespace="test")
ccf32cc…	leo	78	path = cache.get_cache_path("my_key")
ccf32cc…	leo	79	assert path.suffix == ".json"
ccf32cc…	leo	80	assert path.parent.name == "test"

PlanOpticon

Keyboard Shortcuts