Lukas
51b7f349f8
SchedulerConfig.__init__ maps legacy auto_download/folder_scan keys to the primary auto_download_after_rescan/folder_scan_enabled fields. However, model_dump() was including auto_download=null and folder_scan=null in serialised output. When this was written to config.json and reloaded, those keys were present (albeit null), so the alias mapping was skipped and the primary fields retained default False values instead of the configured True values. Fix: - Override SchedulerConfig.model_dump() to drop None-valued alias fields before returning the serialised dict. - ConfigService.save_config() re-serialises the scheduler field through its overridden model_dump() so the fix applies when writing to disk. Tests added: - test_roundtrip_excludes_none_alias_fields: verifies model_dump omits null auto_download/folder_scan keys. - test_save_and_load_scheduler_flags_roundtrip: end-to-end roundtrip through ConfigService confirms raw JSON and loaded values match. Pre-existing failure in test_core_error_handler.py is unrelated.
382 lines
14 KiB
Python
382 lines
14 KiB
Python
"""Unit tests for ConfigService."""
|
|
|
|
import json
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from src.server.models.config import (
|
|
AppConfig,
|
|
BackupConfig,
|
|
ConfigUpdate,
|
|
LoggingConfig,
|
|
SchedulerConfig,
|
|
)
|
|
from src.server.services.config_service import (
|
|
ConfigBackupError,
|
|
ConfigService,
|
|
ConfigServiceError,
|
|
ConfigValidationError,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_dir():
|
|
"""Create temporary directory for test config files."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
yield Path(tmpdir)
|
|
|
|
|
|
@pytest.fixture
|
|
def config_service(temp_dir):
|
|
"""Create ConfigService instance with temporary paths."""
|
|
config_path = temp_dir / "config.json"
|
|
backup_dir = temp_dir / "backups"
|
|
return ConfigService(
|
|
config_path=config_path, backup_dir=backup_dir, max_backups=3
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_config():
|
|
"""Create sample configuration."""
|
|
return AppConfig(
|
|
name="TestApp",
|
|
data_dir="test_data",
|
|
scheduler=SchedulerConfig(enabled=True, interval_minutes=30),
|
|
logging=LoggingConfig(level="DEBUG", file="test.log"),
|
|
backup=BackupConfig(enabled=False),
|
|
other={"custom_key": "custom_value"},
|
|
)
|
|
|
|
|
|
class TestConfigServiceInitialization:
|
|
"""Test ConfigService initialization and directory creation."""
|
|
|
|
def test_initialization_creates_directories(self, temp_dir):
|
|
"""Test that initialization creates necessary directories."""
|
|
config_path = temp_dir / "subdir" / "config.json"
|
|
backup_dir = temp_dir / "subdir" / "backups"
|
|
|
|
service = ConfigService(config_path=config_path, backup_dir=backup_dir)
|
|
|
|
assert config_path.parent.exists()
|
|
assert backup_dir.exists()
|
|
assert service.config_path == config_path
|
|
assert service.backup_dir == backup_dir
|
|
|
|
def test_initialization_with_existing_directories(self, config_service):
|
|
"""Test initialization with existing directories works."""
|
|
assert config_service.config_path.parent.exists()
|
|
assert config_service.backup_dir.exists()
|
|
|
|
|
|
class TestConfigServiceLoadSave:
|
|
"""Test configuration loading and saving."""
|
|
|
|
def test_load_creates_default_config_if_not_exists(self, config_service):
|
|
"""Test that load creates default config if file doesn't exist."""
|
|
config = config_service.load_config()
|
|
|
|
assert isinstance(config, AppConfig)
|
|
assert config.name == "Aniworld"
|
|
assert config_service.config_path.exists()
|
|
|
|
def test_save_and_load_config(self, config_service, sample_config):
|
|
"""Test saving and loading configuration."""
|
|
config_service.save_config(sample_config, create_backup=False)
|
|
|
|
loaded_config = config_service.load_config()
|
|
|
|
assert loaded_config.name == sample_config.name
|
|
assert loaded_config.data_dir == sample_config.data_dir
|
|
assert loaded_config.scheduler.enabled == sample_config.scheduler.enabled
|
|
assert loaded_config.logging.level == sample_config.logging.level
|
|
assert loaded_config.other == sample_config.other
|
|
|
|
def test_save_and_load_scheduler_flags_roundtrip(self, config_service):
|
|
"""Scheduler auto_download_after_rescan and folder_scan_enabled must
|
|
survive a full save/load roundtrip through ConfigService.
|
|
|
|
Regression test for a bug where null legacy alias fields
|
|
(auto_download=None, folder_scan=None) were written to config.json
|
|
on save. On reload the alias mapping was skipped (because the keys
|
|
were present), causing the primary boolean fields to reset to False.
|
|
"""
|
|
original = AppConfig(
|
|
scheduler=SchedulerConfig(
|
|
enabled=True,
|
|
auto_download_after_rescan=True,
|
|
folder_scan_enabled=True,
|
|
)
|
|
)
|
|
config_service.save_config(original, create_backup=False)
|
|
|
|
# Verify raw JSON does not contain legacy alias keys
|
|
with open(config_service.config_path, "r", encoding="utf-8") as f:
|
|
raw = json.load(f)
|
|
assert "auto_download" not in raw["scheduler"]
|
|
assert "folder_scan" not in raw["scheduler"]
|
|
assert raw["scheduler"]["auto_download_after_rescan"] is True
|
|
assert raw["scheduler"]["folder_scan_enabled"] is True
|
|
|
|
# Verify loaded config preserves values
|
|
loaded = config_service.load_config()
|
|
assert loaded.scheduler.auto_download_after_rescan is True
|
|
assert loaded.scheduler.folder_scan_enabled is True
|
|
|
|
def test_save_includes_version(self, config_service, sample_config):
|
|
"""Test that saved config includes version field."""
|
|
config_service.save_config(sample_config, create_backup=False)
|
|
|
|
with open(config_service.config_path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
|
|
assert "version" in data
|
|
assert data["version"] == ConfigService.CONFIG_VERSION
|
|
|
|
def test_save_creates_backup_by_default(self, config_service, sample_config):
|
|
"""Test that save creates backup by default if file exists."""
|
|
# Save initial config
|
|
config_service.save_config(sample_config, create_backup=False)
|
|
|
|
# Modify and save again (should create backup)
|
|
sample_config.name = "Modified"
|
|
config_service.save_config(sample_config, create_backup=True)
|
|
|
|
backups = list(config_service.backup_dir.glob("*.json"))
|
|
assert len(backups) == 1
|
|
|
|
def test_save_atomic_operation(self, config_service, sample_config):
|
|
"""Test that save is atomic (uses temp file)."""
|
|
# Mock exception during JSON dump by using invalid data
|
|
# This should not corrupt existing config
|
|
config_service.save_config(sample_config, create_backup=False)
|
|
|
|
# Verify temp file is cleaned up after successful save
|
|
temp_files = list(config_service.config_path.parent.glob("*.tmp"))
|
|
assert len(temp_files) == 0
|
|
|
|
def test_load_invalid_json_raises_error(self, config_service):
|
|
"""Test that loading invalid JSON raises ConfigValidationError."""
|
|
# Write invalid JSON
|
|
config_service.config_path.write_text("invalid json {")
|
|
|
|
with pytest.raises(ConfigValidationError, match="Invalid JSON"):
|
|
config_service.load_config()
|
|
|
|
|
|
class TestConfigServiceValidation:
|
|
"""Test configuration validation."""
|
|
|
|
def test_validate_valid_config(self, config_service, sample_config):
|
|
"""Test validation of valid configuration."""
|
|
result = config_service.validate_config(sample_config)
|
|
|
|
assert result.valid is True
|
|
assert result.errors == []
|
|
|
|
def test_validate_invalid_config(self, config_service):
|
|
"""Test validation of invalid configuration."""
|
|
# Create config with backups enabled but no path
|
|
invalid_config = AppConfig(
|
|
backup=BackupConfig(enabled=True, path=None)
|
|
)
|
|
|
|
result = config_service.validate_config(invalid_config)
|
|
|
|
assert result.valid is False
|
|
assert len(result.errors or []) > 0
|
|
|
|
def test_save_invalid_config_raises_error(self, config_service):
|
|
"""Test that saving invalid config raises error."""
|
|
invalid_config = AppConfig(
|
|
backup=BackupConfig(enabled=True, path=None)
|
|
)
|
|
|
|
with pytest.raises(ConfigValidationError, match="Cannot save invalid"):
|
|
config_service.save_config(invalid_config)
|
|
|
|
|
|
class TestConfigServiceUpdate:
|
|
"""Test configuration updates."""
|
|
|
|
def test_update_config(self, config_service, sample_config):
|
|
"""Test updating configuration."""
|
|
config_service.save_config(sample_config, create_backup=False)
|
|
|
|
update = ConfigUpdate(
|
|
scheduler=SchedulerConfig(enabled=False, interval_minutes=60),
|
|
logging=LoggingConfig(level="INFO"),
|
|
)
|
|
|
|
updated_config = config_service.update_config(update)
|
|
|
|
assert updated_config.scheduler.enabled is False
|
|
assert updated_config.scheduler.interval_minutes == 60
|
|
assert updated_config.logging.level == "INFO"
|
|
# Other fields should remain unchanged
|
|
assert updated_config.name == sample_config.name
|
|
assert updated_config.data_dir == sample_config.data_dir
|
|
|
|
def test_update_persists_changes(self, config_service, sample_config):
|
|
"""Test that updates are persisted to disk."""
|
|
config_service.save_config(sample_config, create_backup=False)
|
|
|
|
update = ConfigUpdate(logging=LoggingConfig(level="ERROR"))
|
|
config_service.update_config(update)
|
|
|
|
# Load fresh config from disk
|
|
loaded = config_service.load_config()
|
|
assert loaded.logging.level == "ERROR"
|
|
|
|
|
|
class TestConfigServiceBackups:
|
|
"""Test configuration backup functionality."""
|
|
|
|
def test_create_backup(self, config_service, sample_config):
|
|
"""Test creating configuration backup."""
|
|
config_service.save_config(sample_config, create_backup=False)
|
|
|
|
backup_path = config_service.create_backup()
|
|
|
|
assert backup_path.exists()
|
|
assert backup_path.suffix == ".json"
|
|
assert "config_backup_" in backup_path.name
|
|
|
|
def test_create_backup_with_custom_name(
|
|
self, config_service, sample_config
|
|
):
|
|
"""Test creating backup with custom name."""
|
|
config_service.save_config(sample_config, create_backup=False)
|
|
|
|
backup_path = config_service.create_backup(name="my_backup")
|
|
|
|
assert backup_path.name == "my_backup.json"
|
|
|
|
def test_create_backup_without_config_raises_error(self, config_service):
|
|
"""Test that creating backup without config file raises error."""
|
|
with pytest.raises(ConfigBackupError, match="Cannot backup non-existent"):
|
|
config_service.create_backup()
|
|
|
|
def test_list_backups(self, config_service, sample_config):
|
|
"""Test listing configuration backups."""
|
|
config_service.save_config(sample_config, create_backup=False)
|
|
|
|
# Create multiple backups
|
|
config_service.create_backup(name="backup1")
|
|
config_service.create_backup(name="backup2")
|
|
config_service.create_backup(name="backup3")
|
|
|
|
backups = config_service.list_backups()
|
|
|
|
assert len(backups) == 3
|
|
assert all("name" in b for b in backups)
|
|
assert all("size_bytes" in b for b in backups)
|
|
assert all("created_at" in b for b in backups)
|
|
|
|
# Should be sorted by creation time (newest first)
|
|
backup_names = [b["name"] for b in backups]
|
|
assert "backup3.json" in backup_names
|
|
|
|
def test_list_backups_empty(self, config_service):
|
|
"""Test listing backups when none exist."""
|
|
backups = config_service.list_backups()
|
|
assert backups == []
|
|
|
|
def test_restore_backup(self, config_service, sample_config):
|
|
"""Test restoring configuration from backup."""
|
|
# Save initial config and create backup
|
|
config_service.save_config(sample_config, create_backup=False)
|
|
config_service.create_backup(name="original")
|
|
|
|
# Modify and save config
|
|
sample_config.name = "Modified"
|
|
config_service.save_config(sample_config, create_backup=False)
|
|
|
|
# Restore from backup
|
|
restored = config_service.restore_backup("original.json")
|
|
|
|
assert restored.name == "TestApp" # Original name
|
|
|
|
def test_restore_backup_creates_pre_restore_backup(
|
|
self, config_service, sample_config
|
|
):
|
|
"""Test that restore creates pre-restore backup."""
|
|
config_service.save_config(sample_config, create_backup=False)
|
|
config_service.create_backup(name="backup1")
|
|
|
|
sample_config.name = "Modified"
|
|
config_service.save_config(sample_config, create_backup=False)
|
|
|
|
config_service.restore_backup("backup1.json")
|
|
|
|
backups = config_service.list_backups()
|
|
backup_names = [b["name"] for b in backups]
|
|
|
|
assert any("pre_restore" in name for name in backup_names)
|
|
|
|
def test_restore_nonexistent_backup_raises_error(self, config_service):
|
|
"""Test that restoring non-existent backup raises error."""
|
|
with pytest.raises(ConfigBackupError, match="Backup not found"):
|
|
config_service.restore_backup("nonexistent.json")
|
|
|
|
def test_delete_backup(self, config_service, sample_config):
|
|
"""Test deleting configuration backup."""
|
|
config_service.save_config(sample_config, create_backup=False)
|
|
config_service.create_backup(name="to_delete")
|
|
|
|
config_service.delete_backup("to_delete.json")
|
|
|
|
backups = config_service.list_backups()
|
|
assert len(backups) == 0
|
|
|
|
def test_delete_nonexistent_backup_raises_error(self, config_service):
|
|
"""Test that deleting non-existent backup raises error."""
|
|
with pytest.raises(ConfigBackupError, match="Backup not found"):
|
|
config_service.delete_backup("nonexistent.json")
|
|
|
|
def test_cleanup_old_backups(self, config_service, sample_config):
|
|
"""Test that old backups are cleaned up when limit exceeded."""
|
|
config_service.save_config(sample_config, create_backup=False)
|
|
|
|
# Create more backups than max_backups (3)
|
|
for i in range(5):
|
|
config_service.create_backup(name=f"backup{i}")
|
|
|
|
backups = config_service.list_backups()
|
|
assert len(backups) == 3 # Should only keep max_backups
|
|
|
|
|
|
class TestConfigServiceSingleton:
|
|
"""Test singleton instance management."""
|
|
|
|
def test_get_config_service_returns_singleton(self):
|
|
"""Test that get_config_service returns same instance."""
|
|
from src.server.services.config_service import get_config_service
|
|
|
|
service1 = get_config_service()
|
|
service2 = get_config_service()
|
|
|
|
assert service1 is service2
|
|
|
|
|
|
class TestConfigServiceErrorHandling:
|
|
"""Test error handling in ConfigService."""
|
|
|
|
def test_save_config_creates_temp_file(
|
|
self, config_service, sample_config
|
|
):
|
|
"""Test that save operation uses temporary file."""
|
|
# Save config and verify temp file is cleaned up
|
|
config_service.save_config(sample_config, create_backup=False)
|
|
|
|
# Verify no temp files remain
|
|
temp_files = list(config_service.config_path.parent.glob("*.tmp"))
|
|
assert len(temp_files) == 0
|
|
|
|
# Verify config was saved successfully
|
|
loaded = config_service.load_config()
|
|
assert loaded.name == sample_config.name
|