Lukas
396b243d59
Task 9: Clean up legacy code - Added deprecation warnings to Serie.save_to_file() and load_from_file() - Updated infrastructure.md with Data Storage section documenting: - SQLite database as primary storage - Legacy file storage as deprecated - Data migration process - Added deprecation warning tests for Serie class - Updated existing tests to handle new warnings - All 1012 tests pass (872 unit + 55 API + 85 integration)
546 lines
20 KiB
Python
546 lines
20 KiB
Python
"""Tests for SerieList class - identifier standardization."""
|
|
|
|
import os
|
|
import tempfile
|
|
import warnings
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from src.core.entities.SerieList import SerieList
|
|
from src.core.entities.series import Serie
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_directory():
|
|
"""Create a temporary directory for testing."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
yield tmpdir
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_serie():
|
|
"""Create a sample Serie for testing."""
|
|
return Serie(
|
|
key="attack-on-titan",
|
|
name="Attack on Titan",
|
|
site="https://aniworld.to/anime/stream/attack-on-titan",
|
|
folder="Attack on Titan (2013)",
|
|
episodeDict={1: [1, 2, 3]}
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_db_session():
|
|
"""Create a mock async database session."""
|
|
session = AsyncMock()
|
|
return session
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_anime_series():
|
|
"""Create a mock AnimeSeries database model."""
|
|
anime_series = MagicMock()
|
|
anime_series.key = "test-series"
|
|
anime_series.name = "Test Series"
|
|
anime_series.site = "https://aniworld.to/anime/stream/test-series"
|
|
anime_series.folder = "Test Series (2020)"
|
|
anime_series.episode_dict = {"1": [1, 2, 3], "2": [1, 2]}
|
|
return anime_series
|
|
|
|
|
|
class TestSerieListKeyBasedStorage:
|
|
"""Test SerieList uses key for internal storage."""
|
|
|
|
def test_init_creates_empty_keydict(self, temp_directory):
|
|
"""Test initialization creates keyDict."""
|
|
serie_list = SerieList(temp_directory)
|
|
assert hasattr(serie_list, 'keyDict')
|
|
assert isinstance(serie_list.keyDict, dict)
|
|
|
|
def test_add_stores_by_key(self, temp_directory, sample_serie):
|
|
"""Test add() stores series by key."""
|
|
serie_list = SerieList(temp_directory)
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", DeprecationWarning)
|
|
serie_list.add(sample_serie)
|
|
|
|
# Verify stored by key, not folder
|
|
assert sample_serie.key in serie_list.keyDict
|
|
assert serie_list.keyDict[sample_serie.key] == sample_serie
|
|
|
|
def test_contains_checks_by_key(self, temp_directory, sample_serie):
|
|
"""Test contains() checks by key."""
|
|
serie_list = SerieList(temp_directory)
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", DeprecationWarning)
|
|
serie_list.add(sample_serie)
|
|
|
|
assert serie_list.contains(sample_serie.key)
|
|
assert not serie_list.contains("nonexistent-key")
|
|
|
|
def test_add_prevents_duplicates_by_key(
|
|
self, temp_directory, sample_serie
|
|
):
|
|
"""Test add() prevents duplicates based on key."""
|
|
serie_list = SerieList(temp_directory)
|
|
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", DeprecationWarning)
|
|
# Add same serie twice
|
|
serie_list.add(sample_serie)
|
|
initial_count = len(serie_list.keyDict)
|
|
|
|
serie_list.add(sample_serie)
|
|
|
|
# Should still have only one entry
|
|
assert len(serie_list.keyDict) == initial_count
|
|
assert len(serie_list.keyDict) == 1
|
|
|
|
def test_get_by_key_returns_correct_serie(
|
|
self, temp_directory, sample_serie
|
|
):
|
|
"""Test get_by_key() retrieves series correctly."""
|
|
serie_list = SerieList(temp_directory)
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", DeprecationWarning)
|
|
serie_list.add(sample_serie)
|
|
|
|
result = serie_list.get_by_key(sample_serie.key)
|
|
assert result is not None
|
|
assert result.key == sample_serie.key
|
|
assert result.name == sample_serie.name
|
|
|
|
def test_get_by_key_returns_none_for_missing(self, temp_directory):
|
|
"""Test get_by_key() returns None for nonexistent key."""
|
|
serie_list = SerieList(temp_directory)
|
|
|
|
result = serie_list.get_by_key("nonexistent-key")
|
|
assert result is None
|
|
|
|
def test_get_by_folder_backward_compatibility(
|
|
self, temp_directory, sample_serie
|
|
):
|
|
"""Test get_by_folder() provides backward compatibility."""
|
|
serie_list = SerieList(temp_directory)
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", DeprecationWarning)
|
|
serie_list.add(sample_serie)
|
|
result = serie_list.get_by_folder(sample_serie.folder)
|
|
|
|
assert result is not None
|
|
assert result.key == sample_serie.key
|
|
assert result.folder == sample_serie.folder
|
|
|
|
def test_get_by_folder_returns_none_for_missing(self, temp_directory):
|
|
"""Test get_by_folder() returns None for nonexistent folder."""
|
|
serie_list = SerieList(temp_directory)
|
|
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", DeprecationWarning)
|
|
result = serie_list.get_by_folder("Nonexistent Folder")
|
|
assert result is None
|
|
|
|
def test_get_all_returns_all_series(self, temp_directory, sample_serie):
|
|
"""Test get_all() returns all series from keyDict."""
|
|
serie_list = SerieList(temp_directory)
|
|
|
|
serie2 = Serie(
|
|
key="naruto",
|
|
name="Naruto",
|
|
site="https://aniworld.to/anime/stream/naruto",
|
|
folder="Naruto (2002)",
|
|
episodeDict={1: [1, 2]}
|
|
)
|
|
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", DeprecationWarning)
|
|
serie_list.add(sample_serie)
|
|
serie_list.add(serie2)
|
|
|
|
all_series = serie_list.get_all()
|
|
assert len(all_series) == 2
|
|
assert sample_serie in all_series
|
|
assert serie2 in all_series
|
|
|
|
def test_get_missing_episodes_filters_by_episode_dict(
|
|
self, temp_directory
|
|
):
|
|
"""Test get_missing_episodes() returns only series with episodes."""
|
|
serie_list = SerieList(temp_directory)
|
|
|
|
# Serie with missing episodes
|
|
serie_with_episodes = Serie(
|
|
key="serie-with-episodes",
|
|
name="Serie With Episodes",
|
|
site="https://aniworld.to/anime/stream/serie-with-episodes",
|
|
folder="Serie With Episodes (2020)",
|
|
episodeDict={1: [1, 2, 3]}
|
|
)
|
|
|
|
# Serie without missing episodes
|
|
serie_without_episodes = Serie(
|
|
key="serie-without-episodes",
|
|
name="Serie Without Episodes",
|
|
site="https://aniworld.to/anime/stream/serie-without-episodes",
|
|
folder="Serie Without Episodes (2021)",
|
|
episodeDict={}
|
|
)
|
|
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", DeprecationWarning)
|
|
serie_list.add(serie_with_episodes)
|
|
serie_list.add(serie_without_episodes)
|
|
|
|
missing = serie_list.get_missing_episodes()
|
|
assert len(missing) == 1
|
|
assert serie_with_episodes in missing
|
|
assert serie_without_episodes not in missing
|
|
|
|
def test_load_series_stores_by_key(self, temp_directory, sample_serie):
|
|
"""Test load_series() stores series by key when loading from disk."""
|
|
# Create directory structure and save serie
|
|
folder_path = os.path.join(temp_directory, sample_serie.folder)
|
|
os.makedirs(folder_path, exist_ok=True)
|
|
data_path = os.path.join(folder_path, "data")
|
|
sample_serie.save_to_file(data_path)
|
|
|
|
# Create new SerieList (triggers load_series in __init__)
|
|
serie_list = SerieList(temp_directory)
|
|
|
|
# Verify loaded by key
|
|
assert sample_serie.key in serie_list.keyDict
|
|
loaded_serie = serie_list.keyDict[sample_serie.key]
|
|
assert loaded_serie.key == sample_serie.key
|
|
assert loaded_serie.name == sample_serie.name
|
|
|
|
|
|
class TestSerieListPublicAPI:
|
|
"""Test that public API still works correctly."""
|
|
|
|
def test_public_methods_work(self, temp_directory, sample_serie):
|
|
"""Test that all public methods work correctly after refactoring."""
|
|
serie_list = SerieList(temp_directory)
|
|
|
|
# Test add (suppress deprecation warning for test)
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", DeprecationWarning)
|
|
serie_list.add(sample_serie)
|
|
|
|
# Test contains
|
|
assert serie_list.contains(sample_serie.key)
|
|
|
|
# Test GetList/get_all
|
|
assert len(serie_list.GetList()) == 1
|
|
assert len(serie_list.get_all()) == 1
|
|
|
|
# Test GetMissingEpisode/get_missing_episodes
|
|
assert len(serie_list.GetMissingEpisode()) == 1
|
|
assert len(serie_list.get_missing_episodes()) == 1
|
|
|
|
# Test new helper methods
|
|
assert serie_list.get_by_key(sample_serie.key) is not None
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", DeprecationWarning)
|
|
assert serie_list.get_by_folder(sample_serie.folder) is not None
|
|
|
|
|
|
class TestSerieListDatabaseMode:
|
|
"""Test SerieList database-backed storage functionality."""
|
|
|
|
def test_init_with_db_session_skips_file_load(
|
|
self, temp_directory, mock_db_session
|
|
):
|
|
"""Test initialization with db_session skips file-based loading."""
|
|
# Create a data file that should NOT be loaded
|
|
folder_path = os.path.join(temp_directory, "Test Folder")
|
|
os.makedirs(folder_path, exist_ok=True)
|
|
data_path = os.path.join(folder_path, "data")
|
|
|
|
serie = Serie(
|
|
key="test-key",
|
|
name="Test",
|
|
site="https://test.com",
|
|
folder="Test Folder",
|
|
episodeDict={}
|
|
)
|
|
serie.save_to_file(data_path)
|
|
|
|
# Initialize with db_session - should skip file loading
|
|
serie_list = SerieList(
|
|
temp_directory,
|
|
db_session=mock_db_session
|
|
)
|
|
|
|
# Should have empty keyDict (file loading skipped)
|
|
assert len(serie_list.keyDict) == 0
|
|
|
|
def test_init_with_skip_load(self, temp_directory):
|
|
"""Test initialization with skip_load=True skips loading."""
|
|
serie_list = SerieList(temp_directory, skip_load=True)
|
|
assert len(serie_list.keyDict) == 0
|
|
|
|
def test_convert_from_db_basic(self, mock_anime_series):
|
|
"""Test _convert_from_db converts AnimeSeries to Serie correctly."""
|
|
serie = SerieList._convert_from_db(mock_anime_series)
|
|
|
|
assert serie.key == mock_anime_series.key
|
|
assert serie.name == mock_anime_series.name
|
|
assert serie.site == mock_anime_series.site
|
|
assert serie.folder == mock_anime_series.folder
|
|
# Season keys should be converted from string to int
|
|
assert 1 in serie.episodeDict
|
|
assert 2 in serie.episodeDict
|
|
assert serie.episodeDict[1] == [1, 2, 3]
|
|
assert serie.episodeDict[2] == [1, 2]
|
|
|
|
def test_convert_from_db_empty_episode_dict(self, mock_anime_series):
|
|
"""Test _convert_from_db handles empty episode_dict."""
|
|
mock_anime_series.episode_dict = None
|
|
|
|
serie = SerieList._convert_from_db(mock_anime_series)
|
|
|
|
assert serie.episodeDict == {}
|
|
|
|
def test_convert_from_db_handles_invalid_season_keys(
|
|
self, mock_anime_series
|
|
):
|
|
"""Test _convert_from_db handles invalid season keys gracefully."""
|
|
mock_anime_series.episode_dict = {
|
|
"1": [1, 2],
|
|
"invalid": [3, 4], # Invalid key - not an integer
|
|
"2": [5, 6]
|
|
}
|
|
|
|
serie = SerieList._convert_from_db(mock_anime_series)
|
|
|
|
# Valid keys should be converted
|
|
assert 1 in serie.episodeDict
|
|
assert 2 in serie.episodeDict
|
|
# Invalid key should be skipped
|
|
assert "invalid" not in serie.episodeDict
|
|
|
|
def test_convert_to_db_dict(self, sample_serie):
|
|
"""Test _convert_to_db_dict creates correct dictionary."""
|
|
result = SerieList._convert_to_db_dict(sample_serie)
|
|
|
|
assert result["key"] == sample_serie.key
|
|
assert result["name"] == sample_serie.name
|
|
assert result["site"] == sample_serie.site
|
|
assert result["folder"] == sample_serie.folder
|
|
# Keys should be converted to strings for JSON
|
|
assert "1" in result["episode_dict"]
|
|
assert result["episode_dict"]["1"] == [1, 2, 3]
|
|
|
|
def test_convert_to_db_dict_empty_episode_dict(self):
|
|
"""Test _convert_to_db_dict handles empty episode_dict."""
|
|
serie = Serie(
|
|
key="test",
|
|
name="Test",
|
|
site="https://test.com",
|
|
folder="Test",
|
|
episodeDict={}
|
|
)
|
|
|
|
result = SerieList._convert_to_db_dict(serie)
|
|
|
|
assert result["episode_dict"] is None
|
|
|
|
|
|
class TestSerieListDatabaseAsync:
|
|
"""Test async database methods of SerieList."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_series_from_db(
|
|
self, temp_directory, mock_db_session, mock_anime_series
|
|
):
|
|
"""Test load_series_from_db loads from database."""
|
|
# Setup mock to return list of anime series
|
|
with patch(
|
|
'src.server.database.service.AnimeSeriesService'
|
|
) as mock_service:
|
|
mock_service.get_all = AsyncMock(return_value=[mock_anime_series])
|
|
|
|
serie_list = SerieList(temp_directory, skip_load=True)
|
|
count = await serie_list.load_series_from_db(mock_db_session)
|
|
|
|
assert count == 1
|
|
assert mock_anime_series.key in serie_list.keyDict
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_series_from_db_clears_existing(
|
|
self, temp_directory, mock_db_session, mock_anime_series
|
|
):
|
|
"""Test load_series_from_db clears existing data."""
|
|
serie_list = SerieList(temp_directory, skip_load=True)
|
|
# Add an existing entry
|
|
serie_list.keyDict["old-key"] = MagicMock()
|
|
|
|
with patch(
|
|
'src.server.database.service.AnimeSeriesService'
|
|
) as mock_service:
|
|
mock_service.get_all = AsyncMock(return_value=[mock_anime_series])
|
|
|
|
await serie_list.load_series_from_db(mock_db_session)
|
|
|
|
# Old entry should be cleared
|
|
assert "old-key" not in serie_list.keyDict
|
|
assert mock_anime_series.key in serie_list.keyDict
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_to_db_creates_new_series(
|
|
self, temp_directory, mock_db_session, sample_serie
|
|
):
|
|
"""Test add_to_db creates new series in database."""
|
|
with patch(
|
|
'src.server.database.service.AnimeSeriesService'
|
|
) as mock_service:
|
|
mock_service.get_by_key = AsyncMock(return_value=None)
|
|
mock_created = MagicMock()
|
|
mock_created.id = 1
|
|
mock_service.create = AsyncMock(return_value=mock_created)
|
|
|
|
serie_list = SerieList(temp_directory, skip_load=True)
|
|
result = await serie_list.add_to_db(sample_serie, mock_db_session)
|
|
|
|
assert result is mock_created
|
|
mock_service.create.assert_called_once()
|
|
# Should also add to in-memory collection
|
|
assert sample_serie.key in serie_list.keyDict
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_to_db_skips_existing(
|
|
self, temp_directory, mock_db_session, sample_serie
|
|
):
|
|
"""Test add_to_db skips if series already exists."""
|
|
with patch(
|
|
'src.server.database.service.AnimeSeriesService'
|
|
) as mock_service:
|
|
existing = MagicMock()
|
|
mock_service.get_by_key = AsyncMock(return_value=existing)
|
|
|
|
serie_list = SerieList(temp_directory, skip_load=True)
|
|
result = await serie_list.add_to_db(sample_serie, mock_db_session)
|
|
|
|
assert result is None
|
|
mock_service.create.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_contains_in_db_returns_true_when_exists(
|
|
self, temp_directory, mock_db_session
|
|
):
|
|
"""Test contains_in_db returns True when series exists."""
|
|
with patch(
|
|
'src.server.database.service.AnimeSeriesService'
|
|
) as mock_service:
|
|
mock_service.get_by_key = AsyncMock(return_value=MagicMock())
|
|
|
|
serie_list = SerieList(temp_directory, skip_load=True)
|
|
result = await serie_list.contains_in_db(
|
|
"test-key", mock_db_session
|
|
)
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_contains_in_db_returns_false_when_not_exists(
|
|
self, temp_directory, mock_db_session
|
|
):
|
|
"""Test contains_in_db returns False when series doesn't exist."""
|
|
with patch(
|
|
'src.server.database.service.AnimeSeriesService'
|
|
) as mock_service:
|
|
mock_service.get_by_key = AsyncMock(return_value=None)
|
|
|
|
serie_list = SerieList(temp_directory, skip_load=True)
|
|
result = await serie_list.contains_in_db(
|
|
"nonexistent", mock_db_session
|
|
)
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestSerieListDeprecationWarnings:
|
|
"""Test deprecation warnings are raised for file-based methods."""
|
|
|
|
def test_add_raises_deprecation_warning(
|
|
self, temp_directory, sample_serie
|
|
):
|
|
"""Test add() raises deprecation warning."""
|
|
serie_list = SerieList(temp_directory, skip_load=True)
|
|
|
|
with warnings.catch_warnings(record=True) as w:
|
|
warnings.simplefilter("always")
|
|
serie_list.add(sample_serie)
|
|
|
|
# Check at least one deprecation warning was raised for add()
|
|
# (Note: save_to_file also raises a warning, so we may get 2)
|
|
deprecation_warnings = [
|
|
warning for warning in w
|
|
if issubclass(warning.category, DeprecationWarning)
|
|
]
|
|
assert len(deprecation_warnings) >= 1
|
|
# Check that one of them is from add()
|
|
add_warnings = [
|
|
warning for warning in deprecation_warnings
|
|
if "add_to_db()" in str(warning.message)
|
|
]
|
|
assert len(add_warnings) == 1
|
|
|
|
def test_get_by_folder_raises_deprecation_warning(
|
|
self, temp_directory, sample_serie
|
|
):
|
|
"""Test get_by_folder() raises deprecation warning."""
|
|
serie_list = SerieList(temp_directory, skip_load=True)
|
|
serie_list.keyDict[sample_serie.key] = sample_serie
|
|
|
|
with warnings.catch_warnings(record=True) as w:
|
|
warnings.simplefilter("always")
|
|
serie_list.get_by_folder(sample_serie.folder)
|
|
|
|
# Check deprecation warning was raised
|
|
assert len(w) == 1
|
|
assert issubclass(w[0].category, DeprecationWarning)
|
|
assert "get_by_key()" in str(w[0].message)
|
|
|
|
|
|
class TestSerieListBackwardCompatibility:
|
|
"""Test backward compatibility of file-based operations."""
|
|
|
|
def test_file_based_mode_still_works(
|
|
self, temp_directory, sample_serie
|
|
):
|
|
"""Test file-based mode still works without db_session."""
|
|
serie_list = SerieList(temp_directory)
|
|
|
|
# Add should still work (with deprecation warning)
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter("ignore", DeprecationWarning)
|
|
serie_list.add(sample_serie)
|
|
|
|
# File should be created
|
|
data_path = os.path.join(
|
|
temp_directory, sample_serie.folder, "data"
|
|
)
|
|
assert os.path.isfile(data_path)
|
|
|
|
# Series should be in memory
|
|
assert serie_list.contains(sample_serie.key)
|
|
|
|
def test_load_from_file_still_works(
|
|
self, temp_directory, sample_serie
|
|
):
|
|
"""Test loading from files still works."""
|
|
# Create directory and save file
|
|
folder_path = os.path.join(temp_directory, sample_serie.folder)
|
|
os.makedirs(folder_path, exist_ok=True)
|
|
data_path = os.path.join(folder_path, "data")
|
|
sample_serie.save_to_file(data_path)
|
|
|
|
# New SerieList should load it
|
|
serie_list = SerieList(temp_directory)
|
|
|
|
assert serie_list.contains(sample_serie.key)
|
|
loaded = serie_list.get_by_key(sample_serie.key)
|
|
assert loaded.name == sample_serie.name
|