116 lines
4.5 KiB
Python
116 lines
4.5 KiB
Python
"""Integration tests verifying perform_nfo_repair_scan is wired into app startup.
|
|
|
|
These tests confirm that:
|
|
1. The lifespan calls perform_nfo_repair_scan after perform_media_scan_if_needed.
|
|
2. Series with incomplete NFO files are queued via the background_loader.
|
|
"""
|
|
from unittest.mock import AsyncMock, MagicMock, call, patch
|
|
|
|
import pytest
|
|
|
|
|
|
class TestNfoRepairScanCalledOnStartup:
|
|
"""Verify perform_nfo_repair_scan is invoked during the FastAPI lifespan."""
|
|
|
|
def test_perform_nfo_repair_scan_imported_in_lifespan(self):
|
|
"""fastapi_app.py lifespan imports perform_nfo_repair_scan."""
|
|
import importlib
|
|
|
|
import src.server.fastapi_app as app_module
|
|
|
|
source = importlib.util.find_spec("src.server.fastapi_app").origin
|
|
with open(source, "r", encoding="utf-8") as fh:
|
|
content = fh.read()
|
|
|
|
assert "perform_nfo_repair_scan" in content, (
|
|
"perform_nfo_repair_scan must be imported and called in fastapi_app.py"
|
|
)
|
|
|
|
def test_perform_nfo_repair_scan_called_after_media_scan(self):
|
|
"""perform_nfo_repair_scan must appear after perform_media_scan_if_needed."""
|
|
import importlib
|
|
|
|
source = importlib.util.find_spec("src.server.fastapi_app").origin
|
|
with open(source, "r", encoding="utf-8") as fh:
|
|
content = fh.read()
|
|
|
|
media_scan_pos = content.find("perform_media_scan_if_needed(background_loader)")
|
|
repair_scan_pos = content.find("perform_nfo_repair_scan(background_loader)")
|
|
|
|
assert media_scan_pos != -1, "perform_media_scan_if_needed call not found"
|
|
assert repair_scan_pos != -1, "perform_nfo_repair_scan call not found"
|
|
assert repair_scan_pos > media_scan_pos, (
|
|
"perform_nfo_repair_scan must be called AFTER perform_media_scan_if_needed"
|
|
)
|
|
|
|
|
|
class TestNfoRepairScanIntegrationWithBackgroundLoader:
|
|
"""Integration test: incomplete NFO series are queued via background_loader."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_incomplete_nfo_series_queued(self, tmp_path):
|
|
"""Series whose tvshow.nfo is missing required tags get queued."""
|
|
from src.server.services.initialization_service import perform_nfo_repair_scan
|
|
|
|
# Create a series directory with a minimal (incomplete) NFO
|
|
series_dir = tmp_path / "IncompleteAnime"
|
|
series_dir.mkdir()
|
|
(series_dir / "tvshow.nfo").write_text(
|
|
"<tvshow><title>IncompleteAnime</title></tvshow>"
|
|
)
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.tmdb_api_key = "test-key"
|
|
mock_settings.anime_directory = str(tmp_path)
|
|
|
|
mock_loader = AsyncMock()
|
|
mock_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:
|
|
mock_factory.return_value.create.return_value = MagicMock()
|
|
await perform_nfo_repair_scan(background_loader=mock_loader)
|
|
|
|
mock_loader.add_series_loading_task.assert_called_once_with(
|
|
key="IncompleteAnime",
|
|
folder="IncompleteAnime",
|
|
name="IncompleteAnime",
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_complete_nfo_series_not_queued(self, tmp_path):
|
|
"""Series whose tvshow.nfo has all required tags are not queued."""
|
|
from src.server.services.initialization_service import perform_nfo_repair_scan
|
|
|
|
series_dir = tmp_path / "CompleteAnime"
|
|
series_dir.mkdir()
|
|
(series_dir / "tvshow.nfo").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_loader = AsyncMock()
|
|
mock_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:
|
|
mock_factory.return_value.create.return_value = MagicMock()
|
|
await perform_nfo_repair_scan(background_loader=mock_loader)
|
|
|
|
mock_loader.add_series_loading_task.assert_not_called()
|