121 lines
3.7 KiB
Python
121 lines
3.7 KiB
Python
"""Unit tests for cache manager."""
|
|
|
|
import pytest
|
|
import time
|
|
from src.data.cache_manager import CacheManager, CacheEntry
|
|
|
|
|
|
class TestCacheEntry:
|
|
"""Tests for CacheEntry."""
|
|
|
|
def test_init(self):
|
|
"""Test cache entry initialization."""
|
|
entry = CacheEntry("test_data", 60.0)
|
|
assert entry.data == "test_data"
|
|
assert entry.expires_at > time.time()
|
|
assert entry.access_count == 0
|
|
|
|
def test_is_expired(self):
|
|
"""Test expiration checking."""
|
|
entry = CacheEntry("test_data", 0.01) # Very short TTL
|
|
assert not entry.is_expired()
|
|
time.sleep(0.02)
|
|
assert entry.is_expired()
|
|
|
|
def test_touch(self):
|
|
"""Test access tracking."""
|
|
entry = CacheEntry("test_data", 60.0)
|
|
initial_count = entry.access_count
|
|
entry.touch()
|
|
assert entry.access_count == initial_count + 1
|
|
|
|
|
|
class TestCacheManager:
|
|
"""Tests for CacheManager."""
|
|
|
|
@pytest.fixture
|
|
def cache(self):
|
|
"""Create a cache manager instance."""
|
|
return CacheManager(default_ttl=1.0, max_size=10)
|
|
|
|
def test_get_set(self, cache):
|
|
"""Test basic get and set operations."""
|
|
cache.set("key1", "value1")
|
|
assert cache.get("key1") == "value1"
|
|
|
|
def test_get_missing(self, cache):
|
|
"""Test getting non-existent key."""
|
|
assert cache.get("missing") is None
|
|
|
|
def test_expiration(self, cache):
|
|
"""Test cache entry expiration."""
|
|
cache.set("key1", "value1", ttl=0.1)
|
|
assert cache.get("key1") == "value1"
|
|
time.sleep(0.2)
|
|
assert cache.get("key1") is None
|
|
|
|
def test_lru_eviction(self, cache):
|
|
"""Test LRU eviction when max size reached."""
|
|
# Fill cache to max size
|
|
for i in range(10):
|
|
cache.set(f"key{i}", f"value{i}")
|
|
|
|
# Add one more - should evict oldest
|
|
cache.set("key10", "value10")
|
|
|
|
# Oldest key should be evicted
|
|
assert cache.get("key0") is None
|
|
assert cache.get("key10") == "value10"
|
|
|
|
def test_type_specific_ttl(self, cache):
|
|
"""Test type-specific TTL."""
|
|
cache.set("ticker1", {"price": 100}, cache_type='ticker')
|
|
cache.set("ohlcv1", [[1, 2, 3, 4, 5, 6]], cache_type='ohlcv')
|
|
|
|
# Both should be cached
|
|
assert cache.get("ticker1") is not None
|
|
assert cache.get("ohlcv1") is not None
|
|
|
|
def test_delete(self, cache):
|
|
"""Test cache entry deletion."""
|
|
cache.set("key1", "value1")
|
|
assert cache.get("key1") == "value1"
|
|
|
|
cache.delete("key1")
|
|
assert cache.get("key1") is None
|
|
|
|
def test_clear(self, cache):
|
|
"""Test cache clearing."""
|
|
cache.set("key1", "value1")
|
|
cache.set("key2", "value2")
|
|
|
|
cache.clear()
|
|
|
|
assert cache.get("key1") is None
|
|
assert cache.get("key2") is None
|
|
|
|
def test_stats(self, cache):
|
|
"""Test cache statistics."""
|
|
cache.set("key1", "value1")
|
|
cache.get("key1") # Hit
|
|
cache.get("missing") # Miss
|
|
|
|
stats = cache.get_stats()
|
|
|
|
assert stats['hits'] >= 1
|
|
assert stats['misses'] >= 1
|
|
assert stats['size'] == 1
|
|
assert 'hit_rate' in stats
|
|
|
|
def test_invalidate_pattern(self, cache):
|
|
"""Test pattern-based invalidation."""
|
|
cache.set("ticker:BTC/USD", "value1")
|
|
cache.set("ticker:ETH/USD", "value2")
|
|
cache.set("ohlcv:BTC/USD", "value3")
|
|
|
|
cache.invalidate_pattern("ticker:")
|
|
|
|
assert cache.get("ticker:BTC/USD") is None
|
|
assert cache.get("ticker:ETH/USD") is None
|
|
assert cache.get("ohlcv:BTC/USD") is not None
|