Lukas
756731cd5d
- Remove NFO repair scan step from ARCHITECTURE.md startup sequence - Update CHANGELOG.md: rephrase perform_nfo_repair_scan as scheduled scan - Add test verifying perform_nfo_repair_scan is NOT called in lifespan - Keep existing folder scan wiring tests and unit tests intact - NFO_GUIDE.md already correctly describes scheduled scan behavior
295 lines
11 KiB
Python
295 lines
11 KiB
Python
"""Integration tests for poster check service wiring.
|
|
|
|
These tests verify that:
|
|
1. FolderScanService.run_folder_scan calls check_and_download_missing_posters.
|
|
2. The poster check logic is properly integrated into the scheduled folder scan.
|
|
3. Missing posters are downloaded, valid posters are skipped, and errors are handled.
|
|
"""
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
class TestPosterCheckScanCalledInFolderScan:
|
|
"""Verify check_and_download_missing_posters is invoked from FolderScanService."""
|
|
|
|
def test_check_and_download_missing_posters_imported_in_folder_scan_service(self):
|
|
"""folder_scan_service.py imports check_and_download_missing_posters."""
|
|
import importlib
|
|
|
|
source = importlib.util.find_spec(
|
|
"src.server.services.folder_scan_service"
|
|
).origin
|
|
with open(source, "r", encoding="utf-8") as fh:
|
|
content = fh.read()
|
|
|
|
assert "check_and_download_missing_posters" in content, (
|
|
"check_and_download_missing_posters must be imported in folder_scan_service.py"
|
|
)
|
|
|
|
def test_check_and_download_missing_posters_called_in_run_folder_scan(self):
|
|
"""check_and_download_missing_posters must be called inside run_folder_scan."""
|
|
import importlib
|
|
|
|
source = importlib.util.find_spec(
|
|
"src.server.services.folder_scan_service"
|
|
).origin
|
|
with open(source, "r", encoding="utf-8") as fh:
|
|
content = fh.read()
|
|
|
|
run_folder_scan_pos = content.find("def run_folder_scan")
|
|
poster_call_pos = content.find("check_and_download_missing_posters()")
|
|
|
|
assert run_folder_scan_pos != -1, "run_folder_scan method not found"
|
|
assert poster_call_pos != -1, "check_and_download_missing_posters call not found"
|
|
assert poster_call_pos > run_folder_scan_pos, (
|
|
"check_and_download_missing_posters must be called INSIDE run_folder_scan"
|
|
)
|
|
|
|
|
|
class TestPosterCheckIntegration:
|
|
"""Integration test: poster check is triggered during folder scan."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_poster_check_downloads_missing_poster(self, tmp_path):
|
|
"""When poster.jpg is missing, the scan downloads it from the NFO thumb URL."""
|
|
from src.server.services.folder_scan_service import FolderScanService
|
|
|
|
anime_dir = tmp_path / "anime"
|
|
anime_dir.mkdir()
|
|
series_dir = anime_dir / "Attack on Titan (2013)"
|
|
series_dir.mkdir()
|
|
(series_dir / "tvshow.nfo").write_text(
|
|
"<?xml version='1.0' encoding='UTF-8'?>\n"
|
|
"<tvshow>\n"
|
|
" <title>Attack on Titan</title>\n"
|
|
" <year>2013</year>\n"
|
|
' <thumb aspect="poster">https://example.com/poster.jpg</thumb>\n'
|
|
"</tvshow>\n"
|
|
)
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.tmdb_api_key = "test-key"
|
|
mock_settings.anime_directory = str(anime_dir)
|
|
|
|
call_log = []
|
|
|
|
class MockDownloader:
|
|
"""Fake ImageDownloader that records calls."""
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *args):
|
|
return False
|
|
|
|
async def download_poster(self, url, folder, skip_existing=True):
|
|
call_log.append({"url": url, "folder": folder, "skip_existing": skip_existing})
|
|
return True
|
|
|
|
with patch(
|
|
"src.config.settings.settings", mock_settings
|
|
), patch(
|
|
"src.server.services.folder_scan_service.perform_nfo_repair_scan",
|
|
new_callable=AsyncMock,
|
|
), patch(
|
|
"src.server.services.folder_rename_service.validate_and_rename_series_folders",
|
|
new_callable=AsyncMock,
|
|
return_value={"scanned": 0, "renamed": 0, "skipped": 0, "errors": 0},
|
|
), patch(
|
|
"src.server.services.folder_scan_service.ImageDownloader",
|
|
new=MockDownloader,
|
|
):
|
|
service = FolderScanService()
|
|
await service.run_folder_scan()
|
|
|
|
assert len(call_log) == 1, f"Expected 1 download call, got {len(call_log)}"
|
|
assert call_log[0]["url"] == "https://example.com/poster.jpg"
|
|
assert call_log[0]["folder"] == series_dir
|
|
assert call_log[0]["skip_existing"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_poster_check_skips_valid_poster(self, tmp_path):
|
|
"""When poster.jpg exists and is large enough, the scan skips it."""
|
|
from src.server.services.folder_scan_service import FolderScanService
|
|
|
|
anime_dir = tmp_path / "anime"
|
|
anime_dir.mkdir()
|
|
series_dir = anime_dir / "Attack on Titan (2013)"
|
|
series_dir.mkdir()
|
|
(series_dir / "tvshow.nfo").write_text(
|
|
"<tvshow>"
|
|
"<title>Attack on Titan</title>"
|
|
"<year>2013</year>"
|
|
"<thumb aspect='poster'>https://example.com/poster.jpg</thumb>"
|
|
"</tvshow>"
|
|
)
|
|
# Create a valid poster.jpg (larger than 1 KB)
|
|
poster_path = series_dir / "poster.jpg"
|
|
poster_path.write_bytes(b"x" * 2048)
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.tmdb_api_key = "test-key"
|
|
mock_settings.anime_directory = str(anime_dir)
|
|
|
|
with patch(
|
|
"src.config.settings.settings", mock_settings
|
|
), patch(
|
|
"src.server.services.folder_scan_service.perform_nfo_repair_scan",
|
|
new_callable=AsyncMock,
|
|
), patch(
|
|
"src.server.services.folder_rename_service.validate_and_rename_series_folders",
|
|
new_callable=AsyncMock,
|
|
return_value={"scanned": 0, "renamed": 0, "skipped": 0, "errors": 0},
|
|
), patch(
|
|
"src.server.services.folder_scan_service.ImageDownloader"
|
|
) as mock_downloader_cls:
|
|
service = FolderScanService()
|
|
await service.run_folder_scan()
|
|
|
|
mock_downloader_cls.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_poster_check_skips_when_no_thumb_url(self, tmp_path):
|
|
"""When NFO has no thumb URL, the scan skips the folder."""
|
|
from src.server.services.folder_scan_service import FolderScanService
|
|
|
|
anime_dir = tmp_path / "anime"
|
|
anime_dir.mkdir()
|
|
series_dir = anime_dir / "Attack on Titan (2013)"
|
|
series_dir.mkdir()
|
|
(series_dir / "tvshow.nfo").write_text(
|
|
"<tvshow>"
|
|
"<title>Attack on Titan</title>"
|
|
"<year>2013</year>"
|
|
"</tvshow>"
|
|
)
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.tmdb_api_key = "test-key"
|
|
mock_settings.anime_directory = str(anime_dir)
|
|
|
|
with patch(
|
|
"src.config.settings.settings", mock_settings
|
|
), patch(
|
|
"src.server.services.folder_scan_service.perform_nfo_repair_scan",
|
|
new_callable=AsyncMock,
|
|
), patch(
|
|
"src.server.services.folder_rename_service.validate_and_rename_series_folders",
|
|
new_callable=AsyncMock,
|
|
return_value={"scanned": 0, "renamed": 0, "skipped": 0, "errors": 0},
|
|
), patch(
|
|
"src.server.services.folder_scan_service.ImageDownloader"
|
|
) as mock_downloader_cls:
|
|
service = FolderScanService()
|
|
await service.run_folder_scan()
|
|
|
|
mock_downloader_cls.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_poster_check_skipped_when_prerequisites_not_met(self, tmp_path):
|
|
"""If anime directory is missing, poster check logic is skipped gracefully."""
|
|
from src.server.services.folder_scan_service import FolderScanService
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.tmdb_api_key = "test-key"
|
|
mock_settings.anime_directory = str(tmp_path / "nonexistent")
|
|
|
|
with patch(
|
|
"src.config.settings.settings", mock_settings
|
|
), patch(
|
|
"src.server.services.folder_scan_service.perform_nfo_repair_scan",
|
|
new_callable=AsyncMock,
|
|
), patch(
|
|
"src.server.services.folder_rename_service.validate_and_rename_series_folders"
|
|
) as mock_rename, patch(
|
|
"src.server.services.folder_scan_service.ImageDownloader"
|
|
) as mock_downloader_cls:
|
|
service = FolderScanService()
|
|
await service.run_folder_scan()
|
|
|
|
mock_downloader_cls.assert_not_called()
|
|
|
|
|
|
class TestPosterCheckSemaphore:
|
|
"""Verify the poster download semaphore limits concurrency."""
|
|
|
|
def test_poster_download_semaphore_defined(self):
|
|
"""_POSTER_DOWNLOAD_SEMAPHORE must be defined in folder_scan_service.py."""
|
|
import importlib
|
|
|
|
source = importlib.util.find_spec(
|
|
"src.server.services.folder_scan_service"
|
|
).origin
|
|
with open(source, "r", encoding="utf-8") as fh:
|
|
content = fh.read()
|
|
|
|
assert "_POSTER_DOWNLOAD_SEMAPHORE" in content, (
|
|
"_POSTER_DOWNLOAD_SEMAPHORE must be defined in folder_scan_service.py"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_poster_download_uses_semaphore(self, tmp_path):
|
|
"""Poster downloads are gated by the semaphore."""
|
|
from src.server.services.folder_scan_service import (
|
|
_POSTER_DOWNLOAD_SEMAPHORE,
|
|
FolderScanService,
|
|
)
|
|
|
|
anime_dir = tmp_path / "anime"
|
|
anime_dir.mkdir()
|
|
|
|
# Create multiple series folders
|
|
for i in range(5):
|
|
series_dir = anime_dir / f"Series {i}"
|
|
series_dir.mkdir()
|
|
(series_dir / "tvshow.nfo").write_text(
|
|
f"<tvshow>"
|
|
f"<title>Series {i}</title>"
|
|
f"<year>202{i}</year>"
|
|
f"<thumb aspect='poster'>https://example.com/poster{i}.jpg</thumb>"
|
|
f"</tvshow>"
|
|
)
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.tmdb_api_key = "test-key"
|
|
mock_settings.anime_directory = str(anime_dir)
|
|
|
|
active_count = 0
|
|
max_active = 0
|
|
|
|
async def tracked_download(*args, **kwargs):
|
|
nonlocal active_count, max_active
|
|
active_count += 1
|
|
max_active = max(max_active, active_count)
|
|
await asyncio.sleep(0.05)
|
|
active_count -= 1
|
|
return True
|
|
|
|
with patch(
|
|
"src.config.settings.settings", mock_settings
|
|
), patch(
|
|
"src.server.services.folder_scan_service.perform_nfo_repair_scan",
|
|
new_callable=AsyncMock,
|
|
), patch(
|
|
"src.server.services.folder_rename_service.validate_and_rename_series_folders",
|
|
new_callable=AsyncMock,
|
|
return_value={"scanned": 0, "renamed": 0, "skipped": 0, "errors": 0},
|
|
), patch(
|
|
"src.server.services.folder_scan_service.ImageDownloader"
|
|
) as mock_downloader_cls:
|
|
mock_downloader = AsyncMock()
|
|
mock_downloader.download_poster = AsyncMock(side_effect=tracked_download)
|
|
mock_downloader_cls.return_value.__aenter__ = AsyncMock(
|
|
return_value=mock_downloader
|
|
)
|
|
mock_downloader_cls.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
service = FolderScanService()
|
|
await service.run_folder_scan()
|
|
|
|
assert max_active <= 3, (
|
|
f"Expected max concurrent downloads <= 3, got {max_active}"
|
|
)
|