""" 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)