Files
Aniworld/tests/unit/test_nfo_service.py
Lukas d596902ca3 Parse existing NFO for TMDB ID to skip redundant search
Check existing tvshow.nfo for TMDB ID before querying TMDB API.
If found, fetch details directly using cached ID instead of searching.
Reduces API calls and improves performance for already-indexed series.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
2026-05-27 21:22:24 +02:00

1916 lines
80 KiB
Python

"""Unit tests for NFO service."""
import time
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from src.core.services.nfo_service import NFOService
from src.core.services.tmdb_client import TMDBAPIError
from src.core.utils.nfo_mapper import _extract_fsk_rating, tmdb_to_nfo_model
@pytest.fixture
def nfo_service(tmp_path):
"""Create NFO service with test directory."""
service = NFOService(
tmdb_api_key="test_api_key",
anime_directory=str(tmp_path),
image_size="w500",
auto_create=True
)
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."""
return {
"id": 1429,
"name": "Attack on Titan",
"original_name": "進撃の巨人",
"first_air_date": "2013-04-07",
"overview": "Several hundred years ago, humans were nearly...",
"vote_average": 8.6,
"vote_count": 5000,
"status": "Ended",
"episode_run_time": [24],
"genres": [{"id": 16, "name": "Animation"}, {"id": 10765, "name": "Sci-Fi & Fantasy"}],
"networks": [{"id": 1, "name": "MBS"}],
"production_countries": [{"name": "Japan"}],
"poster_path": "/poster.jpg",
"backdrop_path": "/backdrop.jpg",
"external_ids": {
"imdb_id": "tt2560140",
"tvdb_id": 267440
},
"credits": {
"cast": [
{
"id": 1,
"name": "Yuki Kaji",
"character": "Eren Yeager",
"profile_path": "/actor.jpg"
}
]
},
"images": {
"logos": [{"file_path": "/logo.png"}]
}
}
@pytest.fixture
def mock_content_ratings_de():
"""Mock TMDB content ratings with German FSK rating."""
return {
"results": [
{"iso_3166_1": "DE", "rating": "16"},
{"iso_3166_1": "US", "rating": "TV-MA"}
]
}
@pytest.fixture
def mock_content_ratings_no_de():
"""Mock TMDB content ratings without German rating."""
return {
"results": [
{"iso_3166_1": "US", "rating": "TV-MA"},
{"iso_3166_1": "GB", "rating": "15"}
]
}
class TestFSKRatingExtraction:
"""Test FSK rating extraction from TMDB content ratings."""
def test_extract_fsk_rating_de(self, nfo_service, mock_content_ratings_de):
"""Test extraction of German FSK rating."""
fsk = _extract_fsk_rating(mock_content_ratings_de)
assert fsk == "FSK 16"
def test_extract_fsk_rating_no_de(self, nfo_service, mock_content_ratings_no_de):
"""Test extraction when no German rating available."""
fsk = _extract_fsk_rating(mock_content_ratings_no_de)
assert fsk is None
def test_extract_fsk_rating_empty(self, nfo_service):
"""Test extraction with empty content ratings."""
fsk = _extract_fsk_rating({})
assert fsk is None
def test_extract_fsk_rating_none(self, nfo_service):
"""Test extraction with None input."""
fsk = _extract_fsk_rating(None)
assert fsk is None
def test_extract_fsk_all_values(self, nfo_service):
"""Test extraction of all FSK values."""
fsk_mappings = {
"0": "FSK 0",
"6": "FSK 6",
"12": "FSK 12",
"16": "FSK 16",
"18": "FSK 18"
}
for rating_value, expected_fsk in fsk_mappings.items():
content_ratings = {
"results": [{"iso_3166_1": "DE", "rating": rating_value}]
}
fsk = _extract_fsk_rating(content_ratings)
assert fsk == expected_fsk
def test_extract_fsk_already_formatted(self, nfo_service):
"""Test extraction when rating is already in FSK format."""
content_ratings = {
"results": [{"iso_3166_1": "DE", "rating": "FSK 12"}]
}
fsk = _extract_fsk_rating(content_ratings)
assert fsk == "FSK 12"
def test_extract_fsk_partial_match(self, nfo_service):
"""Test extraction with partial number match."""
content_ratings = {
"results": [{"iso_3166_1": "DE", "rating": "Ab 16 Jahren"}]
}
fsk = _extract_fsk_rating(content_ratings)
assert fsk == "FSK 16"
def test_extract_fsk_unmapped_value(self, nfo_service):
"""Test extraction with unmapped rating value."""
content_ratings = {
"results": [{"iso_3166_1": "DE", "rating": "Unknown"}]
}
fsk = _extract_fsk_rating(content_ratings)
assert fsk is None
class TestYearExtraction:
"""Test year extraction from series names."""
def test_extract_year_with_year(self, nfo_service):
"""Test extraction when year is present in format (YYYY)."""
clean_name, year = nfo_service._extract_year_from_name("Attack on Titan (2013)")
assert clean_name == "Attack on Titan"
assert year == 2013
def test_extract_year_without_year(self, nfo_service):
"""Test extraction when no year is present."""
clean_name, year = nfo_service._extract_year_from_name("Attack on Titan")
assert clean_name == "Attack on Titan"
assert year is None
def test_extract_year_multiple_parentheses(self, nfo_service):
"""Test extraction with multiple parentheses - only last one with year."""
clean_name, year = nfo_service._extract_year_from_name("Series (Part 1) (2023)")
assert clean_name == "Series (Part 1)"
assert year == 2023
def test_extract_year_with_trailing_spaces(self, nfo_service):
"""Test extraction with trailing spaces."""
clean_name, year = nfo_service._extract_year_from_name("Attack on Titan (2013) ")
assert clean_name == "Attack on Titan"
assert year == 2013
def test_extract_year_parentheses_not_year(self, nfo_service):
"""Test extraction when parentheses don't contain a year."""
clean_name, year = nfo_service._extract_year_from_name("Series (Special Edition)")
assert clean_name == "Series (Special Edition)"
assert year is None
def test_extract_year_invalid_year_format(self, nfo_service):
"""Test extraction with invalid year format (not 4 digits)."""
clean_name, year = nfo_service._extract_year_from_name("Series (23)")
assert clean_name == "Series (23)"
assert year is None
def test_extract_year_future_year(self, nfo_service):
"""Test extraction with future year."""
clean_name, year = nfo_service._extract_year_from_name("Future Series (2050)")
assert clean_name == "Future Series"
assert year == 2050
def test_extract_year_old_year(self, nfo_service):
"""Test extraction with old year."""
clean_name, year = nfo_service._extract_year_from_name("Classic Series (1990)")
assert clean_name == "Classic Series"
assert year == 1990
def test_extract_year_real_world_example(self, nfo_service):
"""Test extraction with the real-world example from the bug report."""
clean_name, year = nfo_service._extract_year_from_name("The Dreaming Boy is a Realist (2023)")
assert clean_name == "The Dreaming Boy is a Realist"
assert year == 2023
def test_extract_year_uebel_blatt(self, nfo_service):
"""Test extraction with Übel Blatt example."""
clean_name, year = nfo_service._extract_year_from_name("Übel Blatt (2025)")
assert clean_name == "Übel Blatt"
assert year == 2025
class TestTMDBToNFOModel:
"""Test conversion of TMDB data to NFO model."""
@patch('src.core.utils.nfo_mapper._extract_fsk_rating')
def test_tmdb_to_nfo_with_fsk(self, mock_extract_fsk, nfo_service, mock_tmdb_data, mock_content_ratings_de):
"""Test conversion includes FSK rating."""
mock_extract_fsk.return_value = "FSK 16"
nfo_model = tmdb_to_nfo_model(
mock_tmdb_data, mock_content_ratings_de,
nfo_service.tmdb_client.get_image_url, nfo_service.image_size
)
assert nfo_model.title == "Attack on Titan"
assert nfo_model.fsk == "FSK 16"
assert nfo_model.year == 2013
mock_extract_fsk.assert_called_once_with(mock_content_ratings_de)
def test_tmdb_to_nfo_without_content_ratings(self, nfo_service, mock_tmdb_data):
"""Test conversion without content ratings."""
nfo_model = tmdb_to_nfo_model(
mock_tmdb_data, None,
nfo_service.tmdb_client.get_image_url, nfo_service.image_size
)
assert nfo_model.title == "Attack on Titan"
assert nfo_model.fsk is None
assert nfo_model.tmdbid == 1429
def test_tmdb_to_nfo_basic_fields(self, nfo_service, mock_tmdb_data):
"""Test that all basic fields are correctly mapped."""
nfo_model = tmdb_to_nfo_model(
mock_tmdb_data, None,
nfo_service.tmdb_client.get_image_url, nfo_service.image_size
)
assert nfo_model.title == "Attack on Titan"
assert nfo_model.originaltitle == "進撃の巨人"
assert nfo_model.year == 2013
assert nfo_model.plot == "Several hundred years ago, humans were nearly..."
assert nfo_model.status == "Ended"
assert nfo_model.runtime == 24
assert nfo_model.premiered == "2013-04-07"
def test_tmdb_to_nfo_ids(self, nfo_service, mock_tmdb_data):
"""Test that all IDs are correctly mapped."""
nfo_model = tmdb_to_nfo_model(
mock_tmdb_data, None,
nfo_service.tmdb_client.get_image_url, nfo_service.image_size
)
assert nfo_model.tmdbid == 1429
assert nfo_model.imdbid == "tt2560140"
assert nfo_model.tvdbid == 267440
assert len(nfo_model.uniqueid) == 3
def test_tmdb_to_nfo_genres_studios(self, nfo_service, mock_tmdb_data):
"""Test that genres and studios are correctly mapped."""
nfo_model = tmdb_to_nfo_model(
mock_tmdb_data, None,
nfo_service.tmdb_client.get_image_url, nfo_service.image_size
)
assert "Animation" in nfo_model.genre
assert "Sci-Fi & Fantasy" in nfo_model.genre
assert "MBS" in nfo_model.studio
assert "Japan" in nfo_model.country
def test_tmdb_to_nfo_ratings(self, nfo_service, mock_tmdb_data):
"""Test that ratings are correctly mapped."""
nfo_model = tmdb_to_nfo_model(
mock_tmdb_data, None,
nfo_service.tmdb_client.get_image_url, nfo_service.image_size
)
assert len(nfo_model.ratings) == 1
assert nfo_model.ratings[0].name == "themoviedb"
assert nfo_model.ratings[0].value == 8.6
assert nfo_model.ratings[0].votes == 5000
def test_tmdb_to_nfo_cast(self, nfo_service, mock_tmdb_data):
"""Test that cast is correctly mapped."""
nfo_model = tmdb_to_nfo_model(
mock_tmdb_data, None,
nfo_service.tmdb_client.get_image_url, nfo_service.image_size
)
assert len(nfo_model.actors) == 1
assert nfo_model.actors[0].name == "Yuki Kaji"
assert nfo_model.actors[0].role == "Eren Yeager"
assert nfo_model.actors[0].tmdbid == 1
class TestCreateTVShowNFO:
"""Test NFO creation workflow."""
@pytest.mark.asyncio
async def test_create_nfo_with_year_in_name(self, nfo_service, tmp_path, mock_tmdb_data, mock_content_ratings_de):
"""Test NFO creation when year is included in series name.
This test addresses the bug where searching TMDB with year in the name
(e.g., "The Dreaming Boy is a Realist (2023)") fails to find results.
"""
# Setup
serie_name = "The Dreaming Boy is a Realist (2023)"
serie_folder = "The Dreaming Boy is a Realist (2023)"
(tmp_path / serie_folder).mkdir()
# Mock TMDB responses
search_results = {"results": [mock_tmdb_data]}
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.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):
mock_search.return_value = search_results
mock_details.return_value = mock_tmdb_data
mock_ratings.return_value = mock_content_ratings_de
# Act
nfo_path = await nfo_service.create_tvshow_nfo(
serie_name=serie_name,
serie_folder=serie_folder,
year=None # Year should be auto-extracted
)
# Assert - should search with clean name "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()
assert nfo_path.name == "tvshow.nfo"
@pytest.mark.asyncio
async def test_create_nfo_year_parameter_takes_precedence(self, nfo_service, tmp_path, mock_tmdb_data, mock_content_ratings_de):
"""Test that explicit year parameter takes precedence over extracted year."""
# Setup
serie_name = "Attack on Titan (2013)"
serie_folder = "Attack on Titan"
explicit_year = 2015 # Different from extracted year
(tmp_path / serie_folder).mkdir()
# Mock TMDB responses
search_results = {"results": [mock_tmdb_data]}
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, '_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, '_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):
"""Test error handling when TMDB returns no results even with clean name."""
# Setup
serie_name = "Nonexistent Series (2023)"
serie_folder = "Nonexistent Series (2023)"
(tmp_path / serie_folder).mkdir()
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, '_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:
await nfo_service.create_tvshow_nfo(
serie_name=serie_name,
serie_folder=serie_folder
)
# Should use clean name in error message
assert "No results found for: Nonexistent Series" in str(exc_info.value)
@pytest.mark.asyncio
async def test_create_nfo_with_fsk(self, nfo_service, tmp_path, mock_tmdb_data, mock_content_ratings_de):
"""Test NFO creation includes FSK rating."""
# Create series folder
series_folder = tmp_path / "Attack on Titan"
series_folder.mkdir()
# Mock TMDB client methods
with patch.object(nfo_service.tmdb_client, 'search_tv_show', new_callable=AsyncMock) as mock_search, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_details', new_callable=AsyncMock) as mock_details, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_content_ratings', new_callable=AsyncMock) as mock_ratings, \
patch.object(nfo_service, '_download_media_files', new_callable=AsyncMock):
mock_search.return_value = {
"results": [{
"id": 1429,
"name": "Attack on Titan",
"first_air_date": "2013-04-07",
"overview": "Several hundred years ago, humans were nearly...",
}]
}
mock_details.return_value = mock_tmdb_data
mock_ratings.return_value = mock_content_ratings_de
# Create NFO
nfo_path = await nfo_service.create_tvshow_nfo(
"Attack on Titan",
"Attack on Titan",
year=2013,
download_poster=False,
download_logo=False,
download_fanart=False
)
# Verify NFO was created
assert nfo_path.exists()
nfo_content = nfo_path.read_text(encoding="utf-8")
# Check that FSK rating is in the NFO
assert "<mpaa>FSK 16</mpaa>" in nfo_content
# Verify TMDB methods were called
mock_search.assert_called_once()
mock_details.assert_called_once_with(1429, append_to_response="credits,external_ids,images")
mock_ratings.assert_called_once_with(1429)
@pytest.mark.asyncio
async def test_create_nfo_without_fsk(self, nfo_service, tmp_path, mock_tmdb_data, mock_content_ratings_no_de):
"""Test NFO creation fallback when no FSK available."""
# Create series folder
series_folder = tmp_path / "Attack on Titan"
series_folder.mkdir()
# Mock TMDB client methods
with patch.object(nfo_service.tmdb_client, 'search_tv_show', new_callable=AsyncMock) as mock_search, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_details', new_callable=AsyncMock) as mock_details, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_content_ratings', new_callable=AsyncMock) as mock_ratings, \
patch.object(nfo_service, '_download_media_files', new_callable=AsyncMock):
mock_search.return_value = {
"results": [{
"id": 1429,
"name": "Attack on Titan",
"first_air_date": "2013-04-07",
"overview": "Several hundred years ago, humans were nearly...",
}]
}
mock_details.return_value = mock_tmdb_data
mock_ratings.return_value = mock_content_ratings_no_de
# Create NFO
nfo_path = await nfo_service.create_tvshow_nfo(
"Attack on Titan",
"Attack on Titan",
year=2013,
download_poster=False,
download_logo=False,
download_fanart=False
)
# Verify NFO was created
assert nfo_path.exists()
nfo_content = nfo_path.read_text(encoding="utf-8")
# FSK should not be in the NFO
assert "FSK" not in nfo_content
@pytest.mark.asyncio
async def test_update_nfo_with_fsk(self, nfo_service, tmp_path, mock_tmdb_data, mock_content_ratings_de):
"""Test NFO update includes FSK rating."""
# Create series folder with existing NFO
series_folder = tmp_path / "Attack on Titan"
series_folder.mkdir()
nfo_path = series_folder / "tvshow.nfo"
nfo_path.write_text("""<?xml version="1.0" encoding="UTF-8"?>
<tvshow>
<title>Attack on Titan</title>
<tmdbid>1429</tmdbid>
</tvshow>
""", encoding="utf-8")
# Mock TMDB client methods
with patch.object(nfo_service.tmdb_client, 'get_tv_show_details', new_callable=AsyncMock) as mock_details, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_content_ratings', new_callable=AsyncMock) as mock_ratings, \
patch.object(nfo_service, '_download_media_files', new_callable=AsyncMock):
mock_details.return_value = mock_tmdb_data
mock_ratings.return_value = mock_content_ratings_de
# Update NFO
updated_path = await nfo_service.update_tvshow_nfo(
"Attack on Titan",
download_media=False
)
# Verify NFO was updated
assert updated_path.exists()
nfo_content = updated_path.read_text(encoding="utf-8")
# Check that FSK rating is in the updated NFO
assert "<mpaa>FSK 16</mpaa>" in nfo_content
# Verify TMDB methods were called
mock_details.assert_called_once_with(1429, append_to_response="credits,external_ids,images")
mock_ratings.assert_called_once_with(1429)
class TestEnrichDetailsWithFallback:
"""Tests for English fallback when German overview is empty."""
@pytest.mark.asyncio
async def test_create_nfo_uses_english_fallback_for_empty_overview(
self, nfo_service, tmp_path
):
"""When the German overview is empty, create_tvshow_nfo should
fetch the English overview from TMDB and include it as <plot>."""
series_folder = tmp_path / "Basilisk"
series_folder.mkdir()
# German TMDB data with empty overview
de_data = {
"id": 35014, "name": "Basilisk",
"original_name": "甲賀忍法帖", "first_air_date": "2005-04-13",
"overview": "", # <-- empty German overview
"vote_average": 7.2, "vote_count": 200,
"status": "Ended", "episode_run_time": [24],
"genres": [{"id": 16, "name": "Animation"}],
"networks": [{"id": 1, "name": "MBS"}],
"production_countries": [{"name": "Japan"}],
"poster_path": "/poster.jpg", "backdrop_path": "/backdrop.jpg",
"external_ids": {"imdb_id": "tt0464064", "tvdb_id": 79604},
"credits": {"cast": []},
"images": {"logos": []},
}
# English TMDB data with overview
en_data = {
"id": 35014,
"overview": "The year is 1614 and two warring ninja clans collide.",
"tagline": "Blood spills when ninja clans clash.",
}
async def side_effect(tv_id, **kwargs):
if kwargs.get("language") == "en-US":
return en_data
return de_data
with patch.object(nfo_service.tmdb_client, 'search_tv_show', new_callable=AsyncMock) as mock_search, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_details', new_callable=AsyncMock) as mock_details, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_content_ratings', new_callable=AsyncMock) as mock_ratings, \
patch.object(nfo_service, '_download_media_files', new_callable=AsyncMock):
mock_search.return_value = {
"results": [{"id": 35014, "name": "Basilisk", "first_air_date": "2005-04-13"}]
}
mock_details.side_effect = side_effect
mock_ratings.return_value = {"results": []}
nfo_path = await nfo_service.create_tvshow_nfo(
"Basilisk", "Basilisk", year=2005,
download_poster=False, download_logo=False, download_fanart=False,
)
content = nfo_path.read_text(encoding="utf-8")
assert "<plot>The year is 1614" in content
# Details called twice: once for de-DE, once for en-US fallback
assert mock_details.call_count == 2
@pytest.mark.asyncio
async def test_update_nfo_uses_english_fallback_for_empty_overview(
self, nfo_service, tmp_path
):
"""update_tvshow_nfo should also use the English fallback."""
series_folder = tmp_path / "Basilisk"
series_folder.mkdir()
nfo_path = series_folder / "tvshow.nfo"
nfo_path.write_text(
'<?xml version="1.0"?>\n<tvshow><title>Basilisk</title>'
"<tmdbid>35014</tmdbid></tvshow>",
encoding="utf-8",
)
de_data = {
"id": 35014, "name": "Basilisk",
"original_name": "甲賀忍法帖", "first_air_date": "2005-04-13",
"overview": "",
"vote_average": 7.2, "vote_count": 200,
"status": "Ended", "episode_run_time": [24],
"genres": [{"id": 16, "name": "Animation"}],
"networks": [{"id": 1, "name": "MBS"}],
"production_countries": [{"name": "Japan"}],
"poster_path": "/poster.jpg", "backdrop_path": "/backdrop.jpg",
"external_ids": {"imdb_id": "tt0464064", "tvdb_id": 79604},
"credits": {"cast": []},
"images": {"logos": []},
}
en_data = {
"id": 35014,
"overview": "English fallback overview for Basilisk.",
}
async def side_effect(tv_id, **kwargs):
if kwargs.get("language") == "en-US":
return en_data
return de_data
with patch.object(nfo_service.tmdb_client, 'get_tv_show_details', new_callable=AsyncMock) as mock_details, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_content_ratings', new_callable=AsyncMock) as mock_ratings, \
patch.object(nfo_service, '_download_media_files', new_callable=AsyncMock):
mock_details.side_effect = side_effect
mock_ratings.return_value = {"results": []}
updated_path = await nfo_service.update_tvshow_nfo(
"Basilisk", download_media=False,
)
content = updated_path.read_text(encoding="utf-8")
assert "<plot>English fallback overview" in content
assert mock_details.call_count == 2
@pytest.mark.asyncio
async def test_no_fallback_when_german_overview_exists(
self, nfo_service, tmp_path
):
"""No English fallback call when German overview is present."""
series_folder = tmp_path / "Attack on Titan"
series_folder.mkdir()
de_data = {
"id": 1429, "name": "Attack on Titan",
"original_name": "進撃の巨人", "first_air_date": "2013-04-07",
"overview": "Vor mehreren hundert Jahren...",
"vote_average": 8.6, "vote_count": 5000,
"status": "Ended", "episode_run_time": [24],
"genres": [], "networks": [], "production_countries": [],
"poster_path": None, "backdrop_path": None,
"external_ids": {}, "credits": {"cast": []},
"images": {"logos": []},
}
with patch.object(nfo_service.tmdb_client, 'search_tv_show', new_callable=AsyncMock) as mock_search, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_details', new_callable=AsyncMock) as mock_details, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_content_ratings', new_callable=AsyncMock) as mock_ratings, \
patch.object(nfo_service, '_download_media_files', new_callable=AsyncMock):
mock_search.return_value = {
"results": [{"id": 1429, "name": "Attack on Titan", "first_air_date": "2013-04-07"}]
}
mock_details.return_value = de_data
mock_ratings.return_value = {"results": []}
nfo_path = await nfo_service.create_tvshow_nfo(
"Attack on Titan", "Attack on Titan", year=2013,
download_poster=False, download_logo=False, download_fanart=False,
)
content = nfo_path.read_text(encoding="utf-8")
assert "<plot>Vor mehreren hundert Jahren...</plot>" in content
# Only one detail call (German), no fallback needed
mock_details.assert_called_once_with(1429, append_to_response="credits,external_ids,images")
@pytest.mark.asyncio
async def test_plot_tag_always_present_even_if_empty(
self, nfo_service, tmp_path
):
"""<plot> tag should always be present, even when overview is missing
from both German and English TMDB data."""
series_folder = tmp_path / "Unknown Show"
series_folder.mkdir()
empty_data = {
"id": 99999, "name": "Unknown Show",
"original_name": "Unknown", "first_air_date": "2020-01-01",
"overview": "",
"vote_average": 0, "vote_count": 0,
"status": "Ended", "episode_run_time": [],
"genres": [], "networks": [], "production_countries": [],
"poster_path": None, "backdrop_path": None,
"external_ids": {}, "credits": {"cast": []},
"images": {"logos": []},
}
async def side_effect(tv_id, **kwargs):
# English also empty
return empty_data
with patch.object(nfo_service.tmdb_client, 'search_tv_show', new_callable=AsyncMock) as mock_search, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_details', new_callable=AsyncMock) as mock_details, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_content_ratings', new_callable=AsyncMock) as mock_ratings, \
patch.object(nfo_service, '_download_media_files', new_callable=AsyncMock):
mock_search.return_value = {
"results": [{"id": 99999, "name": "Unknown Show", "first_air_date": "2020-01-01"}]
}
mock_details.side_effect = side_effect
mock_ratings.return_value = {"results": []}
nfo_path = await nfo_service.create_tvshow_nfo(
"Unknown Show", "Unknown Show",
download_poster=False, download_logo=False, download_fanart=False,
)
content = nfo_path.read_text(encoding="utf-8")
# <plot/> (self-closing) or <plot></plot> should be present
assert "<plot" in content
class TestNFOServiceEdgeCases:
"""Test edge cases in NFO service."""
@pytest.mark.asyncio
async def test_create_nfo_with_valid_path(self, nfo_service, tmp_path):
"""Test NFO creation succeeds with valid path."""
series_folder = tmp_path / "Test Series"
series_folder.mkdir()
# Mock all necessary TMDB client methods
with patch.object(nfo_service.tmdb_client, 'search_tv_show', new_callable=AsyncMock) as mock_search, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_details', new_callable=AsyncMock) as mock_details, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_content_ratings', new_callable=AsyncMock) as mock_ratings, \
patch.object(nfo_service, '_download_media_files', new_callable=AsyncMock):
tmdb_data = {
"id": 1, "name": "Series", "first_air_date": "2020-01-01",
"original_name": "Original", "overview": "Test", "vote_average": 8.0,
"vote_count": 100, "status": "Continuing", "episode_run_time": [24],
"genres": [], "networks": [], "production_countries": [],
"external_ids": {}, "credits": {"cast": []}, "images": {"logos": []},
"poster_path": None, "backdrop_path": None
}
mock_search.return_value = {
"results": [{
"id": 1,
"name": "Series",
"first_air_date": "2020-01-01",
"overview": "Test overview.",
}]
}
mock_details.return_value = tmdb_data
mock_ratings.return_value = {"results": []}
nfo_path = await nfo_service.create_tvshow_nfo(
"Test Series",
"Test Series",
download_poster=False,
download_logo=False,
download_fanart=False
)
assert nfo_path.exists()
@pytest.mark.asyncio
async def test_create_nfo_no_tmdb_results(self, nfo_service, tmp_path):
"""Test NFO creation when TMDB returns no results."""
series_folder = tmp_path / "Test Series"
series_folder.mkdir()
with patch.object(nfo_service.tmdb_client, 'search_tv_show', new_callable=AsyncMock) as mock_search:
mock_search.return_value = {"results": []}
with pytest.raises(TMDBAPIError, match="No results found"):
await nfo_service.create_tvshow_nfo(
"Test Series",
"Test Series",
download_poster=False,
download_logo=False,
download_fanart=False
)
@pytest.mark.asyncio
async def test_update_nfo_missing_nfo(self, nfo_service, tmp_path):
"""Test NFO update when NFO doesn't exist."""
series_folder = tmp_path / "Test Series"
series_folder.mkdir()
with pytest.raises(FileNotFoundError):
await nfo_service.update_tvshow_nfo("Test Series")
@pytest.mark.asyncio
async def test_update_nfo_no_tmdb_id(self, nfo_service, tmp_path):
"""Test NFO update when NFO has no TMDB ID."""
series_folder = tmp_path / "Test Series"
series_folder.mkdir()
nfo_path = series_folder / "tvshow.nfo"
nfo_path.write_text("""<?xml version="1.0" encoding="UTF-8"?>
<tvshow>
<title>Test Series</title>
</tvshow>
""", encoding="utf-8")
with pytest.raises(TMDBAPIError, match="No TMDB ID found"):
await nfo_service.update_tvshow_nfo("Test Series")
@pytest.mark.asyncio
async def test_check_nfo_exists(self, nfo_service, tmp_path):
"""Test checking if NFO exists."""
series_folder = tmp_path / "Test Series"
series_folder.mkdir()
# NFO doesn't exist yet
exists = await nfo_service.check_nfo_exists("Test Series")
assert not exists
# Create NFO
nfo_path = series_folder / "tvshow.nfo"
nfo_path.write_text("<tvshow></tvshow>", encoding="utf-8")
# NFO now exists
exists = await nfo_service.check_nfo_exists("Test Series")
assert exists
class TestMediaDownloads:
"""Test media file (poster, logo, fanart) download functionality."""
@pytest.mark.asyncio
async def test_download_media_all_enabled(self, nfo_service, tmp_path, mock_tmdb_data):
"""Test downloading all media files when enabled."""
series_folder = tmp_path / "Test Series"
series_folder.mkdir()
with patch.object(nfo_service.image_downloader, 'download_all_media', new_callable=AsyncMock) as mock_download:
mock_download.return_value = {
"poster": True,
"logo": True,
"fanart": True
}
results = await nfo_service._download_media_files(
mock_tmdb_data,
series_folder,
download_poster=True,
download_logo=True,
download_fanart=True
)
assert results["poster"] is True
assert results["logo"] is True
assert results["fanart"] is True
mock_download.assert_called_once()
@pytest.mark.asyncio
async def test_download_media_poster_only(self, nfo_service, tmp_path, mock_tmdb_data):
"""Test downloading only poster."""
series_folder = tmp_path / "Test Series"
series_folder.mkdir()
with patch.object(nfo_service.image_downloader, 'download_all_media', new_callable=AsyncMock) as mock_download:
mock_download.return_value = {"poster": True}
results = await nfo_service._download_media_files(
mock_tmdb_data,
series_folder,
download_poster=True,
download_logo=False,
download_fanart=False
)
# Verify only poster URL was passed
call_args = mock_download.call_args
assert call_args.kwargs['poster_url'] is not None
assert call_args.kwargs['logo_url'] is None
assert call_args.kwargs['fanart_url'] is None
@pytest.mark.asyncio
async def test_download_media_with_image_size(self, nfo_service, tmp_path, mock_tmdb_data):
"""Test that image size configuration is used."""
nfo_service.image_size = "w500"
series_folder = tmp_path / "Test Series"
series_folder.mkdir()
with patch.object(nfo_service.image_downloader, 'download_all_media', new_callable=AsyncMock) as mock_download:
mock_download.return_value = {"poster": True}
await nfo_service._download_media_files(
mock_tmdb_data,
series_folder,
download_poster=True,
download_logo=False,
download_fanart=False
)
# Verify image size was used for poster
call_args = mock_download.call_args
poster_url = call_args.kwargs['poster_url']
assert "w500" in poster_url
@pytest.mark.asyncio
async def test_download_media_missing_poster_path(self, nfo_service, tmp_path):
"""Test media download when poster path is missing."""
series_folder = tmp_path / "Test Series"
series_folder.mkdir()
tmdb_data_no_poster = {
"id": 1,
"name": "Test",
"poster_path": None,
"backdrop_path": "/backdrop.jpg",
"images": {"logos": [{"file_path": "/logo.png"}]}
}
with patch.object(nfo_service.image_downloader, 'download_all_media', new_callable=AsyncMock) as mock_download:
mock_download.return_value = {}
await nfo_service._download_media_files(
tmdb_data_no_poster,
series_folder,
download_poster=True,
download_logo=True,
download_fanart=True
)
# Poster URL should be None
call_args = mock_download.call_args
assert call_args.kwargs['poster_url'] is None
@pytest.mark.asyncio
async def test_download_media_no_logo_available(self, nfo_service, tmp_path):
"""Test media download when logo is not available."""
series_folder = tmp_path / "Test Series"
series_folder.mkdir()
tmdb_data_no_logo = {
"id": 1,
"name": "Test",
"poster_path": "/poster.jpg",
"backdrop_path": "/backdrop.jpg",
"images": {"logos": []} # Empty logos array
}
with patch.object(nfo_service.image_downloader, 'download_all_media', new_callable=AsyncMock) as mock_download:
mock_download.return_value = {"poster": True, "fanart": True}
await nfo_service._download_media_files(
tmdb_data_no_logo,
series_folder,
download_poster=True,
download_logo=True,
download_fanart=True
)
# Logo URL should be None
call_args = mock_download.call_args
assert call_args.kwargs['logo_url'] is None
@pytest.mark.asyncio
async def test_download_media_all_disabled(self, nfo_service, tmp_path, mock_tmdb_data):
"""Test that no downloads occur when all disabled."""
series_folder = tmp_path / "Test Series"
series_folder.mkdir()
with patch.object(nfo_service.image_downloader, 'download_all_media', new_callable=AsyncMock) as mock_download:
mock_download.return_value = {}
await nfo_service._download_media_files(
mock_tmdb_data,
series_folder,
download_poster=False,
download_logo=False,
download_fanart=False
)
# All URLs should be None
call_args = mock_download.call_args
assert call_args.kwargs['poster_url'] is None
assert call_args.kwargs['logo_url'] is None
assert call_args.kwargs['fanart_url'] is None
@pytest.mark.asyncio
async def test_download_media_fanart_uses_original_size(self, nfo_service, tmp_path, mock_tmdb_data):
"""Test that fanart always uses original size regardless of config."""
nfo_service.image_size = "w500"
series_folder = tmp_path / "Test Series"
series_folder.mkdir()
with patch.object(nfo_service.image_downloader, 'download_all_media', new_callable=AsyncMock) as mock_download:
mock_download.return_value = {"fanart": True}
await nfo_service._download_media_files(
mock_tmdb_data,
series_folder,
download_poster=False,
download_logo=False,
download_fanart=True
)
# Fanart should use original size
call_args = mock_download.call_args
fanart_url = call_args.kwargs['fanart_url']
assert "original" in fanart_url
@pytest.mark.asyncio
async def test_download_media_logo_uses_original_size(self, nfo_service, tmp_path, mock_tmdb_data):
"""Test that logo always uses original size."""
nfo_service.image_size = "w500"
series_folder = tmp_path / "Test Series"
series_folder.mkdir()
with patch.object(nfo_service.image_downloader, 'download_all_media', new_callable=AsyncMock) as mock_download:
mock_download.return_value = {"logo": True}
await nfo_service._download_media_files(
mock_tmdb_data,
series_folder,
download_poster=False,
download_logo=True,
download_fanart=False
)
# Logo should use original size
call_args = mock_download.call_args
logo_url = call_args.kwargs['logo_url']
assert "original" in logo_url
class TestNFOServiceConfiguration:
"""Test NFO service with various configuration settings."""
def test_nfo_service_default_config(self, tmp_path):
"""Test NFO service initialization with default config."""
service = NFOService(
tmdb_api_key="test_key",
anime_directory=str(tmp_path)
)
assert service.image_size == "original"
assert service.auto_create is True
def test_nfo_service_custom_config(self, tmp_path):
"""Test NFO service initialization with custom config."""
service = NFOService(
tmdb_api_key="test_key",
anime_directory=str(tmp_path),
image_size="w500",
auto_create=False
)
assert service.image_size == "w500"
assert service.auto_create is False
def test_nfo_service_image_sizes(self, tmp_path):
"""Test NFO service with various image sizes."""
sizes = ["original", "w500", "w780", "w342"]
for size in sizes:
service = NFOService(
tmdb_api_key="test_key",
anime_directory=str(tmp_path),
image_size=size
)
assert service.image_size == size
class TestHasNFOMethod:
"""Test the has_nfo method."""
def test_has_nfo_true(self, nfo_service, tmp_path):
"""Test has_nfo returns True when NFO exists."""
series_folder = tmp_path / "Test Series"
series_folder.mkdir()
nfo_path = series_folder / "tvshow.nfo"
nfo_path.write_text("<tvshow></tvshow>")
assert nfo_service.has_nfo("Test Series") is True
def test_has_nfo_false(self, nfo_service, tmp_path):
"""Test has_nfo returns False when NFO doesn't exist."""
series_folder = tmp_path / "Test Series"
series_folder.mkdir()
assert nfo_service.has_nfo("Test Series") is False
def test_has_nfo_missing_folder(self, nfo_service):
"""Test has_nfo returns False when folder doesn't exist."""
assert nfo_service.has_nfo("Nonexistent Series") is False
class TestFindBestMatchEdgeCases:
"""Test edge cases in _find_best_match."""
def test_find_best_match_no_year_multiple_results(self, nfo_service):
"""Test finding best match returns first result when no year."""
results = [
{"id": 1, "name": "Series", "first_air_date": "2010-01-01"},
{"id": 2, "name": "Series", "first_air_date": "2020-01-01"},
]
match = nfo_service._find_best_match(results, "Series", year=None)
assert match["id"] == 1
def test_find_best_match_year_no_match(self, nfo_service):
"""Test finding best match with year when no exact match returns first."""
results = [
{"id": 1, "name": "Series", "first_air_date": "2010-01-01"},
{"id": 2, "name": "Series", "first_air_date": "2020-01-01"},
]
match = nfo_service._find_best_match(results, "Series", year=2025)
# Should return first result as no year match found
assert match["id"] == 1
def test_find_best_match_empty_results(self, nfo_service):
"""Test finding best match with empty results raises error."""
with pytest.raises(TMDBAPIError, match="No search results"):
nfo_service._find_best_match([], "Series")
def test_find_best_match_no_first_air_date(self, nfo_service):
"""Test finding best match when result has no first_air_date."""
results = [
{"id": 1, "name": "Series"}, # No first_air_date
{"id": 2, "name": "Series", "first_air_date": "2020-01-01"},
]
# With year, should check for first_air_date existence
match = nfo_service._find_best_match(results, "Series", year=2020)
assert match["id"] == 2
class TestParseNFOIDsEdgeCases:
"""Test edge cases in parse_nfo_ids."""
def test_parse_nfo_ids_malformed_ids(self, nfo_service, tmp_path):
"""Test parsing IDs with malformed values."""
nfo_path = tmp_path / "tvshow.nfo"
nfo_path.write_text(
'<tvshow>'
'<uniqueid type="tmdb">not_a_number</uniqueid>'
'<uniqueid type="tvdb">abc123</uniqueid>'
'</tvshow>'
)
ids = nfo_service.parse_nfo_ids(nfo_path)
# Malformed values should be None
assert ids["tmdb_id"] is None
assert ids["tvdb_id"] is None
def test_parse_nfo_ids_multiple_uniqueid(self, nfo_service, tmp_path):
"""Test parsing when multiple uniqueid elements exist."""
nfo_path = tmp_path / "tvshow.nfo"
nfo_path.write_text(
'<tvshow>'
'<uniqueid type="tmdb">1429</uniqueid>'
'<uniqueid type="tvdb">79168</uniqueid>'
'<uniqueid type="imdb">tt2560140</uniqueid>'
'</tvshow>'
)
ids = nfo_service.parse_nfo_ids(nfo_path)
assert ids["tmdb_id"] == 1429
assert ids["tvdb_id"] == 79168
def test_parse_nfo_ids_empty_uniqueid(self, nfo_service, tmp_path):
"""Test parsing with empty uniqueid elements."""
nfo_path = tmp_path / "tvshow.nfo"
nfo_path.write_text(
'<tvshow>'
'<uniqueid type="tmdb"></uniqueid>'
'<uniqueid type="tvdb"></uniqueid>'
'</tvshow>'
)
ids = nfo_service.parse_nfo_ids(nfo_path)
assert ids["tmdb_id"] is None
assert ids["tvdb_id"] is None
class TestTMDBToNFOModelEdgeCases:
"""Test edge cases in _tmdb_to_nfo_model."""
def test_tmdb_to_nfo_minimal_data(self, nfo_service):
"""Test conversion with minimal TMDB data."""
minimal_data = {
"id": 1,
"name": "Series",
"original_name": "Original"
}
nfo_model = tmdb_to_nfo_model(
minimal_data, None,
nfo_service.tmdb_client.get_image_url, nfo_service.image_size
)
assert nfo_model.title == "Series"
assert nfo_model.originaltitle == "Original"
assert nfo_model.year is None
assert nfo_model.tmdbid == 1
def test_tmdb_to_nfo_with_all_cast(self, nfo_service, mock_tmdb_data):
"""Test conversion includes cast members."""
nfo_model = tmdb_to_nfo_model(
mock_tmdb_data, None,
nfo_service.tmdb_client.get_image_url, nfo_service.image_size
)
assert len(nfo_model.actors) >= 1
assert nfo_model.actors[0].name == "Yuki Kaji"
assert nfo_model.actors[0].role == "Eren Yeager"
def test_tmdb_to_nfo_multiple_genres(self, nfo_service, mock_tmdb_data):
"""Test conversion with multiple genres."""
nfo_model = tmdb_to_nfo_model(
mock_tmdb_data, None,
nfo_service.tmdb_client.get_image_url, nfo_service.image_size
)
assert "Animation" in nfo_model.genre
assert "Sci-Fi & Fantasy" in nfo_model.genre
class TestExtractFSKRatingEdgeCases:
"""Test edge cases in _extract_fsk_rating."""
def test_extract_fsk_with_suffix(self, nfo_service):
"""Test extraction when rating has suffix like 'Ab 16 Jahren'."""
content_ratings = {
"results": [{"iso_3166_1": "DE", "rating": "Ab 16 Jahren"}]
}
fsk = _extract_fsk_rating(content_ratings)
assert fsk == "FSK 16"
def test_extract_fsk_multiple_numbers(self, nfo_service):
"""Test extraction with multiple numbers - should pick highest."""
content_ratings = {
"results": [{"iso_3166_1": "DE", "rating": "Rating 6 or 12"}]
}
fsk = _extract_fsk_rating(content_ratings)
# Should find 12 first in the search order
assert fsk == "FSK 12"
def test_extract_fsk_empty_results_list(self, nfo_service):
"""Test extraction with empty results list."""
content_ratings = {"results": []}
fsk = _extract_fsk_rating(content_ratings)
assert fsk is None
def test_extract_fsk_none_input(self, nfo_service):
"""Test extraction with None input."""
fsk = _extract_fsk_rating(None)
assert fsk is None
def test_extract_fsk_missing_results_key(self, nfo_service):
"""Test extraction when results key is missing."""
fsk = _extract_fsk_rating({})
assert fsk is None
class TestDownloadMediaFilesEdgeCases:
"""Test edge cases in _download_media_files."""
@pytest.mark.asyncio
async def test_download_media_empty_tmdb_data(self, nfo_service, tmp_path):
"""Test media download with empty TMDB data."""
series_folder = tmp_path / "Test Series"
series_folder.mkdir()
with patch.object(nfo_service.image_downloader, 'download_all_media', new_callable=AsyncMock) as mock_download:
mock_download.return_value = {}
results = await nfo_service._download_media_files(
{},
series_folder,
download_poster=True,
download_logo=True,
download_fanart=True
)
# Should call download with all None URLs
call_args = mock_download.call_args
assert call_args.kwargs['poster_url'] is None
assert call_args.kwargs['logo_url'] is None
assert call_args.kwargs['fanart_url'] is None
@pytest.mark.asyncio
async def test_download_media_only_poster_available(self, nfo_service, tmp_path):
"""Test media download when only poster is available."""
series_folder = tmp_path / "Test Series"
series_folder.mkdir()
tmdb_data = {
"id": 1,
"poster_path": "/poster.jpg",
"backdrop_path": None,
"images": {"logos": []}
}
with patch.object(nfo_service.image_downloader, 'download_all_media', new_callable=AsyncMock) as mock_download:
mock_download.return_value = {"poster": True}
await nfo_service._download_media_files(
tmdb_data,
series_folder,
download_poster=True,
download_logo=True,
download_fanart=True
)
call_args = mock_download.call_args
assert call_args.kwargs['poster_url'] is not None
assert call_args.kwargs['fanart_url'] is None
assert call_args.kwargs['logo_url'] is None
class TestUpdateNFOEdgeCases:
"""Test edge cases in update_tvshow_nfo."""
@pytest.mark.asyncio
async def test_update_nfo_without_media_download(self, nfo_service, tmp_path, mock_tmdb_data, mock_content_ratings_de):
"""Test NFO update without re-downloading media."""
series_folder = tmp_path / "Attack on Titan"
series_folder.mkdir()
nfo_path = series_folder / "tvshow.nfo"
nfo_path.write_text(
'<tvshow><uniqueid type="tmdb">1429</uniqueid></tvshow>',
encoding="utf-8"
)
with patch.object(nfo_service.tmdb_client, 'get_tv_show_details', new_callable=AsyncMock) as mock_details, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_content_ratings', new_callable=AsyncMock) as mock_ratings, \
patch.object(nfo_service, '_download_media_files', new_callable=AsyncMock) as mock_download:
mock_details.return_value = mock_tmdb_data
mock_ratings.return_value = mock_content_ratings_de
await nfo_service.update_tvshow_nfo("Attack on Titan", download_media=False)
# Verify download was not called
mock_download.assert_not_called()
class TestNFOServiceClose:
"""Test NFO service cleanup and close."""
@pytest.mark.asyncio
async def test_nfo_service_close(self, nfo_service):
"""Test NFO service close."""
with patch.object(nfo_service.tmdb_client, 'close', new_callable=AsyncMock) as mock_close:
await nfo_service.close()
mock_close.assert_called_once()
class TestYearExtractionComprehensive:
"""Comprehensive tests for year extraction."""
def test_extract_year_with_leading_spaces(self, nfo_service):
"""Test extraction with leading spaces - they get stripped."""
clean_name, year = nfo_service._extract_year_from_name(" Attack on Titan (2013)")
assert clean_name == "Attack on Titan" # Leading spaces are stripped
assert year == 2013
def test_extract_year_with_year_in_middle(self, nfo_service):
"""Test that year in middle doesn't get extracted."""
clean_name, year = nfo_service._extract_year_from_name("Attack on Titan 2013")
assert clean_name == "Attack on Titan 2013"
assert year is None
def test_extract_year_three_digit(self, nfo_service):
"""Test that 3-digit number is not extracted."""
clean_name, year = nfo_service._extract_year_from_name("Series (123)")
assert clean_name == "Series (123)"
assert year is None
def test_extract_year_five_digit(self, nfo_service):
"""Test that 5-digit number is not extracted."""
clean_name, year = nfo_service._extract_year_from_name("Series (12345)")
assert clean_name == "Series (12345)"
assert year is None
class TestEnrichFallbackLanguages:
"""Tests for multi-language fallback and search overview fallback."""
@pytest.mark.asyncio
async def test_japanese_fallback_when_english_also_empty(
self, nfo_service, tmp_path,
):
"""ja-JP fallback is tried when both de-DE and en-US are empty."""
series_folder = tmp_path / "Rare Anime"
series_folder.mkdir()
de_data = {
"id": 55555, "name": "Rare Anime",
"original_name": "レアアニメ", "first_air_date": "2024-01-01",
"overview": "",
"vote_average": 7.0, "vote_count": 50,
"status": "Continuing", "episode_run_time": [24],
"genres": [], "networks": [], "production_countries": [],
"poster_path": None, "backdrop_path": None,
"external_ids": {}, "credits": {"cast": []},
"images": {"logos": []},
}
en_data = {"id": 55555, "overview": ""}
ja_data = {"id": 55555, "overview": "日本語のあらすじ"}
async def side_effect(tv_id, **kwargs):
lang = kwargs.get("language")
if lang == "ja-JP":
return ja_data
if lang == "en-US":
return en_data
return de_data
with patch.object(nfo_service.tmdb_client, 'search_tv_show', new_callable=AsyncMock) as mock_search, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_details', new_callable=AsyncMock) as mock_details, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_content_ratings', new_callable=AsyncMock) as mock_ratings, \
patch.object(nfo_service, '_download_media_files', new_callable=AsyncMock):
mock_search.return_value = {
"results": [{"id": 55555, "name": "Rare Anime", "first_air_date": "2024-01-01"}],
}
mock_details.side_effect = side_effect
mock_ratings.return_value = {"results": []}
nfo_path = await nfo_service.create_tvshow_nfo(
"Rare Anime", "Rare Anime",
download_poster=False, download_logo=False, download_fanart=False,
)
content = nfo_path.read_text(encoding="utf-8")
assert "<plot>日本語のあらすじ</plot>" in content
@pytest.mark.asyncio
async def test_search_overview_fallback_when_all_languages_empty(
self, nfo_service, tmp_path,
):
"""Search result overview is used as last resort."""
series_folder = tmp_path / "Brand New Anime"
series_folder.mkdir()
empty_data = {
"id": 77777, "name": "Brand New Anime",
"original_name": "新しいアニメ", "first_air_date": "2025-01-01",
"overview": "",
"vote_average": 0, "vote_count": 0,
"status": "Continuing", "episode_run_time": [],
"genres": [], "networks": [], "production_countries": [],
"poster_path": None, "backdrop_path": None,
"external_ids": {}, "credits": {"cast": []},
"images": {"logos": []},
}
async def side_effect(tv_id, **kwargs):
return empty_data
with patch.object(nfo_service.tmdb_client, 'search_tv_show', new_callable=AsyncMock) as mock_search, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_details', new_callable=AsyncMock) as mock_details, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_content_ratings', new_callable=AsyncMock) as mock_ratings, \
patch.object(nfo_service, '_download_media_files', new_callable=AsyncMock):
# Search result DOES have an overview
mock_search.return_value = {
"results": [{
"id": 77777,
"name": "Brand New Anime",
"first_air_date": "2025-01-01",
"overview": "Search result overview text.",
}],
}
mock_details.side_effect = side_effect
mock_ratings.return_value = {"results": []}
nfo_path = await nfo_service.create_tvshow_nfo(
"Brand New Anime", "Brand New Anime",
download_poster=False, download_logo=False, download_fanart=False,
)
content = nfo_path.read_text(encoding="utf-8")
assert "<plot>Search result overview text.</plot>" in content
@pytest.mark.asyncio
async def test_en_us_search_fallback_when_german_search_overview_empty(
self, nfo_service, tmp_path
):
"""When the German search overview is empty, fallback to en-US search overview."""
series_folder = tmp_path / "Rare Anime"
series_folder.mkdir()
empty_data = {
"id": 77777, "name": "Rare Anime",
"original_name": "新しいアニメ", "first_air_date": "2025-01-01",
"overview": "",
"vote_average": 0, "vote_count": 0,
"status": "Continuing", "episode_run_time": [],
"genres": [], "networks": [], "production_countries": [],
"poster_path": None, "backdrop_path": None,
"external_ids": {}, "credits": {"cast": []},
"images": {"logos": []},
}
async def search_side_effect(query, language="de-DE", page=1):
if language == "en-US":
return {
"results": [{
"id": 77777,
"name": "Rare Anime",
"first_air_date": "2025-01-01",
"overview": "English search overview text.",
}],
}
return {
"results": [{
"id": 77777,
"name": "Rare Anime",
"first_air_date": "2025-01-01",
"overview": "",
}],
}
async def details_side_effect(tv_id, **kwargs):
return empty_data
with patch.object(nfo_service.tmdb_client, 'search_tv_show', new_callable=AsyncMock) as mock_search, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_details', new_callable=AsyncMock) as mock_details, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_content_ratings', new_callable=AsyncMock) as mock_ratings, \
patch.object(nfo_service, '_download_media_files', new_callable=AsyncMock):
mock_search.side_effect = search_side_effect
mock_details.side_effect = details_side_effect
mock_ratings.return_value = {"results": []}
nfo_path = await nfo_service.create_tvshow_nfo(
"Rare Anime", "Rare Anime",
download_poster=False, download_logo=False, download_fanart=False,
)
content = nfo_path.read_text(encoding="utf-8")
assert "<plot>English search overview text.</plot>" in content
assert mock_search.call_count == 2
assert mock_search.call_args_list[1].kwargs['language'] == 'en-US'
@pytest.mark.asyncio
async def test_no_japanese_fallback_when_english_succeeds(
self, nfo_service, tmp_path,
):
"""Stop after en-US if it provides the overview."""
series_folder = tmp_path / "Test Anime"
series_folder.mkdir()
de_data = {
"id": 88888, "name": "Test Anime",
"original_name": "テスト", "first_air_date": "2024-01-01",
"overview": "",
"vote_average": 7.0, "vote_count": 50,
"status": "Continuing", "episode_run_time": [24],
"genres": [], "networks": [], "production_countries": [],
"poster_path": None, "backdrop_path": None,
"external_ids": {}, "credits": {"cast": []},
"images": {"logos": []},
}
en_data = {"id": 88888, "overview": "English overview."}
async def side_effect(tv_id, **kwargs):
lang = kwargs.get("language")
if lang == "en-US":
return en_data
return de_data
with patch.object(nfo_service.tmdb_client, 'search_tv_show', new_callable=AsyncMock) as mock_search, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_details', new_callable=AsyncMock) as mock_details, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_content_ratings', new_callable=AsyncMock) as mock_ratings, \
patch.object(nfo_service, '_download_media_files', new_callable=AsyncMock):
mock_search.return_value = {
"results": [{"id": 88888, "name": "Test Anime", "first_air_date": "2024-01-01"}],
}
mock_details.side_effect = side_effect
mock_ratings.return_value = {"results": []}
nfo_path = await nfo_service.create_tvshow_nfo(
"Test Anime", "Test Anime",
download_poster=False, download_logo=False, download_fanart=False,
)
content = nfo_path.read_text(encoding="utf-8")
assert "<plot>English overview.</plot>" in content
# 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
class TestNFOIDOverride:
"""Tests for manual TMDB ID override via NFO."""
@pytest.mark.asyncio
async def test_create_tvshow_nfo_uses_existing_tmdb_id(self, nfo_service, tmp_path, mock_tmdb_data, mock_content_ratings_de):
"""Test that existing TMDB ID in NFO skips search."""
# Create series folder with existing NFO containing TMDB ID
series_folder = tmp_path / "Attack on Titan"
series_folder.mkdir()
nfo_path = series_folder / "tvshow.nfo"
nfo_path.write_text("""<?xml version="1.0" encoding="UTF-8"?>
<tvshow>
<title>Attack on Titan</title>
<tmdbid>1429</tmdbid>
</tvshow>
""", encoding="utf-8")
with patch.object(nfo_service.tmdb_client, 'get_tv_show_details', new_callable=AsyncMock) as mock_details, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_content_ratings', new_callable=AsyncMock) as mock_ratings, \
patch.object(nfo_service, '_download_media_files', new_callable=AsyncMock), \
patch.object(nfo_service.tmdb_client, '_ensure_session', new_callable=AsyncMock):
mock_details.return_value = mock_tmdb_data
mock_ratings.return_value = mock_content_ratings_de
nfo_path_result = await nfo_service.create_tvshow_nfo(
"Attack on Titan",
"Attack on Titan",
download_poster=False, download_logo=False, download_fanart=False
)
# Verify NFO was created
assert nfo_path_result.exists()
# Verify get_tv_show_details was called directly with the ID (no search)
mock_details.assert_called_once_with(1429, append_to_response="credits,external_ids,images")
# Verify search was NOT called
# (we can check by verifying no search_tv_show mock was set up)
@pytest.mark.asyncio
async def test_create_tvshow_nfo_searches_when_no_tmdb_id(self, nfo_service, tmp_path, mock_tmdb_data, mock_content_ratings_de):
"""Test that search is used when NFO has no TMDB ID."""
# Create series folder without existing NFO
series_folder = tmp_path / "Test Anime"
series_folder.mkdir()
with patch.object(nfo_service.tmdb_client, 'search_tv_show', new_callable=AsyncMock) as mock_search, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_details', new_callable=AsyncMock) as mock_details, \
patch.object(nfo_service.tmdb_client, 'get_tv_show_content_ratings', new_callable=AsyncMock) as mock_ratings, \
patch.object(nfo_service, '_download_media_files', new_callable=AsyncMock), \
patch.object(nfo_service.tmdb_client, '_ensure_session', new_callable=AsyncMock):
mock_search.return_value = {
"results": [{
"id": 1429,
"name": "Test Anime",
"first_air_date": "2024-01-01",
"overview": "Test overview"
}]
}
mock_details.return_value = mock_tmdb_data
mock_ratings.return_value = mock_content_ratings_de
nfo_path = await nfo_service.create_tvshow_nfo(
"Test Anime",
"Test Anime",
download_poster=False, download_logo=False, download_fanart=False
)
# Verify search was called
mock_search.assert_called()
class TestSearchMultiStrategy:
"""Tests for search/multi fallback strategy."""
@pytest.mark.asyncio
async def test_search_multi_strategy_used_as_fallback(self, nfo_service, mock_tmdb_data):
"""Test that search/multi is tried after regular search fails."""
mock_search = AsyncMock()
mock_multi = AsyncMock()
# First: regular search fails
# Second: multi search returns TV result
mock_search.return_value = {"results": []}
mock_multi.return_value = {
"results": [
{"media_type": "movie", "id": 123},
{"media_type": "tv", "id": 456, "name": "Found Show", "first_air_date": "2024-01-01"}
]
}
with patch.object(nfo_service.tmdb_client, 'search_tv_show', mock_search), \
patch.object(nfo_service.tmdb_client, 'search_multi', mock_multi):
result, source = await nfo_service._search_with_fallback(
"Unknown Show", 2024, None
)
assert result["id"] == 456
assert source == "multi_search"