322 lines
11 KiB
Python
322 lines
11 KiB
Python
"""Unit tests for BackgroundLoaderService.
|
|
|
|
Tests task queuing, status tracking, and worker logic in isolation.
|
|
"""
|
|
import asyncio
|
|
from datetime import datetime
|
|
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():
|
|
"""Mock SeriesApp."""
|
|
app = Mock()
|
|
app.directory_to_search = "/test/anime"
|
|
app.nfo_service = Mock()
|
|
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 TestBackgroundLoaderService:
|
|
"""Test suite for BackgroundLoaderService."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_service_initialization(self, background_loader):
|
|
"""Test service initializes correctly."""
|
|
assert background_loader.task_queue is not None
|
|
assert isinstance(background_loader.active_tasks, dict)
|
|
assert len(background_loader.active_tasks) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_worker(self, background_loader):
|
|
"""Test worker starts successfully."""
|
|
await background_loader.start()
|
|
assert background_loader.worker_task is not None
|
|
assert not background_loader.worker_task.done()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_worker_gracefully(self, background_loader):
|
|
"""Test worker stops gracefully."""
|
|
await background_loader.start()
|
|
await background_loader.stop()
|
|
assert background_loader.worker_task.done()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_series_loading_task(self, background_loader):
|
|
"""Test adding a series to the loading queue."""
|
|
await background_loader.add_series_loading_task(
|
|
key="test-series",
|
|
folder="Test Series",
|
|
name="Test Series",
|
|
year=2024
|
|
)
|
|
|
|
# Verify task in active tasks
|
|
assert "test-series" in background_loader.active_tasks
|
|
task = background_loader.active_tasks["test-series"]
|
|
assert task.key == "test-series"
|
|
assert task.status == LoadingStatus.PENDING
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_duplicate_task_handling(self, background_loader):
|
|
"""Test that duplicate tasks for same series are handled correctly."""
|
|
key = "test-series"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_nfo_skips_if_exists(self, background_loader, mock_series_app):
|
|
"""Test that NFO creation is skipped if NFO already exists."""
|
|
# Mock has_nfo to return True (NFO exists)
|
|
mock_series_app.nfo_service.has_nfo.return_value = True
|
|
|
|
task = SeriesLoadingTask(
|
|
key="test-series",
|
|
folder="Test Series",
|
|
name="Test Series",
|
|
year=2023
|
|
)
|
|
|
|
# Mock database
|
|
mock_db = AsyncMock()
|
|
mock_series_db = Mock()
|
|
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:
|
|
mock_get.return_value = mock_series_db
|
|
|
|
# Execute
|
|
await background_loader._load_nfo_and_images(task, mock_db)
|
|
|
|
# Verify NFO creation was NOT called
|
|
mock_series_app.nfo_service.create_tvshow_nfo.assert_not_called()
|
|
|
|
# Verify progress was updated
|
|
assert task.progress["nfo"] is True
|
|
assert task.progress["logo"] is True
|
|
assert task.progress["images"] is True
|
|
|
|
# Verify database was updated
|
|
assert mock_series_db.has_nfo is True
|
|
assert mock_series_db.logo_loaded is True
|
|
assert mock_series_db.images_loaded is True
|
|
mock_db.commit.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_nfo_creates_if_not_exists(self, background_loader, mock_series_app):
|
|
"""Test that NFO is created if it doesn't exist."""
|
|
# Mock has_nfo to return False (NFO doesn't exist)
|
|
mock_series_app.nfo_service.has_nfo.return_value = False
|
|
mock_series_app.nfo_service.create_tvshow_nfo.return_value = "/test/anime/Test Series/tvshow.nfo"
|
|
|
|
task = SeriesLoadingTask(
|
|
key="test-series",
|
|
folder="Test Series",
|
|
name="Test Series",
|
|
year=2023
|
|
)
|
|
|
|
# Mock database
|
|
mock_db = AsyncMock()
|
|
mock_series_db = Mock()
|
|
mock_series_db.has_nfo = False
|
|
|
|
with patch('src.server.database.service.AnimeSeriesService.get_by_key') as mock_get:
|
|
mock_get.return_value = mock_series_db
|
|
|
|
# Execute
|
|
await background_loader._load_nfo_and_images(task, mock_db)
|
|
|
|
# Verify NFO creation WAS called
|
|
mock_series_app.nfo_service.create_tvshow_nfo.assert_called_once_with(
|
|
serie_name="Test Series",
|
|
serie_folder="Test Series",
|
|
year=2023,
|
|
download_poster=True,
|
|
download_logo=True,
|
|
download_fanart=True
|
|
)
|
|
|
|
# Verify progress was updated
|
|
assert task.progress["nfo"] is True
|
|
assert task.progress["logo"] is True
|
|
assert task.progress["images"] is True
|
|
|
|
# Verify database was updated
|
|
assert mock_series_db.has_nfo is True
|
|
mock_db.commit.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_nfo_doesnt_update_if_already_marked(self, background_loader, mock_series_app):
|
|
"""Test that database is not updated if NFO is already marked in DB."""
|
|
# Mock has_nfo to return True (NFO exists)
|
|
mock_series_app.nfo_service.has_nfo.return_value = True
|
|
|
|
task = SeriesLoadingTask(
|
|
key="test-series",
|
|
folder="Test Series",
|
|
name="Test Series"
|
|
)
|
|
|
|
# Mock database - NFO already marked
|
|
mock_db = AsyncMock()
|
|
mock_series_db = Mock()
|
|
mock_series_db.has_nfo = True # Already marked
|
|
mock_series_db.logo_loaded = True
|
|
mock_series_db.images_loaded = True
|
|
|
|
with patch('src.server.database.service.AnimeSeriesService.get_by_key') as mock_get:
|
|
mock_get.return_value = mock_series_db
|
|
|
|
# Execute
|
|
await background_loader._load_nfo_and_images(task, mock_db)
|
|
|
|
# Verify database commit was still called
|
|
mock_db.commit.assert_called_once()
|
|
|
|
# Verify NFO creation was NOT called
|
|
mock_series_app.nfo_service.create_tvshow_nfo.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_duplicate_task_handling_continued(self, background_loader):
|
|
"""Test that duplicate tasks for same series are handled correctly."""
|
|
key = "test-series"
|
|
|
|
await background_loader.add_series_loading_task(
|
|
key=key,
|
|
folder="Test Series",
|
|
name="Test Series"
|
|
)
|
|
await background_loader.add_series_loading_task(
|
|
key=key,
|
|
folder="Test Series",
|
|
name="Test Series"
|
|
)
|
|
|
|
# Verify only one task exists
|
|
assert len([k for k in background_loader.active_tasks if k == key]) == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_missing_data_all_missing(
|
|
self,
|
|
background_loader,
|
|
mock_series_app
|
|
):
|
|
"""Test checking for missing data when all data is missing."""
|
|
with patch('src.server.database.service.AnimeSeriesService.get_by_key') as mock_get:
|
|
mock_series = Mock()
|
|
mock_series.episodes_loaded = False
|
|
mock_series.has_nfo = False
|
|
mock_series.logo_loaded = False
|
|
mock_series.images_loaded = False
|
|
mock_get.return_value = mock_series
|
|
|
|
mock_db = AsyncMock()
|
|
|
|
missing_data = await background_loader.check_missing_data(
|
|
key="test-series",
|
|
folder="Test Series",
|
|
anime_directory="/test/anime",
|
|
db=mock_db
|
|
)
|
|
|
|
assert missing_data["episodes"] is True
|
|
assert missing_data["nfo"] is True
|
|
assert missing_data["logo"] is True
|
|
assert missing_data["images"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_status(self, background_loader, mock_websocket_service):
|
|
"""Test status broadcasting via WebSocket."""
|
|
task = SeriesLoadingTask(
|
|
key="test-series",
|
|
folder="Test Series",
|
|
name="Test Series",
|
|
status=LoadingStatus.LOADING_EPISODES
|
|
)
|
|
|
|
await background_loader._broadcast_status(task)
|
|
|
|
# Verify broadcast was called
|
|
mock_websocket_service.broadcast.assert_called_once()
|
|
|
|
# Verify message structure
|
|
call_args = mock_websocket_service.broadcast.call_args[0][0]
|
|
assert call_args["type"] == "series_loading_update"
|
|
assert call_args["key"] == "test-series"
|
|
assert call_args["loading_status"] == "loading_episodes"
|
|
|
|
|
|
class TestSeriesLoadingTask:
|
|
"""Test SeriesLoadingTask model."""
|
|
|
|
def test_task_initialization(self):
|
|
"""Test task initializes with correct defaults."""
|
|
task = SeriesLoadingTask(
|
|
key="test",
|
|
folder="Test",
|
|
name="Test"
|
|
)
|
|
|
|
assert task.key == "test"
|
|
assert task.status == LoadingStatus.PENDING
|
|
assert not any(task.progress.values())
|
|
|
|
def test_task_progress_tracking(self):
|
|
"""Test progress tracking updates correctly."""
|
|
task = SeriesLoadingTask(
|
|
key="test",
|
|
folder="Test",
|
|
name="Test"
|
|
)
|
|
|
|
task.progress["episodes"] = True
|
|
assert task.progress["episodes"] is True
|
|
assert not task.progress["nfo"]
|
|
assert not task.progress["logo"]
|
|
assert not task.progress["images"]
|
|
|
|
def test_loading_status_enum(self):
|
|
"""Test LoadingStatus enum values."""
|
|
assert LoadingStatus.PENDING == "pending"
|
|
assert LoadingStatus.LOADING_EPISODES == "loading_episodes"
|
|
assert LoadingStatus.LOADING_NFO == "loading_nfo"
|
|
assert LoadingStatus.COMPLETED == "completed"
|
|
assert LoadingStatus.FAILED == "failed"
|