619 lines
22 KiB
Python
619 lines
22 KiB
Python
"""
|
|
Integration Tests for Web Interface
|
|
|
|
This module contains integration tests for the Flask web application,
|
|
testing the complete workflow from HTTP requests to database operations.
|
|
"""
|
|
|
|
import unittest
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import shutil
|
|
import json
|
|
import sqlite3
|
|
from unittest.mock import Mock, MagicMock, patch
|
|
import threading
|
|
import time
|
|
|
|
# Add parent directory to path for imports
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
|
|
|
|
# Import Flask app and components
|
|
from app import app, socketio, init_series_app
|
|
from database_manager import DatabaseManager, AnimeMetadata
|
|
from auth import session_manager
|
|
from config import config
|
|
|
|
|
|
class TestWebInterface(unittest.TestCase):
|
|
"""Integration tests for the web interface."""
|
|
|
|
def setUp(self):
|
|
"""Set up test environment."""
|
|
# Create temporary directory for test files
|
|
self.test_dir = tempfile.mkdtemp()
|
|
|
|
# Configure Flask app for testing
|
|
app.config['TESTING'] = True
|
|
app.config['WTF_CSRF_ENABLED'] = False
|
|
app.config['SECRET_KEY'] = 'test-secret-key'
|
|
|
|
self.app = app
|
|
self.client = app.test_client()
|
|
|
|
# Create test database
|
|
self.test_db_path = os.path.join(self.test_dir, 'test.db')
|
|
|
|
# Mock configuration
|
|
self.original_config = {}
|
|
for attr in ['anime_directory', 'master_password', 'database_path']:
|
|
if hasattr(config, attr):
|
|
self.original_config[attr] = getattr(config, attr)
|
|
|
|
config.anime_directory = self.test_dir
|
|
config.master_password = 'test123'
|
|
config.database_path = self.test_db_path
|
|
|
|
def tearDown(self):
|
|
"""Clean up test environment."""
|
|
# Restore original configuration
|
|
for attr, value in self.original_config.items():
|
|
setattr(config, attr, value)
|
|
|
|
# Clean up temporary files
|
|
shutil.rmtree(self.test_dir, ignore_errors=True)
|
|
|
|
# Clear sessions
|
|
session_manager.clear_all_sessions()
|
|
|
|
def test_index_page_unauthenticated(self):
|
|
"""Test index page redirects to login when unauthenticated."""
|
|
response = self.client.get('/')
|
|
|
|
# Should redirect to login
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertIn('/login', response.location)
|
|
|
|
def test_login_page_loads(self):
|
|
"""Test login page loads correctly."""
|
|
response = self.client.get('/login')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn(b'login', response.data.lower())
|
|
|
|
def test_successful_login(self):
|
|
"""Test successful login flow."""
|
|
# Attempt login with correct password
|
|
response = self.client.post('/login', data={
|
|
'password': 'test123'
|
|
}, follow_redirects=True)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
# Should be redirected to main page after successful login
|
|
|
|
def test_failed_login(self):
|
|
"""Test failed login with wrong password."""
|
|
response = self.client.post('/login', data={
|
|
'password': 'wrong_password'
|
|
})
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
# Should return to login page with error
|
|
|
|
def test_authenticated_index_page(self):
|
|
"""Test index page loads when authenticated."""
|
|
# Login first
|
|
with self.client.session_transaction() as sess:
|
|
sess['authenticated'] = True
|
|
sess['session_id'] = 'test-session'
|
|
session_manager.sessions['test-session'] = {
|
|
'authenticated': True,
|
|
'created_at': time.time(),
|
|
'last_accessed': time.time()
|
|
}
|
|
|
|
response = self.client.get('/')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_api_authentication_required(self):
|
|
"""Test API endpoints require authentication."""
|
|
# Test unauthenticated API call
|
|
response = self.client.get('/api/series/list')
|
|
self.assertEqual(response.status_code, 401)
|
|
|
|
# Test authenticated API call
|
|
with self.client.session_transaction() as sess:
|
|
sess['authenticated'] = True
|
|
sess['session_id'] = 'test-session'
|
|
session_manager.sessions['test-session'] = {
|
|
'authenticated': True,
|
|
'created_at': time.time(),
|
|
'last_accessed': time.time()
|
|
}
|
|
|
|
response = self.client.get('/api/series/list')
|
|
# Should not return 401 (might return other codes based on implementation)
|
|
self.assertNotEqual(response.status_code, 401)
|
|
|
|
def test_config_api_endpoints(self):
|
|
"""Test configuration API endpoints."""
|
|
# Authenticate
|
|
with self.client.session_transaction() as sess:
|
|
sess['authenticated'] = True
|
|
sess['session_id'] = 'test-session'
|
|
session_manager.sessions['test-session'] = {
|
|
'authenticated': True,
|
|
'created_at': time.time(),
|
|
'last_accessed': time.time()
|
|
}
|
|
|
|
# Get current config
|
|
response = self.client.get('/api/config')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
config_data = json.loads(response.data)
|
|
self.assertIn('anime_directory', config_data)
|
|
|
|
def test_download_queue_operations(self):
|
|
"""Test download queue management."""
|
|
# Authenticate
|
|
with self.client.session_transaction() as sess:
|
|
sess['authenticated'] = True
|
|
sess['session_id'] = 'test-session'
|
|
session_manager.sessions['test-session'] = {
|
|
'authenticated': True,
|
|
'created_at': time.time(),
|
|
'last_accessed': time.time()
|
|
}
|
|
|
|
# Get queue status
|
|
response = self.client.get('/api/queue/status')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
queue_data = json.loads(response.data)
|
|
self.assertIn('status', queue_data)
|
|
|
|
def test_process_locking_endpoints(self):
|
|
"""Test process locking API endpoints."""
|
|
# Authenticate
|
|
with self.client.session_transaction() as sess:
|
|
sess['authenticated'] = True
|
|
sess['session_id'] = 'test-session'
|
|
session_manager.sessions['test-session'] = {
|
|
'authenticated': True,
|
|
'created_at': time.time(),
|
|
'last_accessed': time.time()
|
|
}
|
|
|
|
# Check process locks
|
|
response = self.client.get('/api/process/locks')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
locks_data = json.loads(response.data)
|
|
self.assertIn('locks', locks_data)
|
|
|
|
def test_database_api_endpoints(self):
|
|
"""Test database management API endpoints."""
|
|
# Authenticate
|
|
with self.client.session_transaction() as sess:
|
|
sess['authenticated'] = True
|
|
sess['session_id'] = 'test-session'
|
|
session_manager.sessions['test-session'] = {
|
|
'authenticated': True,
|
|
'created_at': time.time(),
|
|
'last_accessed': time.time()
|
|
}
|
|
|
|
# Get database info
|
|
response = self.client.get('/api/database/info')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
db_data = json.loads(response.data)
|
|
self.assertIn('status', db_data)
|
|
|
|
def test_health_monitoring_endpoints(self):
|
|
"""Test health monitoring API endpoints."""
|
|
# Authenticate (health endpoints might be public)
|
|
with self.client.session_transaction() as sess:
|
|
sess['authenticated'] = True
|
|
sess['session_id'] = 'test-session'
|
|
session_manager.sessions['test-session'] = {
|
|
'authenticated': True,
|
|
'created_at': time.time(),
|
|
'last_accessed': time.time()
|
|
}
|
|
|
|
# Get system health
|
|
response = self.client.get('/api/health/system')
|
|
# Health endpoints might be accessible without auth
|
|
self.assertIn(response.status_code, [200, 401])
|
|
|
|
def test_error_handling(self):
|
|
"""Test error handling for invalid requests."""
|
|
# Authenticate
|
|
with self.client.session_transaction() as sess:
|
|
sess['authenticated'] = True
|
|
sess['session_id'] = 'test-session'
|
|
session_manager.sessions['test-session'] = {
|
|
'authenticated': True,
|
|
'created_at': time.time(),
|
|
'last_accessed': time.time()
|
|
}
|
|
|
|
# Test invalid endpoint
|
|
response = self.client.get('/api/nonexistent/endpoint')
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
# Test invalid method
|
|
response = self.client.post('/api/series/list')
|
|
# Should return method not allowed or other appropriate error
|
|
self.assertIn(response.status_code, [405, 400, 404])
|
|
|
|
def test_json_response_format(self):
|
|
"""Test API responses return valid JSON."""
|
|
# Authenticate
|
|
with self.client.session_transaction() as sess:
|
|
sess['authenticated'] = True
|
|
sess['session_id'] = 'test-session'
|
|
session_manager.sessions['test-session'] = {
|
|
'authenticated': True,
|
|
'created_at': time.time(),
|
|
'last_accessed': time.time()
|
|
}
|
|
|
|
# Test various API endpoints for valid JSON
|
|
endpoints = [
|
|
'/api/config',
|
|
'/api/queue/status',
|
|
'/api/process/locks',
|
|
'/api/database/info'
|
|
]
|
|
|
|
for endpoint in endpoints:
|
|
with self.subTest(endpoint=endpoint):
|
|
response = self.client.get(endpoint)
|
|
if response.status_code == 200:
|
|
# Should be valid JSON
|
|
try:
|
|
json.loads(response.data)
|
|
except json.JSONDecodeError:
|
|
self.fail(f"Invalid JSON response from {endpoint}")
|
|
|
|
|
|
class TestSocketIOEvents(unittest.TestCase):
|
|
"""Integration tests for SocketIO events."""
|
|
|
|
def setUp(self):
|
|
"""Set up test environment for SocketIO."""
|
|
app.config['TESTING'] = True
|
|
self.socketio_client = socketio.test_client(app)
|
|
|
|
def tearDown(self):
|
|
"""Clean up SocketIO test environment."""
|
|
if self.socketio_client:
|
|
self.socketio_client.disconnect()
|
|
|
|
def test_socketio_connection(self):
|
|
"""Test SocketIO connection establishment."""
|
|
self.assertTrue(self.socketio_client.is_connected())
|
|
|
|
def test_download_progress_events(self):
|
|
"""Test download progress event handling."""
|
|
# Mock download progress update
|
|
test_progress = {
|
|
'episode': 'Test Episode 1',
|
|
'progress': 50,
|
|
'speed': '1.5 MB/s',
|
|
'eta': '2 minutes'
|
|
}
|
|
|
|
# Emit progress update
|
|
socketio.emit('download_progress', test_progress)
|
|
|
|
# Check if client receives the event
|
|
received = self.socketio_client.get_received()
|
|
# Note: In real tests, you'd check if the client received the event
|
|
|
|
def test_scan_progress_events(self):
|
|
"""Test scan progress event handling."""
|
|
test_scan_data = {
|
|
'status': 'scanning',
|
|
'current_folder': 'Test Anime',
|
|
'progress': 25,
|
|
'total_series': 100,
|
|
'scanned_series': 25
|
|
}
|
|
|
|
# Emit scan progress
|
|
socketio.emit('scan_progress', test_scan_data)
|
|
|
|
# Verify event handling
|
|
received = self.socketio_client.get_received()
|
|
# In real implementation, verify the event was received and processed
|
|
|
|
|
|
class TestDatabaseIntegration(unittest.TestCase):
|
|
"""Integration tests for database operations."""
|
|
|
|
def setUp(self):
|
|
"""Set up database integration test environment."""
|
|
self.test_dir = tempfile.mkdtemp()
|
|
self.test_db = os.path.join(self.test_dir, 'integration_test.db')
|
|
self.db_manager = DatabaseManager(self.test_db)
|
|
|
|
# Configure Flask app for testing
|
|
app.config['TESTING'] = True
|
|
self.client = app.test_client()
|
|
|
|
# Authenticate for API calls
|
|
self.auth_session = {
|
|
'authenticated': True,
|
|
'session_id': 'integration-test-session'
|
|
}
|
|
session_manager.sessions['integration-test-session'] = {
|
|
'authenticated': True,
|
|
'created_at': time.time(),
|
|
'last_accessed': time.time()
|
|
}
|
|
|
|
def tearDown(self):
|
|
"""Clean up database integration test environment."""
|
|
self.db_manager.close()
|
|
shutil.rmtree(self.test_dir, ignore_errors=True)
|
|
session_manager.clear_all_sessions()
|
|
|
|
def test_anime_crud_via_api(self):
|
|
"""Test anime CRUD operations via API endpoints."""
|
|
# Authenticate session
|
|
with self.client.session_transaction() as sess:
|
|
sess.update(self.auth_session)
|
|
|
|
# Create anime via API
|
|
anime_data = {
|
|
'name': 'Integration Test Anime',
|
|
'folder': 'integration_test_folder',
|
|
'key': 'integration-test-key',
|
|
'description': 'Test anime for integration testing',
|
|
'genres': ['Action', 'Adventure'],
|
|
'release_year': 2023,
|
|
'status': 'ongoing'
|
|
}
|
|
|
|
response = self.client.post('/api/database/anime',
|
|
data=json.dumps(anime_data),
|
|
content_type='application/json')
|
|
|
|
self.assertEqual(response.status_code, 201)
|
|
response_data = json.loads(response.data)
|
|
self.assertEqual(response_data['status'], 'success')
|
|
|
|
anime_id = response_data['data']['anime_id']
|
|
|
|
# Read anime via API
|
|
response = self.client.get(f'/api/database/anime/{anime_id}')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
response_data = json.loads(response.data)
|
|
self.assertEqual(response_data['status'], 'success')
|
|
self.assertEqual(response_data['data']['name'], anime_data['name'])
|
|
|
|
# Update anime via API
|
|
update_data = {
|
|
'description': 'Updated description for integration testing'
|
|
}
|
|
|
|
response = self.client.put(f'/api/database/anime/{anime_id}',
|
|
data=json.dumps(update_data),
|
|
content_type='application/json')
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
# Verify update
|
|
response = self.client.get(f'/api/database/anime/{anime_id}')
|
|
response_data = json.loads(response.data)
|
|
self.assertEqual(
|
|
response_data['data']['description'],
|
|
update_data['description']
|
|
)
|
|
|
|
# Delete anime via API
|
|
response = self.client.delete(f'/api/database/anime/{anime_id}')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
# Verify deletion
|
|
response = self.client.get(f'/api/database/anime/{anime_id}')
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
def test_backup_operations_via_api(self):
|
|
"""Test backup operations via API."""
|
|
# Authenticate session
|
|
with self.client.session_transaction() as sess:
|
|
sess.update(self.auth_session)
|
|
|
|
# Create test data
|
|
anime_data = {
|
|
'name': 'Backup Test Anime',
|
|
'folder': 'backup_test_folder',
|
|
'key': 'backup-test-key'
|
|
}
|
|
|
|
response = self.client.post('/api/database/anime',
|
|
data=json.dumps(anime_data),
|
|
content_type='application/json')
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
# Create backup via API
|
|
backup_data = {
|
|
'backup_type': 'full',
|
|
'description': 'Integration test backup'
|
|
}
|
|
|
|
response = self.client.post('/api/database/backups/create',
|
|
data=json.dumps(backup_data),
|
|
content_type='application/json')
|
|
|
|
self.assertEqual(response.status_code, 201)
|
|
response_data = json.loads(response.data)
|
|
self.assertEqual(response_data['status'], 'success')
|
|
|
|
backup_id = response_data['data']['backup_id']
|
|
|
|
# List backups
|
|
response = self.client.get('/api/database/backups')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
response_data = json.loads(response.data)
|
|
self.assertGreater(response_data['data']['count'], 0)
|
|
|
|
# Verify backup exists in list
|
|
backup_found = False
|
|
for backup in response_data['data']['backups']:
|
|
if backup['backup_id'] == backup_id:
|
|
backup_found = True
|
|
break
|
|
self.assertTrue(backup_found)
|
|
|
|
def test_search_functionality(self):
|
|
"""Test search functionality via API."""
|
|
# Authenticate session
|
|
with self.client.session_transaction() as sess:
|
|
sess.update(self.auth_session)
|
|
|
|
# Create test anime for searching
|
|
test_anime = [
|
|
{'name': 'Attack on Titan', 'folder': 'attack_titan', 'key': 'attack-titan'},
|
|
{'name': 'Death Note', 'folder': 'death_note', 'key': 'death-note'},
|
|
{'name': 'Naruto', 'folder': 'naruto', 'key': 'naruto'}
|
|
]
|
|
|
|
for anime_data in test_anime:
|
|
response = self.client.post('/api/database/anime',
|
|
data=json.dumps(anime_data),
|
|
content_type='application/json')
|
|
self.assertEqual(response.status_code, 201)
|
|
|
|
# Test search
|
|
search_queries = [
|
|
('Attack', 1), # Should find "Attack on Titan"
|
|
('Note', 1), # Should find "Death Note"
|
|
('Naruto', 1), # Should find "Naruto"
|
|
('Anime', 0), # Should find nothing
|
|
('', 0) # Empty search should return error
|
|
]
|
|
|
|
for search_term, expected_count in search_queries:
|
|
with self.subTest(search_term=search_term):
|
|
response = self.client.get(f'/api/database/anime/search?q={search_term}')
|
|
|
|
if search_term == '':
|
|
self.assertEqual(response.status_code, 400)
|
|
else:
|
|
self.assertEqual(response.status_code, 200)
|
|
response_data = json.loads(response.data)
|
|
self.assertEqual(response_data['data']['count'], expected_count)
|
|
|
|
|
|
class TestPerformanceIntegration(unittest.TestCase):
|
|
"""Integration tests for performance features."""
|
|
|
|
def setUp(self):
|
|
"""Set up performance integration test environment."""
|
|
app.config['TESTING'] = True
|
|
self.client = app.test_client()
|
|
|
|
# Authenticate
|
|
self.auth_session = {
|
|
'authenticated': True,
|
|
'session_id': 'performance-test-session'
|
|
}
|
|
session_manager.sessions['performance-test-session'] = {
|
|
'authenticated': True,
|
|
'created_at': time.time(),
|
|
'last_accessed': time.time()
|
|
}
|
|
|
|
def tearDown(self):
|
|
"""Clean up performance test environment."""
|
|
session_manager.clear_all_sessions()
|
|
|
|
def test_performance_monitoring_api(self):
|
|
"""Test performance monitoring API endpoints."""
|
|
# Authenticate session
|
|
with self.client.session_transaction() as sess:
|
|
sess.update(self.auth_session)
|
|
|
|
# Test system metrics
|
|
response = self.client.get('/api/performance/system-metrics')
|
|
if response.status_code == 200: # Endpoint might not exist yet
|
|
metrics_data = json.loads(response.data)
|
|
self.assertIn('status', metrics_data)
|
|
|
|
def test_download_speed_limiting(self):
|
|
"""Test download speed limiting configuration."""
|
|
# Authenticate session
|
|
with self.client.session_transaction() as sess:
|
|
sess.update(self.auth_session)
|
|
|
|
# Test speed limit configuration
|
|
speed_config = {'max_speed_mbps': 10}
|
|
|
|
response = self.client.post('/api/performance/speed-limit',
|
|
data=json.dumps(speed_config),
|
|
content_type='application/json')
|
|
|
|
# Endpoint might not exist yet, so check for appropriate response
|
|
self.assertIn(response.status_code, [200, 404, 405])
|
|
|
|
|
|
def run_integration_tests():
|
|
"""Run the integration test suite."""
|
|
# Create test suite
|
|
suite = unittest.TestSuite()
|
|
|
|
# Add integration test cases
|
|
integration_test_classes = [
|
|
TestWebInterface,
|
|
TestSocketIOEvents,
|
|
TestDatabaseIntegration,
|
|
TestPerformanceIntegration
|
|
]
|
|
|
|
for test_class in integration_test_classes:
|
|
tests = unittest.TestLoader().loadTestsFromTestCase(test_class)
|
|
suite.addTests(tests)
|
|
|
|
# Run tests
|
|
runner = unittest.TextTestRunner(verbosity=2)
|
|
result = runner.run(suite)
|
|
|
|
return result
|
|
|
|
|
|
if __name__ == '__main__':
|
|
print("Running AniWorld Integration Tests...")
|
|
print("=" * 50)
|
|
|
|
result = run_integration_tests()
|
|
|
|
print("\n" + "=" * 50)
|
|
print(f"Tests run: {result.testsRun}")
|
|
print(f"Failures: {len(result.failures)}")
|
|
print(f"Errors: {len(result.errors)}")
|
|
|
|
if result.failures:
|
|
print("\nFailures:")
|
|
for test, traceback in result.failures:
|
|
print(f"- {test}")
|
|
|
|
if result.errors:
|
|
print("\nErrors:")
|
|
for test, traceback in result.errors:
|
|
print(f"- {test}")
|
|
|
|
if result.wasSuccessful():
|
|
print("\nAll integration tests passed! ✅")
|
|
sys.exit(0)
|
|
else:
|
|
print("\nSome integration tests failed! ❌")
|
|
sys.exit(1) |