Lukas Pupka-Lipinski
3f98dd6ebb
- Add SetupService for detecting application setup completion - Create ApplicationFlowMiddleware to enforce setup auth main flow - Add setup processing endpoints (/api/auth/setup, /api/auth/setup/status) - Add Pydantic models for setup requests and responses - Integrate middleware into FastAPI application - Fix logging paths to use ./logs consistently - All existing templates (setup.html, login.html) already working
407 lines
16 KiB
Python
407 lines
16 KiB
Python
"""
|
|
End-to-end tests for CLI flows.
|
|
|
|
Tests complete CLI workflows including progress bar functionality,
|
|
retry logic, user interactions, and error scenarios.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
|
|
# Add source directory to path
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..'))
|
|
|
|
# Import after path setup
|
|
from src.cli.Main import SeriesApp # noqa: E402
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_directory():
|
|
"""Create a temporary directory for testing."""
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
yield temp_dir
|
|
|
|
|
|
@pytest.mark.e2e
|
|
class TestCLICompleteWorkflows:
|
|
"""Test complete CLI workflows from user perspective."""
|
|
|
|
def test_search_and_download_workflow(self, temp_directory):
|
|
"""Test complete search -> select -> download workflow."""
|
|
with patch('src.cli.Main.Loaders'), \
|
|
patch('src.cli.Main.SerieScanner'), \
|
|
patch('src.cli.Main.SerieList'):
|
|
|
|
app = SeriesApp(temp_directory)
|
|
|
|
# Mock search results
|
|
mock_search_results = [
|
|
{"name": "Test Anime", "link": "test_link"}
|
|
]
|
|
|
|
# Mock series for download
|
|
mock_episode_dict = {1: [1, 2, 3], 2: [1, 2]}
|
|
mock_series = Mock(
|
|
episodeDict=mock_episode_dict,
|
|
folder="test_anime",
|
|
key="test_key"
|
|
)
|
|
app.series_list = [mock_series]
|
|
|
|
# Mock loader
|
|
mock_loader = Mock()
|
|
mock_loader.Search.return_value = mock_search_results
|
|
mock_loader.IsLanguage.return_value = True
|
|
mock_loader.Download.return_value = None
|
|
app.Loaders.GetLoader.return_value = mock_loader
|
|
|
|
# Test search workflow
|
|
with patch('builtins.input', side_effect=['test query', '1']), \
|
|
patch('builtins.print'):
|
|
|
|
app.search_mode()
|
|
|
|
# Should have called search and add
|
|
mock_loader.Search.assert_called_with('test query')
|
|
app.List.add.assert_called_once()
|
|
|
|
# Test download workflow
|
|
with patch('rich.progress.Progress') as mock_progress_class, \
|
|
patch('time.sleep'), \
|
|
patch('builtins.input', return_value='1'):
|
|
|
|
mock_progress = Mock()
|
|
mock_progress_class.return_value = mock_progress
|
|
|
|
selected_series = app.get_user_selection()
|
|
assert selected_series is not None
|
|
|
|
app.download_series(selected_series)
|
|
|
|
# Should have set up progress tracking
|
|
mock_progress.start.assert_called_once()
|
|
mock_progress.stop.assert_called_once()
|
|
|
|
# Should have attempted downloads for all episodes
|
|
expected_downloads = sum(len(episodes) for episodes in mock_episode_dict.values())
|
|
assert mock_loader.Download.call_count == expected_downloads
|
|
|
|
def test_init_and_rescan_workflow(self, temp_directory):
|
|
"""Test initialization and rescanning workflow."""
|
|
with patch('src.cli.Main.Loaders'), \
|
|
patch('src.cli.Main.SerieScanner') as mock_scanner_class, \
|
|
patch('src.cli.Main.SerieList') as mock_list_class:
|
|
|
|
mock_scanner = Mock()
|
|
mock_scanner_class.return_value = mock_scanner
|
|
mock_list = Mock()
|
|
mock_list_class.return_value = mock_list
|
|
|
|
app = SeriesApp(temp_directory)
|
|
app.SerieScanner = mock_scanner
|
|
|
|
# Test rescan workflow
|
|
with patch('rich.progress.Progress') as mock_progress_class, \
|
|
patch('builtins.print'):
|
|
|
|
mock_progress = Mock()
|
|
mock_progress_class.return_value = mock_progress
|
|
|
|
# Simulate init action
|
|
app.progress = mock_progress
|
|
app.task1 = "task1_id"
|
|
|
|
# Call reinit workflow
|
|
app.SerieScanner.Reinit()
|
|
app.SerieScanner.Scan(app.updateFromReinit)
|
|
|
|
# Should have called scanner methods
|
|
mock_scanner.Reinit.assert_called_once()
|
|
mock_scanner.Scan.assert_called_once()
|
|
|
|
def test_error_recovery_workflow(self, temp_directory):
|
|
"""Test error recovery in CLI workflows."""
|
|
with patch('src.cli.Main.Loaders'), \
|
|
patch('src.cli.Main.SerieScanner'), \
|
|
patch('src.cli.Main.SerieList'):
|
|
|
|
app = SeriesApp(temp_directory)
|
|
|
|
# Test retry mechanism with eventual success
|
|
mock_func = Mock(side_effect=[
|
|
Exception("First failure"),
|
|
Exception("Second failure"),
|
|
None # Success on third try
|
|
])
|
|
|
|
with patch('time.sleep'), patch('builtins.print'):
|
|
result = app.retry(mock_func, max_retries=3, delay=0)
|
|
|
|
assert result is True
|
|
assert mock_func.call_count == 3
|
|
|
|
# Test retry mechanism with persistent failure
|
|
mock_func_fail = Mock(side_effect=Exception("Persistent error"))
|
|
|
|
with patch('time.sleep'), patch('builtins.print'):
|
|
result = app.retry(mock_func_fail, max_retries=2, delay=0)
|
|
|
|
assert result is False
|
|
assert mock_func_fail.call_count == 2
|
|
|
|
|
|
@pytest.mark.e2e
|
|
class TestCLIUserInteractionFlows:
|
|
"""Test CLI user interaction flows."""
|
|
|
|
def test_user_selection_validation_flow(self, temp_directory):
|
|
"""Test user selection with various invalid inputs before success."""
|
|
with patch('src.cli.Main.Loaders'), \
|
|
patch('src.cli.Main.SerieScanner'), \
|
|
patch('src.cli.Main.SerieList'):
|
|
|
|
app = SeriesApp(temp_directory)
|
|
app.series_list = [
|
|
Mock(name="Anime 1", folder="anime1"),
|
|
Mock(name="Anime 2", folder="anime2")
|
|
]
|
|
|
|
# Test sequence: invalid text -> invalid number -> valid selection
|
|
input_sequence = ['invalid_text', '999', '1']
|
|
|
|
with patch('builtins.input', side_effect=input_sequence), \
|
|
patch('builtins.print'):
|
|
|
|
selected = app.get_user_selection()
|
|
|
|
assert selected is not None
|
|
assert len(selected) == 1
|
|
assert selected[0].name == "Anime 1"
|
|
|
|
def test_search_interaction_flow(self, temp_directory):
|
|
"""Test search interaction with various user inputs."""
|
|
with patch('src.cli.Main.Loaders'), \
|
|
patch('src.cli.Main.SerieScanner'), \
|
|
patch('src.cli.Main.SerieList'):
|
|
|
|
app = SeriesApp(temp_directory)
|
|
|
|
mock_search_results = [
|
|
{"name": "Result 1", "link": "link1"},
|
|
{"name": "Result 2", "link": "link2"}
|
|
]
|
|
|
|
mock_loader = Mock()
|
|
mock_loader.Search.return_value = mock_search_results
|
|
app.Loaders.GetLoader.return_value = mock_loader
|
|
|
|
# Test sequence: search -> invalid selection -> valid selection
|
|
with patch('builtins.input', side_effect=['test search', '999', '1']), \
|
|
patch('builtins.print'):
|
|
|
|
app.search_mode()
|
|
|
|
# Should have added the selected item
|
|
app.List.add.assert_called_once()
|
|
|
|
def test_main_loop_interaction_flow(self, temp_directory):
|
|
"""Test main application loop with user interactions."""
|
|
with patch('src.cli.Main.Loaders'), \
|
|
patch('src.cli.Main.SerieScanner'), \
|
|
patch('src.cli.Main.SerieList'):
|
|
|
|
app = SeriesApp(temp_directory)
|
|
app.series_list = [Mock(name="Test Anime", folder="test")]
|
|
|
|
# Mock various components
|
|
with patch.object(app, 'search_mode') as mock_search, \
|
|
patch.object(app, 'get_user_selection', return_value=[Mock()]), \
|
|
patch.object(app, 'download_series') as mock_download, \
|
|
patch('rich.progress.Progress'), \
|
|
patch('builtins.print'):
|
|
|
|
# Test sequence: search -> download -> exit
|
|
with patch('builtins.input', side_effect=['s', 'd', KeyboardInterrupt()]):
|
|
try:
|
|
app.run()
|
|
except KeyboardInterrupt:
|
|
pass
|
|
|
|
mock_search.assert_called_once()
|
|
mock_download.assert_called_once()
|
|
|
|
|
|
@pytest.mark.e2e
|
|
class TestCLIProgressAndFeedback:
|
|
"""Test CLI progress indicators and user feedback."""
|
|
|
|
def test_download_progress_flow(self, temp_directory):
|
|
"""Test download progress tracking throughout workflow."""
|
|
with patch('src.cli.Main.Loaders'), \
|
|
patch('src.cli.Main.SerieScanner'), \
|
|
patch('src.cli.Main.SerieList'):
|
|
|
|
app = SeriesApp(temp_directory)
|
|
|
|
# Mock series with episodes
|
|
mock_series = [
|
|
Mock(
|
|
episodeDict={1: [1, 2], 2: [1]},
|
|
folder="anime1",
|
|
key="key1"
|
|
)
|
|
]
|
|
|
|
# Mock loader
|
|
mock_loader = Mock()
|
|
mock_loader.IsLanguage.return_value = True
|
|
mock_loader.Download.return_value = None
|
|
app.Loaders.GetLoader.return_value = mock_loader
|
|
|
|
with patch('rich.progress.Progress') as mock_progress_class, \
|
|
patch('time.sleep'):
|
|
|
|
mock_progress = Mock()
|
|
mock_progress_class.return_value = mock_progress
|
|
|
|
app.download_series(mock_series)
|
|
|
|
# Verify progress setup
|
|
assert mock_progress.add_task.call_count >= 3 # At least 3 tasks
|
|
mock_progress.start.assert_called_once()
|
|
mock_progress.stop.assert_called_once()
|
|
|
|
# Verify progress updates
|
|
assert mock_progress.update.call_count > 0
|
|
|
|
def test_progress_callback_integration(self, temp_directory):
|
|
"""Test progress callback integration with download system."""
|
|
with patch('src.cli.Main.Loaders'), \
|
|
patch('src.cli.Main.SerieScanner'), \
|
|
patch('src.cli.Main.SerieList'):
|
|
|
|
app = SeriesApp(temp_directory)
|
|
app.progress = Mock()
|
|
app.task3 = "download_task"
|
|
|
|
# Test various progress states
|
|
progress_states = [
|
|
{
|
|
'status': 'downloading',
|
|
'total_bytes': 1000000,
|
|
'downloaded_bytes': 250000
|
|
},
|
|
{
|
|
'status': 'downloading',
|
|
'total_bytes': 1000000,
|
|
'downloaded_bytes': 750000
|
|
},
|
|
{
|
|
'status': 'finished'
|
|
}
|
|
]
|
|
|
|
for state in progress_states:
|
|
app.print_Download_Progress(state)
|
|
|
|
# Should have updated progress for each state
|
|
assert app.progress.update.call_count == len(progress_states)
|
|
|
|
# Last call should indicate completion
|
|
last_call = app.progress.update.call_args_list[-1]
|
|
assert last_call[1].get('completed') == 100
|
|
|
|
def test_scan_progress_integration(self, temp_directory):
|
|
"""Test scanning progress integration."""
|
|
with patch('src.cli.Main.Loaders'), \
|
|
patch('src.cli.Main.SerieScanner'), \
|
|
patch('src.cli.Main.SerieList'):
|
|
|
|
app = SeriesApp(temp_directory)
|
|
app.progress = Mock()
|
|
app.task1 = "scan_task"
|
|
|
|
# Simulate scan progress updates
|
|
for i in range(5):
|
|
app.updateFromReinit("folder", i)
|
|
|
|
# Should have updated progress for each folder
|
|
assert app.progress.update.call_count == 5
|
|
|
|
# Each call should advance by 1
|
|
for call in app.progress.update.call_args_list:
|
|
assert call[1].get('advance') == 1
|
|
|
|
|
|
@pytest.mark.e2e
|
|
class TestCLIErrorScenarios:
|
|
"""Test CLI error scenarios and recovery."""
|
|
|
|
def test_network_error_recovery(self, temp_directory):
|
|
"""Test recovery from network errors during operations."""
|
|
with patch('src.cli.Main.Loaders'), \
|
|
patch('src.cli.Main.SerieScanner'), \
|
|
patch('src.cli.Main.SerieList'):
|
|
|
|
app = SeriesApp(temp_directory)
|
|
|
|
# Mock network failures
|
|
network_error = Exception("Network connection failed")
|
|
mock_func = Mock(side_effect=[network_error, network_error, None])
|
|
|
|
with patch('time.sleep'), patch('builtins.print'):
|
|
result = app.retry(mock_func, max_retries=3, delay=0)
|
|
|
|
assert result is True
|
|
assert mock_func.call_count == 3
|
|
|
|
def test_invalid_directory_handling(self):
|
|
"""Test handling of invalid directory paths."""
|
|
invalid_directory = "/nonexistent/path/that/does/not/exist"
|
|
|
|
with patch('src.cli.Main.Loaders'), \
|
|
patch('src.cli.Main.SerieScanner'), \
|
|
patch('src.cli.Main.SerieList'):
|
|
|
|
# Should not raise exception during initialization
|
|
app = SeriesApp(invalid_directory)
|
|
assert app.directory_to_search == invalid_directory
|
|
|
|
def test_empty_search_results_handling(self, temp_directory):
|
|
"""Test handling of empty search results."""
|
|
with patch('src.cli.Main.Loaders'), \
|
|
patch('src.cli.Main.SerieScanner'), \
|
|
patch('src.cli.Main.SerieList'):
|
|
|
|
app = SeriesApp(temp_directory)
|
|
|
|
# Mock empty search results
|
|
mock_loader = Mock()
|
|
mock_loader.Search.return_value = []
|
|
app.Loaders.GetLoader.return_value = mock_loader
|
|
|
|
with patch('builtins.input', return_value='nonexistent anime'), \
|
|
patch('builtins.print') as mock_print:
|
|
|
|
app.search_mode()
|
|
|
|
# Should print "No results found" message
|
|
print_calls = [call[0][0] for call in mock_print.call_args_list]
|
|
assert any("No results found" in call for call in print_calls)
|
|
|
|
def test_keyboard_interrupt_handling(self, temp_directory):
|
|
"""Test graceful handling of keyboard interrupts."""
|
|
with patch('src.cli.Main.Loaders'), \
|
|
patch('src.cli.Main.SerieScanner'), \
|
|
patch('src.cli.Main.SerieList'):
|
|
|
|
app = SeriesApp(temp_directory)
|
|
|
|
# Test that KeyboardInterrupt propagates correctly
|
|
with patch('builtins.input', side_effect=KeyboardInterrupt()):
|
|
with pytest.raises(KeyboardInterrupt):
|
|
app.run() |