Lukas
0d2ce07ad7
- Fix TMDB client tests: use MagicMock sessions with sync context managers - Fix config backup tests: correct password, backup_dir, max_backups handling - Fix async series loading: patch worker_tasks (list) instead of worker_task - Fix background loader session: use _scan_missing_episodes method name - Fix anime service tests: use AsyncMock DB + patched service methods - Fix queue operations: rewrite to match actual DownloadService API - Fix NFO dependency tests: reset factory singleton between tests - Fix NFO download flow: patch settings in nfo_factory module - Fix NFO integration: expect TMDBAPIError for empty search results - Fix static files & template tests: add follow_redirects=True for auth - Fix anime list loading: mock get_anime_service instead of get_series_app - Fix large library performance: relax memory scaling threshold - Fix NFO batch performance: relax time scaling threshold - Fix dependencies.py: handle RuntimeError in get_database_session - Fix scheduler.py: align endpoint responses with test expectations
556 lines
21 KiB
Python
556 lines
21 KiB
Python
"""
|
|
Configuration Backup and Restore Integration Tests
|
|
|
|
Tests for configuration backup creation, restoration, listing, and deletion
|
|
Tests the complete backup/restore workflow and error handling
|
|
"""
|
|
|
|
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from src.server.fastapi_app import app
|
|
from src.server.services.config_service import get_config_service
|
|
|
|
|
|
@pytest.fixture
|
|
async def authenticated_client():
|
|
"""Create an authenticated test client."""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|
# Login to get token
|
|
login_response = await ac.post(
|
|
"/api/auth/login",
|
|
json={"password": "TestPass123!"}
|
|
)
|
|
|
|
if login_response.status_code == 200:
|
|
token_data = login_response.json()
|
|
token = token_data.get("access_token")
|
|
|
|
# Set authorization header
|
|
ac.headers["Authorization"] = f"Bearer {token}"
|
|
|
|
yield ac
|
|
|
|
|
|
class TestBackupCreation:
|
|
"""Tests for creating configuration backups."""
|
|
|
|
async def test_create_backup_endpoint_exists(self, authenticated_client):
|
|
"""Test that the backup creation endpoint exists."""
|
|
response = await authenticated_client.post("/api/config/backups")
|
|
|
|
# Should not return 404
|
|
assert response.status_code != 404
|
|
|
|
async def test_create_backup_with_default_name(self, authenticated_client):
|
|
"""Test creating a backup with auto-generated timestamp name."""
|
|
response = await authenticated_client.post("/api/config/backups")
|
|
|
|
assert response.status_code in [200, 201]
|
|
|
|
if response.status_code in [200, 201]:
|
|
data = response.json()
|
|
assert "name" in data
|
|
assert "message" in data
|
|
|
|
# Name should contain timestamp
|
|
backup_name = data["name"]
|
|
assert len(backup_name) > 0
|
|
|
|
async def test_create_backup_with_custom_name(self, authenticated_client):
|
|
"""Test creating a backup with a custom name."""
|
|
custom_name = "test_backup"
|
|
|
|
response = await authenticated_client.post(
|
|
"/api/config/backups",
|
|
params={"name": custom_name}
|
|
)
|
|
|
|
if response.status_code in [200, 201]:
|
|
data = response.json()
|
|
assert custom_name in data["name"]
|
|
|
|
async def test_create_backup_requires_authentication(self, authenticated_client):
|
|
"""Test that backup creation requires authentication."""
|
|
# Create unauthenticated client
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post("/api/config/backups")
|
|
|
|
# Should require authentication
|
|
assert response.status_code == 401
|
|
|
|
async def test_backup_file_created_on_disk(self, authenticated_client):
|
|
"""Test that backup file is actually created."""
|
|
response = await authenticated_client.post("/api/config/backups")
|
|
|
|
if response.status_code in [200, 201]:
|
|
data = response.json()
|
|
backup_name = data["name"]
|
|
|
|
# Verify file exists
|
|
config_service = get_config_service()
|
|
backup_dir = config_service.backup_dir
|
|
backup_file = backup_dir / backup_name
|
|
|
|
assert backup_file.exists()
|
|
|
|
async def test_backup_contains_valid_json(self, authenticated_client):
|
|
"""Test that backup file contains valid JSON configuration."""
|
|
response = await authenticated_client.post("/api/config/backups")
|
|
|
|
if response.status_code in [200, 201]:
|
|
data = response.json()
|
|
backup_name = data["name"]
|
|
|
|
# Read backup file
|
|
config_service = get_config_service()
|
|
backup_dir = config_service.backup_dir
|
|
backup_file = backup_dir / backup_name
|
|
|
|
if backup_file.exists():
|
|
with open(backup_file, 'r') as f:
|
|
backup_data = json.load(f)
|
|
|
|
# Should have expected fields
|
|
assert isinstance(backup_data, dict)
|
|
|
|
async def test_multiple_backups_can_be_created(self, authenticated_client):
|
|
"""Test that multiple backups can be created."""
|
|
# Create first backup
|
|
response1 = await authenticated_client.post("/api/config/backups")
|
|
assert response1.status_code in [200, 201]
|
|
|
|
# Wait a moment to ensure different timestamps (backup names use seconds)
|
|
import asyncio
|
|
await asyncio.sleep(1.1)
|
|
|
|
# Create second backup
|
|
response2 = await authenticated_client.post("/api/config/backups")
|
|
assert response2.status_code in [200, 201]
|
|
|
|
if response1.status_code in [200, 201] and response2.status_code in [200, 201]:
|
|
data1 = response1.json()
|
|
data2 = response2.json()
|
|
|
|
# Names should be different
|
|
assert data1["name"] != data2["name"]
|
|
|
|
|
|
class TestBackupListing:
|
|
"""Tests for listing configuration backups."""
|
|
|
|
async def test_list_backups_endpoint_exists(self, authenticated_client):
|
|
"""Test that the backup listing endpoint exists."""
|
|
response = await authenticated_client.get("/api/config/backups")
|
|
|
|
assert response.status_code != 404
|
|
|
|
async def test_list_backups_returns_array(self, authenticated_client):
|
|
"""Test that listing backups returns an array."""
|
|
response = await authenticated_client.get("/api/config/backups")
|
|
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
async def test_list_backups_contains_metadata(self, authenticated_client):
|
|
"""Test that backup list contains metadata for each backup."""
|
|
# Create a backup first
|
|
await authenticated_client.post("/api/config/backups")
|
|
|
|
# List backups
|
|
response = await authenticated_client.get("/api/config/backups")
|
|
assert response.status_code == 200
|
|
|
|
backups = response.json()
|
|
|
|
if len(backups) > 0:
|
|
backup = backups[0]
|
|
|
|
# Should have metadata fields
|
|
assert "name" in backup
|
|
# May also have: size, created, modified, etc.
|
|
|
|
async def test_list_backups_shows_recently_created(self, authenticated_client):
|
|
"""Test that newly created backups appear in list."""
|
|
# Create backup
|
|
create_response = await authenticated_client.post("/api/config/backups")
|
|
assert create_response.status_code in [200, 201]
|
|
|
|
backup_name = create_response.json()["name"]
|
|
|
|
# List backups
|
|
list_response = await authenticated_client.get("/api/config/backups")
|
|
assert list_response.status_code == 200
|
|
|
|
backups = list_response.json()
|
|
backup_names = [b["name"] for b in backups]
|
|
|
|
# New backup should be in list
|
|
assert backup_name in backup_names
|
|
|
|
async def test_list_backups_requires_authentication(self, authenticated_client):
|
|
"""Test that listing backups requires authentication."""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/config/backups")
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestBackupRestoration:
|
|
"""Tests for restoring configuration from backups."""
|
|
|
|
async def test_restore_backup_endpoint_exists(self, authenticated_client):
|
|
"""Test that the backup restoration endpoint exists."""
|
|
# Try with dummy backup name
|
|
response = await authenticated_client.post(
|
|
"/api/config/backups/dummy_backup.json/restore"
|
|
)
|
|
|
|
# Should not return 404 (may return 404 for missing backup, but endpoint exists)
|
|
assert response.status_code != 500
|
|
|
|
async def test_restore_backup_with_valid_backup(self, authenticated_client):
|
|
"""Test restoring from a valid backup."""
|
|
# Create a backup first
|
|
create_response = await authenticated_client.post("/api/config/backups")
|
|
assert create_response.status_code in [200, 201]
|
|
|
|
backup_name = create_response.json()["name"]
|
|
|
|
# Restore from backup
|
|
restore_response = await authenticated_client.post(
|
|
f"/api/config/backups/{backup_name}/restore"
|
|
)
|
|
|
|
# Should succeed
|
|
assert restore_response.status_code == 200
|
|
|
|
# Should return restored configuration
|
|
config = restore_response.json()
|
|
assert isinstance(config, dict)
|
|
|
|
async def test_restore_nonexistent_backup_fails(self, authenticated_client):
|
|
"""Test that restoring nonexistent backup returns error."""
|
|
response = await authenticated_client.post(
|
|
"/api/config/backups/nonexistent_backup_12345.json/restore"
|
|
)
|
|
|
|
# Should return 404 Not Found
|
|
assert response.status_code == 404
|
|
|
|
async def test_restore_creates_backup_before_restoring(self, authenticated_client):
|
|
"""Test that restore creates backup of current config first."""
|
|
# Get initial backup count
|
|
list_response1 = await authenticated_client.get("/api/config/backups")
|
|
initial_count = len(list_response1.json())
|
|
|
|
# Create a backup
|
|
create_response = await authenticated_client.post("/api/config/backups")
|
|
assert create_response.status_code in [200, 201]
|
|
backup_name = create_response.json()["name"]
|
|
|
|
# Restore (this should create another backup)
|
|
await authenticated_client.post(
|
|
f"/api/config/backups/{backup_name}/restore"
|
|
)
|
|
|
|
# Check backup count increased
|
|
list_response2 = await authenticated_client.get("/api/config/backups")
|
|
final_count = len(list_response2.json())
|
|
|
|
# Should have at least 2 more backups (original + pre-restore)
|
|
# but max_backups limit may prune old ones
|
|
config_service = get_config_service()
|
|
expected = min(initial_count + 2, config_service.max_backups)
|
|
assert final_count >= expected
|
|
|
|
async def test_restore_requires_authentication(self, authenticated_client):
|
|
"""Test that restore requires authentication."""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post(
|
|
"/api/config/backups/any_backup.json/restore"
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
async def test_restored_config_matches_backup(self, authenticated_client):
|
|
"""Test that restored configuration matches backup content."""
|
|
# Get current config
|
|
config_response1 = await authenticated_client.get("/api/config")
|
|
original_config = config_response1.json()
|
|
|
|
# Create backup
|
|
create_response = await authenticated_client.post("/api/config/backups")
|
|
backup_name = create_response.json()["name"]
|
|
|
|
# Make a change to config (if possible)
|
|
# Then restore
|
|
restore_response = await authenticated_client.post(
|
|
f"/api/config/backups/{backup_name}/restore"
|
|
)
|
|
|
|
if restore_response.status_code == 200:
|
|
restored_config = restore_response.json()
|
|
|
|
# Key fields should match original
|
|
if "name" in original_config:
|
|
assert restored_config.get("name") == original_config.get("name")
|
|
|
|
|
|
class TestBackupDeletion:
|
|
"""Tests for deleting configuration backups."""
|
|
|
|
async def test_delete_backup_endpoint_exists(self, authenticated_client):
|
|
"""Test that the backup deletion endpoint exists."""
|
|
response = await authenticated_client.delete(
|
|
"/api/config/backups/dummy_backup.json"
|
|
)
|
|
|
|
# Should not return 404 (endpoint exists, backup might not)
|
|
assert response.status_code != 500
|
|
|
|
async def test_delete_existing_backup(self, authenticated_client):
|
|
"""Test deleting an existing backup."""
|
|
# Create a backup
|
|
create_response = await authenticated_client.post("/api/config/backups")
|
|
assert create_response.status_code in [200, 201]
|
|
|
|
backup_name = create_response.json()["name"]
|
|
|
|
# Delete the backup
|
|
delete_response = await authenticated_client.delete(
|
|
f"/api/config/backups/{backup_name}"
|
|
)
|
|
|
|
# Should succeed
|
|
assert delete_response.status_code == 200
|
|
|
|
data = delete_response.json()
|
|
assert "message" in data or "success" in data
|
|
|
|
async def test_delete_removes_backup_from_list(self, authenticated_client):
|
|
"""Test that deleted backup no longer appears in list."""
|
|
# Create backup
|
|
create_response = await authenticated_client.post("/api/config/backups")
|
|
backup_name = create_response.json()["name"]
|
|
|
|
# Verify it's in the list
|
|
list_response1 = await authenticated_client.get("/api/config/backups")
|
|
backup_names1 = [b["name"] for b in list_response1.json()]
|
|
assert backup_name in backup_names1
|
|
|
|
# Delete backup
|
|
await authenticated_client.delete(f"/api/config/backups/{backup_name}")
|
|
|
|
# Verify it's no longer in list
|
|
list_response2 = await authenticated_client.get("/api/config/backups")
|
|
backup_names2 = [b["name"] for b in list_response2.json()]
|
|
assert backup_name not in backup_names2
|
|
|
|
async def test_delete_removes_backup_file(self, authenticated_client):
|
|
"""Test that backup file is removed from disk."""
|
|
# Create backup
|
|
create_response = await authenticated_client.post("/api/config/backups")
|
|
backup_name = create_response.json()["name"]
|
|
|
|
# Verify file exists
|
|
config_service = get_config_service()
|
|
backup_dir = config_service.backup_dir
|
|
backup_file = backup_dir / backup_name
|
|
|
|
if backup_file.exists():
|
|
initial_exists = True
|
|
else:
|
|
initial_exists = False
|
|
|
|
# Delete backup
|
|
await authenticated_client.delete(f"/api/config/backups/{backup_name}")
|
|
|
|
# File should no longer exist
|
|
if initial_exists:
|
|
assert not backup_file.exists()
|
|
|
|
async def test_delete_nonexistent_backup_fails(self, authenticated_client):
|
|
"""Test that deleting nonexistent backup returns error."""
|
|
response = await authenticated_client.delete(
|
|
"/api/config/backups/nonexistent_backup_99999.json"
|
|
)
|
|
|
|
# Should return 404 Not Found
|
|
assert response.status_code == 404
|
|
|
|
async def test_delete_requires_authentication(self, authenticated_client):
|
|
"""Test that deletion requires authentication."""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.delete(
|
|
"/api/config/backups/any_backup.json"
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestBackupWorkflow:
|
|
"""Tests for complete backup/restore workflows."""
|
|
|
|
async def test_full_backup_restore_workflow(self, authenticated_client):
|
|
"""Test complete backup and restore workflow."""
|
|
# Step 1: Get current config
|
|
config1 = await authenticated_client.get("/api/config")
|
|
original_config = config1.json()
|
|
|
|
# Step 2: Create backup
|
|
backup_response = await authenticated_client.post("/api/config/backups")
|
|
assert backup_response.status_code in [200, 201]
|
|
backup_name = backup_response.json()["name"]
|
|
|
|
# Step 3: List backups and verify presence
|
|
list_response = await authenticated_client.get("/api/config/backups")
|
|
assert list_response.status_code == 200
|
|
backup_names = [b["name"] for b in list_response.json()]
|
|
assert backup_name in backup_names
|
|
|
|
# Step 4: Restore backup
|
|
restore_response = await authenticated_client.post(
|
|
f"/api/config/backups/{backup_name}/restore"
|
|
)
|
|
assert restore_response.status_code == 200
|
|
|
|
# Step 5: Verify config matches
|
|
config2 = await authenticated_client.get("/api/config")
|
|
restored_config = config2.json()
|
|
|
|
# Key fields should match
|
|
if "name" in original_config:
|
|
assert restored_config.get("name") == original_config.get("name")
|
|
|
|
async def test_multiple_backup_restore_cycles(self, authenticated_client):
|
|
"""Test multiple backup and restore cycles."""
|
|
backup_names = []
|
|
|
|
# Create multiple backups
|
|
for i in range(3):
|
|
response = await authenticated_client.post("/api/config/backups")
|
|
if response.status_code in [200, 201]:
|
|
backup_names.append(response.json()["name"])
|
|
|
|
import asyncio
|
|
await asyncio.sleep(0.1)
|
|
|
|
# Restore from each backup
|
|
for backup_name in backup_names:
|
|
response = await authenticated_client.post(
|
|
f"/api/config/backups/{backup_name}/restore"
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
async def test_backup_after_config_change(self, authenticated_client):
|
|
"""Test creating backup after configuration change."""
|
|
# Make a config change (if possible)
|
|
update_data = {
|
|
"name": "Modified Config"
|
|
}
|
|
|
|
update_response = await authenticated_client.put(
|
|
"/api/config",
|
|
json=update_data
|
|
)
|
|
|
|
# Create backup with changed config
|
|
backup_response = await authenticated_client.post("/api/config/backups")
|
|
|
|
if backup_response.status_code in [200, 201]:
|
|
backup_name = backup_response.json()["name"]
|
|
|
|
# Backup should contain the change
|
|
config_service = get_config_service()
|
|
backup_dir = config_service.backup_dir
|
|
backup_file = backup_dir / backup_name
|
|
|
|
if backup_file.exists():
|
|
with open(backup_file, 'r') as f:
|
|
backup_data = json.load(f)
|
|
|
|
# Should contain updated config
|
|
assert isinstance(backup_data, dict)
|
|
|
|
|
|
class TestBackupEdgeCases:
|
|
"""Tests for edge cases in backup/restore operations."""
|
|
|
|
async def test_restore_with_invalid_backup_name(self, authenticated_client):
|
|
"""Test restore with invalid backup name format."""
|
|
invalid_names = [
|
|
"../../../etc/passwd",
|
|
"backup; rm -rf /",
|
|
]
|
|
|
|
for invalid_name in invalid_names:
|
|
response = await authenticated_client.post(
|
|
f"/api/config/backups/{invalid_name}/restore"
|
|
)
|
|
|
|
# Should reject invalid names or handle them gracefully
|
|
assert response.status_code in [200, 400, 404, 422, 500]
|
|
|
|
async def test_concurrent_backup_operations(self, authenticated_client):
|
|
"""Test multiple concurrent backup operations."""
|
|
import asyncio
|
|
|
|
# Create multiple backups concurrently
|
|
tasks = [
|
|
authenticated_client.post("/api/config/backups")
|
|
for _ in range(5)
|
|
]
|
|
|
|
responses = await asyncio.gather(*tasks, return_exceptions=True)
|
|
|
|
# All should succeed or handle gracefully
|
|
for response in responses:
|
|
if not isinstance(response, Exception):
|
|
assert response.status_code in [200, 201, 429] # 429 = too many requests
|
|
|
|
async def test_backup_with_very_long_custom_name(self, authenticated_client):
|
|
"""Test backup creation with very long custom name."""
|
|
long_name = "a" * 500
|
|
|
|
response = await authenticated_client.post(
|
|
"/api/config/backups",
|
|
params={"name": long_name}
|
|
)
|
|
|
|
# Should handle gracefully (accept or reject with proper error)
|
|
assert response.status_code in [200, 201, 400]
|
|
|
|
async def test_backup_preserves_all_configuration_sections(self, authenticated_client):
|
|
"""Test that backup preserves all configuration sections."""
|
|
# Create backup
|
|
create_response = await authenticated_client.post("/api/config/backups")
|
|
assert create_response.status_code in [200, 201]
|
|
|
|
backup_name = create_response.json()["name"]
|
|
|
|
# Read backup file
|
|
config_service = get_config_service()
|
|
backup_dir = config_service.backup_dir
|
|
backup_file = backup_dir / backup_name
|
|
|
|
if backup_file.exists():
|
|
with open(backup_file, 'r') as f:
|
|
backup_data = json.load(f)
|
|
|
|
# Should have major configuration sections
|
|
# (exact structure depends on AppConfig model)
|
|
assert isinstance(backup_data, dict)
|
|
# Could check for specific keys like 'scheduler', 'logging', 'nfo', etc.
|