Lukas
a06abaa2e5
- Fix test_data_file_db_sync.py: Remove unused mock logger parameters - Fix test_nfo_workflow.py: Add missing async mocks for TMDB methods * Add get_tv_show_content_ratings mock for FSK rating support * Add get_image_url mock to return proper URL strings * Fix test_nfo_update_workflow to include TMDB ID in existing NFO - Fix DownloadService method calls in test fixtures * Change stop() to stop_downloads() (correct method name) * Change start() to start_queue_processing() * Add exception handling for ProgressServiceError in teardown - All 1380 tests now passing with 0 failures and 0 errors
413 lines
15 KiB
Python
413 lines
15 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.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.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.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.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.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"}]},
|
|
]
|
|
)
|
|
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"},
|
|
]
|
|
)
|
|
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.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
|