feat(scanner): add DB fallback for series key resolution
When SerieScanner encounters a folder without a local key or data file, it now optionally falls back to a database lookup by folder name. This prevents newly-added series from being silently skipped on rescan when their metadata only lives in the DB. Changes: - SerieScanner accepts an optional db_lookup callable - SeriesApp forwards db_lookup to SerieScanner - AnimeSeriesService adds get_by_folder_sync() helper - dependencies.py wires a sync DB lookup into get_series_app() - Unit tests cover fallback hit, miss, and exception paths
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
"""Tests for SerieScanner class - file-based operations."""
|
||||
|
||||
import logging
|
||||
import os
|
||||
import tempfile
|
||||
from unittest.mock import MagicMock, patch
|
||||
@@ -651,4 +652,187 @@ class TestScanProgressEvents:
|
||||
|
||||
error_handler.assert_called_once()
|
||||
call_data = error_handler.call_args[0][0]
|
||||
assert call_data["recoverable"] is True
|
||||
assert call_data["recoverable"] is True
|
||||
|
||||
|
||||
class TestDbLookupFallback:
|
||||
"""Tests for the db_lookup callback in SerieScanner."""
|
||||
|
||||
def _make_scanner(self, tmp_dir, mock_loader, db_lookup=None):
|
||||
"""Create a scanner with an optional db_lookup."""
|
||||
# Create a folder with an mp4 but NO key/data file
|
||||
folder = os.path.join(tmp_dir, "Rooster Fighter (2026)")
|
||||
os.makedirs(folder, exist_ok=True)
|
||||
mp4 = os.path.join(folder, "Rooster Fighter - S01E001 - (German Dub).mp4")
|
||||
with open(mp4, "w") as f:
|
||||
f.write("dummy")
|
||||
return SerieScanner(tmp_dir, mock_loader, db_lookup=db_lookup)
|
||||
|
||||
def test_db_lookup_stored_on_init(self, temp_directory, mock_loader):
|
||||
"""db_lookup callable should be stored as _db_lookup."""
|
||||
lookup = MagicMock(return_value=None)
|
||||
scanner = SerieScanner(temp_directory, mock_loader, db_lookup=lookup)
|
||||
assert scanner._db_lookup is lookup
|
||||
|
||||
def test_no_db_lookup_defaults_to_none(self, temp_directory, mock_loader):
|
||||
"""Without db_lookup, _db_lookup should be None."""
|
||||
scanner = SerieScanner(temp_directory, mock_loader)
|
||||
assert scanner._db_lookup is None
|
||||
|
||||
def test_db_lookup_called_when_no_files(self, mock_loader):
|
||||
"""db_lookup is called when neither key nor data file exists."""
|
||||
import tempfile
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
lookup = MagicMock(return_value=None)
|
||||
scanner = self._make_scanner(tmp_dir, mock_loader, db_lookup=lookup)
|
||||
|
||||
with patch.object(scanner, 'get_total_to_scan', return_value=1), \
|
||||
patch.object(
|
||||
scanner,
|
||||
'_SerieScanner__get_missing_episodes_and_season',
|
||||
return_value=({}, "aniworld.to"),
|
||||
):
|
||||
scanner.scan()
|
||||
|
||||
lookup.assert_called_once_with("Rooster Fighter (2026)")
|
||||
|
||||
def test_db_lookup_not_called_when_key_file_exists(self, mock_loader):
|
||||
"""db_lookup is NOT called when a key file is present."""
|
||||
import tempfile
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
folder = os.path.join(tmp_dir, "Rooster Fighter (2026)")
|
||||
os.makedirs(folder, exist_ok=True)
|
||||
mp4 = os.path.join(folder, "S01E001.mp4")
|
||||
with open(mp4, "w") as f:
|
||||
f.write("dummy")
|
||||
with open(os.path.join(folder, "key"), "w") as f:
|
||||
f.write("rooster-fighter")
|
||||
|
||||
lookup = MagicMock(return_value=None)
|
||||
scanner = SerieScanner(tmp_dir, mock_loader, db_lookup=lookup)
|
||||
|
||||
with patch.object(scanner, 'get_total_to_scan', return_value=1), \
|
||||
patch.object(
|
||||
scanner,
|
||||
'_SerieScanner__get_missing_episodes_and_season',
|
||||
return_value=({1: []}, "aniworld.to"),
|
||||
), \
|
||||
patch.object(
|
||||
SerieScanner,
|
||||
'_SerieScanner__read_data_from_file',
|
||||
return_value=Serie(
|
||||
key="rooster-fighter", name="", site="aniworld.to",
|
||||
folder="Rooster Fighter (2026)", episodeDict={},
|
||||
),
|
||||
):
|
||||
scanner.scan()
|
||||
|
||||
lookup.assert_not_called()
|
||||
|
||||
def test_db_lookup_resolves_serie_and_scans(self, mock_loader):
|
||||
"""When db_lookup returns a Serie, scanning continues normally."""
|
||||
import tempfile
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
resolved = Serie(
|
||||
key="rooster-fighter",
|
||||
name="Rooster Fighter",
|
||||
site="aniworld.to",
|
||||
folder="Rooster Fighter (2026)",
|
||||
episodeDict={},
|
||||
year=2026,
|
||||
)
|
||||
lookup = MagicMock(return_value=resolved)
|
||||
scanner = self._make_scanner(tmp_dir, mock_loader, db_lookup=lookup)
|
||||
|
||||
with patch.object(scanner, 'get_total_to_scan', return_value=1), \
|
||||
patch.object(
|
||||
scanner,
|
||||
'_SerieScanner__get_missing_episodes_and_season',
|
||||
return_value=({1: [1, 2, 3]}, "aniworld.to"),
|
||||
), \
|
||||
patch.object(resolved, 'save_to_file'):
|
||||
scanner.scan()
|
||||
|
||||
assert "rooster-fighter" in scanner.keyDict
|
||||
assert scanner.keyDict["rooster-fighter"].episodeDict == {1: [1, 2, 3]}
|
||||
|
||||
def test_db_lookup_returns_none_folder_skipped(self, mock_loader):
|
||||
"""When db_lookup returns None, the folder is skipped with a warning."""
|
||||
import tempfile
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
lookup = MagicMock(return_value=None)
|
||||
scanner = self._make_scanner(tmp_dir, mock_loader, db_lookup=lookup)
|
||||
|
||||
with patch.object(scanner, 'get_total_to_scan', return_value=1):
|
||||
scanner.scan()
|
||||
|
||||
assert len(scanner.keyDict) == 0
|
||||
|
||||
def test_db_lookup_exception_skips_folder(self, mock_loader):
|
||||
"""When db_lookup raises, the folder is skipped gracefully."""
|
||||
import tempfile
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
lookup = MagicMock(side_effect=RuntimeError("DB offline"))
|
||||
scanner = self._make_scanner(tmp_dir, mock_loader, db_lookup=lookup)
|
||||
|
||||
with patch.object(scanner, 'get_total_to_scan', return_value=1):
|
||||
scanner.scan() # should not raise
|
||||
|
||||
assert len(scanner.keyDict) == 0
|
||||
|
||||
def test_db_lookup_warning_logged_when_no_files(
|
||||
self, mock_loader, caplog
|
||||
):
|
||||
"""A warning is logged for folders without key/data file."""
|
||||
import tempfile
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
scanner = self._make_scanner(tmp_dir, mock_loader, db_lookup=None)
|
||||
|
||||
with caplog.at_level(logging.WARNING, logger="src.core.SerieScanner"):
|
||||
with patch.object(scanner, 'get_total_to_scan', return_value=1):
|
||||
scanner.scan()
|
||||
|
||||
assert any(
|
||||
"Rooster Fighter (2026)" in record.message
|
||||
for record in caplog.records
|
||||
if record.levelname == "WARNING"
|
||||
)
|
||||
|
||||
def test_db_lookup_info_logged_on_resolution(
|
||||
self, mock_loader, caplog
|
||||
):
|
||||
"""An INFO log is emitted when db_lookup resolves a folder."""
|
||||
import tempfile
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
resolved = Serie(
|
||||
key="rooster-fighter",
|
||||
name="",
|
||||
site="aniworld.to",
|
||||
folder="Rooster Fighter (2026)",
|
||||
episodeDict={},
|
||||
)
|
||||
lookup = MagicMock(return_value=resolved)
|
||||
scanner = self._make_scanner(tmp_dir, mock_loader, db_lookup=lookup)
|
||||
|
||||
with caplog.at_level(logging.INFO, logger="src.core.SerieScanner"), \
|
||||
patch.object(scanner, 'get_total_to_scan', return_value=1), \
|
||||
patch.object(
|
||||
scanner,
|
||||
'_SerieScanner__get_missing_episodes_and_season',
|
||||
return_value=({}, "aniworld.to"),
|
||||
), \
|
||||
patch.object(resolved, 'save_to_file'):
|
||||
scanner.scan()
|
||||
|
||||
assert any(
|
||||
"rooster-fighter" in record.message
|
||||
for record in caplog.records
|
||||
if record.levelname == "INFO"
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user