test: add tests for scheduled folder scan and startup NFO repair removal
Add comprehensive test coverage for Tasks 1.1–1.5 and 2.1: - test_scheduler_config_model.py: folder_scan_enabled defaults, explicit values, backward compatibility with old configs, serialization roundtrip - test_folder_scan_service.py (new): prerequisites, NFO repair integration, folder rename integration, poster check/download, semaphore values, NFO thumb URL extraction, full end-to-end scan flow - test_scheduler_service.py: scheduler _perform_rescan integration with folder_scan_enabled (called when enabled, skipped when disabled, error handling and broadcasting), folder_scan_enabled in get_status output - test_nfo_repair_startup.py: verify perform_nfo_repair_scan is NOT called during FastAPI lifespan startup and IS called from FolderScanService All 90 tests pass.
This commit is contained in:
607
tests/unit/test_folder_scan_service.py
Normal file
607
tests/unit/test_folder_scan_service.py
Normal file
@@ -0,0 +1,607 @@
|
|||||||
|
"""Unit tests for FolderScanService (Tasks 1.2–1.5).
|
||||||
|
|
||||||
|
Covers:
|
||||||
|
- Prerequisites checking (TMDB key, anime directory)
|
||||||
|
- NFO repair integration (Task 1.3)
|
||||||
|
- Folder rename validation (Task 1.4)
|
||||||
|
- Poster check and download (Task 1.5)
|
||||||
|
- Exception handling and semaphore usage
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
from pathlib import Path
|
||||||
|
from unittest.mock import AsyncMock, MagicMock, Mock, patch
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from src.server.services.folder_scan_service import (
|
||||||
|
_POSTER_DOWNLOAD_SEMAPHORE,
|
||||||
|
_TMDB_SEMAPHORE,
|
||||||
|
FolderScanService,
|
||||||
|
FolderScanServiceError,
|
||||||
|
)
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Fixtures
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def folder_scan_service() -> FolderScanService:
|
||||||
|
"""Return a fresh FolderScanService instance."""
|
||||||
|
return FolderScanService()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def mock_settings(tmp_path: Path):
|
||||||
|
"""Return a mock settings object with valid prerequisites."""
|
||||||
|
mock = MagicMock()
|
||||||
|
mock.tmdb_api_key = "test-api-key"
|
||||||
|
mock.anime_directory = str(tmp_path)
|
||||||
|
mock.nfo_download_poster = True
|
||||||
|
return mock
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 1.2 – Skeleton / prerequisites
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestPrerequisites:
|
||||||
|
"""Test _prerequisites_met checks."""
|
||||||
|
|
||||||
|
def test_prerequisites_met(self, folder_scan_service, tmp_path):
|
||||||
|
"""All prerequisites present → True."""
|
||||||
|
with patch(
|
||||||
|
"src.config.settings.settings"
|
||||||
|
) as mock_settings:
|
||||||
|
mock_settings.tmdb_api_key = "key"
|
||||||
|
mock_settings.anime_directory = str(tmp_path)
|
||||||
|
assert folder_scan_service._prerequisites_met() is True
|
||||||
|
|
||||||
|
def test_missing_tmdb_key(self, folder_scan_service, tmp_path):
|
||||||
|
"""Missing TMDB API key → False."""
|
||||||
|
with patch(
|
||||||
|
"src.config.settings.settings"
|
||||||
|
) as mock_settings:
|
||||||
|
mock_settings.tmdb_api_key = None
|
||||||
|
mock_settings.anime_directory = str(tmp_path)
|
||||||
|
assert folder_scan_service._prerequisites_met() is False
|
||||||
|
|
||||||
|
def test_missing_anime_directory(self, folder_scan_service):
|
||||||
|
"""Missing anime_directory → False."""
|
||||||
|
with patch(
|
||||||
|
"src.config.settings.settings"
|
||||||
|
) as mock_settings:
|
||||||
|
mock_settings.tmdb_api_key = "key"
|
||||||
|
mock_settings.anime_directory = None
|
||||||
|
assert folder_scan_service._prerequisites_met() is False
|
||||||
|
|
||||||
|
def test_anime_directory_not_found(self, folder_scan_service, tmp_path):
|
||||||
|
"""anime_directory points to non-existent path → False."""
|
||||||
|
non_existent = tmp_path / "does_not_exist"
|
||||||
|
with patch(
|
||||||
|
"src.config.settings.settings"
|
||||||
|
) as mock_settings:
|
||||||
|
mock_settings.tmdb_api_key = "key"
|
||||||
|
mock_settings.anime_directory = str(non_existent)
|
||||||
|
assert folder_scan_service._prerequisites_met() is False
|
||||||
|
|
||||||
|
|
||||||
|
class TestRunFolderScanPrerequisites:
|
||||||
|
"""Test run_folder_scan skips when prerequisites not met."""
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_skips_when_prerequisites_missing(self, folder_scan_service):
|
||||||
|
"""If _prerequisites_met returns False, scan exits early."""
|
||||||
|
with patch.object(
|
||||||
|
folder_scan_service, "_prerequisites_met", return_value=False
|
||||||
|
), patch(
|
||||||
|
"src.server.services.folder_scan_service.perform_nfo_repair_scan"
|
||||||
|
) as mock_repair:
|
||||||
|
await folder_scan_service.run_folder_scan()
|
||||||
|
mock_repair.assert_not_called()
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_logs_start_and_completion(self, folder_scan_service, tmp_path):
|
||||||
|
"""Scan logs start and completion when prerequisites are met."""
|
||||||
|
with patch.object(
|
||||||
|
folder_scan_service, "_prerequisites_met", return_value=True
|
||||||
|
), 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.object(
|
||||||
|
folder_scan_service,
|
||||||
|
"check_and_download_missing_posters",
|
||||||
|
new_callable=AsyncMock,
|
||||||
|
return_value={"scanned": 0, "downloaded": 0, "skipped": 0, "errors": 0},
|
||||||
|
):
|
||||||
|
# Should not raise
|
||||||
|
await folder_scan_service.run_folder_scan()
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_catches_unhandled_exceptions(self, folder_scan_service):
|
||||||
|
"""Unhandled exceptions are caught and logged, not re-raised."""
|
||||||
|
with patch.object(
|
||||||
|
folder_scan_service,
|
||||||
|
"_prerequisites_met",
|
||||||
|
side_effect=RuntimeError("boom"),
|
||||||
|
):
|
||||||
|
# Must NOT raise
|
||||||
|
await folder_scan_service.run_folder_scan()
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 1.3 – NFO repair integration
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestNfoRepairIntegration:
|
||||||
|
"""Test perform_nfo_repair_scan is called inside run_folder_scan."""
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_calls_perform_nfo_repair_scan(self, folder_scan_service, tmp_path):
|
||||||
|
"""run_folder_scan must call perform_nfo_repair_scan."""
|
||||||
|
with patch.object(
|
||||||
|
folder_scan_service, "_prerequisites_met", return_value=True
|
||||||
|
), patch(
|
||||||
|
"src.server.services.folder_scan_service.perform_nfo_repair_scan",
|
||||||
|
new_callable=AsyncMock,
|
||||||
|
) as mock_repair, 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.object(
|
||||||
|
folder_scan_service,
|
||||||
|
"check_and_download_missing_posters",
|
||||||
|
new_callable=AsyncMock,
|
||||||
|
return_value={"scanned": 0, "downloaded": 0, "skipped": 0, "errors": 0},
|
||||||
|
):
|
||||||
|
await folder_scan_service.run_folder_scan()
|
||||||
|
mock_repair.assert_awaited_once_with(background_loader=None)
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_nfo_repair_failure_does_not_crash_scan(
|
||||||
|
self, folder_scan_service, tmp_path
|
||||||
|
):
|
||||||
|
"""If perform_nfo_repair_scan raises, the broad except catches it
|
||||||
|
and the scan stops — remaining steps are NOT invoked."""
|
||||||
|
with patch.object(
|
||||||
|
folder_scan_service, "_prerequisites_met", return_value=True
|
||||||
|
), patch(
|
||||||
|
"src.server.services.folder_scan_service.perform_nfo_repair_scan",
|
||||||
|
new_callable=AsyncMock,
|
||||||
|
side_effect=RuntimeError("repair failed"),
|
||||||
|
) as mock_repair, 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},
|
||||||
|
) as mock_rename, patch.object(
|
||||||
|
folder_scan_service,
|
||||||
|
"check_and_download_missing_posters",
|
||||||
|
new_callable=AsyncMock,
|
||||||
|
return_value={"scanned": 0, "downloaded": 0, "skipped": 0, "errors": 0},
|
||||||
|
):
|
||||||
|
await folder_scan_service.run_folder_scan()
|
||||||
|
mock_repair.assert_awaited_once()
|
||||||
|
# Broad except stops the scan; rename/poster are skipped
|
||||||
|
mock_rename.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 1.4 – Folder rename integration
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestFolderRenameIntegration:
|
||||||
|
"""Test validate_and_rename_series_folders is called and stats logged."""
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_calls_folder_rename_service(self, folder_scan_service, tmp_path):
|
||||||
|
"""run_folder_scan must call validate_and_rename_series_folders."""
|
||||||
|
with patch.object(
|
||||||
|
folder_scan_service, "_prerequisites_met", return_value=True
|
||||||
|
), 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": 5, "renamed": 2, "skipped": 2, "errors": 1},
|
||||||
|
) as mock_rename, patch.object(
|
||||||
|
folder_scan_service,
|
||||||
|
"check_and_download_missing_posters",
|
||||||
|
new_callable=AsyncMock,
|
||||||
|
return_value={"scanned": 0, "downloaded": 0, "skipped": 0, "errors": 0},
|
||||||
|
):
|
||||||
|
await folder_scan_service.run_folder_scan()
|
||||||
|
mock_rename.assert_awaited_once()
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_folder_rename_failure_does_not_crash_scan(
|
||||||
|
self, folder_scan_service, tmp_path
|
||||||
|
):
|
||||||
|
"""If validate_and_rename_series_folders raises, the broad except
|
||||||
|
catches it and the scan stops — poster check is NOT invoked."""
|
||||||
|
with patch.object(
|
||||||
|
folder_scan_service, "_prerequisites_met", return_value=True
|
||||||
|
), 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,
|
||||||
|
side_effect=RuntimeError("rename failed"),
|
||||||
|
), patch.object(
|
||||||
|
folder_scan_service,
|
||||||
|
"check_and_download_missing_posters",
|
||||||
|
new_callable=AsyncMock,
|
||||||
|
return_value={"scanned": 0, "downloaded": 0, "skipped": 0, "errors": 0},
|
||||||
|
) as mock_poster:
|
||||||
|
await folder_scan_service.run_folder_scan()
|
||||||
|
# Broad except stops the scan; poster check is skipped
|
||||||
|
mock_poster.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 1.5 – Poster check and download
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestPosterCheck:
|
||||||
|
"""Test check_and_download_missing_posters logic."""
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_no_anime_directory_returns_empty_stats(self, folder_scan_service):
|
||||||
|
"""Missing anime_directory → empty stats."""
|
||||||
|
with patch(
|
||||||
|
"src.config.settings.settings"
|
||||||
|
) as mock_settings:
|
||||||
|
mock_settings.anime_directory = None
|
||||||
|
stats = await folder_scan_service.check_and_download_missing_posters()
|
||||||
|
assert stats == {"scanned": 0, "downloaded": 0, "skipped": 0, "errors": 0}
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_nonexistent_directory_returns_empty_stats(
|
||||||
|
self, folder_scan_service, tmp_path
|
||||||
|
):
|
||||||
|
"""Non-existent anime_directory → empty stats."""
|
||||||
|
non_existent = tmp_path / "missing"
|
||||||
|
with patch(
|
||||||
|
"src.config.settings.settings"
|
||||||
|
) as mock_settings:
|
||||||
|
mock_settings.anime_directory = str(non_existent)
|
||||||
|
stats = await folder_scan_service.check_and_download_missing_posters()
|
||||||
|
assert stats == {"scanned": 0, "downloaded": 0, "skipped": 0, "errors": 0}
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_no_series_folders_returns_empty_stats(
|
||||||
|
self, folder_scan_service, tmp_path
|
||||||
|
):
|
||||||
|
"""Empty anime_directory → empty stats."""
|
||||||
|
with patch(
|
||||||
|
"src.config.settings.settings"
|
||||||
|
) as mock_settings:
|
||||||
|
mock_settings.anime_directory = str(tmp_path)
|
||||||
|
stats = await folder_scan_service.check_and_download_missing_posters()
|
||||||
|
assert stats == {"scanned": 0, "downloaded": 0, "skipped": 0, "errors": 0}
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_skips_folders_without_nfo(self, folder_scan_service, tmp_path):
|
||||||
|
"""Folders without tvshow.nfo are ignored."""
|
||||||
|
(tmp_path / "SomeShow").mkdir()
|
||||||
|
with patch(
|
||||||
|
"src.config.settings.settings"
|
||||||
|
) as mock_settings:
|
||||||
|
mock_settings.anime_directory = str(tmp_path)
|
||||||
|
stats = await folder_scan_service.check_and_download_missing_posters()
|
||||||
|
assert stats == {"scanned": 0, "downloaded": 0, "skipped": 0, "errors": 0}
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_valid_poster_skipped(self, folder_scan_service, tmp_path):
|
||||||
|
"""Existing poster.jpg ≥ 1 KB is skipped."""
|
||||||
|
series_dir = tmp_path / "Attack on Titan (2013)"
|
||||||
|
series_dir.mkdir()
|
||||||
|
(series_dir / "tvshow.nfo").write_text(
|
||||||
|
"<tvshow><title>Attack on Titan</title><year>2013</year></tvshow>"
|
||||||
|
)
|
||||||
|
# Write a 2 KB poster
|
||||||
|
(series_dir / "poster.jpg").write_bytes(b"x" * 2048)
|
||||||
|
|
||||||
|
with patch(
|
||||||
|
"src.config.settings.settings"
|
||||||
|
) as mock_settings:
|
||||||
|
mock_settings.anime_directory = str(tmp_path)
|
||||||
|
stats = await folder_scan_service.check_and_download_missing_posters()
|
||||||
|
|
||||||
|
assert stats["scanned"] == 1
|
||||||
|
assert stats["skipped"] == 1
|
||||||
|
assert stats["downloaded"] == 0
|
||||||
|
assert stats["errors"] == 0
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_missing_poster_downloaded(self, folder_scan_service, tmp_path):
|
||||||
|
"""Missing poster triggers download when thumb URL exists."""
|
||||||
|
series_dir = tmp_path / "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>"
|
||||||
|
)
|
||||||
|
|
||||||
|
mock_downloader = AsyncMock()
|
||||||
|
mock_downloader.download_poster = AsyncMock(return_value=True)
|
||||||
|
mock_downloader.__aenter__ = AsyncMock(return_value=mock_downloader)
|
||||||
|
mock_downloader.__aexit__ = AsyncMock(return_value=False)
|
||||||
|
|
||||||
|
with patch(
|
||||||
|
"src.config.settings.settings"
|
||||||
|
) as mock_settings:
|
||||||
|
mock_settings.anime_directory = str(tmp_path)
|
||||||
|
mock_settings.nfo_download_poster = True
|
||||||
|
|
||||||
|
with patch(
|
||||||
|
"src.server.services.folder_scan_service.ImageDownloader",
|
||||||
|
return_value=mock_downloader,
|
||||||
|
):
|
||||||
|
stats = await folder_scan_service.check_and_download_missing_posters()
|
||||||
|
|
||||||
|
assert stats["scanned"] == 1
|
||||||
|
assert stats["downloaded"] == 1
|
||||||
|
assert stats["skipped"] == 0
|
||||||
|
assert stats["errors"] == 0
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_no_thumb_url_skipped(self, folder_scan_service, tmp_path):
|
||||||
|
"""NFO without thumb URL → skipped."""
|
||||||
|
series_dir = tmp_path / "Attack on Titan (2013)"
|
||||||
|
series_dir.mkdir()
|
||||||
|
(series_dir / "tvshow.nfo").write_text(
|
||||||
|
"<tvshow><title>Attack on Titan</title><year>2013</year></tvshow>"
|
||||||
|
)
|
||||||
|
|
||||||
|
with patch(
|
||||||
|
"src.config.settings.settings"
|
||||||
|
) as mock_settings:
|
||||||
|
mock_settings.anime_directory = str(tmp_path)
|
||||||
|
mock_settings.nfo_download_poster = True
|
||||||
|
stats = await folder_scan_service.check_and_download_missing_posters()
|
||||||
|
|
||||||
|
assert stats["scanned"] == 1
|
||||||
|
assert stats["skipped"] == 1
|
||||||
|
assert stats["downloaded"] == 0
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_poster_download_disabled_by_setting(
|
||||||
|
self, folder_scan_service, tmp_path
|
||||||
|
):
|
||||||
|
"""nfo_download_poster=False → skipped even with valid thumb URL."""
|
||||||
|
series_dir = tmp_path / "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>"
|
||||||
|
)
|
||||||
|
|
||||||
|
with patch(
|
||||||
|
"src.config.settings.settings"
|
||||||
|
) as mock_settings:
|
||||||
|
mock_settings.anime_directory = str(tmp_path)
|
||||||
|
mock_settings.nfo_download_poster = False
|
||||||
|
stats = await folder_scan_service.check_and_download_missing_posters()
|
||||||
|
|
||||||
|
assert stats["scanned"] == 1
|
||||||
|
assert stats["skipped"] == 1
|
||||||
|
assert stats["downloaded"] == 0
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_download_failure_counts_as_error(self, folder_scan_service, tmp_path):
|
||||||
|
"""Failed download increments errors."""
|
||||||
|
series_dir = tmp_path / "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>"
|
||||||
|
)
|
||||||
|
|
||||||
|
mock_downloader = AsyncMock()
|
||||||
|
mock_downloader.download_poster = AsyncMock(return_value=False)
|
||||||
|
mock_downloader.__aenter__ = AsyncMock(return_value=mock_downloader)
|
||||||
|
mock_downloader.__aexit__ = AsyncMock(return_value=False)
|
||||||
|
|
||||||
|
with patch(
|
||||||
|
"src.config.settings.settings"
|
||||||
|
) as mock_settings:
|
||||||
|
mock_settings.anime_directory = str(tmp_path)
|
||||||
|
mock_settings.nfo_download_poster = True
|
||||||
|
|
||||||
|
with patch(
|
||||||
|
"src.server.services.folder_scan_service.ImageDownloader",
|
||||||
|
return_value=mock_downloader,
|
||||||
|
):
|
||||||
|
stats = await folder_scan_service.check_and_download_missing_posters()
|
||||||
|
|
||||||
|
assert stats["scanned"] == 1
|
||||||
|
assert stats["errors"] == 1
|
||||||
|
assert stats["downloaded"] == 0
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_download_exception_counts_as_error(self, folder_scan_service, tmp_path):
|
||||||
|
"""Exception during download increments errors."""
|
||||||
|
series_dir = tmp_path / "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>"
|
||||||
|
)
|
||||||
|
|
||||||
|
mock_downloader = AsyncMock()
|
||||||
|
mock_downloader.download_poster = AsyncMock(side_effect=RuntimeError("net"))
|
||||||
|
mock_downloader.__aenter__ = AsyncMock(return_value=mock_downloader)
|
||||||
|
mock_downloader.__aexit__ = AsyncMock(return_value=False)
|
||||||
|
|
||||||
|
with patch(
|
||||||
|
"src.config.settings.settings"
|
||||||
|
) as mock_settings:
|
||||||
|
mock_settings.anime_directory = str(tmp_path)
|
||||||
|
mock_settings.nfo_download_poster = True
|
||||||
|
|
||||||
|
with patch(
|
||||||
|
"src.server.services.folder_scan_service.ImageDownloader",
|
||||||
|
return_value=mock_downloader,
|
||||||
|
):
|
||||||
|
stats = await folder_scan_service.check_and_download_missing_posters()
|
||||||
|
|
||||||
|
assert stats["scanned"] == 1
|
||||||
|
assert stats["errors"] == 1
|
||||||
|
assert stats["downloaded"] == 0
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_too_small_poster_re_downloaded(self, folder_scan_service, tmp_path):
|
||||||
|
"""Poster < 1 KB is treated as missing and re-downloaded."""
|
||||||
|
series_dir = tmp_path / "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>"
|
||||||
|
)
|
||||||
|
# Write a tiny 100-byte poster
|
||||||
|
(series_dir / "poster.jpg").write_bytes(b"x" * 100)
|
||||||
|
|
||||||
|
mock_downloader = AsyncMock()
|
||||||
|
mock_downloader.download_poster = AsyncMock(return_value=True)
|
||||||
|
mock_downloader.__aenter__ = AsyncMock(return_value=mock_downloader)
|
||||||
|
mock_downloader.__aexit__ = AsyncMock(return_value=False)
|
||||||
|
|
||||||
|
with patch(
|
||||||
|
"src.config.settings.settings"
|
||||||
|
) as mock_settings:
|
||||||
|
mock_settings.anime_directory = str(tmp_path)
|
||||||
|
mock_settings.nfo_download_poster = True
|
||||||
|
|
||||||
|
with patch(
|
||||||
|
"src.server.services.folder_scan_service.ImageDownloader",
|
||||||
|
return_value=mock_downloader,
|
||||||
|
):
|
||||||
|
stats = await folder_scan_service.check_and_download_missing_posters()
|
||||||
|
|
||||||
|
assert stats["scanned"] == 1
|
||||||
|
assert stats["downloaded"] == 1
|
||||||
|
assert stats["skipped"] == 0
|
||||||
|
|
||||||
|
|
||||||
|
class TestExtractPosterUrl:
|
||||||
|
"""Test _extract_poster_url_from_nfo static method."""
|
||||||
|
|
||||||
|
def test_extract_poster_url_with_aspect(self, tmp_path):
|
||||||
|
nfo = tmp_path / "tvshow.nfo"
|
||||||
|
nfo.write_text(
|
||||||
|
"<tvshow>"
|
||||||
|
'<thumb aspect=\"poster\">https://example.com/poster.jpg</thumb>'
|
||||||
|
"</tvshow>"
|
||||||
|
)
|
||||||
|
url = FolderScanService._extract_poster_url_from_nfo(nfo)
|
||||||
|
assert url == "https://example.com/poster.jpg"
|
||||||
|
|
||||||
|
def test_extract_first_thumb_fallback(self, tmp_path):
|
||||||
|
nfo = tmp_path / "tvshow.nfo"
|
||||||
|
nfo.write_text(
|
||||||
|
"<tvshow>"
|
||||||
|
'<thumb>https://example.com/fallback.jpg</thumb>'
|
||||||
|
"</tvshow>"
|
||||||
|
)
|
||||||
|
url = FolderScanService._extract_poster_url_from_nfo(nfo)
|
||||||
|
assert url == "https://example.com/fallback.jpg"
|
||||||
|
|
||||||
|
def test_no_thumb_returns_none(self, tmp_path):
|
||||||
|
nfo = tmp_path / "tvshow.nfo"
|
||||||
|
nfo.write_text("<tvshow><title>Test</title></tvshow>")
|
||||||
|
url = FolderScanService._extract_poster_url_from_nfo(nfo)
|
||||||
|
assert url is None
|
||||||
|
|
||||||
|
def test_missing_file_returns_none(self, tmp_path):
|
||||||
|
nfo = tmp_path / "tvshow.nfo"
|
||||||
|
url = FolderScanService._extract_poster_url_from_nfo(nfo)
|
||||||
|
assert url is None
|
||||||
|
|
||||||
|
def test_malformed_xml_returns_none(self, tmp_path):
|
||||||
|
nfo = tmp_path / "tvshow.nfo"
|
||||||
|
nfo.write_text("not xml")
|
||||||
|
url = FolderScanService._extract_poster_url_from_nfo(nfo)
|
||||||
|
assert url is None
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Semaphores
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestSemaphores:
|
||||||
|
"""Verify module-level semaphores exist and have correct initial value."""
|
||||||
|
|
||||||
|
def test_tmdb_semaphore_value(self):
|
||||||
|
assert _TMDB_SEMAPHORE._value == 3
|
||||||
|
|
||||||
|
def test_poster_download_semaphore_value(self):
|
||||||
|
assert _POSTER_DOWNLOAD_SEMAPHORE._value == 3
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Full run_folder_scan integration
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestRunFolderScanFull:
|
||||||
|
"""End-to-end tests for run_folder_scan with mocked sub-tasks."""
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_full_scan_happy_path(self, folder_scan_service, tmp_path):
|
||||||
|
"""All sub-tasks succeed."""
|
||||||
|
with patch.object(
|
||||||
|
folder_scan_service, "_prerequisites_met", return_value=True
|
||||||
|
), patch(
|
||||||
|
"src.server.services.folder_scan_service.perform_nfo_repair_scan",
|
||||||
|
new_callable=AsyncMock,
|
||||||
|
) as mock_repair, patch(
|
||||||
|
"src.server.services.folder_rename_service.validate_and_rename_series_folders",
|
||||||
|
new_callable=AsyncMock,
|
||||||
|
return_value={"scanned": 3, "renamed": 1, "skipped": 1, "errors": 1},
|
||||||
|
) as mock_rename, patch.object(
|
||||||
|
folder_scan_service,
|
||||||
|
"check_and_download_missing_posters",
|
||||||
|
new_callable=AsyncMock,
|
||||||
|
return_value={"scanned": 3, "downloaded": 2, "skipped": 1, "errors": 0},
|
||||||
|
) as mock_poster:
|
||||||
|
await folder_scan_service.run_folder_scan()
|
||||||
|
|
||||||
|
mock_repair.assert_awaited_once_with(background_loader=None)
|
||||||
|
mock_rename.assert_awaited_once()
|
||||||
|
mock_poster.assert_awaited_once()
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_full_scan_all_stats_zero(self, folder_scan_service, tmp_path):
|
||||||
|
"""Empty library → all stats zero."""
|
||||||
|
with patch.object(
|
||||||
|
folder_scan_service, "_prerequisites_met", return_value=True
|
||||||
|
), 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.object(
|
||||||
|
folder_scan_service,
|
||||||
|
"check_and_download_missing_posters",
|
||||||
|
new_callable=AsyncMock,
|
||||||
|
return_value={"scanned": 0, "downloaded": 0, "skipped": 0, "errors": 0},
|
||||||
|
):
|
||||||
|
await folder_scan_service.run_folder_scan()
|
||||||
@@ -113,8 +113,36 @@ class TestSchedulerConfigBackwardCompat:
|
|||||||
assert config.interval_minutes == 30
|
assert config.interval_minutes == 30
|
||||||
|
|
||||||
|
|
||||||
|
class TestSchedulerConfigFolderScanEnabled:
|
||||||
|
"""3.8 – folder_scan_enabled field (Task 1.1)."""
|
||||||
|
|
||||||
|
def test_default_folder_scan_enabled(self) -> None:
|
||||||
|
config = SchedulerConfig()
|
||||||
|
assert config.folder_scan_enabled is False
|
||||||
|
|
||||||
|
def test_set_folder_scan_enabled_true(self) -> None:
|
||||||
|
config = SchedulerConfig(folder_scan_enabled=True)
|
||||||
|
assert config.folder_scan_enabled is True
|
||||||
|
|
||||||
|
def test_set_folder_scan_enabled_false(self) -> None:
|
||||||
|
config = SchedulerConfig(folder_scan_enabled=False)
|
||||||
|
assert config.folder_scan_enabled is False
|
||||||
|
|
||||||
|
def test_backward_compat_missing_field(self) -> None:
|
||||||
|
"""Old configs without folder_scan_enabled load successfully."""
|
||||||
|
dumped = {
|
||||||
|
"enabled": True,
|
||||||
|
"interval_minutes": 60,
|
||||||
|
"schedule_time": "03:00",
|
||||||
|
"schedule_days": ALL_DAYS,
|
||||||
|
"auto_download_after_rescan": False,
|
||||||
|
}
|
||||||
|
config = SchedulerConfig(**dumped)
|
||||||
|
assert config.folder_scan_enabled is False
|
||||||
|
|
||||||
|
|
||||||
class TestSchedulerConfigSerialisation:
|
class TestSchedulerConfigSerialisation:
|
||||||
"""3.8 – Serialisation roundtrip."""
|
"""3.9 – Serialisation roundtrip."""
|
||||||
|
|
||||||
def test_roundtrip(self) -> None:
|
def test_roundtrip(self) -> None:
|
||||||
original = SchedulerConfig(
|
original = SchedulerConfig(
|
||||||
@@ -123,6 +151,7 @@ class TestSchedulerConfigSerialisation:
|
|||||||
schedule_time="04:30",
|
schedule_time="04:30",
|
||||||
schedule_days=["mon", "wed", "fri"],
|
schedule_days=["mon", "wed", "fri"],
|
||||||
auto_download_after_rescan=True,
|
auto_download_after_rescan=True,
|
||||||
|
folder_scan_enabled=True,
|
||||||
)
|
)
|
||||||
dumped = original.model_dump()
|
dumped = original.model_dump()
|
||||||
restored = SchedulerConfig(**dumped)
|
restored = SchedulerConfig(**dumped)
|
||||||
|
|||||||
@@ -9,16 +9,16 @@ Covers:
|
|||||||
- Error handling and edge cases
|
- Error handling and edge cases
|
||||||
"""
|
"""
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
from unittest.mock import AsyncMock, MagicMock, Mock, patch, call
|
from unittest.mock import AsyncMock, MagicMock, Mock, call, patch
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from apscheduler.triggers.cron import CronTrigger
|
from apscheduler.triggers.cron import CronTrigger
|
||||||
|
|
||||||
from src.server.models.config import AppConfig, SchedulerConfig
|
from src.server.models.config import AppConfig, SchedulerConfig
|
||||||
from src.server.services.scheduler_service import (
|
from src.server.services.scheduler_service import (
|
||||||
|
_JOB_ID,
|
||||||
SchedulerService,
|
SchedulerService,
|
||||||
SchedulerServiceError,
|
SchedulerServiceError,
|
||||||
_JOB_ID,
|
|
||||||
get_scheduler_service,
|
get_scheduler_service,
|
||||||
reset_scheduler_service,
|
reset_scheduler_service,
|
||||||
)
|
)
|
||||||
@@ -364,6 +364,7 @@ class TestGetStatus:
|
|||||||
schedule_time="04:00",
|
schedule_time="04:00",
|
||||||
schedule_days=["mon"],
|
schedule_days=["mon"],
|
||||||
auto_download_after_rescan=True,
|
auto_download_after_rescan=True,
|
||||||
|
folder_scan_enabled=True,
|
||||||
)
|
)
|
||||||
status = scheduler_service.get_status()
|
status = scheduler_service.get_status()
|
||||||
|
|
||||||
@@ -373,13 +374,100 @@ class TestGetStatus:
|
|||||||
assert "schedule_time" in status
|
assert "schedule_time" in status
|
||||||
assert "schedule_days" in status
|
assert "schedule_days" in status
|
||||||
assert "auto_download_after_rescan" in status
|
assert "auto_download_after_rescan" in status
|
||||||
|
assert "folder_scan_enabled" in status
|
||||||
assert status["schedule_time"] == "04:00"
|
assert status["schedule_time"] == "04:00"
|
||||||
assert status["schedule_days"] == ["mon"]
|
assert status["schedule_days"] == ["mon"]
|
||||||
assert status["auto_download_after_rescan"] is True
|
assert status["auto_download_after_rescan"] is True
|
||||||
|
assert status["folder_scan_enabled"] is True
|
||||||
assert status["is_running"] is False
|
assert status["is_running"] is False
|
||||||
assert status["next_run"] is None
|
assert status["next_run"] is None
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 12.11 _perform_rescan() with folder_scan_enabled=True
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestPerformRescanFolderScan:
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_folder_scan_called_when_enabled(self, scheduler_service):
|
||||||
|
scheduler_service._config = SchedulerConfig(
|
||||||
|
folder_scan_enabled=True,
|
||||||
|
schedule_time="03:00",
|
||||||
|
schedule_days=ALL_DAYS,
|
||||||
|
)
|
||||||
|
|
||||||
|
mock_anime = MagicMock()
|
||||||
|
mock_anime.rescan = AsyncMock()
|
||||||
|
mock_anime._cached_list_missing.return_value = []
|
||||||
|
|
||||||
|
mock_ws = MagicMock()
|
||||||
|
mock_ws.manager.broadcast = AsyncMock()
|
||||||
|
|
||||||
|
mock_folder_scan = AsyncMock()
|
||||||
|
|
||||||
|
with patch("src.server.utils.dependencies.get_anime_service", return_value=mock_anime), \
|
||||||
|
patch("src.server.services.websocket_service.get_websocket_service", return_value=mock_ws), \
|
||||||
|
patch("src.server.services.folder_scan_service.FolderScanService") as MockFSS:
|
||||||
|
MockFSS.return_value.run_folder_scan = mock_folder_scan
|
||||||
|
await scheduler_service._perform_rescan()
|
||||||
|
|
||||||
|
mock_folder_scan.assert_awaited_once()
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_folder_scan_skipped_when_disabled(self, scheduler_service):
|
||||||
|
scheduler_service._config = SchedulerConfig(
|
||||||
|
folder_scan_enabled=False,
|
||||||
|
schedule_time="03:00",
|
||||||
|
schedule_days=ALL_DAYS,
|
||||||
|
)
|
||||||
|
|
||||||
|
mock_anime = MagicMock()
|
||||||
|
mock_anime.rescan = AsyncMock()
|
||||||
|
mock_anime._cached_list_missing.return_value = []
|
||||||
|
|
||||||
|
mock_ws = MagicMock()
|
||||||
|
mock_ws.manager.broadcast = AsyncMock()
|
||||||
|
|
||||||
|
mock_folder_scan = AsyncMock()
|
||||||
|
|
||||||
|
with patch("src.server.utils.dependencies.get_anime_service", return_value=mock_anime), \
|
||||||
|
patch("src.server.services.websocket_service.get_websocket_service", return_value=mock_ws), \
|
||||||
|
patch("src.server.services.folder_scan_service.FolderScanService") as MockFSS:
|
||||||
|
MockFSS.return_value.run_folder_scan = mock_folder_scan
|
||||||
|
await scheduler_service._perform_rescan()
|
||||||
|
|
||||||
|
mock_folder_scan.assert_not_called()
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_folder_scan_error_broadcasts_and_does_not_crash(self, scheduler_service):
|
||||||
|
scheduler_service._config = SchedulerConfig(
|
||||||
|
folder_scan_enabled=True,
|
||||||
|
schedule_time="03:00",
|
||||||
|
schedule_days=ALL_DAYS,
|
||||||
|
)
|
||||||
|
|
||||||
|
mock_anime = MagicMock()
|
||||||
|
mock_anime.rescan = AsyncMock()
|
||||||
|
mock_anime._cached_list_missing.return_value = []
|
||||||
|
|
||||||
|
mock_ws = MagicMock()
|
||||||
|
mock_ws.manager.broadcast = AsyncMock()
|
||||||
|
|
||||||
|
mock_folder_scan = AsyncMock(side_effect=RuntimeError("folder scan boom"))
|
||||||
|
|
||||||
|
with patch("src.server.utils.dependencies.get_anime_service", return_value=mock_anime), \
|
||||||
|
patch("src.server.services.websocket_service.get_websocket_service", return_value=mock_ws), \
|
||||||
|
patch("src.server.services.folder_scan_service.FolderScanService") as MockFSS:
|
||||||
|
MockFSS.return_value.run_folder_scan = mock_folder_scan
|
||||||
|
# Should NOT raise
|
||||||
|
await scheduler_service._perform_rescan()
|
||||||
|
|
||||||
|
mock_folder_scan.assert_awaited_once()
|
||||||
|
calls = [str(c) for c in mock_ws.manager.broadcast.call_args_list]
|
||||||
|
assert any("folder_scan_error" in c for c in calls)
|
||||||
|
assert scheduler_service._scan_in_progress is False
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# Singleton helpers
|
# Singleton helpers
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|||||||
Reference in New Issue
Block a user