lukas.pupkalipinski/Aniworld
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(NFO): add TMDB search fallback with alt_titles support

Browse Source 
- New _search_with_fallback() method tries multiple strategies:
  1. Primary query with year filter (de-DE locale)
  2. Alternative titles with ja-JP / en-US locales
  3. English search (en-US)
  4. Search without year constraint
  5. Punctuation-normalized query
- create_nfo() accepts new alt_titles param for Japanese/title fallback
- Better match rate for anime with non-English titles

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Lukas
2026-05-23 21:57:00 +02:00
parent 3f7651404d
commit 9a20541598
7 changed files with 588 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

245
tests/unit/test_nfo_service.py
View File

@@ -1,5 +1,6 @@
"""Unit tests for NFO service."""
import time
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
@@ -22,6 +23,14 @@ def nfo_service(tmp_path):
return service
@pytest.fixture
def tmdb_client():
"""Create TMDB client with test API key."""
from src.core.services.tmdb_client import TMDBClient
client = TMDBClient(api_key="test_api_key")
return client
@pytest.fixture
def mock_tmdb_data():
"""Mock TMDB API response data."""
@@ -342,7 +351,7 @@ class TestCreateTVShowNFO:
)
# Assert - should search with clean name "The Dreaming Boy is a Realist"
mock_search.assert_called_once_with("The Dreaming Boy is a Realist")
mock_search.assert_called_once_with("The Dreaming Boy is a Realist", language="de-DE")
# Verify NFO file was created
assert nfo_path.exists()
@@ -362,29 +371,28 @@ class TestCreateTVShowNFO:
with patch.object(nfo_service.tmdb_client, '__aenter__', return_value=nfo_service.tmdb_client):
with patch.object(nfo_service.tmdb_client, '__aexit__', return_value=None):
with patch.object(nfo_service.tmdb_client, 'search_tv_show', new_callable=AsyncMock) as mock_search:
with patch.object(nfo_service, '_search_with_fallback', new_callable=AsyncMock) as mock_search_fallback:
with patch.object(nfo_service.tmdb_client, 'get_tv_show_details', new_callable=AsyncMock) as mock_details:
with patch.object(nfo_service.tmdb_client, 'get_tv_show_content_ratings', new_callable=AsyncMock) as mock_ratings:
with patch.object(nfo_service.image_downloader, 'download_poster', new_callable=AsyncMock):
with patch.object(nfo_service.image_downloader, 'download_logo', new_callable=AsyncMock):
with patch.object(nfo_service.image_downloader, 'download_fanart', new_callable=AsyncMock):
with patch.object(nfo_service, '_find_best_match') as mock_find_match:
mock_search.return_value = search_results
mock_details.return_value = mock_tmdb_data
mock_ratings.return_value = mock_content_ratings_de
mock_find_match.return_value = mock_tmdb_data
# Act
await nfo_service.create_tvshow_nfo(
serie_name=serie_name,
serie_folder=serie_folder,
year=explicit_year # Explicit year provided
)
# Assert - should use explicit year, not extracted year
mock_find_match.assert_called_once()
call_args = mock_find_match.call_args
assert call_args[0][2] == explicit_year # Third argument is year
with patch.object(nfo_service, '_enrich_details_with_fallback', new_callable=AsyncMock) as mock_enrich:
with patch.object(nfo_service, '_download_media_files', new_callable=AsyncMock):
mock_search_fallback.return_value = (mock_tmdb_data, "primary")
mock_details.return_value = mock_tmdb_data
mock_ratings.return_value = mock_content_ratings_de
mock_enrich.return_value = mock_tmdb_data
# Act
await nfo_service.create_tvshow_nfo(
serie_name=serie_name,
serie_folder=serie_folder,
year=explicit_year # Explicit year provided
)
# Assert - _search_with_fallback should be called with explicit year
mock_search_fallback.assert_called_once()
call_args = mock_search_fallback.call_args
assert call_args[0][0] == "Attack on Titan" # clean name
assert call_args[0][1] == explicit_year # explicit year
@pytest.mark.asyncio
async def test_create_nfo_no_results_with_clean_name(self, nfo_service, tmp_path):
@@ -396,8 +404,8 @@ class TestCreateTVShowNFO:
with patch.object(nfo_service.tmdb_client, '__aenter__', return_value=nfo_service.tmdb_client):
with patch.object(nfo_service.tmdb_client, '__aexit__', return_value=None):
with patch.object(nfo_service.tmdb_client, 'search_tv_show', new_callable=AsyncMock) as mock_search:
mock_search.return_value = {"results": []}
with patch.object(nfo_service, '_search_with_fallback', new_callable=AsyncMock) as mock_search_fallback:
mock_search_fallback.side_effect = TMDBAPIError("No results found for: Nonexistent Series")
# Act & Assert
with pytest.raises(TMDBAPIError) as exc_info:
@@ -408,8 +416,6 @@ class TestCreateTVShowNFO:
# Should use clean name in error message
assert "No results found for: Nonexistent Series" in str(exc_info.value)
# Should have searched with clean name
mock_search.assert_called_once_with("Nonexistent Series")
@pytest.mark.asyncio
async def test_create_nfo_with_fsk(self, nfo_service, tmp_path, mock_tmdb_data, mock_content_ratings_de):
@@ -1616,3 +1622,190 @@ class TestEnrichFallbackLanguages:
# de-DE + en-US = 2 calls (no ja-JP needed)
assert mock_details.call_count == 2
class TestSearchWithFallback:
"""Tests for TMDB search fallback functionality."""
@pytest.mark.asyncio
async def test_search_with_fallback_primary_success(self, nfo_service, mock_tmdb_data):
"""Test that primary query succeeds without fallback."""
with patch.object(nfo_service.tmdb_client, 'search_tv_show', new_callable=AsyncMock) as mock_search:
mock_search.return_value = {"results": [mock_tmdb_data]}
result, source = await nfo_service._search_with_fallback(
"Attack on Titan", 2013, None
)
assert result["id"] == mock_tmdb_data["id"]
assert source == "primary"
assert mock_search.call_count == 1
@pytest.mark.asyncio
async def test_search_with_fallback_uses_alt_titles(self, nfo_service, mock_tmdb_data):
"""Test that alternative titles are tried when primary fails."""
mock_search = AsyncMock()
# First call returns empty, second (with Japanese title) returns result
mock_search.side_effect = [
{"results": []},
{"results": [mock_tmdb_data]}
]
with patch.object(nfo_service.tmdb_client, 'search_tv_show', mock_search):
result, source = await nfo_service._search_with_fallback(
"Suzume", 2022, alt_titles=["すずめの戸締まり"]
)
assert result["id"] == mock_tmdb_data["id"]
assert "alt_title" in source
@pytest.mark.asyncio
async def test_search_with_fallback_year_not_matched(self, nfo_service, mock_tmdb_data):
"""Test fallback when year doesn't match but first result is used anyway."""
# First result doesn't match year, but should still be returned
different_year_data = {**mock_tmdb_data, "first_air_date": "2020-01-01"}
mock_search = AsyncMock(return_value={"results": [different_year_data]})
with patch.object(nfo_service.tmdb_client, 'search_tv_show', mock_search):
result, source = await nfo_service._search_with_fallback(
"Attack on Titan", 2013, None
)
assert result["id"] == mock_tmdb_data["id"]
@pytest.mark.asyncio
async def test_search_with_fallback_no_year_strategy(self, nfo_service, mock_tmdb_data):
"""Test that search without year is attempted when year-filtered fails."""
mock_search = AsyncMock()
# First call with year fails, second (without year) succeeds
mock_search.side_effect = [
{"results": []},
{"results": [mock_tmdb_data]}
]
with patch.object(nfo_service.tmdb_client, 'search_tv_show', mock_search):
result, source = await nfo_service._search_with_fallback(
"Attack on Titan", 2013, None
)
assert result["id"] == mock_tmdb_data["id"]
# Strategy order: primary -> english -> no_year (english comes before no_year)
assert mock_search.call_count == 2
@pytest.mark.asyncio
async def test_search_with_fallback_all_strategies_fail(self, nfo_service):
"""Test that TMDBAPIError is raised when all strategies fail."""
mock_search = AsyncMock(return_value={"results": []})
with patch.object(nfo_service.tmdb_client, 'search_tv_show', mock_search):
with pytest.raises(TMDBAPIError) as exc_info:
await nfo_service._search_with_fallback(
"Nonexistent Anime", 2023, None
)
assert "Nonexistent Anime" in str(exc_info.value)
# Should have tried multiple strategies
assert mock_search.call_count >= 3
@pytest.mark.asyncio
async def test_search_with_fallback_normalizes_punctuation(self, nfo_service, mock_tmdb_data):
"""Test that punctuation-normalized search is attempted."""
mock_search = AsyncMock()
# First call fails, normalized version succeeds
mock_search.side_effect = [
{"results": []},
{"results": [mock_tmdb_data]}
]
with patch.object(nfo_service.tmdb_client, 'search_tv_show', mock_search):
result, source = await nfo_service._search_with_fallback(
"Attack on Titan:", 2013, None
)
assert result["id"] == mock_tmdb_data["id"]
def test_normalize_query_for_search(self, nfo_service):
"""Test punctuation normalization in queries."""
# Test normal punctuation removal
assert nfo_service._normalize_query_for_search("Attack on Titan:") == "Attack on Titan"
assert nfo_service._normalize_query_for_search("Suzume no Tojimari.") == "Suzume no Tojimari"
# Test CJK characters are preserved
assert "すずめ" in nfo_service._normalize_query_for_search("すずめの戸締まり")
# Test multiple spaces are collapsed
assert nfo_service._normalize_query_for_search("Attack on Titan") == "Attack on Titan"
class TestNegativeCache:
"""Tests for negative result caching in TMDB client."""
@pytest.mark.asyncio
async def test_negative_result_cached(self, tmdb_client):
"""Test that empty search results are cached."""
import time
mock_session = MagicMock()
mock_response = AsyncMock()
mock_response.status = 200
mock_response.json = AsyncMock(return_value={"results": []})
mock_response.__aenter__ = AsyncMock(return_value=mock_response)
mock_response.__aexit__ = AsyncMock(return_value=None)
mock_session.get = MagicMock(return_value=mock_response)
tmdb_client.session = mock_session
with patch.object(tmdb_client, '_ensure_session', new_callable=AsyncMock):
# First call
result = await tmdb_client.search_tv_show("Nonexistent")
assert result["results"] == []
# Negative cache should be set
assert len(tmdb_client._negative_cache) > 0
@pytest.mark.asyncio
async def test_negative_cache_prevents_duplicate_call(self, tmdb_client):
"""Test that negative cache prevents second API call within 24 hours."""
import time
mock_session = MagicMock()
mock_response = AsyncMock()
mock_response.status = 200
mock_response.json = AsyncMock(return_value={"results": []})
mock_response.__aenter__ = AsyncMock(return_value=mock_response)
mock_response.__aexit__ = AsyncMock(return_value=None)
mock_session.get = MagicMock(return_value=mock_response)
tmdb_client.session = mock_session
with patch.object(tmdb_client, '_ensure_session', new_callable=AsyncMock):
# First call - should hit API
await tmdb_client.search_tv_show("Nonexistent")
first_call_count = mock_session.get.call_count
# Second call with same query - should use negative cache, not hit API
await tmdb_client.search_tv_show("Nonexistent")
second_call_count = mock_session.get.call_count
# Should not have made second API call
assert first_call_count == second_call_count
def test_clear_negative_cache(self, tmdb_client):
"""Test clearing negative cache."""
# Add some negative cache entries
tmdb_client._negative_cache["test_key"] = time.monotonic()
assert len(tmdb_client._negative_cache) > 0
tmdb_client.clear_negative_cache()
assert len(tmdb_client._negative_cache) == 0
def test_cleanup_expired_negative_cache(self, tmdb_client):
"""Test cleanup of expired negative cache entries."""
# Add an expired entry
old_timestamp = time.monotonic() - (tmdb_client.NEGATIVE_CACHE_TTL + 1)
tmdb_client._negative_cache["expired_key"] = old_timestamp
tmdb_client._negative_cache["valid_key"] = time.monotonic()
removed = tmdb_client.cleanup_expired_negative_cache()
assert removed == 1
assert "expired_key" not in tmdb_client._negative_cache
assert "valid_key" in tmdb_client._negative_cache

74
tests/unit/test_scheduler_service.py
View File

@@ -117,6 +117,8 @@ class TestStart:
call_kwargs = mock_sched.add_job.call_args
assert call_kwargs[1]["id"] == _JOB_ID
assert isinstance(call_kwargs[1]["trigger"], CronTrigger)
assert call_kwargs[1]["misfire_grace_time"] == 3600
assert call_kwargs[1]["coalesce"] is True
mock_sched.start.assert_called_once()
assert scheduler_service._is_running is True
@@ -485,3 +487,75 @@ class TestSingletonHelpers:
svc = get_scheduler_service()
assert svc is not None # fresh instance
# ---------------------------------------------------------------------------
# 12.12 Persistent job store — SQLAlchemyJobStore passed to AsyncIOScheduler
# ---------------------------------------------------------------------------
class TestPersistentJobStore:
@pytest.mark.asyncio
async def test_start_creates_scheduler_with_sqlalchemy_jobstore(
self, scheduler_service, mock_config_service
):
with patch(
"src.server.services.scheduler_service.AsyncIOScheduler"
) as MockScheduler:
mock_sched = MagicMock()
mock_sched.running = False
MockScheduler.return_value = mock_sched
await scheduler_service.start()
MockScheduler.assert_called_once()
call_kwargs = MockScheduler.call_args
jobstores = call_kwargs[1]["jobstores"]
assert "default" in jobstores
# Verify it's a SQLAlchemyJobStore (class check via module name)
assert "sqlalchemy" in type(jobstores["default"]).__module__
@pytest.mark.asyncio
async def test_job_options_include_misfire_grace_and_coalesce(
self, scheduler_service, mock_config_service
):
with patch(
"src.server.services.scheduler_service.AsyncIOScheduler"
) as MockScheduler:
mock_sched = MagicMock()
mock_sched.running = False
MockScheduler.return_value = mock_sched
await scheduler_service.start()
call_kwargs = mock_sched.add_job.call_args
assert call_kwargs[1]["misfire_grace_time"] == 3600
assert call_kwargs[1]["coalesce"] is True
# ---------------------------------------------------------------------------
# 12.13 Startup recovery — next run logged after start()
# ---------------------------------------------------------------------------
class TestStartupRecovery:
@pytest.mark.asyncio
async def test_start_logs_next_run_time(
self, scheduler_service, mock_config_service
):
with patch(
"src.server.services.scheduler_service.AsyncIOScheduler"
) as MockScheduler:
mock_job = MagicMock()
next_run_dt = datetime(2026, 5, 25, 3, 0, tzinfo=timezone.utc)
mock_job.next_run_time = next_run_dt
mock_sched = MagicMock()
mock_sched.running = False
mock_sched.get_job.return_value = mock_job
MockScheduler.return_value = mock_sched
with patch(
"src.server.services.scheduler_service.logger"
) as mock_logger:
await scheduler_service.start()
# Check that next_run was logged
info_calls = [str(c) for c in mock_logger.info.call_args_list]
assert any("next_run" in c for c in info_calls)