Lukas
b2379e05cf
- Root cause: Server needed restart to complete initialization - Startup process syncs data files to DB and loads into memory - Verified: GET /api/anime returns 192 anime with full metadata
490 lines
16 KiB
Python
490 lines
16 KiB
Python
"""Unit tests for background loader service optimization (no full rescans)."""
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
import pytest
|
|
|
|
from src.server.services.background_loader_service import (
|
|
BackgroundLoaderService,
|
|
LoadingStatus,
|
|
SeriesLoadingTask,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_websocket_service():
|
|
"""Mock WebSocket service."""
|
|
service = Mock()
|
|
service.broadcast = AsyncMock()
|
|
return service
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_anime_service():
|
|
"""Mock anime service."""
|
|
service = Mock()
|
|
service.rescan = AsyncMock()
|
|
return service
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_series_app(tmp_path):
|
|
"""Mock SeriesApp."""
|
|
app = Mock()
|
|
app.directory_to_search = str(tmp_path)
|
|
app.nfo_service = Mock()
|
|
app.nfo_service.has_nfo = Mock(return_value=False)
|
|
app.nfo_service.create_tvshow_nfo = AsyncMock()
|
|
return app
|
|
|
|
|
|
@pytest.fixture
|
|
async def background_loader(mock_websocket_service, mock_anime_service, mock_series_app):
|
|
"""Create BackgroundLoaderService instance."""
|
|
service = BackgroundLoaderService(
|
|
websocket_service=mock_websocket_service,
|
|
anime_service=mock_anime_service,
|
|
series_app=mock_series_app
|
|
)
|
|
yield service
|
|
await service.stop()
|
|
|
|
|
|
class TestFindSeriesDirectory:
|
|
"""Test finding series directory without full rescan."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_find_existing_directory(self, background_loader, tmp_path):
|
|
"""Test finding a series directory that exists."""
|
|
# Create series directory
|
|
series_dir = tmp_path / "Test Series"
|
|
series_dir.mkdir()
|
|
|
|
task = SeriesLoadingTask(
|
|
key="test-series",
|
|
folder="Test Series",
|
|
name="Test Series"
|
|
)
|
|
|
|
# Find directory
|
|
result = await background_loader._find_series_directory(task)
|
|
|
|
# Verify
|
|
assert result is not None
|
|
assert result == series_dir
|
|
assert result.exists()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_find_nonexistent_directory(self, background_loader, tmp_path):
|
|
"""Test finding a series directory that doesn't exist."""
|
|
task = SeriesLoadingTask(
|
|
key="nonexistent",
|
|
folder="Nonexistent Series",
|
|
name="Nonexistent Series"
|
|
)
|
|
|
|
# Find directory
|
|
result = await background_loader._find_series_directory(task)
|
|
|
|
# Verify
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_find_directory_with_special_characters(self, background_loader, tmp_path):
|
|
"""Test finding directory with special characters in name."""
|
|
# Create series directory with special characters
|
|
series_dir = tmp_path / "Series (2023) - Special!"
|
|
series_dir.mkdir()
|
|
|
|
task = SeriesLoadingTask(
|
|
key="special-series",
|
|
folder="Series (2023) - Special!",
|
|
name="Series (2023) - Special!"
|
|
)
|
|
|
|
# Find directory
|
|
result = await background_loader._find_series_directory(task)
|
|
|
|
# Verify
|
|
assert result is not None
|
|
assert result == series_dir
|
|
|
|
|
|
class TestScanSeriesEpisodes:
|
|
"""Test scanning episodes for a specific series."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_single_season(self, background_loader, tmp_path):
|
|
"""Test scanning a series with one season."""
|
|
# Create series structure
|
|
series_dir = tmp_path / "Test Series"
|
|
season_dir = series_dir / "Season 1"
|
|
season_dir.mkdir(parents=True)
|
|
(season_dir / "episode1.mp4").touch()
|
|
(season_dir / "episode2.mp4").touch()
|
|
(season_dir / "episode3.mp4").touch()
|
|
|
|
task = SeriesLoadingTask(
|
|
key="test-series",
|
|
folder="Test Series",
|
|
name="Test Series"
|
|
)
|
|
|
|
# Scan episodes
|
|
episodes = await background_loader._scan_series_episodes(series_dir, task)
|
|
|
|
# Verify
|
|
assert "Season 1" in episodes
|
|
assert len(episodes["Season 1"]) == 3
|
|
assert "episode1.mp4" in episodes["Season 1"]
|
|
assert "episode2.mp4" in episodes["Season 1"]
|
|
assert "episode3.mp4" in episodes["Season 1"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_multiple_seasons(self, background_loader, tmp_path):
|
|
"""Test scanning a series with multiple seasons."""
|
|
# Create series structure
|
|
series_dir = tmp_path / "Multi Season Series"
|
|
|
|
for season_num in range(1, 4):
|
|
season_dir = series_dir / f"Season {season_num}"
|
|
season_dir.mkdir(parents=True)
|
|
|
|
for episode_num in range(1, 6):
|
|
(season_dir / f"episode{episode_num}.mp4").touch()
|
|
|
|
task = SeriesLoadingTask(
|
|
key="multi-season",
|
|
folder="Multi Season Series",
|
|
name="Multi Season Series"
|
|
)
|
|
|
|
# Scan episodes
|
|
episodes = await background_loader._scan_series_episodes(series_dir, task)
|
|
|
|
# Verify
|
|
assert len(episodes) == 3
|
|
assert "Season 1" in episodes
|
|
assert "Season 2" in episodes
|
|
assert "Season 3" in episodes
|
|
assert all(len(eps) == 5 for eps in episodes.values())
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_ignores_non_mp4_files(self, background_loader, tmp_path):
|
|
"""Test that only .mp4 files are counted as episodes."""
|
|
# Create series structure
|
|
series_dir = tmp_path / "Test Series"
|
|
season_dir = series_dir / "Season 1"
|
|
season_dir.mkdir(parents=True)
|
|
|
|
(season_dir / "episode1.mp4").touch()
|
|
(season_dir / "episode2.mkv").touch() # Should be ignored
|
|
(season_dir / "subtitle.srt").touch() # Should be ignored
|
|
(season_dir / "readme.txt").touch() # Should be ignored
|
|
|
|
task = SeriesLoadingTask(
|
|
key="test-series",
|
|
folder="Test Series",
|
|
name="Test Series"
|
|
)
|
|
|
|
# Scan episodes
|
|
episodes = await background_loader._scan_series_episodes(series_dir, task)
|
|
|
|
# Verify - only .mp4 file should be counted
|
|
assert "Season 1" in episodes
|
|
assert len(episodes["Season 1"]) == 1
|
|
assert episodes["Season 1"][0] == "episode1.mp4"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_empty_seasons_ignored(self, background_loader, tmp_path):
|
|
"""Test that seasons with no episodes are ignored."""
|
|
# Create series structure
|
|
series_dir = tmp_path / "Test Series"
|
|
season1_dir = series_dir / "Season 1"
|
|
season2_dir = series_dir / "Season 2"
|
|
season1_dir.mkdir(parents=True)
|
|
season2_dir.mkdir(parents=True)
|
|
|
|
# Only add episodes to Season 1
|
|
(season1_dir / "episode1.mp4").touch()
|
|
# Season 2 is empty
|
|
|
|
task = SeriesLoadingTask(
|
|
key="test-series",
|
|
folder="Test Series",
|
|
name="Test Series"
|
|
)
|
|
|
|
# Scan episodes
|
|
episodes = await background_loader._scan_series_episodes(series_dir, task)
|
|
|
|
# Verify - Season 2 should not be included
|
|
assert len(episodes) == 1
|
|
assert "Season 1" in episodes
|
|
assert "Season 2" not in episodes
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_ignores_files_in_series_root(self, background_loader, tmp_path):
|
|
"""Test that files directly in series root are ignored."""
|
|
# Create series structure
|
|
series_dir = tmp_path / "Test Series"
|
|
series_dir.mkdir()
|
|
season_dir = series_dir / "Season 1"
|
|
season_dir.mkdir()
|
|
|
|
# Add episode in season folder
|
|
(season_dir / "episode1.mp4").touch()
|
|
|
|
# Add file in series root (should be ignored)
|
|
(series_dir / "random.mp4").touch()
|
|
(series_dir / "info.txt").touch()
|
|
|
|
task = SeriesLoadingTask(
|
|
key="test-series",
|
|
folder="Test Series",
|
|
name="Test Series"
|
|
)
|
|
|
|
# Scan episodes
|
|
episodes = await background_loader._scan_series_episodes(series_dir, task)
|
|
|
|
# Verify - only episode in season folder should be counted
|
|
assert len(episodes) == 1
|
|
assert "Season 1" in episodes
|
|
assert len(episodes["Season 1"]) == 1
|
|
|
|
|
|
class TestLoadEpisodesOptimization:
|
|
"""Test that loading episodes doesn't trigger full rescans."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_episodes_no_full_rescan(
|
|
self, background_loader, mock_anime_service, tmp_path
|
|
):
|
|
"""Test that loading episodes doesn't call anime_service.rescan()."""
|
|
# Create series structure
|
|
series_dir = tmp_path / "Test Series"
|
|
season_dir = series_dir / "Season 1"
|
|
season_dir.mkdir(parents=True)
|
|
(season_dir / "episode1.mp4").touch()
|
|
|
|
task = SeriesLoadingTask(
|
|
key="test-series",
|
|
folder="Test Series",
|
|
name="Test Series"
|
|
)
|
|
|
|
# Mock database
|
|
mock_db = AsyncMock()
|
|
mock_series_db = Mock()
|
|
|
|
with patch('src.server.database.service.AnimeSeriesService.get_by_key') as mock_get:
|
|
mock_get.return_value = mock_series_db
|
|
|
|
# Load episodes
|
|
await background_loader._load_episodes(task, mock_db)
|
|
|
|
# Verify rescan was NOT called
|
|
mock_anime_service.rescan.assert_not_called()
|
|
|
|
# Verify progress was updated
|
|
assert task.progress["episodes"] is True
|
|
|
|
# Verify database was updated
|
|
assert mock_series_db.episodes_loaded is True
|
|
mock_db.commit.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_episodes_handles_missing_directory(
|
|
self, background_loader, tmp_path
|
|
):
|
|
"""Test that loading episodes handles missing directory gracefully."""
|
|
task = SeriesLoadingTask(
|
|
key="nonexistent",
|
|
folder="Nonexistent Series",
|
|
name="Nonexistent Series"
|
|
)
|
|
|
|
mock_db = AsyncMock()
|
|
|
|
# Load episodes
|
|
await background_loader._load_episodes(task, mock_db)
|
|
|
|
# Verify progress was marked as failed
|
|
assert task.progress["episodes"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_episodes_handles_empty_directory(
|
|
self, background_loader, tmp_path
|
|
):
|
|
"""Test that loading episodes handles directory with no episodes."""
|
|
# Create empty series directory
|
|
series_dir = tmp_path / "Empty Series"
|
|
series_dir.mkdir()
|
|
|
|
task = SeriesLoadingTask(
|
|
key="empty-series",
|
|
folder="Empty Series",
|
|
name="Empty Series"
|
|
)
|
|
|
|
mock_db = AsyncMock()
|
|
|
|
# Load episodes
|
|
await background_loader._load_episodes(task, mock_db)
|
|
|
|
# Verify progress was marked as failed
|
|
assert task.progress["episodes"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_episodes_updates_database_correctly(
|
|
self, background_loader, tmp_path
|
|
):
|
|
"""Test that loading episodes updates database with correct information."""
|
|
# Create series structure
|
|
series_dir = tmp_path / "Test Series"
|
|
season_dir = series_dir / "Season 1"
|
|
season_dir.mkdir(parents=True)
|
|
(season_dir / "episode1.mp4").touch()
|
|
(season_dir / "episode2.mp4").touch()
|
|
|
|
task = SeriesLoadingTask(
|
|
key="test-series",
|
|
folder="Test Series",
|
|
name="Test Series"
|
|
)
|
|
|
|
# Mock database
|
|
mock_db = AsyncMock()
|
|
mock_series_db = Mock()
|
|
mock_series_db.episodes_loaded = False
|
|
mock_series_db.loading_status = None
|
|
|
|
with patch('src.server.database.service.AnimeSeriesService.get_by_key') as mock_get:
|
|
mock_get.return_value = mock_series_db
|
|
|
|
# Load episodes
|
|
await background_loader._load_episodes(task, mock_db)
|
|
|
|
# Verify database fields were updated
|
|
assert mock_series_db.episodes_loaded is True
|
|
assert mock_series_db.loading_status == "loading_episodes"
|
|
mock_db.commit.assert_called_once()
|
|
|
|
|
|
class TestIntegrationNoFullRescan:
|
|
"""Integration tests verifying no full rescans occur."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_loading_workflow_no_rescan(
|
|
self, background_loader, mock_anime_service, tmp_path
|
|
):
|
|
"""Test complete loading workflow doesn't trigger rescan."""
|
|
# Create series structure
|
|
series_dir = tmp_path / "Complete Series"
|
|
season_dir = series_dir / "Season 1"
|
|
season_dir.mkdir(parents=True)
|
|
(season_dir / "episode1.mp4").touch()
|
|
(season_dir / "episode2.mp4").touch()
|
|
|
|
task = SeriesLoadingTask(
|
|
key="complete-series",
|
|
folder="Complete Series",
|
|
name="Complete Series"
|
|
)
|
|
|
|
# Mock database
|
|
mock_db = AsyncMock()
|
|
mock_series_db = Mock()
|
|
mock_series_db.episodes_loaded = False
|
|
mock_series_db.has_nfo = False
|
|
mock_series_db.logo_loaded = False
|
|
mock_series_db.images_loaded = False
|
|
|
|
with patch('src.server.database.service.AnimeSeriesService.get_by_key') as mock_get:
|
|
with patch('src.server.database.connection.get_db_session') as mock_get_db:
|
|
mock_get.return_value = mock_series_db
|
|
mock_get_db.return_value.__aenter__.return_value = mock_db
|
|
|
|
# Check missing data and load
|
|
missing = await background_loader.check_missing_data(
|
|
task.key,
|
|
task.folder,
|
|
str(tmp_path),
|
|
mock_db
|
|
)
|
|
|
|
if missing["episodes"]:
|
|
await background_loader._load_episodes(task, mock_db)
|
|
|
|
# Verify NO full rescan was triggered
|
|
mock_anime_service.rescan.assert_not_called()
|
|
|
|
# Verify task completed successfully
|
|
assert task.progress["episodes"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multiple_series_no_cross_contamination(
|
|
self, background_loader, tmp_path
|
|
):
|
|
"""Test loading multiple series doesn't cause cross-contamination."""
|
|
# Create multiple series
|
|
for series_name in ["Series A", "Series B", "Series C"]:
|
|
series_dir = tmp_path / series_name
|
|
season_dir = series_dir / "Season 1"
|
|
season_dir.mkdir(parents=True)
|
|
(season_dir / "episode1.mp4").touch()
|
|
|
|
tasks = [
|
|
SeriesLoadingTask(key=f"series-{i}", folder=f"Series {chr(65+i)}", name=f"Series {chr(65+i)}")
|
|
for i in range(3)
|
|
]
|
|
|
|
mock_db = AsyncMock()
|
|
|
|
# Load all series
|
|
for task in tasks:
|
|
series_dir = await background_loader._find_series_directory(task)
|
|
assert series_dir is not None
|
|
|
|
episodes = await background_loader._scan_series_episodes(series_dir, task)
|
|
assert len(episodes) == 1
|
|
assert "Season 1" in episodes
|
|
|
|
|
|
class TestPerformanceComparison:
|
|
"""Tests to demonstrate performance improvement."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scan_single_series_is_fast(self, background_loader, tmp_path):
|
|
"""Test that scanning a single series is fast."""
|
|
import time
|
|
|
|
# Create series structure
|
|
series_dir = tmp_path / "Performance Test"
|
|
for season_num in range(1, 6):
|
|
season_dir = series_dir / f"Season {season_num}"
|
|
season_dir.mkdir(parents=True)
|
|
|
|
for episode_num in range(1, 26):
|
|
(season_dir / f"episode{episode_num}.mp4").touch()
|
|
|
|
task = SeriesLoadingTask(
|
|
key="performance-test",
|
|
folder="Performance Test",
|
|
name="Performance Test"
|
|
)
|
|
|
|
# Measure time
|
|
start_time = time.time()
|
|
episodes = await background_loader._scan_series_episodes(series_dir, task)
|
|
elapsed_time = time.time() - start_time
|
|
|
|
# Verify it's fast (should be under 1 second even for 125 episodes)
|
|
assert elapsed_time < 1.0
|
|
assert len(episodes) == 5
|
|
assert all(len(eps) == 25 for eps in episodes.values())
|