Lukas
6a6ae7e059
- Add missing src/server/api/__init__.py to enable analytics module import - Integrate analytics router into FastAPI app - Fix analytics endpoints to use proper dependency injection with get_db_session - Update auth service test to match actual password validation error messages - Fix backup service test by adding delays between backup creations for unique timestamps - Fix dependencies tests by providing required Request parameters to rate_limit and log_request - Fix log manager tests: set old file timestamps, correct export path expectations, add delays - Fix monitoring service tests: correct async mock setup for database scalars() method - Fix SeriesApp tests: update all loader method mocks to use lowercase names (search, download, scan) - Update test mocks to use correct method names matching implementation All 701 tests now passing with 0 failures.
260 lines
7.2 KiB
Python
260 lines
7.2 KiB
Python
"""Unit tests for backup service."""
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from src.server.services.backup_service import BackupService, get_backup_service
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_backup_env():
|
|
"""Create temporary directories for testing."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
backup_dir = Path(tmpdir) / "backups"
|
|
config_dir = Path(tmpdir) / "config"
|
|
config_dir.mkdir()
|
|
|
|
# Create mock config files
|
|
(config_dir / "config.json").write_text('{"test": "config"}')
|
|
(config_dir / "download_queue.json").write_text('{"queue": []}')
|
|
|
|
yield {
|
|
"backup_dir": str(backup_dir),
|
|
"config_dir": str(config_dir),
|
|
"tmpdir": tmpdir,
|
|
}
|
|
|
|
|
|
def test_backup_service_initialization(temp_backup_env):
|
|
"""Test backup service initialization."""
|
|
service = BackupService(
|
|
backup_dir=temp_backup_env["backup_dir"],
|
|
config_dir=temp_backup_env["config_dir"],
|
|
)
|
|
|
|
assert service is not None
|
|
assert service.backup_dir.exists()
|
|
|
|
|
|
def test_backup_configuration(temp_backup_env):
|
|
"""Test configuration backup creation."""
|
|
service = BackupService(
|
|
backup_dir=temp_backup_env["backup_dir"],
|
|
config_dir=temp_backup_env["config_dir"],
|
|
)
|
|
|
|
backup_info = service.backup_configuration("Test backup")
|
|
|
|
assert backup_info is not None
|
|
assert backup_info.backup_type == "config"
|
|
assert backup_info.size_bytes > 0
|
|
assert "config_" in backup_info.name
|
|
|
|
|
|
def test_backup_configuration_no_config(temp_backup_env):
|
|
"""Test configuration backup with missing config file."""
|
|
service = BackupService(
|
|
backup_dir=temp_backup_env["backup_dir"],
|
|
config_dir=temp_backup_env["config_dir"],
|
|
)
|
|
|
|
# Remove config file
|
|
(Path(temp_backup_env["config_dir"]) / "config.json").unlink()
|
|
|
|
# Should still create backup (empty tar)
|
|
backup_info = service.backup_configuration()
|
|
|
|
assert backup_info is not None
|
|
|
|
|
|
def test_backup_database(temp_backup_env):
|
|
"""Test database backup creation."""
|
|
# Create mock database file
|
|
db_path = Path(temp_backup_env["tmpdir"]) / "aniworld.db"
|
|
db_path.write_bytes(b"mock database content")
|
|
|
|
service = BackupService(
|
|
backup_dir=temp_backup_env["backup_dir"],
|
|
config_dir=temp_backup_env["config_dir"],
|
|
database_path=str(db_path),
|
|
)
|
|
|
|
backup_info = service.backup_database("DB backup")
|
|
|
|
assert backup_info is not None
|
|
assert backup_info.backup_type == "data"
|
|
assert backup_info.size_bytes > 0
|
|
assert "database_" in backup_info.name
|
|
|
|
|
|
def test_backup_database_not_found(temp_backup_env):
|
|
"""Test database backup with missing database."""
|
|
service = BackupService(
|
|
backup_dir=temp_backup_env["backup_dir"],
|
|
config_dir=temp_backup_env["config_dir"],
|
|
database_path="/nonexistent/database.db",
|
|
)
|
|
|
|
backup_info = service.backup_database()
|
|
|
|
assert backup_info is None
|
|
|
|
|
|
def test_backup_full(temp_backup_env):
|
|
"""Test full system backup."""
|
|
service = BackupService(
|
|
backup_dir=temp_backup_env["backup_dir"],
|
|
config_dir=temp_backup_env["config_dir"],
|
|
)
|
|
|
|
backup_info = service.backup_full("Full backup")
|
|
|
|
assert backup_info is not None
|
|
assert backup_info.backup_type == "full"
|
|
assert backup_info.size_bytes > 0
|
|
|
|
|
|
def test_list_backups(temp_backup_env):
|
|
"""Test listing backups."""
|
|
service = BackupService(
|
|
backup_dir=temp_backup_env["backup_dir"],
|
|
config_dir=temp_backup_env["config_dir"],
|
|
)
|
|
|
|
# Create several backups
|
|
service.backup_configuration()
|
|
service.backup_full()
|
|
|
|
backups = service.list_backups()
|
|
|
|
assert len(backups) >= 2
|
|
assert all("name" in b for b in backups)
|
|
assert all("type" in b for b in backups)
|
|
|
|
|
|
def test_list_backups_by_type(temp_backup_env):
|
|
"""Test listing backups filtered by type."""
|
|
service = BackupService(
|
|
backup_dir=temp_backup_env["backup_dir"],
|
|
config_dir=temp_backup_env["config_dir"],
|
|
)
|
|
|
|
# Create different types of backups
|
|
service.backup_configuration()
|
|
service.backup_full()
|
|
|
|
config_backups = service.list_backups("config")
|
|
|
|
assert all(b["type"] == "config" for b in config_backups)
|
|
|
|
|
|
def test_delete_backup(temp_backup_env):
|
|
"""Test backup deletion."""
|
|
service = BackupService(
|
|
backup_dir=temp_backup_env["backup_dir"],
|
|
config_dir=temp_backup_env["config_dir"],
|
|
)
|
|
|
|
backup_info = service.backup_configuration()
|
|
assert backup_info is not None
|
|
|
|
backups_before = service.list_backups()
|
|
assert len(backups_before) > 0
|
|
|
|
result = service.delete_backup(backup_info.name)
|
|
|
|
assert result is True
|
|
backups_after = service.list_backups()
|
|
assert len(backups_after) < len(backups_before)
|
|
|
|
|
|
def test_delete_backup_not_found(temp_backup_env):
|
|
"""Test deleting non-existent backup."""
|
|
service = BackupService(
|
|
backup_dir=temp_backup_env["backup_dir"],
|
|
config_dir=temp_backup_env["config_dir"],
|
|
)
|
|
|
|
result = service.delete_backup("nonexistent_backup.tar.gz")
|
|
|
|
assert result is False
|
|
|
|
|
|
def test_cleanup_old_backups(temp_backup_env):
|
|
"""Test cleanup of old backups."""
|
|
import time
|
|
|
|
service = BackupService(
|
|
backup_dir=temp_backup_env["backup_dir"],
|
|
config_dir=temp_backup_env["config_dir"],
|
|
)
|
|
|
|
# Create multiple backups with small delays to ensure unique timestamps
|
|
for i in range(5):
|
|
service.backup_configuration()
|
|
time.sleep(1) # Ensure different timestamps
|
|
|
|
backups_before = service.list_backups()
|
|
assert len(backups_before) == 5
|
|
|
|
# Keep only 2 backups
|
|
deleted = service.cleanup_old_backups(max_backups=2)
|
|
|
|
backups_after = service.list_backups()
|
|
assert len(backups_after) <= 2
|
|
assert deleted == 3
|
|
|
|
|
|
def test_export_anime_data(temp_backup_env):
|
|
"""Test anime data export."""
|
|
service = BackupService(
|
|
backup_dir=temp_backup_env["backup_dir"],
|
|
config_dir=temp_backup_env["config_dir"],
|
|
)
|
|
|
|
export_file = Path(temp_backup_env["tmpdir"]) / "anime_export.json"
|
|
result = service.export_anime_data(str(export_file))
|
|
|
|
assert result is True
|
|
assert export_file.exists()
|
|
assert "timestamp" in export_file.read_text()
|
|
|
|
|
|
def test_import_anime_data(temp_backup_env):
|
|
"""Test anime data import."""
|
|
service = BackupService(
|
|
backup_dir=temp_backup_env["backup_dir"],
|
|
config_dir=temp_backup_env["config_dir"],
|
|
)
|
|
|
|
# Create import file
|
|
import_file = Path(temp_backup_env["tmpdir"]) / "anime_import.json"
|
|
import_file.write_text('{"timestamp": "2025-01-01T00:00:00", "data": []}')
|
|
|
|
result = service.import_anime_data(str(import_file))
|
|
|
|
assert result is True
|
|
|
|
|
|
def test_import_anime_data_not_found(temp_backup_env):
|
|
"""Test anime data import with missing file."""
|
|
service = BackupService(
|
|
backup_dir=temp_backup_env["backup_dir"],
|
|
config_dir=temp_backup_env["config_dir"],
|
|
)
|
|
|
|
result = service.import_anime_data("/nonexistent/file.json")
|
|
|
|
assert result is False
|
|
|
|
|
|
def test_get_backup_service():
|
|
"""Test singleton backup service."""
|
|
service1 = get_backup_service()
|
|
service2 = get_backup_service()
|
|
|
|
assert service1 is service2
|
|
assert isinstance(service1, BackupService)
|