Lukas
ceac22fc34
- Add missing TMDB async mock methods (_ensure_session, close) to all TMDB mocks in test_nfo_workflow.py - Refactor test_anime_add_nfo_isolation.py to mock get_nfo_factory() instead of asserting on series_app.nfo_service directly - Patch get_nfo_factory in test_background_loader_service.py to align with factory-based NFOService creation Fixes test failures caused by NFOService refactoring that introduced explicit TMDB session lifecycle and NFO factory pattern.
429 lines
16 KiB
Python
429 lines
16 KiB
Python
"""
|
|
Integration test for complete NFO workflow.
|
|
|
|
Tests the end-to-end NFO creation process including:
|
|
- TMDB metadata retrieval
|
|
- NFO file generation
|
|
- Image downloads
|
|
- Database updates
|
|
"""
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestCompleteNFOWorkflow:
|
|
"""Test complete NFO creation workflow from start to finish."""
|
|
|
|
async def test_complete_nfo_workflow_with_all_features(self):
|
|
"""
|
|
Test complete NFO workflow:
|
|
1. Create NFO service with valid config
|
|
2. Fetch metadata from TMDB
|
|
3. Generate NFO files
|
|
4. Download images
|
|
5. Update database
|
|
"""
|
|
from src.core.services.nfo_service import NFOService
|
|
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
# Initialize database
|
|
|
|
# Create anime directory structure
|
|
anime_dir = Path(tmp_dir) / "Attack on Titan"
|
|
season1_dir = anime_dir / "Season 1"
|
|
season1_dir.mkdir(parents=True)
|
|
|
|
# Create dummy episode files
|
|
(season1_dir / "S01E01.mkv").touch()
|
|
(season1_dir / "S01E02.mkv").touch()
|
|
|
|
# Mock TMDB responses
|
|
mock_tmdb_show = {
|
|
"id": 1429,
|
|
"name": "Attack on Titan",
|
|
"original_name": "進撃の巨人",
|
|
"overview": "Humans are nearly exterminated...",
|
|
"first_air_date": "2013-04-07",
|
|
"vote_average": 8.5,
|
|
"vote_count": 5000,
|
|
"genres": [
|
|
{"id": 16, "name": "Animation"},
|
|
{"id": 10759, "name": "Action & Adventure"},
|
|
],
|
|
"origin_country": ["JP"],
|
|
"original_language": "ja",
|
|
"popularity": 250.0,
|
|
"status": "Ended",
|
|
"type": "Scripted",
|
|
"poster_path": "/poster.jpg",
|
|
"backdrop_path": "/fanart.jpg",
|
|
}
|
|
|
|
mock_tmdb_season = {
|
|
"id": 59321,
|
|
"season_number": 1,
|
|
"episode_count": 25,
|
|
"episodes": [
|
|
{
|
|
"id": 63056,
|
|
"episode_number": 1,
|
|
"name": "To You, in 2000 Years",
|
|
"overview": "After a hundred years...",
|
|
"air_date": "2013-04-07",
|
|
"vote_average": 8.2,
|
|
"vote_count": 100,
|
|
"still_path": "/episode1.jpg",
|
|
},
|
|
{
|
|
"id": 63057,
|
|
"episode_number": 2,
|
|
"name": "That Day",
|
|
"overview": "Eren begins training...",
|
|
"air_date": "2013-04-14",
|
|
"vote_average": 8.1,
|
|
"vote_count": 95,
|
|
"still_path": "/episode2.jpg",
|
|
},
|
|
],
|
|
}
|
|
|
|
# Mock TMDB client
|
|
mock_tmdb = Mock()
|
|
mock_tmdb.__aenter__ = AsyncMock(return_value=mock_tmdb)
|
|
mock_tmdb.__aexit__ = AsyncMock(return_value=None)
|
|
mock_tmdb._ensure_session = AsyncMock()
|
|
mock_tmdb.close = AsyncMock()
|
|
mock_tmdb.search_tv_show = AsyncMock(return_value={"results": [mock_tmdb_show]})
|
|
mock_tmdb.get_tv_show = AsyncMock(return_value=mock_tmdb_show)
|
|
mock_tmdb.get_tv_show_details = AsyncMock(return_value=mock_tmdb_show)
|
|
mock_tmdb.get_tv_show_content_ratings = AsyncMock(return_value={"results": []})
|
|
mock_tmdb.get_image_url = Mock(return_value="https://image.tmdb.org/t/p/original/test.jpg")
|
|
|
|
# Create NFO service with mocked TMDB
|
|
with patch(
|
|
"src.core.services.nfo_service.TMDBClient",
|
|
return_value=mock_tmdb,
|
|
):
|
|
nfo_service = NFOService(
|
|
tmdb_api_key="test_key",
|
|
anime_directory=tmp_dir,
|
|
image_size="w500",
|
|
)
|
|
|
|
# Step 1: Create tvshow.nfo
|
|
_ = await nfo_service.create_tvshow_nfo(
|
|
serie_name="Attack on Titan",
|
|
serie_folder="Attack on Titan",
|
|
year=2013,
|
|
download_poster=True,
|
|
download_fanart=True,
|
|
download_logo=False,
|
|
)
|
|
|
|
# Step 2: Verify NFO file created
|
|
tvshow_nfo = anime_dir / "tvshow.nfo"
|
|
assert tvshow_nfo.exists()
|
|
assert tvshow_nfo.stat().st_size > 0
|
|
|
|
# Step 3: Verify NFO content
|
|
with open(tvshow_nfo, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
assert "Attack on Titan" in content
|
|
assert "進撃の巨人" in content
|
|
assert "<tvshow>" in content
|
|
assert "</tvshow>" in content
|
|
assert "1429" in content # TMDB ID
|
|
assert "Animation" in content
|
|
|
|
# Step 4: Verify check_nfo_exists works
|
|
assert await nfo_service.check_nfo_exists("Attack on Titan")
|
|
|
|
async def test_nfo_workflow_handles_missing_episodes(self):
|
|
"""Test NFO creation with basic workflow."""
|
|
from src.core.services.nfo_service import NFOService
|
|
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
# Create anime directory with episodes
|
|
anime_dir = Path(tmp_dir) / "Test Anime"
|
|
season1_dir = anime_dir / "Season 1"
|
|
season1_dir.mkdir(parents=True)
|
|
|
|
# Create episode files
|
|
(season1_dir / "S01E01.mkv").touch()
|
|
(season1_dir / "S01E03.mkv").touch()
|
|
|
|
mock_tmdb = Mock()
|
|
mock_tmdb.__aenter__ = AsyncMock(return_value=mock_tmdb)
|
|
mock_tmdb.__aexit__ = AsyncMock(return_value=None)
|
|
mock_tmdb._ensure_session = AsyncMock()
|
|
mock_tmdb.close = AsyncMock()
|
|
mock_tmdb.search_tv_show = AsyncMock(
|
|
return_value={"results": [{
|
|
"id": 999,
|
|
"name": "Test Anime",
|
|
"first_air_date": "2020-01-01",
|
|
}]}
|
|
)
|
|
mock_tmdb.get_tv_show_details = AsyncMock(
|
|
return_value={
|
|
"id": 999,
|
|
"name": "Test Anime",
|
|
"first_air_date": "2020-01-01",
|
|
}
|
|
)
|
|
mock_tmdb.get_tv_show_content_ratings = AsyncMock(return_value={"results": []})
|
|
|
|
with patch(
|
|
"src.core.services.nfo_service.TMDBClient",
|
|
return_value=mock_tmdb,
|
|
):
|
|
nfo_service = NFOService(
|
|
tmdb_api_key="test_key",
|
|
anime_directory=tmp_dir
|
|
)
|
|
|
|
# Create tvshow.nfo
|
|
await nfo_service.create_tvshow_nfo(
|
|
serie_name="Test Anime",
|
|
serie_folder="Test Anime",
|
|
download_poster=False,
|
|
download_logo=False,
|
|
download_fanart=False,
|
|
)
|
|
|
|
# Should create tvshow.nfo
|
|
assert (anime_dir / "tvshow.nfo").exists()
|
|
|
|
async def test_nfo_workflow_error_recovery(self):
|
|
"""Test NFO workflow handles TMDB errors gracefully."""
|
|
from src.core.services.nfo_service import NFOService
|
|
from src.core.services.tmdb_client import TMDBAPIError
|
|
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
anime_dir = Path(tmp_dir) / "Test Anime"
|
|
anime_dir.mkdir(parents=True)
|
|
|
|
# Mock TMDB to fail
|
|
mock_tmdb = Mock()
|
|
mock_tmdb.__aenter__ = AsyncMock(return_value=mock_tmdb)
|
|
mock_tmdb.__aexit__ = AsyncMock(return_value=None)
|
|
mock_tmdb._ensure_session = AsyncMock()
|
|
mock_tmdb.close = AsyncMock()
|
|
mock_tmdb.search_tv_show = AsyncMock(
|
|
side_effect=TMDBAPIError("API error")
|
|
)
|
|
|
|
with patch(
|
|
"src.core.services.nfo_service.TMDBClient",
|
|
return_value=mock_tmdb,
|
|
):
|
|
nfo_service = NFOService(
|
|
tmdb_api_key="test_key",
|
|
anime_directory=tmp_dir
|
|
)
|
|
|
|
# Should raise TMDBAPIError
|
|
with pytest.raises(TMDBAPIError):
|
|
await nfo_service.create_tvshow_nfo(
|
|
serie_name="Test Anime",
|
|
serie_folder="Test Anime",
|
|
download_poster=False,
|
|
download_logo=False,
|
|
download_fanart=False,
|
|
)
|
|
|
|
async def test_nfo_update_workflow(self):
|
|
"""Test updating existing NFO files with new metadata."""
|
|
from src.core.services.nfo_service import NFOService
|
|
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
anime_dir = Path(tmp_dir) / "Test Anime"
|
|
anime_dir.mkdir(parents=True)
|
|
|
|
# Create initial NFO file
|
|
tvshow_nfo = anime_dir / "tvshow.nfo"
|
|
tvshow_nfo.write_text(
|
|
"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<tvshow>
|
|
<title>Test Anime</title>
|
|
<year>2020</year>
|
|
<uniqueid type="tmdb" default="true">999</uniqueid>
|
|
</tvshow>"""
|
|
)
|
|
|
|
mock_tmdb = Mock()
|
|
mock_tmdb.__aenter__ = AsyncMock(return_value=mock_tmdb)
|
|
mock_tmdb.__aexit__ = AsyncMock(return_value=None)
|
|
mock_tmdb._ensure_session = AsyncMock()
|
|
mock_tmdb.close = AsyncMock()
|
|
mock_tmdb.search_tv_show = AsyncMock(
|
|
return_value={"results": [{
|
|
"id": 999,
|
|
"name": "Test Anime Updated",
|
|
"overview": "New description",
|
|
"first_air_date": "2020-01-01",
|
|
"vote_average": 9.0,
|
|
}]}
|
|
)
|
|
mock_tmdb.get_tv_show_details = AsyncMock(
|
|
return_value={
|
|
"id": 999,
|
|
"name": "Test Anime Updated",
|
|
"overview": "New description",
|
|
"first_air_date": "2020-01-01",
|
|
"vote_average": 9.0,
|
|
}
|
|
)
|
|
mock_tmdb.get_tv_show_content_ratings = AsyncMock(return_value={"results": []})
|
|
|
|
with patch(
|
|
"src.core.services.nfo_service.TMDBClient",
|
|
return_value=mock_tmdb,
|
|
):
|
|
nfo_service = NFOService(
|
|
tmdb_api_key="test_key",
|
|
anime_directory=tmp_dir
|
|
)
|
|
|
|
# Update NFO
|
|
await nfo_service.update_tvshow_nfo(
|
|
serie_folder="Test Anime"
|
|
)
|
|
|
|
# Verify NFO updated
|
|
content = tvshow_nfo.read_text()
|
|
assert "Test Anime Updated" in content
|
|
assert "New description" in content
|
|
|
|
async def test_nfo_batch_creation_workflow(self):
|
|
"""Test creating NFOs for multiple anime in batch."""
|
|
from src.core.services.nfo_service import NFOService
|
|
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
# Create multiple anime directories
|
|
anime1_dir = Path(tmp_dir) / "Anime 1"
|
|
anime1_dir.mkdir(parents=True)
|
|
|
|
anime2_dir = Path(tmp_dir) / "Anime 2"
|
|
anime2_dir.mkdir(parents=True)
|
|
|
|
mock_tmdb = Mock()
|
|
mock_tmdb.__aenter__ = AsyncMock(return_value=mock_tmdb)
|
|
mock_tmdb.__aexit__ = AsyncMock(return_value=None)
|
|
mock_tmdb._ensure_session = AsyncMock()
|
|
mock_tmdb.close = AsyncMock()
|
|
mock_tmdb.search_tv_show = AsyncMock(
|
|
side_effect=[
|
|
{"results": [{"id": 1, "name": "Anime 1", "first_air_date": "2020-01-01"}]},
|
|
{"results": [{"id": 2, "name": "Anime 2", "first_air_date": "2021-01-01"}]},
|
|
{"results": [{"id": 1, "name": "Anime 1", "first_air_date": "2020-01-01"}]},
|
|
{"results": [{"id": 2, "name": "Anime 2", "first_air_date": "2021-01-01"}]},
|
|
]
|
|
)
|
|
mock_tmdb.get_tv_show_details = AsyncMock(
|
|
side_effect=[
|
|
{"id": 1, "name": "Anime 1", "first_air_date": "2020-01-01"},
|
|
{"id": 2, "name": "Anime 2", "first_air_date": "2021-01-01"},
|
|
{"id": 1, "name": "Anime 1", "first_air_date": "2020-01-01"},
|
|
{"id": 2, "name": "Anime 2", "first_air_date": "2021-01-01"},
|
|
]
|
|
)
|
|
mock_tmdb.get_tv_show_content_ratings = AsyncMock(return_value={"results": []})
|
|
|
|
with patch(
|
|
"src.core.services.nfo_service.TMDBClient",
|
|
return_value=mock_tmdb,
|
|
):
|
|
nfo_service = NFOService(
|
|
tmdb_api_key="test_key",
|
|
anime_directory=tmp_dir
|
|
)
|
|
|
|
# Create NFOs for both
|
|
await nfo_service.create_tvshow_nfo(
|
|
serie_name="Anime 1",
|
|
serie_folder="Anime 1",
|
|
download_poster=False,
|
|
download_logo=False,
|
|
download_fanart=False,
|
|
)
|
|
await nfo_service.create_tvshow_nfo(
|
|
serie_name="Anime 2",
|
|
serie_folder="Anime 2",
|
|
download_poster=False,
|
|
download_logo=False,
|
|
download_fanart=False,
|
|
)
|
|
|
|
assert (anime1_dir / "tvshow.nfo").exists()
|
|
assert (anime2_dir / "tvshow.nfo").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestNFOWorkflowWithDownloads:
|
|
"""Test NFO creation integrated with download workflow."""
|
|
|
|
async def test_nfo_created_during_download(self):
|
|
"""Test NFO creation works with the actual service."""
|
|
from src.core.services.nfo_service import NFOService
|
|
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
anime_dir = Path(tmp_dir) / "Test Anime"
|
|
anime_dir.mkdir(parents=True)
|
|
|
|
# Create NFO service
|
|
mock_tmdb = Mock()
|
|
mock_tmdb.__aenter__ = AsyncMock(return_value=mock_tmdb)
|
|
mock_tmdb.__aexit__ = AsyncMock(return_value=None)
|
|
mock_tmdb._ensure_session = AsyncMock()
|
|
mock_tmdb.close = AsyncMock()
|
|
mock_tmdb.search_tv_show = AsyncMock(
|
|
return_value={"results": [{
|
|
"id": 999,
|
|
"name": "Test Anime",
|
|
"first_air_date": "2020-01-01",
|
|
}]}
|
|
)
|
|
mock_tmdb.get_tv_show_details = AsyncMock(
|
|
return_value={
|
|
"id": 999,
|
|
"name": "Test Anime",
|
|
"first_air_date": "2020-01-01",
|
|
}
|
|
)
|
|
mock_tmdb.get_tv_show_content_ratings = AsyncMock(return_value={"results": []})
|
|
|
|
with patch(
|
|
"src.core.services.nfo_service.TMDBClient",
|
|
return_value=mock_tmdb,
|
|
):
|
|
nfo_service = NFOService(
|
|
tmdb_api_key="test_key",
|
|
anime_directory=tmp_dir
|
|
)
|
|
|
|
# Simulate download completion - create episode file
|
|
season_dir = anime_dir / "Season 1"
|
|
season_dir.mkdir()
|
|
(season_dir / "S01E01.mkv").touch()
|
|
|
|
# Create tvshow.nfo
|
|
await nfo_service.create_tvshow_nfo(
|
|
serie_name="Test Anime",
|
|
serie_folder="Test Anime",
|
|
download_poster=False,
|
|
download_logo=False,
|
|
download_fanart=False,
|
|
)
|
|
|
|
# Verify NFO created
|
|
tvshow_nfo = anime_dir / "tvshow.nfo"
|
|
assert tvshow_nfo.exists()
|
|
content = tvshow_nfo.read_text()
|
|
assert "Test Anime" in content
|