- Add sanitize_folder_name utility for filesystem-safe folder names - Add sanitized_folder property to Serie entity - Update SerieList.add() to use sanitized display names for folders - Add scan_single_series() method for targeted episode scanning - Enhance add_series endpoint: DB save -> folder create -> targeted scan - Update response to include missing_episodes and total_missing - Add comprehensive unit tests for new functionality - Update API tests with proper mock support
320 lines
11 KiB
Python
320 lines
11 KiB
Python
"""Tests for SerieScanner class - file-based operations."""
|
|
|
|
import os
|
|
import tempfile
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from src.core.entities.series import Serie
|
|
from src.core.SerieScanner import SerieScanner
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_directory():
|
|
"""Create a temporary directory with subdirectories for testing."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# Create an anime folder with an mp4 file
|
|
anime_folder = os.path.join(tmpdir, "Attack on Titan (2013)")
|
|
os.makedirs(anime_folder, exist_ok=True)
|
|
|
|
# Create a dummy mp4 file
|
|
mp4_path = os.path.join(
|
|
anime_folder, "Attack on Titan - S01E001 - (German Dub).mp4"
|
|
)
|
|
with open(mp4_path, "w") as f:
|
|
f.write("dummy mp4")
|
|
|
|
yield tmpdir
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_loader():
|
|
"""Create a mock Loader instance."""
|
|
loader = MagicMock()
|
|
loader.get_season_episode_count = MagicMock(return_value={1: 25})
|
|
loader.is_language = MagicMock(return_value=True)
|
|
return loader
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_serie():
|
|
"""Create a sample Serie for testing."""
|
|
return Serie(
|
|
key="attack-on-titan",
|
|
name="Attack on Titan",
|
|
site="aniworld.to",
|
|
folder="Attack on Titan (2013)",
|
|
episodeDict={1: [2, 3, 4]}
|
|
)
|
|
|
|
|
|
class TestSerieScannerInitialization:
|
|
"""Test SerieScanner initialization."""
|
|
|
|
def test_init_success(self, temp_directory, mock_loader):
|
|
"""Test successful initialization."""
|
|
scanner = SerieScanner(temp_directory, mock_loader)
|
|
|
|
assert scanner.directory == os.path.abspath(temp_directory)
|
|
assert scanner.loader == mock_loader
|
|
assert scanner.keyDict == {}
|
|
|
|
def test_init_empty_path_raises_error(self, mock_loader):
|
|
"""Test initialization with empty path raises ValueError."""
|
|
with pytest.raises(ValueError, match="empty"):
|
|
SerieScanner("", mock_loader)
|
|
|
|
def test_init_nonexistent_path_raises_error(self, mock_loader):
|
|
"""Test initialization with non-existent path raises ValueError."""
|
|
with pytest.raises(ValueError, match="does not exist"):
|
|
SerieScanner("/nonexistent/path", mock_loader)
|
|
|
|
|
|
class TestSerieScannerScan:
|
|
"""Test file-based scan operations."""
|
|
|
|
def test_file_based_scan_works(
|
|
self, temp_directory, mock_loader, sample_serie
|
|
):
|
|
"""Test file-based scan works properly."""
|
|
scanner = SerieScanner(temp_directory, mock_loader)
|
|
|
|
with patch.object(scanner, 'get_total_to_scan', return_value=1):
|
|
with patch.object(
|
|
scanner,
|
|
'_SerieScanner__find_mp4_files',
|
|
return_value=iter([
|
|
("Attack on Titan (2013)", ["S01E001.mp4"])
|
|
])
|
|
):
|
|
with patch.object(
|
|
scanner,
|
|
'_SerieScanner__read_data_from_file',
|
|
return_value=sample_serie
|
|
):
|
|
with patch.object(
|
|
scanner,
|
|
'_SerieScanner__get_missing_episodes_and_season',
|
|
return_value=({1: [2, 3]}, "aniworld.to")
|
|
):
|
|
with patch.object(
|
|
sample_serie, 'save_to_file'
|
|
) as mock_save:
|
|
scanner.scan()
|
|
|
|
# Verify file was saved
|
|
mock_save.assert_called_once()
|
|
|
|
def test_keydict_populated_after_scan(
|
|
self, temp_directory, mock_loader, sample_serie
|
|
):
|
|
"""Test keyDict is populated after scan."""
|
|
scanner = SerieScanner(temp_directory, mock_loader)
|
|
|
|
with patch.object(scanner, 'get_total_to_scan', return_value=1):
|
|
with patch.object(
|
|
scanner,
|
|
'_SerieScanner__find_mp4_files',
|
|
return_value=iter([
|
|
("Attack on Titan (2013)", ["S01E001.mp4"])
|
|
])
|
|
):
|
|
with patch.object(
|
|
scanner,
|
|
'_SerieScanner__read_data_from_file',
|
|
return_value=sample_serie
|
|
):
|
|
with patch.object(
|
|
scanner,
|
|
'_SerieScanner__get_missing_episodes_and_season',
|
|
return_value=({1: [2, 3]}, "aniworld.to")
|
|
):
|
|
with patch.object(sample_serie, 'save_to_file'):
|
|
scanner.scan()
|
|
|
|
assert sample_serie.key in scanner.keyDict
|
|
|
|
|
|
class TestSerieScannerSingleSeries:
|
|
"""Test scan_single_series method for targeted scanning."""
|
|
|
|
def test_scan_single_series_basic(
|
|
self, temp_directory, mock_loader
|
|
):
|
|
"""Test basic scan_single_series functionality."""
|
|
scanner = SerieScanner(temp_directory, mock_loader)
|
|
|
|
# Mock the missing episodes calculation
|
|
with patch.object(
|
|
scanner,
|
|
'_SerieScanner__get_missing_episodes_and_season',
|
|
return_value=({1: [5, 6, 7], 2: [1, 2]}, "aniworld.to")
|
|
):
|
|
result = scanner.scan_single_series(
|
|
key="attack-on-titan",
|
|
folder="Attack on Titan (2013)"
|
|
)
|
|
|
|
# Verify result structure
|
|
assert isinstance(result, dict)
|
|
assert 1 in result
|
|
assert 2 in result
|
|
assert result[1] == [5, 6, 7]
|
|
assert result[2] == [1, 2]
|
|
|
|
def test_scan_single_series_updates_keydict(
|
|
self, temp_directory, mock_loader
|
|
):
|
|
"""Test that scan_single_series updates keyDict."""
|
|
scanner = SerieScanner(temp_directory, mock_loader)
|
|
|
|
with patch.object(
|
|
scanner,
|
|
'_SerieScanner__get_missing_episodes_and_season',
|
|
return_value=({1: [1, 2, 3]}, "aniworld.to")
|
|
):
|
|
scanner.scan_single_series(
|
|
key="test-anime",
|
|
folder="Test Anime"
|
|
)
|
|
|
|
# Verify keyDict was updated
|
|
assert "test-anime" in scanner.keyDict
|
|
assert scanner.keyDict["test-anime"].episodeDict == {1: [1, 2, 3]}
|
|
|
|
def test_scan_single_series_existing_entry(
|
|
self, temp_directory, mock_loader, sample_serie
|
|
):
|
|
"""Test scan_single_series updates existing entry in keyDict."""
|
|
scanner = SerieScanner(temp_directory, mock_loader)
|
|
|
|
# Pre-populate keyDict
|
|
scanner.keyDict[sample_serie.key] = sample_serie
|
|
old_episode_dict = sample_serie.episodeDict.copy()
|
|
|
|
with patch.object(
|
|
scanner,
|
|
'_SerieScanner__get_missing_episodes_and_season',
|
|
return_value=({1: [10, 11, 12]}, "aniworld.to")
|
|
):
|
|
scanner.scan_single_series(
|
|
key=sample_serie.key,
|
|
folder=sample_serie.folder
|
|
)
|
|
|
|
# Verify existing entry was updated
|
|
assert scanner.keyDict[sample_serie.key].episodeDict != old_episode_dict
|
|
assert scanner.keyDict[sample_serie.key].episodeDict == {1: [10, 11, 12]}
|
|
|
|
def test_scan_single_series_empty_key_raises_error(
|
|
self, temp_directory, mock_loader
|
|
):
|
|
"""Test that empty key raises ValueError."""
|
|
scanner = SerieScanner(temp_directory, mock_loader)
|
|
|
|
with pytest.raises(ValueError, match="key cannot be empty"):
|
|
scanner.scan_single_series(key="", folder="Test Folder")
|
|
|
|
def test_scan_single_series_empty_folder_raises_error(
|
|
self, temp_directory, mock_loader
|
|
):
|
|
"""Test that empty folder raises ValueError."""
|
|
scanner = SerieScanner(temp_directory, mock_loader)
|
|
|
|
with pytest.raises(ValueError, match="folder cannot be empty"):
|
|
scanner.scan_single_series(key="test-key", folder="")
|
|
|
|
def test_scan_single_series_nonexistent_folder(
|
|
self, temp_directory, mock_loader
|
|
):
|
|
"""Test scanning a series with non-existent folder."""
|
|
scanner = SerieScanner(temp_directory, mock_loader)
|
|
|
|
# Mock to return some episodes (as if from provider)
|
|
with patch.object(
|
|
scanner,
|
|
'_SerieScanner__get_missing_episodes_and_season',
|
|
return_value=({1: [1, 2, 3, 4, 5]}, "aniworld.to")
|
|
):
|
|
result = scanner.scan_single_series(
|
|
key="new-anime",
|
|
folder="NonExistent Folder"
|
|
)
|
|
|
|
# Should still return missing episodes from provider
|
|
assert result == {1: [1, 2, 3, 4, 5]}
|
|
|
|
def test_scan_single_series_error_handling(
|
|
self, temp_directory, mock_loader
|
|
):
|
|
"""Test that errors during scan return empty dict."""
|
|
scanner = SerieScanner(temp_directory, mock_loader)
|
|
|
|
with patch.object(
|
|
scanner,
|
|
'_SerieScanner__get_missing_episodes_and_season',
|
|
side_effect=Exception("Provider error")
|
|
):
|
|
result = scanner.scan_single_series(
|
|
key="test-anime",
|
|
folder="Test Folder"
|
|
)
|
|
|
|
# Should return empty dict on error
|
|
assert result == {}
|
|
|
|
def test_scan_single_series_no_missing_episodes(
|
|
self, temp_directory, mock_loader
|
|
):
|
|
"""Test scan when no episodes are missing."""
|
|
scanner = SerieScanner(temp_directory, mock_loader)
|
|
|
|
with patch.object(
|
|
scanner,
|
|
'_SerieScanner__get_missing_episodes_and_season',
|
|
return_value=({}, "aniworld.to")
|
|
):
|
|
result = scanner.scan_single_series(
|
|
key="complete-anime",
|
|
folder="Complete Anime"
|
|
)
|
|
|
|
assert result == {}
|
|
assert "complete-anime" in scanner.keyDict
|
|
assert scanner.keyDict["complete-anime"].episodeDict == {}
|
|
|
|
def test_scan_single_series_with_existing_files(
|
|
self, temp_directory, mock_loader
|
|
):
|
|
"""Test scan with existing MP4 files in folder."""
|
|
# Create folder with some files
|
|
anime_folder = os.path.join(temp_directory, "Test Anime")
|
|
os.makedirs(anime_folder, exist_ok=True)
|
|
season_folder = os.path.join(anime_folder, "Season 1")
|
|
os.makedirs(season_folder, exist_ok=True)
|
|
|
|
# Create dummy MP4 files
|
|
for ep in [1, 2, 3]:
|
|
mp4_path = os.path.join(
|
|
season_folder, f"Test Anime - S01E{ep:03d} - (German Dub).mp4"
|
|
)
|
|
with open(mp4_path, "w") as f:
|
|
f.write("dummy")
|
|
|
|
scanner = SerieScanner(temp_directory, mock_loader)
|
|
|
|
# Mock to return missing episodes (4, 5, 6)
|
|
with patch.object(
|
|
scanner,
|
|
'_SerieScanner__get_missing_episodes_and_season',
|
|
return_value=({1: [4, 5, 6]}, "aniworld.to")
|
|
):
|
|
result = scanner.scan_single_series(
|
|
key="test-anime",
|
|
folder="Test Anime"
|
|
)
|
|
|
|
# Should only show missing episodes
|
|
assert result == {1: [4, 5, 6]}
|