mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-05-15 18:03:32 +00:00
55161f309b
Remove unused imports across 86 files in src/, web_interface/, test/, and scripts/ using autoflake. No logic changes — only dead import statements and unused names in from-imports are removed. Also remove bare exception aliases where the variable is never referenced in the handler body: - src/cache/disk_cache.py: except (IOError, OSError, PermissionError) as e - src/cache_manager.py: except (OSError, IOError, PermissionError) as perm_error - src/plugin_system/resource_monitor.py: except Exception as e - web_interface/app.py: except Exception as read_err 86 files changed, 205 lines removed, 18 pre-existing test failures unchanged. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
390 lines
13 KiB
Python
390 lines
13 KiB
Python
"""
|
|
Tests for CacheManager and cache components.
|
|
|
|
Tests cache functionality including memory cache, disk cache, strategy, and metrics.
|
|
"""
|
|
|
|
import pytest
|
|
import time
|
|
from unittest.mock import patch
|
|
from src.cache_manager import CacheManager
|
|
from src.cache.memory_cache import MemoryCache
|
|
from src.cache.disk_cache import DiskCache
|
|
from src.cache.cache_strategy import CacheStrategy
|
|
from src.cache.cache_metrics import CacheMetrics
|
|
from datetime import datetime
|
|
|
|
|
|
class TestCacheManager:
|
|
"""Test CacheManager functionality."""
|
|
|
|
def test_init(self, tmp_path):
|
|
"""Test CacheManager initialization."""
|
|
with patch('src.cache_manager.CacheManager._get_writable_cache_dir', return_value=str(tmp_path)):
|
|
cm = CacheManager()
|
|
assert cm.cache_dir == str(tmp_path)
|
|
assert hasattr(cm, '_memory_cache_component')
|
|
assert hasattr(cm, '_disk_cache_component')
|
|
assert hasattr(cm, '_strategy_component')
|
|
assert hasattr(cm, '_metrics_component')
|
|
|
|
def test_set_and_get(self, tmp_path):
|
|
"""Test basic set and get operations."""
|
|
with patch('src.cache_manager.CacheManager._get_writable_cache_dir', return_value=str(tmp_path)):
|
|
cm = CacheManager()
|
|
test_data = {"key": "value", "number": 42}
|
|
|
|
cm.set("test_key", test_data)
|
|
result = cm.get("test_key")
|
|
|
|
assert result == test_data
|
|
|
|
def test_get_expired(self, tmp_path):
|
|
"""Test getting expired cache entry."""
|
|
with patch('src.cache_manager.CacheManager._get_writable_cache_dir', return_value=str(tmp_path)):
|
|
cm = CacheManager()
|
|
cm.set("test_key", {"data": "value"})
|
|
|
|
# Get with max_age=0 to force expiration
|
|
result = cm.get("test_key", max_age=0)
|
|
assert result is None
|
|
|
|
|
|
class TestCacheStrategy:
|
|
"""Test CacheStrategy functionality."""
|
|
|
|
def test_get_cache_strategy_default(self):
|
|
"""Test getting default cache strategy."""
|
|
strategy = CacheStrategy()
|
|
result = strategy.get_cache_strategy("unknown_type")
|
|
|
|
assert "max_age" in result
|
|
assert "memory_ttl" in result
|
|
assert result["max_age"] == 300 # Default
|
|
|
|
def test_get_cache_strategy_live(self):
|
|
"""Test getting live sports cache strategy."""
|
|
strategy = CacheStrategy()
|
|
result = strategy.get_cache_strategy("sports_live")
|
|
|
|
assert "max_age" in result
|
|
assert result["max_age"] <= 60 # Live data should be short
|
|
|
|
def test_get_data_type_from_key(self):
|
|
"""Test data type detection from cache key."""
|
|
strategy = CacheStrategy()
|
|
|
|
assert strategy.get_data_type_from_key("nba_live_scores") == "sports_live"
|
|
# "weather_current" contains "current" which matches live sports pattern first
|
|
# Use "weather" without "current" to test weather detection
|
|
assert strategy.get_data_type_from_key("weather") == "weather_current"
|
|
assert strategy.get_data_type_from_key("weather_data") == "weather_current"
|
|
assert strategy.get_data_type_from_key("unknown_key") == "default"
|
|
|
|
|
|
class TestMemoryCache:
|
|
"""Test MemoryCache functionality."""
|
|
|
|
def test_init(self):
|
|
"""Test MemoryCache initialization."""
|
|
cache = MemoryCache(max_size=100, cleanup_interval=60.0)
|
|
|
|
assert cache._max_size == 100
|
|
assert cache._cleanup_interval == 60.0
|
|
assert cache.size() == 0
|
|
|
|
def test_set_and_get(self):
|
|
"""Test basic set and get operations."""
|
|
cache = MemoryCache()
|
|
test_data = {"key": "value", "number": 42}
|
|
|
|
cache.set("test_key", test_data)
|
|
result = cache.get("test_key")
|
|
|
|
assert result == test_data
|
|
|
|
def test_get_expired(self):
|
|
"""Test getting expired cache entry."""
|
|
cache = MemoryCache()
|
|
cache.set("test_key", {"data": "value"})
|
|
|
|
# Get with max_age=0 to force expiration
|
|
result = cache.get("test_key", max_age=0)
|
|
assert result is None
|
|
|
|
def test_get_nonexistent(self):
|
|
"""Test getting non-existent key."""
|
|
cache = MemoryCache()
|
|
result = cache.get("nonexistent_key")
|
|
assert result is None
|
|
|
|
def test_clear_specific_key(self):
|
|
"""Test clearing a specific cache key."""
|
|
cache = MemoryCache()
|
|
cache.set("key1", {"data": "value1"})
|
|
cache.set("key2", {"data": "value2"})
|
|
|
|
cache.clear("key1")
|
|
|
|
assert cache.get("key1") is None
|
|
assert cache.get("key2") is not None
|
|
|
|
def test_clear_all(self):
|
|
"""Test clearing all cache entries."""
|
|
cache = MemoryCache()
|
|
cache.set("key1", {"data": "value1"})
|
|
cache.set("key2", {"data": "value2"})
|
|
|
|
cache.clear()
|
|
|
|
assert cache.size() == 0
|
|
assert cache.get("key1") is None
|
|
assert cache.get("key2") is None
|
|
|
|
def test_cleanup_expired(self):
|
|
"""Test cleanup removes expired entries."""
|
|
cache = MemoryCache()
|
|
cache.set("key1", {"data": "value1"})
|
|
# Force expiration by manipulating timestamp (older than 1 hour cleanup threshold)
|
|
# Cleanup uses max_age_for_cleanup = 3600 (1 hour)
|
|
cache._timestamps["key1"] = time.time() - 4000 # More than 1 hour
|
|
|
|
removed = cache.cleanup(force=True)
|
|
|
|
# Cleanup should remove expired entries (older than 3600 seconds)
|
|
# The key should be gone after cleanup
|
|
assert cache.get("key1") is None or removed >= 0
|
|
|
|
def test_cleanup_size_limit(self):
|
|
"""Test cleanup enforces size limits."""
|
|
cache = MemoryCache(max_size=3)
|
|
# Add more entries than max_size
|
|
for i in range(5):
|
|
cache.set(f"key{i}", {"data": f"value{i}"})
|
|
|
|
removed = cache.cleanup(force=True)
|
|
|
|
assert cache.size() <= cache._max_size
|
|
assert removed >= 0
|
|
|
|
def test_size(self):
|
|
"""Test size reporting."""
|
|
cache = MemoryCache()
|
|
assert cache.size() == 0
|
|
|
|
cache.set("key1", {"data": "value1"})
|
|
cache.set("key2", {"data": "value2"})
|
|
|
|
assert cache.size() == 2
|
|
|
|
def test_max_size(self):
|
|
"""Test max_size property."""
|
|
cache = MemoryCache(max_size=500)
|
|
assert cache.max_size() == 500
|
|
|
|
def test_get_stats(self):
|
|
"""Test getting cache statistics."""
|
|
cache = MemoryCache()
|
|
cache.set("key1", {"data": "value1"})
|
|
cache.set("key2", {"data": "value2"})
|
|
|
|
stats = cache.get_stats()
|
|
|
|
assert "size" in stats
|
|
assert "max_size" in stats
|
|
assert stats["size"] == 2
|
|
assert stats["max_size"] == 1000 # default
|
|
|
|
|
|
class TestCacheMetrics:
|
|
"""Test CacheMetrics functionality."""
|
|
|
|
def test_record_hit(self):
|
|
"""Test recording cache hit."""
|
|
metrics = CacheMetrics()
|
|
metrics.record_hit()
|
|
stats = metrics.get_metrics()
|
|
|
|
# get_metrics() returns calculated values, not raw hits/misses
|
|
assert stats['total_requests'] == 1
|
|
assert stats['cache_hit_rate'] == 1.0 # 1 hit out of 1 request
|
|
|
|
def test_record_miss(self):
|
|
"""Test recording cache miss."""
|
|
metrics = CacheMetrics()
|
|
metrics.record_miss()
|
|
stats = metrics.get_metrics()
|
|
|
|
# get_metrics() returns calculated values, not raw hits/misses
|
|
assert stats['total_requests'] == 1
|
|
assert stats['cache_hit_rate'] == 0.0 # 0 hits out of 1 request
|
|
|
|
def test_record_fetch_time(self):
|
|
"""Test recording fetch time."""
|
|
metrics = CacheMetrics()
|
|
metrics.record_fetch_time(0.5)
|
|
stats = metrics.get_metrics()
|
|
|
|
assert stats['fetch_count'] == 1
|
|
assert stats['total_fetch_time'] == 0.5
|
|
assert stats['average_fetch_time'] == 0.5
|
|
|
|
def test_cache_hit_rate(self):
|
|
"""Test cache hit rate calculation."""
|
|
metrics = CacheMetrics()
|
|
metrics.record_hit()
|
|
metrics.record_hit()
|
|
metrics.record_miss()
|
|
|
|
stats = metrics.get_metrics()
|
|
assert stats['cache_hit_rate'] == pytest.approx(0.666, abs=0.01)
|
|
|
|
|
|
class TestDiskCache:
|
|
"""Test DiskCache functionality."""
|
|
|
|
def test_init_with_dir(self, tmp_path):
|
|
"""Test DiskCache initialization with directory."""
|
|
cache = DiskCache(cache_dir=str(tmp_path))
|
|
assert cache.cache_dir == str(tmp_path)
|
|
|
|
def test_init_without_dir(self):
|
|
"""Test DiskCache initialization without directory."""
|
|
cache = DiskCache(cache_dir=None)
|
|
assert cache.cache_dir is None
|
|
|
|
def test_get_cache_path(self, tmp_path):
|
|
"""Test getting cache file path."""
|
|
cache = DiskCache(cache_dir=str(tmp_path))
|
|
path = cache.get_cache_path("test_key")
|
|
assert path == str(tmp_path / "test_key.json")
|
|
|
|
def test_get_cache_path_disabled(self):
|
|
"""Test getting cache path when disabled."""
|
|
cache = DiskCache(cache_dir=None)
|
|
path = cache.get_cache_path("test_key")
|
|
assert path is None
|
|
|
|
def test_set_and_get(self, tmp_path):
|
|
"""Test basic set and get operations."""
|
|
cache = DiskCache(cache_dir=str(tmp_path))
|
|
test_data = {"key": "value", "number": 42}
|
|
|
|
cache.set("test_key", test_data)
|
|
result = cache.get("test_key")
|
|
|
|
assert result == test_data
|
|
|
|
def test_get_expired(self, tmp_path):
|
|
"""Test getting expired cache entry."""
|
|
cache = DiskCache(cache_dir=str(tmp_path))
|
|
cache.set("test_key", {"data": "value"})
|
|
|
|
# Get with max_age=0 to force expiration
|
|
result = cache.get("test_key", max_age=0)
|
|
assert result is None
|
|
|
|
def test_get_nonexistent(self, tmp_path):
|
|
"""Test getting non-existent key."""
|
|
cache = DiskCache(cache_dir=str(tmp_path))
|
|
result = cache.get("nonexistent_key")
|
|
assert result is None
|
|
|
|
def test_clear_specific_key(self, tmp_path):
|
|
"""Test clearing a specific cache key."""
|
|
cache = DiskCache(cache_dir=str(tmp_path))
|
|
cache.set("key1", {"data": "value1"})
|
|
cache.set("key2", {"data": "value2"})
|
|
|
|
cache.clear("key1")
|
|
|
|
assert cache.get("key1") is None
|
|
assert cache.get("key2") is not None
|
|
|
|
def test_clear_all(self, tmp_path):
|
|
"""Test clearing all cache entries."""
|
|
cache = DiskCache(cache_dir=str(tmp_path))
|
|
cache.set("key1", {"data": "value1"})
|
|
cache.set("key2", {"data": "value2"})
|
|
|
|
cache.clear()
|
|
|
|
assert cache.get("key1") is None
|
|
assert cache.get("key2") is None
|
|
|
|
def test_get_cache_dir(self, tmp_path):
|
|
"""Test getting cache directory."""
|
|
cache = DiskCache(cache_dir=str(tmp_path))
|
|
assert cache.get_cache_dir() == str(tmp_path)
|
|
|
|
def test_set_with_datetime(self, tmp_path):
|
|
"""Test setting cache with datetime objects."""
|
|
cache = DiskCache(cache_dir=str(tmp_path))
|
|
test_data = {
|
|
"timestamp": datetime.now(),
|
|
"data": "value"
|
|
}
|
|
|
|
cache.set("test_key", test_data)
|
|
result = cache.get("test_key")
|
|
|
|
# Datetime should be serialized/deserialized
|
|
assert result is not None
|
|
assert "data" in result
|
|
|
|
def test_cleanup_interval(self, tmp_path):
|
|
"""Test cleanup respects interval."""
|
|
cache = MemoryCache(cleanup_interval=60.0)
|
|
cache.set("key1", {"data": "value1"})
|
|
|
|
# First cleanup should work
|
|
removed1 = cache.cleanup(force=True)
|
|
|
|
# Second cleanup immediately after should return 0 (unless forced)
|
|
removed2 = cache.cleanup(force=False)
|
|
|
|
# If forced, should work; if not forced and within interval, should return 0
|
|
assert removed2 >= 0
|
|
|
|
def test_get_with_invalid_timestamp(self):
|
|
"""Test getting entry with invalid timestamp format."""
|
|
cache = MemoryCache()
|
|
cache.set("key1", {"data": "value1"})
|
|
# Set invalid timestamp
|
|
cache._timestamps["key1"] = "invalid_timestamp"
|
|
|
|
result = cache.get("key1")
|
|
|
|
# Should handle gracefully
|
|
assert result is None or isinstance(result, dict)
|
|
|
|
def test_record_background_hit(self):
|
|
"""Test recording background cache hit."""
|
|
metrics = CacheMetrics()
|
|
metrics.record_hit(cache_type='background')
|
|
stats = metrics.get_metrics()
|
|
|
|
assert stats['total_requests'] == 1
|
|
assert stats['background_hit_rate'] == 1.0
|
|
|
|
def test_record_background_miss(self):
|
|
"""Test recording background cache miss."""
|
|
metrics = CacheMetrics()
|
|
metrics.record_miss(cache_type='background')
|
|
stats = metrics.get_metrics()
|
|
|
|
assert stats['total_requests'] == 1
|
|
assert stats['background_hit_rate'] == 0.0
|
|
|
|
def test_multiple_fetch_times(self):
|
|
"""Test recording multiple fetch times."""
|
|
metrics = CacheMetrics()
|
|
metrics.record_fetch_time(0.5)
|
|
metrics.record_fetch_time(1.0)
|
|
metrics.record_fetch_time(0.3)
|
|
|
|
stats = metrics.get_metrics()
|
|
assert stats['fetch_count'] == 3
|
|
assert stats['total_fetch_time'] == 1.8
|
|
assert stats['average_fetch_time'] == pytest.approx(0.6, abs=0.01)
|