feat: Add comprehensive settings modal and backup/restore tests
E2E Tests (tests/frontend/e2e/settings_modal.spec.js): - Modal open/close: button, overlay, Escape (5 tests) - Configuration sections: all sections display (5 tests) - Load configuration: directory, count, scheduler, status (4 tests) - Edit fields: name, directory, scheduler, interval (6 tests) - Save configuration: main, scheduler, feedback (4 tests) - Reset configuration to original values (2 tests) - Browse directory functionality (2 tests) - Connection test and status update (2 tests) - Scheduler status: next/last rescan, running (3 tests) - Accessibility: labels, keyboard nav, focus trap (4 tests) - Edge cases: rapid changes, long inputs, multiple opens (5 tests) - Theme integration: respect theme, toggle (2 tests) Total: 44 E2E tests Integration Tests (tests/integration/test_config_backup_restore.py): - Backup creation: default/custom name, auth, file creation (6 tests) - Backup listing: array, metadata, recent, auth (5 tests) - Backup restoration: valid, nonexistent, pre-backup, auth (6 tests) - Backup deletion: existing, removes file/list, auth (5 tests) - Complete workflows: full cycle, multiple cycles (3 tests) - Edge cases: invalid names, concurrent ops, long names (4 tests) Total: 29 integration tests Updated instructions.md marking settings modal tests complete
This commit is contained in:
553
tests/integration/test_config_backup_restore.py
Normal file
553
tests/integration/test_config_backup_restore.py
Normal file
@@ -0,0 +1,553 @@
|
||||
"""
|
||||
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": "Hallo123!"}
|
||||
)
|
||||
|
||||
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 = Path(config_service.data_dir) / "config_backups"
|
||||
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 = Path(config_service.data_dir) / "config_backups"
|
||||
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
|
||||
import asyncio
|
||||
await asyncio.sleep(0.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)
|
||||
assert final_count >= initial_count + 2
|
||||
|
||||
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 = Path(config_service.data_dir) / "config_backups"
|
||||
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 = Path(config_service.data_dir) / "config_backups"
|
||||
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 /",
|
||||
"backup\x00.json"
|
||||
]
|
||||
|
||||
for invalid_name in invalid_names:
|
||||
response = await authenticated_client.post(
|
||||
f"/api/config/backups/{invalid_name}/restore"
|
||||
)
|
||||
|
||||
# Should reject invalid names
|
||||
assert response.status_code in [400, 404]
|
||||
|
||||
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 = Path(config_service.data_dir) / "config_backups"
|
||||
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.
|
||||
Reference in New Issue
Block a user