rick/LEDMatrix
1
0
Fork 0
mirror of https://github.com/ChuckBuilds/LEDMatrix.git synced 2026-08-01 08:48:05 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
fix/mock-cache-manager-strategy-method
LEDMatrix/test/test_cache_manager.py
T
ChuckGitHubClaude Opus 4.8
 3b93024993 feat: activate dormant plugin health/metrics subsystem and surface it in the web UI (#388) 
* feat(plugin-system): activate dormant plugin health & metrics subsystem

PluginManager shipped a fully-built health tracker, resource monitor and
circuit breaker that were never instantiated (health_tracker/resource_monitor
were left as None), so the circuit breaker never engaged and the existing
health/metrics API routes always returned "not available".

- DisplayController now wires a PluginHealthTracker and PluginResourceMonitor
  onto the plugin manager, enabling the circuit breaker (a repeatedly-failing
  plugin's update() is skipped after consecutive failures, then retried after
  a cooldown) and per-plugin execution-time metrics. Both persist to the
  shared cache.
- load_plugin() now validates each plugin's config against its JSON schema in
  a strictly warn/degrade-only way: a violation logs a warning and flags the
  plugin degraded in the health tracker, but never changes whether the plugin
  loads or its pass/fail behaviour. Adds PluginHealthTracker.set_degraded(),
  which never touches the circuit breaker.
- ResourceMonitor CPU/memory sampling now reuses a cached psutil.Process and
  reads cpu_percent(interval=None), so monitoring no longer blocks ~100ms per
  call on the display loop's update path.
- Fix DiskCache.get() raising TypeError for max_age=None ("never expires"),
  which silently discarded persisted plugin health/metrics on read and thus
  broke cross-process and post-restart surfacing.
- Fix two dead PluginManager helpers that called non-existent tracker methods.

Tests: new test_resource_monitor, test_plugin_health,
test_plugin_manager_schema_soft; extended test_cache_manager and
test_display_controller.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UvTav268UXv44ub9K11LYq

* feat(web-ui): surface plugin health, metrics and load state

With the health/metrics subsystem now active in the display service, expose it
in the web UI (which runs as a separate process from the display loop):

- Wire a health tracker / resource monitor backed by the shared on-disk cache
  into the web process so /api/v3/plugins/health and /plugins/metrics read the
  data the display service persists.
- Build those route responses per installed plugin id (the tracker's in-memory
  view is empty in a fresh web process) so cross-process data is included.
- Add state + error_info to /plugins/installed entries so the UI can show why a
  plugin isn't running instead of just loaded:false.
- Add a "Plugin Health" panel to the Tools page (circuit status, avg/max update
  time, update count, last error) plus PluginAPI.getPluginMetrics().

Tests: route-level tests for the health/metrics endpoints in test_web_api.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UvTav268UXv44ub9K11LYq

* fix(plugin-metrics): refresh cross-process health/metrics reads; type hints

Addresses CodeRabbit review on #388:

- Major: the web process's health/resource trackers cached the first persisted
  read in an in-memory dict (and the CacheManager memory tier held max_age=None
  entries indefinitely), so a long-lived web process showed the first snapshot
  and never reflected the display service's later updates. Add an opt-in
  force_reload path (get_health_summary/get_health_state/_load_health_state and
  get_metrics_summary/get_metrics) that bypasses the in-memory copy and, via a
  new memory_ttl passthrough on CacheManager.get, the cache manager's memory
  tier — so each /plugins/health and /plugins/metrics poll reads fresh persisted
  state. Default behaviour (force_reload=False) is unchanged for the display
  process and existing callers.
- Minor: DiskCache.get type hint is now Optional[int] with the None ("never
  expires") semantics documented, matching MemoryCache.get.

Tests: new force_reload staleness cases in test_plugin_health and
test_resource_monitor.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UvTav268UXv44ub9K11LYq

---------

Co-authored-by: Claude <noreply@anthropic.com>
2026-07-09 07:54:18 -04:00

403 lines
14 KiB
Python
Raw Permalink Blame History

"""
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_max_age_none_never_expires(self, tmp_path):
"""max_age=None must return persisted records regardless of age.
Regression: the age comparison raised TypeError for max_age=None,
which was swallowed and treated as a miss — silently breaking
long-lived state (plugin health/metrics) read across processes.
"""
cache = DiskCache(cache_dir=str(tmp_path))
cache.set("test_key", {"data": "value", "timestamp": 0}) # epoch → very old
result = cache.get("test_key", max_age=None)
assert result is not None
assert result["data"] == "value"
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)
Reference in New Issue View Git Blame Copy Permalink