881 lines
38 KiB
Python
881 lines
38 KiB
Python
"""Unit tests for initialization service.
|
|
|
|
This module tests application startup orchestration, configuration loading,
|
|
and initialization sequences to ensure proper system setup.
|
|
|
|
Coverage Target: 85%+
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, call, patch
|
|
|
|
import pytest
|
|
|
|
from src.server.services.initialization_service import (
|
|
_check_initial_scan_status,
|
|
_check_media_scan_status,
|
|
_check_nfo_scan_status,
|
|
_check_scan_status,
|
|
_execute_media_scan,
|
|
_execute_nfo_scan,
|
|
_is_nfo_scan_configured,
|
|
_load_series_into_memory,
|
|
_mark_initial_scan_completed,
|
|
_mark_media_scan_completed,
|
|
_mark_nfo_scan_completed,
|
|
_mark_scan_completed,
|
|
_sync_anime_folders,
|
|
_validate_anime_directory,
|
|
perform_initial_setup,
|
|
perform_media_scan_if_needed,
|
|
perform_nfo_scan_if_needed,
|
|
perform_nfo_repair_scan,
|
|
)
|
|
|
|
|
|
class TestCheckScanStatus:
|
|
"""Test _check_scan_status generic function."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_scan_status_completed(self):
|
|
"""Test checking scan status when completed."""
|
|
mock_check_method = AsyncMock(return_value=True)
|
|
|
|
with patch('src.server.database.connection.get_db_session') as mock_get_db:
|
|
mock_db = AsyncMock()
|
|
mock_get_db.return_value.__aenter__.return_value = mock_db
|
|
|
|
result = await _check_scan_status(
|
|
check_method=mock_check_method,
|
|
scan_type="test",
|
|
log_completed_msg="Completed",
|
|
log_not_completed_msg="Not completed"
|
|
)
|
|
|
|
assert result is True
|
|
mock_check_method.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_scan_status_not_completed(self):
|
|
"""Test checking scan status when not completed."""
|
|
mock_check_method = AsyncMock(return_value=False)
|
|
|
|
with patch('src.server.database.connection.get_db_session') as mock_get_db:
|
|
mock_db = AsyncMock()
|
|
mock_get_db.return_value.__aenter__.return_value = mock_db
|
|
|
|
result = await _check_scan_status(
|
|
check_method=mock_check_method,
|
|
scan_type="test"
|
|
)
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_scan_status_exception(self):
|
|
"""Test checking scan status with exception returns False."""
|
|
mock_check_method = AsyncMock(side_effect=Exception("Database error"))
|
|
|
|
with patch('src.server.database.connection.get_db_session') as mock_get_db:
|
|
mock_db = AsyncMock()
|
|
mock_get_db.return_value.__aenter__.return_value = mock_db
|
|
|
|
result = await _check_scan_status(
|
|
check_method=mock_check_method,
|
|
scan_type="test"
|
|
)
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestMarkScanCompleted:
|
|
"""Test _mark_scan_completed generic function."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mark_scan_completed_success(self):
|
|
"""Test marking scan as completed successfully."""
|
|
mock_mark_method = AsyncMock()
|
|
|
|
with patch('src.server.database.connection.get_db_session') as mock_get_db:
|
|
mock_db = AsyncMock()
|
|
mock_get_db.return_value.__aenter__.return_value = mock_db
|
|
|
|
await _mark_scan_completed(
|
|
mark_method=mock_mark_method,
|
|
scan_type="test"
|
|
)
|
|
|
|
mock_mark_method.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mark_scan_completed_exception(self):
|
|
"""Test marking scan as completed with exception logs warning."""
|
|
mock_mark_method = AsyncMock(side_effect=Exception("Database error"))
|
|
|
|
with patch('src.server.database.connection.get_db_session') as mock_get_db:
|
|
mock_db = AsyncMock()
|
|
mock_get_db.return_value.__aenter__.return_value = mock_db
|
|
|
|
# Should not raise exception
|
|
await _mark_scan_completed(
|
|
mark_method=mock_mark_method,
|
|
scan_type="test"
|
|
)
|
|
|
|
|
|
class TestInitialScanFunctions:
|
|
"""Test initial scan status check and marking functions."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_initial_scan_status_completed(self):
|
|
"""Test checking initial scan when completed."""
|
|
with patch('src.server.services.initialization_service._check_scan_status',
|
|
new_callable=AsyncMock, return_value=True) as mock_check:
|
|
result = await _check_initial_scan_status()
|
|
|
|
assert result is True
|
|
mock_check.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_initial_scan_status_not_completed(self):
|
|
"""Test checking initial scan when not completed."""
|
|
with patch('src.server.services.initialization_service._check_scan_status',
|
|
new_callable=AsyncMock, return_value=False) as mock_check:
|
|
result = await _check_initial_scan_status()
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mark_initial_scan_completed(self):
|
|
"""Test marking initial scan as completed."""
|
|
with patch('src.server.services.initialization_service._mark_scan_completed',
|
|
new_callable=AsyncMock) as mock_mark:
|
|
await _mark_initial_scan_completed()
|
|
|
|
mock_mark.assert_called_once()
|
|
|
|
|
|
class TestSyncAnimeFolders:
|
|
"""Test anime folder scanning and syncing."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_anime_folders_without_progress(self):
|
|
"""Test syncing anime folders without progress service."""
|
|
with patch('src.server.services.initialization_service.sync_series_from_data_files',
|
|
new_callable=AsyncMock, return_value=42) as mock_sync:
|
|
result = await _sync_anime_folders()
|
|
|
|
assert result == 42
|
|
mock_sync.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_anime_folders_with_progress(self):
|
|
"""Test syncing anime folders with progress updates."""
|
|
mock_progress = AsyncMock()
|
|
|
|
with patch('src.server.services.initialization_service.sync_series_from_data_files',
|
|
new_callable=AsyncMock, return_value=10) as mock_sync:
|
|
result = await _sync_anime_folders(progress_service=mock_progress)
|
|
|
|
assert result == 10
|
|
# Verify progress updates were called
|
|
assert mock_progress.update_progress.call_count == 2
|
|
mock_progress.update_progress.assert_any_call(
|
|
progress_id="series_sync",
|
|
current=25,
|
|
message="Scanning anime folders...",
|
|
metadata={"step_id": "series_sync"}
|
|
)
|
|
mock_progress.update_progress.assert_any_call(
|
|
progress_id="series_sync",
|
|
current=75,
|
|
message="Synced 10 series from data files",
|
|
metadata={"step_id": "series_sync"}
|
|
)
|
|
|
|
|
|
class TestLoadSeriesIntoMemory:
|
|
"""Test series loading into memory."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_series_without_progress(self):
|
|
"""Test loading series without progress service."""
|
|
mock_anime_service = MagicMock()
|
|
mock_anime_service._load_series_from_db = AsyncMock()
|
|
|
|
with patch('src.server.utils.dependencies.get_anime_service',
|
|
return_value=mock_anime_service):
|
|
await _load_series_into_memory()
|
|
|
|
mock_anime_service._load_series_from_db.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_series_with_progress(self):
|
|
"""Test loading series with progress service."""
|
|
mock_anime_service = MagicMock()
|
|
mock_anime_service._load_series_from_db = AsyncMock()
|
|
mock_progress = AsyncMock()
|
|
|
|
with patch('src.server.utils.dependencies.get_anime_service',
|
|
return_value=mock_anime_service):
|
|
await _load_series_into_memory(progress_service=mock_progress)
|
|
|
|
mock_anime_service._load_series_from_db.assert_called_once()
|
|
mock_progress.complete_progress.assert_called_once_with(
|
|
progress_id="series_sync",
|
|
message="Series loaded into memory",
|
|
metadata={"step_id": "series_sync"}
|
|
)
|
|
|
|
|
|
class TestValidateAnimeDirectory:
|
|
"""Test anime directory validation."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_directory_configured(self):
|
|
"""Test validation when directory is configured."""
|
|
with patch('src.server.services.initialization_service.settings') as mock_settings:
|
|
mock_settings.anime_directory = "/path/to/anime"
|
|
|
|
result = await _validate_anime_directory()
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_directory_not_configured(self):
|
|
"""Test validation when directory is not configured."""
|
|
with patch('src.server.services.initialization_service.settings') as mock_settings:
|
|
mock_settings.anime_directory = None
|
|
|
|
result = await _validate_anime_directory()
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_directory_not_configured_with_progress(self):
|
|
"""Test validation when directory not configured with progress service."""
|
|
mock_progress = AsyncMock()
|
|
|
|
with patch('src.server.services.initialization_service.settings') as mock_settings:
|
|
mock_settings.anime_directory = ""
|
|
|
|
result = await _validate_anime_directory(progress_service=mock_progress)
|
|
|
|
assert result is False
|
|
mock_progress.complete_progress.assert_called_once_with(
|
|
progress_id="series_sync",
|
|
message="No anime directory configured",
|
|
metadata={"step_id": "series_sync"}
|
|
)
|
|
|
|
|
|
class TestPerformInitialSetup:
|
|
"""Test complete initial setup orchestration."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initial_setup_already_completed(self):
|
|
"""Test setup skips when already completed."""
|
|
with patch('src.server.services.initialization_service._check_initial_scan_status',
|
|
new_callable=AsyncMock, return_value=True):
|
|
result = await perform_initial_setup()
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initial_setup_already_completed_with_progress(self):
|
|
"""Test setup skips when already completed with progress updates."""
|
|
mock_progress = AsyncMock()
|
|
|
|
with patch('src.server.services.initialization_service._check_initial_scan_status',
|
|
new_callable=AsyncMock, return_value=True):
|
|
result = await perform_initial_setup(progress_service=mock_progress)
|
|
|
|
assert result is False
|
|
mock_progress.start_progress.assert_called_once()
|
|
mock_progress.complete_progress.assert_called_once_with(
|
|
progress_id="series_sync",
|
|
message="Already completed",
|
|
metadata={"step_id": "series_sync"}
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initial_setup_directory_not_configured(self):
|
|
"""Test setup fails when directory not configured."""
|
|
with patch('src.server.services.initialization_service._check_initial_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._validate_anime_directory',
|
|
new_callable=AsyncMock, return_value=False):
|
|
result = await perform_initial_setup()
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initial_setup_success(self):
|
|
"""Test successful initial setup."""
|
|
with patch('src.server.services.initialization_service._check_initial_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._validate_anime_directory',
|
|
new_callable=AsyncMock, return_value=True), \
|
|
patch('src.server.services.initialization_service._sync_anime_folders',
|
|
new_callable=AsyncMock, return_value=5), \
|
|
patch('src.server.services.initialization_service._mark_initial_scan_completed',
|
|
new_callable=AsyncMock), \
|
|
patch('src.server.services.initialization_service._load_series_into_memory',
|
|
new_callable=AsyncMock):
|
|
result = await perform_initial_setup()
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initial_setup_with_progress_service(self):
|
|
"""Test successful initial setup with progress updates."""
|
|
mock_progress = AsyncMock()
|
|
|
|
with patch('src.server.services.initialization_service._check_initial_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._validate_anime_directory',
|
|
new_callable=AsyncMock, return_value=True), \
|
|
patch('src.server.services.initialization_service._sync_anime_folders',
|
|
new_callable=AsyncMock, return_value=5), \
|
|
patch('src.server.services.initialization_service._mark_initial_scan_completed',
|
|
new_callable=AsyncMock), \
|
|
patch('src.server.services.initialization_service._load_series_into_memory',
|
|
new_callable=AsyncMock):
|
|
result = await perform_initial_setup(progress_service=mock_progress)
|
|
|
|
assert result is True
|
|
mock_progress.start_progress.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initial_setup_os_error(self):
|
|
"""Test setup handles OSError gracefully."""
|
|
with patch('src.server.services.initialization_service._check_initial_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._validate_anime_directory',
|
|
new_callable=AsyncMock, return_value=True), \
|
|
patch('src.server.services.initialization_service._sync_anime_folders',
|
|
new_callable=AsyncMock, side_effect=OSError("Disk error")):
|
|
result = await perform_initial_setup()
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initial_setup_runtime_error(self):
|
|
"""Test setup handles RuntimeError gracefully."""
|
|
with patch('src.server.services.initialization_service._check_initial_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._validate_anime_directory',
|
|
new_callable=AsyncMock, return_value=True), \
|
|
patch('src.server.services.initialization_service._sync_anime_folders',
|
|
new_callable=AsyncMock, side_effect=RuntimeError("Runtime error")):
|
|
result = await perform_initial_setup()
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initial_setup_value_error(self):
|
|
"""Test setup handles ValueError gracefully."""
|
|
with patch('src.server.services.initialization_service._check_initial_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._validate_anime_directory',
|
|
new_callable=AsyncMock, return_value=True), \
|
|
patch('src.server.services.initialization_service._sync_anime_folders',
|
|
new_callable=AsyncMock, side_effect=ValueError("Invalid value")):
|
|
result = await perform_initial_setup()
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestNFOScanFunctions:
|
|
"""Test NFO scan status and configuration checks."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_nfo_scan_status(self):
|
|
"""Test checking NFO scan status."""
|
|
with patch('src.server.services.initialization_service._check_scan_status',
|
|
new_callable=AsyncMock, return_value=True) as mock_check:
|
|
result = await _check_nfo_scan_status()
|
|
|
|
assert result is True
|
|
mock_check.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mark_nfo_scan_completed(self):
|
|
"""Test marking NFO scan as completed."""
|
|
with patch('src.server.services.initialization_service._mark_scan_completed',
|
|
new_callable=AsyncMock) as mock_mark:
|
|
await _mark_nfo_scan_completed()
|
|
|
|
mock_mark.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_nfo_scan_configured_with_api_key_and_auto_create(self):
|
|
"""Test NFO scan configured with API key and auto create."""
|
|
with patch('src.server.services.initialization_service.settings') as mock_settings:
|
|
mock_settings.tmdb_api_key = "test_api_key"
|
|
mock_settings.nfo_auto_create = True
|
|
mock_settings.nfo_update_on_scan = False
|
|
|
|
result = await _is_nfo_scan_configured()
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_nfo_scan_configured_with_api_key_and_update_on_scan(self):
|
|
"""Test NFO scan configured with API key and update on scan."""
|
|
with patch('src.server.services.initialization_service.settings') as mock_settings:
|
|
mock_settings.tmdb_api_key = "test_api_key"
|
|
mock_settings.nfo_auto_create = False
|
|
mock_settings.nfo_update_on_scan = True
|
|
|
|
result = await _is_nfo_scan_configured()
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_nfo_scan_not_configured_no_api_key(self):
|
|
"""Test NFO scan not configured without API key."""
|
|
with patch('src.server.services.initialization_service.settings') as mock_settings:
|
|
mock_settings.tmdb_api_key = None
|
|
mock_settings.nfo_auto_create = True
|
|
mock_settings.nfo_update_on_scan = True
|
|
|
|
result = await _is_nfo_scan_configured()
|
|
|
|
# Result should be falsy (None or False) when API key is None
|
|
assert not result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_nfo_scan_not_configured_features_disabled(self):
|
|
"""Test NFO scan not configured when features disabled."""
|
|
with patch('src.server.services.initialization_service.settings') as mock_settings:
|
|
mock_settings.tmdb_api_key = "test_api_key"
|
|
mock_settings.nfo_auto_create = False
|
|
mock_settings.nfo_update_on_scan = False
|
|
|
|
result = await _is_nfo_scan_configured()
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestExecuteNFOScan:
|
|
"""Test NFO scan execution."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_nfo_scan_without_progress(self):
|
|
"""Test executing NFO scan without progress service."""
|
|
mock_manager = MagicMock()
|
|
mock_manager.scan_and_process_nfo = AsyncMock()
|
|
mock_manager.close = AsyncMock()
|
|
|
|
with patch('src.core.services.series_manager_service.SeriesManagerService') as mock_sms:
|
|
mock_sms.from_settings.return_value = mock_manager
|
|
|
|
await _execute_nfo_scan()
|
|
|
|
mock_manager.scan_and_process_nfo.assert_called_once()
|
|
mock_manager.close.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_nfo_scan_with_progress(self):
|
|
"""Test executing NFO scan with progress updates."""
|
|
mock_manager = MagicMock()
|
|
mock_manager.scan_and_process_nfo = AsyncMock()
|
|
mock_manager.close = AsyncMock()
|
|
mock_progress = AsyncMock()
|
|
|
|
with patch('src.core.services.series_manager_service.SeriesManagerService') as mock_sms:
|
|
mock_sms.from_settings.return_value = mock_manager
|
|
|
|
await _execute_nfo_scan(progress_service=mock_progress)
|
|
|
|
mock_manager.scan_and_process_nfo.assert_called_once()
|
|
mock_manager.close.assert_called_once()
|
|
assert mock_progress.update_progress.call_count == 2
|
|
mock_progress.complete_progress.assert_called_once()
|
|
|
|
|
|
class TestPerformNFOScan:
|
|
"""Test complete NFO scan orchestration."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nfo_scan_already_completed(self):
|
|
"""Test NFO scan skips when already completed."""
|
|
with patch('src.server.services.initialization_service._check_nfo_scan_status',
|
|
new_callable=AsyncMock, return_value=True), \
|
|
patch('src.server.services.initialization_service._is_nfo_scan_configured',
|
|
new_callable=AsyncMock, return_value=True):
|
|
await perform_nfo_scan_if_needed()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nfo_scan_not_configured_no_api_key(self):
|
|
"""Test NFO scan skips when API key not configured."""
|
|
mock_progress = AsyncMock()
|
|
|
|
with patch('src.server.services.initialization_service._check_nfo_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._is_nfo_scan_configured',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service.settings') as mock_settings:
|
|
mock_settings.tmdb_api_key = None
|
|
|
|
await perform_nfo_scan_if_needed(progress_service=mock_progress)
|
|
|
|
mock_progress.start_progress.assert_called_once()
|
|
mock_progress.complete_progress.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nfo_scan_not_configured_features_disabled(self):
|
|
"""Test NFO scan skips when features disabled."""
|
|
mock_progress = AsyncMock()
|
|
|
|
with patch('src.server.services.initialization_service._check_nfo_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._is_nfo_scan_configured',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service.settings') as mock_settings:
|
|
mock_settings.tmdb_api_key = "test_key"
|
|
|
|
await perform_nfo_scan_if_needed(progress_service=mock_progress)
|
|
|
|
mock_progress.complete_progress.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nfo_scan_success(self):
|
|
"""Test successful NFO scan execution."""
|
|
with patch('src.server.services.initialization_service._check_nfo_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._is_nfo_scan_configured',
|
|
new_callable=AsyncMock, return_value=True), \
|
|
patch('src.server.services.initialization_service._execute_nfo_scan',
|
|
new_callable=AsyncMock), \
|
|
patch('src.server.services.initialization_service._mark_nfo_scan_completed',
|
|
new_callable=AsyncMock) as mock_mark:
|
|
await perform_nfo_scan_if_needed()
|
|
|
|
mock_mark.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nfo_scan_with_progress_service(self):
|
|
"""Test NFO scan with progress service updates."""
|
|
mock_progress = AsyncMock()
|
|
|
|
with patch('src.server.services.initialization_service._check_nfo_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._is_nfo_scan_configured',
|
|
new_callable=AsyncMock, return_value=True), \
|
|
patch('src.server.services.initialization_service._execute_nfo_scan',
|
|
new_callable=AsyncMock), \
|
|
patch('src.server.services.initialization_service._mark_nfo_scan_completed',
|
|
new_callable=AsyncMock):
|
|
await perform_nfo_scan_if_needed(progress_service=mock_progress)
|
|
|
|
mock_progress.start_progress.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nfo_scan_exception_handling(self):
|
|
"""Test NFO scan handles exceptions and updates progress."""
|
|
mock_progress = AsyncMock()
|
|
|
|
with patch('src.server.services.initialization_service._check_nfo_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._is_nfo_scan_configured',
|
|
new_callable=AsyncMock, return_value=True), \
|
|
patch('src.server.services.initialization_service._execute_nfo_scan',
|
|
new_callable=AsyncMock, side_effect=Exception("TMDB API error")):
|
|
await perform_nfo_scan_if_needed(progress_service=mock_progress)
|
|
|
|
mock_progress.fail_progress.assert_called_once()
|
|
|
|
|
|
class TestMediaScanFunctions:
|
|
"""Test media scan status and execution."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_media_scan_status(self):
|
|
"""Test checking media scan status."""
|
|
with patch('src.server.services.initialization_service._check_scan_status',
|
|
new_callable=AsyncMock, return_value=True) as mock_check:
|
|
result = await _check_media_scan_status()
|
|
|
|
assert result is True
|
|
mock_check.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mark_media_scan_completed(self):
|
|
"""Test marking media scan as completed."""
|
|
with patch('src.server.services.initialization_service._mark_scan_completed',
|
|
new_callable=AsyncMock) as mock_mark:
|
|
await _mark_media_scan_completed()
|
|
|
|
mock_mark.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_media_scan(self):
|
|
"""Test executing media scan."""
|
|
mock_loader = MagicMock()
|
|
|
|
with patch('src.server.fastapi_app._check_incomplete_series_on_startup',
|
|
new_callable=AsyncMock) as mock_check:
|
|
await _execute_media_scan(mock_loader)
|
|
|
|
mock_check.assert_called_once_with(mock_loader)
|
|
|
|
|
|
class TestPerformMediaScan:
|
|
"""Test complete media scan orchestration."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_media_scan_already_completed(self):
|
|
"""Test media scan skips when already completed."""
|
|
mock_loader = MagicMock()
|
|
|
|
with patch('src.server.services.initialization_service._check_media_scan_status',
|
|
new_callable=AsyncMock, return_value=True):
|
|
await perform_media_scan_if_needed(mock_loader)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_media_scan_success(self):
|
|
"""Test successful media scan execution."""
|
|
mock_loader = MagicMock()
|
|
|
|
with patch('src.server.services.initialization_service._check_media_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._execute_media_scan',
|
|
new_callable=AsyncMock), \
|
|
patch('src.server.services.initialization_service._mark_media_scan_completed',
|
|
new_callable=AsyncMock) as mock_mark:
|
|
await perform_media_scan_if_needed(mock_loader)
|
|
|
|
mock_mark.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_media_scan_exception_handling(self):
|
|
"""Test media scan handles exceptions gracefully."""
|
|
mock_loader = MagicMock()
|
|
|
|
with patch('src.server.services.initialization_service._check_media_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._execute_media_scan',
|
|
new_callable=AsyncMock, side_effect=Exception("Media scan error")):
|
|
# Should not raise exception
|
|
await perform_media_scan_if_needed(mock_loader)
|
|
|
|
|
|
class TestInitializationIntegration:
|
|
"""Test integration scenarios for initialization service."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_initialization_sequence(self):
|
|
"""Test complete initialization sequence."""
|
|
mock_progress = AsyncMock()
|
|
mock_loader = MagicMock()
|
|
|
|
# Initial setup
|
|
with patch('src.server.services.initialization_service._check_initial_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._validate_anime_directory',
|
|
new_callable=AsyncMock, return_value=True), \
|
|
patch('src.server.services.initialization_service._sync_anime_folders',
|
|
new_callable=AsyncMock, return_value=10), \
|
|
patch('src.server.services.initialization_service._mark_initial_scan_completed',
|
|
new_callable=AsyncMock), \
|
|
patch('src.server.services.initialization_service._load_series_into_memory',
|
|
new_callable=AsyncMock):
|
|
result = await perform_initial_setup(progress_service=mock_progress)
|
|
|
|
assert result is True
|
|
|
|
# NFO scan
|
|
with patch('src.server.services.initialization_service._check_nfo_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._is_nfo_scan_configured',
|
|
new_callable=AsyncMock, return_value=True), \
|
|
patch('src.server.services.initialization_service._execute_nfo_scan',
|
|
new_callable=AsyncMock), \
|
|
patch('src.server.services.initialization_service._mark_nfo_scan_completed',
|
|
new_callable=AsyncMock):
|
|
await perform_nfo_scan_if_needed(progress_service=mock_progress)
|
|
|
|
# Media scan
|
|
with patch('src.server.services.initialization_service._check_media_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._execute_media_scan',
|
|
new_callable=AsyncMock), \
|
|
patch('src.server.services.initialization_service._mark_media_scan_completed',
|
|
new_callable=AsyncMock):
|
|
await perform_media_scan_if_needed(mock_loader)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_partial_initialization_recovery(self):
|
|
"""Test recovery from partial initialization."""
|
|
# Simulate initial scan failed
|
|
with patch('src.server.services.initialization_service._check_initial_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._validate_anime_directory',
|
|
new_callable=AsyncMock, return_value=True), \
|
|
patch('src.server.services.initialization_service._sync_anime_folders',
|
|
new_callable=AsyncMock, side_effect=OSError("Disk full")):
|
|
result = await perform_initial_setup()
|
|
|
|
assert result is False
|
|
|
|
# Retry should work
|
|
with patch('src.server.services.initialization_service._check_initial_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._validate_anime_directory',
|
|
new_callable=AsyncMock, return_value=True), \
|
|
patch('src.server.services.initialization_service._sync_anime_folders',
|
|
new_callable=AsyncMock, return_value=5), \
|
|
patch('src.server.services.initialization_service._mark_initial_scan_completed',
|
|
new_callable=AsyncMock), \
|
|
patch('src.server.services.initialization_service._load_series_into_memory',
|
|
new_callable=AsyncMock):
|
|
result = await perform_initial_setup()
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_idempotent_initialization(self):
|
|
"""Test initialization is idempotent."""
|
|
# First run
|
|
with patch('src.server.services.initialization_service._check_initial_scan_status',
|
|
new_callable=AsyncMock, return_value=False), \
|
|
patch('src.server.services.initialization_service._validate_anime_directory',
|
|
new_callable=AsyncMock, return_value=True), \
|
|
patch('src.server.services.initialization_service._sync_anime_folders',
|
|
new_callable=AsyncMock, return_value=5), \
|
|
patch('src.server.services.initialization_service._mark_initial_scan_completed',
|
|
new_callable=AsyncMock), \
|
|
patch('src.server.services.initialization_service._load_series_into_memory',
|
|
new_callable=AsyncMock):
|
|
result1 = await perform_initial_setup()
|
|
|
|
assert result1 is True
|
|
|
|
# Second run should skip
|
|
with patch('src.server.services.initialization_service._check_initial_scan_status',
|
|
new_callable=AsyncMock, return_value=True):
|
|
result2 = await perform_initial_setup()
|
|
|
|
assert result2 is False
|
|
|
|
|
|
class TestPerformNfoRepairScan:
|
|
"""Tests for the perform_nfo_repair_scan startup hook."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skips_without_tmdb_api_key(self, tmp_path):
|
|
"""Should return immediately when no TMDB API key is configured."""
|
|
mock_settings = MagicMock()
|
|
mock_settings.tmdb_api_key = ""
|
|
mock_settings.anime_directory = str(tmp_path)
|
|
|
|
with patch(
|
|
"src.server.services.initialization_service.settings", mock_settings
|
|
):
|
|
await perform_nfo_repair_scan()
|
|
|
|
# No exception means guard worked — nothing was iterated
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skips_without_anime_directory(self, tmp_path):
|
|
"""Should return immediately when no anime directory is configured."""
|
|
mock_settings = MagicMock()
|
|
mock_settings.tmdb_api_key = "some-key"
|
|
mock_settings.anime_directory = ""
|
|
|
|
with patch(
|
|
"src.server.services.initialization_service.settings", mock_settings
|
|
):
|
|
await perform_nfo_repair_scan()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_queues_deficient_series_via_background_loader(self, tmp_path):
|
|
"""Series with incomplete NFO should be queued via background_loader."""
|
|
# Create a fake series directory with a tvshow.nfo file
|
|
series_dir = tmp_path / "MyAnime"
|
|
series_dir.mkdir()
|
|
nfo_file = series_dir / "tvshow.nfo"
|
|
nfo_file.write_text("<tvshow><title>MyAnime</title></tvshow>")
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.tmdb_api_key = "test-key"
|
|
mock_settings.anime_directory = str(tmp_path)
|
|
|
|
mock_background_loader = AsyncMock()
|
|
mock_background_loader.add_series_loading_task = AsyncMock()
|
|
|
|
with patch(
|
|
"src.server.services.initialization_service.settings", mock_settings
|
|
), patch(
|
|
"src.core.services.nfo_repair_service.nfo_needs_repair",
|
|
return_value=True,
|
|
), patch(
|
|
"src.core.services.nfo_factory.NFOServiceFactory"
|
|
) as mock_factory_cls:
|
|
mock_factory_cls.return_value.create.return_value = MagicMock()
|
|
await perform_nfo_repair_scan(background_loader=mock_background_loader)
|
|
|
|
mock_background_loader.add_series_loading_task.assert_called_once_with(
|
|
key="MyAnime", folder="MyAnime", name="MyAnime"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skips_complete_series(self, tmp_path):
|
|
"""Series with complete NFO should not be queued or repaired."""
|
|
series_dir = tmp_path / "CompleteAnime"
|
|
series_dir.mkdir()
|
|
nfo_file = series_dir / "tvshow.nfo"
|
|
nfo_file.write_text("<tvshow><title>CompleteAnime</title></tvshow>")
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.tmdb_api_key = "test-key"
|
|
mock_settings.anime_directory = str(tmp_path)
|
|
|
|
mock_background_loader = AsyncMock()
|
|
mock_background_loader.add_series_loading_task = AsyncMock()
|
|
|
|
with patch(
|
|
"src.server.services.initialization_service.settings", mock_settings
|
|
), patch(
|
|
"src.core.services.nfo_repair_service.nfo_needs_repair",
|
|
return_value=False,
|
|
), patch(
|
|
"src.core.services.nfo_factory.NFOServiceFactory"
|
|
) as mock_factory_cls:
|
|
mock_factory_cls.return_value.create.return_value = MagicMock()
|
|
await perform_nfo_repair_scan(background_loader=mock_background_loader)
|
|
|
|
mock_background_loader.add_series_loading_task.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_repairs_directly_without_background_loader(self, tmp_path):
|
|
"""When no background_loader provided, repair_series is called directly."""
|
|
series_dir = tmp_path / "NeedsRepair"
|
|
series_dir.mkdir()
|
|
nfo_file = series_dir / "tvshow.nfo"
|
|
nfo_file.write_text("<tvshow><title>NeedsRepair</title></tvshow>")
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.tmdb_api_key = "test-key"
|
|
mock_settings.anime_directory = str(tmp_path)
|
|
|
|
mock_repair_service = AsyncMock()
|
|
mock_repair_service.repair_series = AsyncMock(return_value=True)
|
|
|
|
with patch(
|
|
"src.server.services.initialization_service.settings", mock_settings
|
|
), patch(
|
|
"src.core.services.nfo_repair_service.nfo_needs_repair",
|
|
return_value=True,
|
|
), patch(
|
|
"src.core.services.nfo_factory.NFOServiceFactory"
|
|
) as mock_factory_cls, patch(
|
|
"src.core.services.nfo_repair_service.NfoRepairService",
|
|
return_value=mock_repair_service,
|
|
):
|
|
mock_factory_cls.return_value.create.return_value = MagicMock()
|
|
await perform_nfo_repair_scan(background_loader=None)
|
|
|
|
mock_repair_service.repair_series.assert_called_once()
|