3ae5d736ce
- T55: Setup APScheduler with AsyncIOScheduler and @scheduled_job decorator - T56: Implement hourly usage stats sync from OpenRouter API - T57: Implement daily API key validation job - T58: Implement weekly cleanup job for old usage stats - Add usage_stats_retention_days config option - Integrate scheduler with FastAPI lifespan events - Add 26 unit tests for scheduler, sync, and cleanup tasks - Add apscheduler to requirements.txt The background tasks now automatically: - Sync usage stats every hour from OpenRouter - Validate API keys daily at 2 AM UTC - Clean up old data weekly on Sunday at 3 AM UTC
108 lines
3.9 KiB
Python
108 lines
3.9 KiB
Python
"""Tests for cleanup tasks.
|
|
|
|
T58: Task to clean up old usage stats data.
|
|
"""
|
|
import pytest
|
|
from datetime import datetime, date, timedelta
|
|
from unittest.mock import Mock, patch, MagicMock, AsyncMock
|
|
from apscheduler.triggers.cron import CronTrigger
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCleanupOldUsageStats:
|
|
"""Test suite for cleanup_old_usage_stats task."""
|
|
|
|
def test_cleanup_has_correct_decorator(self):
|
|
"""Test that cleanup_old_usage_stats has correct scheduled_job decorator."""
|
|
# Arrange
|
|
from openrouter_monitor.tasks.cleanup import cleanup_old_usage_stats
|
|
from openrouter_monitor.tasks.scheduler import get_scheduler
|
|
|
|
# Act
|
|
scheduler = get_scheduler()
|
|
job = scheduler.get_job('cleanup_old_usage_stats')
|
|
|
|
# Assert
|
|
assert job is not None
|
|
assert job.func == cleanup_old_usage_stats
|
|
assert isinstance(job.trigger, CronTrigger)
|
|
|
|
def test_cleanup_is_async_function(self):
|
|
"""Test that cleanup_old_usage_stats is an async function."""
|
|
# Arrange
|
|
from openrouter_monitor.tasks.cleanup import cleanup_old_usage_stats
|
|
import inspect
|
|
|
|
# Assert
|
|
assert inspect.iscoroutinefunction(cleanup_old_usage_stats)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cleanup_handles_errors_gracefully(self):
|
|
"""Test that cleanup handles errors without crashing."""
|
|
# Arrange
|
|
from openrouter_monitor.tasks.cleanup import cleanup_old_usage_stats
|
|
|
|
with patch('openrouter_monitor.tasks.cleanup.SessionLocal') as mock_session:
|
|
# Simulate database error
|
|
mock_session.side_effect = Exception("Database connection failed")
|
|
|
|
# Act & Assert - should not raise
|
|
await cleanup_old_usage_stats()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cleanup_uses_retention_days_from_config(self):
|
|
"""Test that cleanup uses retention days from settings."""
|
|
# Arrange
|
|
from openrouter_monitor.tasks.cleanup import cleanup_old_usage_stats
|
|
from openrouter_monitor.config import get_settings
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.rowcount = 0
|
|
|
|
async def mock_execute(*args, **kwargs):
|
|
return mock_result
|
|
|
|
mock_db = MagicMock()
|
|
mock_db.execute = mock_execute
|
|
mock_db.commit = Mock()
|
|
|
|
# Get actual retention days from config
|
|
settings = get_settings()
|
|
expected_retention = settings.usage_stats_retention_days
|
|
|
|
with patch('openrouter_monitor.tasks.cleanup.SessionLocal') as mock_session:
|
|
mock_session.return_value.__enter__ = Mock(return_value=mock_db)
|
|
mock_session.return_value.__exit__ = Mock(return_value=False)
|
|
|
|
# Act
|
|
await cleanup_old_usage_stats()
|
|
|
|
# Assert - verify retention days is reasonable (default 365)
|
|
assert expected_retention > 0
|
|
assert expected_retention <= 365 * 5 # Max 5 years
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCleanupConfiguration:
|
|
"""Test suite for cleanup configuration."""
|
|
|
|
def test_retention_days_configurable(self):
|
|
"""Test that retention days is configurable."""
|
|
from openrouter_monitor.config import get_settings
|
|
|
|
settings = get_settings()
|
|
|
|
# Should have a default value
|
|
assert hasattr(settings, 'usage_stats_retention_days')
|
|
assert isinstance(settings.usage_stats_retention_days, int)
|
|
assert settings.usage_stats_retention_days > 0
|
|
|
|
def test_default_retention_is_one_year(self):
|
|
"""Test that default retention period is approximately one year."""
|
|
from openrouter_monitor.config import get_settings
|
|
|
|
settings = get_settings()
|
|
|
|
# Default should be 365 days (1 year)
|
|
assert settings.usage_stats_retention_days == 365
|