Spaces:
Running
Running
File size: 3,926 Bytes
2ed2bd7 |
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 |
"""
Tests for the cache service.
"""
import time
import pytest
from app.core.cache import SimpleCache
def test_cache_initialization():
"""Test cache is initialized with correct settings."""
cache = SimpleCache(ttl_seconds=3600, max_size=100)
assert cache._ttl == 3600
assert cache._max_size == 100
stats = cache.stats()
assert stats["size"] == 0
assert stats["hits"] == 0
assert stats["misses"] == 0
def test_cache_set_and_get():
"""Test setting and getting cache entries."""
cache = SimpleCache(ttl_seconds=60)
test_data = {"text": "Test article", "title": "Test"}
cache.set("http://example.com", test_data)
result = cache.get("http://example.com")
assert result is not None
assert result["text"] == "Test article"
assert result["title"] == "Test"
def test_cache_miss():
"""Test cache miss returns None."""
cache = SimpleCache()
result = cache.get("http://nonexistent.com")
assert result is None
def test_cache_expiration():
"""Test cache entries expire after TTL."""
cache = SimpleCache(ttl_seconds=1) # 1 second TTL
test_data = {"text": "Test article"}
cache.set("http://example.com", test_data)
# Should be in cache immediately
assert cache.get("http://example.com") is not None
# Wait for expiration
time.sleep(1.5)
# Should be expired now
assert cache.get("http://example.com") is None
def test_cache_max_size():
"""Test cache enforces max size by removing oldest entries."""
cache = SimpleCache(ttl_seconds=3600, max_size=3)
cache.set("url1", {"data": "1"})
cache.set("url2", {"data": "2"})
cache.set("url3", {"data": "3"})
assert cache.stats()["size"] == 3
# Adding a 4th entry should remove the oldest
cache.set("url4", {"data": "4"})
assert cache.stats()["size"] == 3
assert cache.get("url1") is None # Oldest should be removed
assert cache.get("url4") is not None
def test_cache_stats():
"""Test cache statistics tracking."""
cache = SimpleCache()
cache.set("url1", {"data": "1"})
cache.set("url2", {"data": "2"})
# Generate some hits and misses
cache.get("url1") # hit
cache.get("url1") # hit
cache.get("url3") # miss
stats = cache.stats()
assert stats["size"] == 2
assert stats["hits"] == 2
assert stats["misses"] == 1
assert stats["hit_rate"] == 66.67
def test_cache_clear_expired():
"""Test clearing expired entries."""
cache = SimpleCache(ttl_seconds=1)
cache.set("url1", {"data": "1"})
cache.set("url2", {"data": "2"})
# Wait for expiration
time.sleep(1.5)
# Add a fresh entry
cache.set("url3", {"data": "3"})
# Clear expired entries
removed = cache.clear_expired()
assert removed == 2
assert cache.stats()["size"] == 1
assert cache.get("url3") is not None
def test_cache_clear_all():
"""Test clearing all cache entries."""
cache = SimpleCache()
cache.set("url1", {"data": "1"})
cache.set("url2", {"data": "2"})
cache.get("url1") # Generate some stats
cache.clear_all()
stats = cache.stats()
assert stats["size"] == 0
assert stats["hits"] == 0
assert stats["misses"] == 0
def test_cache_thread_safety():
"""Test cache thread safety with concurrent access."""
import threading
cache = SimpleCache()
def set_values():
for i in range(10):
cache.set(f"url{i}", {"data": str(i)})
def get_values():
for i in range(10):
cache.get(f"url{i}")
threads = []
for _ in range(5):
threads.append(threading.Thread(target=set_values))
threads.append(threading.Thread(target=get_values))
for t in threads:
t.start()
for t in threads:
t.join()
# No assertion needed - test passes if no race condition errors occur
assert cache.stats()["size"] <= 10
|