feat(SerieScanner): DB lookup primary, deprecate key file fallback

DB now source of truth for folder -> Serie resolution.

Changes:
- AnimeSeriesService.get_by_folder(): new async lookup by folder name
- SerieScanner.__read_data_from_file(): query DB first, then provider callback, then legacy key file (temporary, removed v3.0.0)
- Serie: reconstruct from DB record with episode dict
- Key file: warn on use, scheduled removal v3.0.0

Add unit tests for DB hit/miss/callback/fallback edge cases

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
2026-05-26 17:56:37 +02:00
parent cbd53ef2a0
commit 841368bf85
3 changed files with 219 additions and 32 deletions

View File

@@ -0,0 +1,124 @@
"""Tests for SerieScanner DB lookup functionality."""
import logging
import os
import tempfile
from unittest.mock import MagicMock, patch
import pytest
from src.core.entities.series import Serie
from src.core.SerieScanner import SerieScanner
@pytest.fixture
def mock_loader():
"""Create a mock Loader instance."""
loader = MagicMock()
loader.get_season_episode_count = MagicMock(return_value={1: 12})
loader.is_language = MagicMock(return_value=True)
loader.get_year = MagicMock(return_value=2026)
return loader
@pytest.fixture
def temp_directory():
"""Create a temporary directory with subdirectories for testing."""
with tempfile.TemporaryDirectory() as tmpdir:
anime_folder = os.path.join(tmpdir, "Rooster Fighter (2026)")
os.makedirs(anime_folder, exist_ok=True)
mp4_path = os.path.join(anime_folder, "S01E001.mp4")
with open(mp4_path, "w") as f:
f.write("dummy mp4")
yield tmpdir
class TestGetSerieFromFolderDbLookup:
"""Test __read_data_from_file DB lookup behavior."""
def test_db_hit_returns_serie_from_db(self, temp_directory, mock_loader):
"""DB lookup resolves folder -> Serie returned."""
from src.server.database.models import AnimeSeries
from src.server.database import service as anime_series_service
mock_session = MagicMock()
mock_anime_series = MagicMock()
mock_anime_series.key = "rooster-fighter"
mock_anime_series.name = "Rooster Fighter"
mock_anime_series.site = "aniworld.to"
mock_anime_series.folder = "Rooster Fighter (2026)"
mock_anime_series.year = 2026
mock_anime_series.episodes = []
mock_session.execute.return_value.scalar_one_or_none.return_value = mock_anime_series
with patch("src.core.SerieScanner.get_sync_session", return_value=mock_session):
scanner = SerieScanner(temp_directory, mock_loader)
result = scanner._SerieScanner__read_data_from_file("Rooster Fighter (2026)")
assert result is not None
assert result.key == "rooster-fighter"
assert result.name == "Rooster Fighter"
assert result.year == 2026
def test_db_miss_falls_back_to_provider_callback(self, temp_directory, mock_loader):
"""DB miss -> _db_lookup callback called."""
lookup = MagicMock(return_value=Serie(
key="rooster-fighter",
name="Rooster Fighter",
site="aniworld.to",
folder="Rooster Fighter (2026)",
episodeDict={},
))
scanner = SerieScanner(temp_directory, mock_loader, db_lookup=lookup)
result = scanner._SerieScanner__read_data_from_file("Rooster Fighter (2026)")
assert result is not None
assert result.key == "rooster-fighter"
lookup.assert_called_once_with("Rooster Fighter (2026)")
def test_legacy_key_file_as_last_resort(self, temp_directory, mock_loader):
"""No DB, no callback -> legacy 'key' file used with deprecation warning."""
folder = os.path.join(temp_directory, "Legacy Series")
os.makedirs(folder, exist_ok=True)
with open(os.path.join(folder, "key"), "w") as f:
f.write("legacy-key")
scanner = SerieScanner(temp_directory, mock_loader)
with patch.object(logging.getLogger("src.core.SerieScanner"), "warning") as mock_warning:
result = scanner._SerieScanner__read_data_from_file("Legacy Series")
assert result is not None
assert result.key == "legacy-key"
mock_warning.assert_called()
warning_calls = [str(c) for c in mock_warning.call_args_list]
assert any("deprecated" in c or "v3.0.0" in c for c in warning_calls)
def test_db_lookup_exception_caught_and_logged(self, temp_directory, mock_loader):
"""DB exception -> fallback to provider callback."""
def bad_lookup(folder):
raise RuntimeError("DB connection failed")
scanner = SerieScanner(temp_directory, mock_loader, db_lookup=bad_lookup)
with patch.object(logging.getLogger("src.core.SerieScanner"), "warning") as mock_warning:
result = scanner._SerieScanner__read_data_from_file("Rooster Fighter (2026)")
mock_warning.assert_called()
assert any("DB lookup failed" in str(c) for c in mock_warning.call_args_list)
class TestGetSerieFromFolderEdgeCases:
"""Edge case tests for __read_data_from_file."""
def test_empty_folder_name_returns_none(self, temp_directory, mock_loader):
"""Empty folder name -> returns None (no DB lookup attempted)."""
scanner = SerieScanner(temp_directory, mock_loader)
result = scanner._SerieScanner__read_data_from_file("")
assert result is None
def test_nonexistent_folder_no_exception(self, temp_directory, mock_loader):
"""Folder doesn't exist -> returns None without raising."""
scanner = SerieScanner(temp_directory, mock_loader)
result = scanner._SerieScanner__read_data_from_file("Nonexistent Folder")
assert result is None