Lukas
596476f9ac
- Remove db_session parameter from SeriesApp, SerieList, SerieScanner - Move all database operations to AnimeService (service layer) - Add add_series_to_db, contains_in_db methods to AnimeService - Update sync_series_from_data_files to use inline DB operations - Remove obsolete test classes for removed DB methods - Fix pylint issues: add broad-except comments, fix line lengths - Core layer (src/core/) now has zero database imports 722 unit tests pass
275 lines
9.9 KiB
Python
275 lines
9.9 KiB
Python
"""Integration tests for data file to database synchronization.
|
|
|
|
This module verifies that the data file to database sync functionality
|
|
works correctly, including:
|
|
- Loading series from data files
|
|
- Adding series to the database
|
|
- Preventing duplicate entries
|
|
- Handling corrupt or missing files gracefully
|
|
- End-to-end startup sync behavior
|
|
|
|
The sync functionality allows existing series metadata stored in
|
|
data files to be automatically imported into the database during
|
|
application startup.
|
|
"""
|
|
import json
|
|
import os
|
|
import tempfile
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
import pytest
|
|
|
|
from src.core.entities.series import Serie
|
|
from src.core.SeriesApp import SeriesApp
|
|
|
|
|
|
class TestGetAllSeriesFromDataFiles:
|
|
"""Test SeriesApp.get_all_series_from_data_files() method."""
|
|
|
|
def test_returns_empty_list_for_empty_directory(self):
|
|
"""Test that empty directory returns empty list."""
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
with patch('src.core.SeriesApp.Loaders'), \
|
|
patch('src.core.SeriesApp.SerieScanner'):
|
|
app = SeriesApp(tmp_dir)
|
|
result = app.get_all_series_from_data_files()
|
|
|
|
assert isinstance(result, list)
|
|
assert len(result) == 0
|
|
|
|
def test_returns_series_from_data_files(self):
|
|
"""Test that valid data files are loaded correctly."""
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
# Create test data files
|
|
_create_test_data_file(
|
|
tmp_dir,
|
|
folder="Anime Test 1",
|
|
key="anime-test-1",
|
|
name="Anime Test 1",
|
|
episodes={1: [1, 2, 3]}
|
|
)
|
|
_create_test_data_file(
|
|
tmp_dir,
|
|
folder="Anime Test 2",
|
|
key="anime-test-2",
|
|
name="Anime Test 2",
|
|
episodes={1: [1]}
|
|
)
|
|
|
|
with patch('src.core.SeriesApp.Loaders'), \
|
|
patch('src.core.SeriesApp.SerieScanner'):
|
|
app = SeriesApp(tmp_dir)
|
|
result = app.get_all_series_from_data_files()
|
|
|
|
assert isinstance(result, list)
|
|
assert len(result) == 2
|
|
keys = {s.key for s in result}
|
|
assert "anime-test-1" in keys
|
|
assert "anime-test-2" in keys
|
|
|
|
def test_handles_corrupt_data_files_gracefully(self):
|
|
"""Test that corrupt data files don't crash the sync."""
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
# Create a valid data file
|
|
_create_test_data_file(
|
|
tmp_dir,
|
|
folder="Valid Anime",
|
|
key="valid-anime",
|
|
name="Valid Anime",
|
|
episodes={1: [1]}
|
|
)
|
|
|
|
# Create a corrupt data file (invalid JSON)
|
|
corrupt_dir = os.path.join(tmp_dir, "Corrupt Anime")
|
|
os.makedirs(corrupt_dir, exist_ok=True)
|
|
with open(os.path.join(corrupt_dir, "data"), "w") as f:
|
|
f.write("this is not valid json {{{")
|
|
|
|
with patch('src.core.SeriesApp.Loaders'), \
|
|
patch('src.core.SeriesApp.SerieScanner'):
|
|
app = SeriesApp(tmp_dir)
|
|
result = app.get_all_series_from_data_files()
|
|
|
|
# Should still return the valid series
|
|
assert isinstance(result, list)
|
|
assert len(result) >= 1
|
|
# The valid anime should be loaded
|
|
keys = {s.key for s in result}
|
|
assert "valid-anime" in keys
|
|
|
|
def test_handles_missing_directory_gracefully(self):
|
|
"""Test that non-existent directory returns empty list."""
|
|
non_existent_dir = "/non/existent/directory/path"
|
|
|
|
with patch('src.core.SeriesApp.Loaders'), \
|
|
patch('src.core.SeriesApp.SerieScanner'):
|
|
app = SeriesApp(non_existent_dir)
|
|
result = app.get_all_series_from_data_files()
|
|
|
|
assert isinstance(result, list)
|
|
assert len(result) == 0
|
|
|
|
|
|
class TestSyncSeriesToDatabase:
|
|
"""Test sync_series_from_data_files function from anime_service."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_with_empty_directory(self):
|
|
"""Test sync with empty anime directory."""
|
|
from src.server.services.anime_service import sync_series_from_data_files
|
|
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
mock_logger = Mock()
|
|
|
|
with patch('src.core.SeriesApp.Loaders'), \
|
|
patch('src.core.SeriesApp.SerieScanner'):
|
|
count = await sync_series_from_data_files(tmp_dir, mock_logger)
|
|
|
|
assert count == 0
|
|
# Should log that no series were found
|
|
mock_logger.info.assert_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_adds_new_series_to_database(self):
|
|
"""Test that sync adds new series to database.
|
|
|
|
This is a more realistic test that verifies series data is loaded
|
|
from files and the sync function attempts to add them to the DB.
|
|
The actual DB interaction is tested in test_add_to_db_creates_record.
|
|
"""
|
|
from src.server.services.anime_service import sync_series_from_data_files
|
|
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
# Create test data files
|
|
_create_test_data_file(
|
|
tmp_dir,
|
|
folder="Sync Test Anime",
|
|
key="sync-test-anime",
|
|
name="Sync Test Anime",
|
|
episodes={1: [1, 2]}
|
|
)
|
|
|
|
mock_logger = Mock()
|
|
|
|
# First verify that we can load the series from files
|
|
with patch('src.core.SeriesApp.Loaders'), \
|
|
patch('src.core.SeriesApp.SerieScanner'):
|
|
app = SeriesApp(tmp_dir)
|
|
series = app.get_all_series_from_data_files()
|
|
assert len(series) == 1
|
|
assert series[0].key == "sync-test-anime"
|
|
|
|
# Now test that the sync function loads series and handles DB
|
|
# gracefully (even if DB operations fail, it should not crash)
|
|
with patch('src.core.SeriesApp.Loaders'), \
|
|
patch('src.core.SeriesApp.SerieScanner'):
|
|
# The function should return 0 because DB isn't available
|
|
# but should not crash
|
|
count = await sync_series_from_data_files(tmp_dir, mock_logger)
|
|
|
|
# Since no real DB, it will fail gracefully
|
|
assert isinstance(count, int)
|
|
# Should have logged something
|
|
assert mock_logger.info.called or mock_logger.warning.called
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_handles_exceptions_gracefully(self):
|
|
"""Test that sync handles exceptions without crashing."""
|
|
from src.server.services.anime_service import sync_series_from_data_files
|
|
|
|
mock_logger = Mock()
|
|
|
|
# Make SeriesApp raise an exception during initialization
|
|
with patch('src.core.SeriesApp.Loaders'), \
|
|
patch('src.core.SeriesApp.SerieScanner'), \
|
|
patch(
|
|
'src.core.SeriesApp.SerieList',
|
|
side_effect=Exception("Test error")
|
|
):
|
|
count = await sync_series_from_data_files(
|
|
"/fake/path", mock_logger
|
|
)
|
|
|
|
assert count == 0
|
|
# Should log the warning
|
|
mock_logger.warning.assert_called()
|
|
|
|
|
|
class TestEndToEndSync:
|
|
"""End-to-end tests for the sync functionality."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_startup_sync_integration(self):
|
|
"""Test end-to-end startup sync behavior."""
|
|
# This test verifies the integration of all components
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
# Create test data
|
|
_create_test_data_file(
|
|
tmp_dir,
|
|
folder="E2E Test Anime 1",
|
|
key="e2e-test-anime-1",
|
|
name="E2E Test Anime 1",
|
|
episodes={1: [1, 2, 3]}
|
|
)
|
|
_create_test_data_file(
|
|
tmp_dir,
|
|
folder="E2E Test Anime 2",
|
|
key="e2e-test-anime-2",
|
|
name="E2E Test Anime 2",
|
|
episodes={1: [1], 2: [1, 2]}
|
|
)
|
|
|
|
# Use SeriesApp to load series from files
|
|
with patch('src.core.SeriesApp.Loaders'), \
|
|
patch('src.core.SeriesApp.SerieScanner'):
|
|
app = SeriesApp(tmp_dir)
|
|
all_series = app.get_all_series_from_data_files()
|
|
|
|
# Verify all series were loaded
|
|
assert len(all_series) == 2
|
|
|
|
# Verify series data is correct
|
|
series_by_key = {s.key: s for s in all_series}
|
|
assert "e2e-test-anime-1" in series_by_key
|
|
assert "e2e-test-anime-2" in series_by_key
|
|
|
|
# Verify episode data
|
|
anime1 = series_by_key["e2e-test-anime-1"]
|
|
assert anime1.episodeDict == {1: [1, 2, 3]}
|
|
|
|
anime2 = series_by_key["e2e-test-anime-2"]
|
|
assert anime2.episodeDict == {1: [1], 2: [1, 2]}
|
|
|
|
|
|
def _create_test_data_file(
|
|
base_dir: str,
|
|
folder: str,
|
|
key: str,
|
|
name: str,
|
|
episodes: dict
|
|
) -> None:
|
|
"""
|
|
Create a test data file in the anime directory.
|
|
|
|
Args:
|
|
base_dir: Base directory for anime folders
|
|
folder: Folder name for the anime
|
|
key: Unique key for the series
|
|
name: Display name of the series
|
|
episodes: Dictionary mapping season to list of episode numbers
|
|
"""
|
|
anime_dir = os.path.join(base_dir, folder)
|
|
os.makedirs(anime_dir, exist_ok=True)
|
|
|
|
data = {
|
|
"key": key,
|
|
"name": name,
|
|
"site": "https://aniworld.to",
|
|
"folder": folder,
|
|
"episodeDict": {str(k): v for k, v in episodes.items()}
|
|
}
|
|
|
|
data_file = os.path.join(anime_dir, "data")
|
|
with open(data_file, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, indent=2)
|