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