"""
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 pytest
import tempfile
from pathlib import Path
from unittest.mock import Mock, patch, AsyncMock
import json
@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
from src.server.database.connection import init_db, get_db_session
from src.server.database.models import AnimeSeries
with tempfile.TemporaryDirectory() as tmp_dir:
# Initialize database
db_path = Path(tmp_dir) / "test.db"
await init_db(f"sqlite:///{db_path}")
# 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.get_tv_show = AsyncMock(return_value=mock_tmdb_show)
mock_tmdb.get_tv_season = AsyncMock(
return_value=mock_tmdb_season
)
mock_tmdb.download_image = AsyncMock(return_value=True)
# 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",
download_poster=True,
download_fanart=True,
download_logo=False,
image_size="w500",
)
# Create anime series in database
async with get_db_session() as db:
anime = AnimeSeries(
tmdb_id=None,
key="attack-on-titan",
folder="Attack on Titan",
name="Attack on Titan",
original_name="進撃の巨人",
status="Ended",
year=2013,
)
db.add(anime)
await db.commit()
await db.refresh(anime)
anime_id = anime.id
# Step 1: Create NFO files
result = await nfo_service.create_nfo_files(
folder_path=str(anime_dir),
anime_id=anime_id,
anime_name="Attack on Titan",
year=2013,
)
assert result["success"] is True
assert "files_created" in result
assert len(result["files_created"]) >= 3 # tvshow + 2 episodes
# Step 2: Verify NFO files created
tvshow_nfo = anime_dir / "tvshow.nfo"
assert tvshow_nfo.exists()
assert tvshow_nfo.stat().st_size > 0
episode1_nfo = season1_dir / "S01E01.nfo"
assert episode1_nfo.exists()
episode2_nfo = season1_dir / "S01E02.nfo"
assert episode2_nfo.exists()
# 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 "" in content
assert "" in content
assert "1429" in content # TMDB ID
assert "Animation" in content
# Step 4: Verify images downloaded
poster = anime_dir / "poster.jpg"
# Image download was mocked, check it was called
assert mock_tmdb.download_image.call_count >= 2
# Step 5: Verify database updated
async with get_db_session() as db:
anime = await db.get(AnimeSeries, anime_id)
assert anime is not None
assert anime.tmdb_id == 1429
assert anime.has_nfo is True
assert anime.nfo_file_count >= 3
async def test_nfo_workflow_handles_missing_episodes(self):
"""Test NFO creation only for episodes that have video files."""
from src.core.services.nfo_service import NFOService
from src.server.database.connection import init_db
with tempfile.TemporaryDirectory() as tmp_dir:
db_path = Path(tmp_dir) / "test.db"
await init_db(f"sqlite:///{db_path}")
# Create anime directory with partial episodes
anime_dir = Path(tmp_dir) / "Test Anime"
season1_dir = anime_dir / "Season 1"
season1_dir.mkdir(parents=True)
# Only create episode 1 and 3, skip 2
(season1_dir / "S01E01.mkv").touch()
(season1_dir / "S01E03.mkv").touch()
mock_tmdb = Mock()
mock_tmdb.get_tv_show = AsyncMock(
return_value={
"id": 999,
"name": "Test Anime",
"first_air_date": "2020-01-01",
}
)
mock_tmdb.get_tv_season = AsyncMock(
return_value={
"season_number": 1,
"episodes": [
{
"episode_number": 1,
"name": "Episode 1",
"air_date": "2020-01-01",
},
{
"episode_number": 2,
"name": "Episode 2",
"air_date": "2020-01-08",
},
{
"episode_number": 3,
"name": "Episode 3",
"air_date": "2020-01-15",
},
],
}
)
with patch(
"src.core.services.nfo_service.TMDBClient",
return_value=mock_tmdb,
):
nfo_service = NFOService(tmdb_api_key="test_key")
result = await nfo_service.create_nfo_files(
folder_path=str(anime_dir),
anime_id=999,
anime_name="Test Anime",
)
# Should create NFOs only for existing episodes
assert result["success"] is True
assert (season1_dir / "S01E01.nfo").exists()
assert not (season1_dir / "S01E02.nfo").exists()
assert (season1_dir / "S01E03.nfo").exists()
async def test_nfo_workflow_error_recovery(self):
"""Test NFO workflow continues on partial failures."""
from src.core.services.nfo_service import NFOService
from src.server.database.connection import init_db
with tempfile.TemporaryDirectory() as tmp_dir:
db_path = Path(tmp_dir) / "test.db"
await init_db(f"sqlite:///{db_path}")
anime_dir = Path(tmp_dir) / "Test Anime"
season1_dir = anime_dir / "Season 1"
season1_dir.mkdir(parents=True)
(season1_dir / "S01E01.mkv").touch()
# Mock TMDB to fail on episode but succeed on show
mock_tmdb = Mock()
mock_tmdb.get_tv_show = AsyncMock(
return_value={
"id": 999,
"name": "Test Anime",
"first_air_date": "2020-01-01",
}
)
mock_tmdb.get_tv_season = AsyncMock(
side_effect=Exception("TMDB API error")
)
with patch(
"src.core.services.nfo_service.TMDBClient",
return_value=mock_tmdb,
):
nfo_service = NFOService(tmdb_api_key="test_key")
result = await nfo_service.create_nfo_files(
folder_path=str(anime_dir),
anime_id=999,
anime_name="Test Anime",
)
# Should succeed for tvshow.nfo even if episodes fail
assert (anime_dir / "tvshow.nfo").exists()
# Episode NFO should not exist due to API error
assert not (season1_dir / "S01E01.nfo").exists()
async def test_nfo_update_workflow(self):
"""Test updating existing NFO files with new metadata."""
from src.core.services.nfo_service import NFOService
from src.server.database.connection import init_db
with tempfile.TemporaryDirectory() as tmp_dir:
db_path = Path(tmp_dir) / "test.db"
await init_db(f"sqlite:///{db_path}")
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(
"""
Test Anime
2020
"""
)
mock_tmdb = Mock()
mock_tmdb.get_tv_show = AsyncMock(
return_value={
"id": 999,
"name": "Test Anime Updated",
"overview": "New description",
"first_air_date": "2020-01-01",
"vote_average": 9.0,
}
)
with patch(
"src.core.services.nfo_service.TMDBClient",
return_value=mock_tmdb,
):
nfo_service = NFOService(tmdb_api_key="test_key")
result = await nfo_service.update_nfo_files(
folder_path=str(anime_dir),
anime_id=999,
force=True,
)
assert result["success"] is True
# Verify NFO updated
content = tvshow_nfo.read_text()
assert "Test Anime Updated" in content
assert "New description" in content
assert "9.0" 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
from src.server.database.connection import init_db, get_db_session
from src.server.database.models import AnimeSeries
with tempfile.TemporaryDirectory() as tmp_dir:
db_path = Path(tmp_dir) / "test.db"
await init_db(f"sqlite:///{db_path}")
# 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)
# Create anime in database
async with get_db_session() as db:
anime1 = AnimeSeries(
key="anime-1",
folder="Anime 1",
name="Anime 1",
)
anime2 = AnimeSeries(
key="anime-2",
folder="Anime 2",
name="Anime 2",
)
db.add_all([anime1, anime2])
await db.commit()
mock_tmdb = Mock()
mock_tmdb.get_tv_show = AsyncMock(
return_value={
"id": 999,
"name": "Test Anime",
"first_air_date": "2020-01-01",
}
)
with patch(
"src.core.services.nfo_service.TMDBClient",
return_value=mock_tmdb,
):
nfo_service = NFOService(tmdb_api_key="test_key")
# Batch create NFOs
# Note: This would typically be done through the API
result1 = await nfo_service.create_nfo_files(
folder_path=str(anime1_dir),
anime_id=1,
anime_name="Anime 1",
)
result2 = await nfo_service.create_nfo_files(
folder_path=str(anime2_dir),
anime_id=2,
anime_name="Anime 2",
)
assert result1["success"] is True
assert result2["success"] is True
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 files created automatically during episode download."""
from src.server.services.download_service import DownloadService
from src.server.services.anime_service import AnimeService
from src.server.services.progress_service import ProgressService
from src.core.services.nfo_service import NFOService
from src.server.database.connection import init_db
with tempfile.TemporaryDirectory() as tmp_dir:
db_path = Path(tmp_dir) / "test.db"
await init_db(f"sqlite:///{db_path}")
anime_dir = Path(tmp_dir) / "Test Anime"
anime_dir.mkdir(parents=True)
# Create NFO service
mock_tmdb = Mock()
mock_tmdb.get_tv_show = AsyncMock(
return_value={
"id": 999,
"name": "Test Anime",
"first_air_date": "2020-01-01",
}
)
mock_tmdb.get_tv_season = AsyncMock(
return_value={
"season_number": 1,
"episodes": [
{
"episode_number": 1,
"name": "Episode 1",
"air_date": "2020-01-01",
}
],
}
)
with patch(
"src.core.services.nfo_service.TMDBClient",
return_value=mock_tmdb,
):
nfo_service = NFOService(tmdb_api_key="test_key")
# Simulate download completion
season_dir = anime_dir / "Season 1"
season_dir.mkdir()
(season_dir / "S01E01.mkv").touch()
# Create NFO after download
result = await nfo_service.create_episode_nfo(
folder_path=str(anime_dir),
season=1,
episode=1,
tmdb_id=999,
)
# Verify NFO created
episode_nfo = season_dir / "S01E01.nfo"
assert episode_nfo.exists()
content = episode_nfo.read_text()
assert "Episode 1" in content
assert "1" in content
assert "1" in content