""" Live Flask App API Tests These tests actually start the Flask application and make real HTTP requests to test the API endpoints end-to-end. """ import unittest import json import sys import os from unittest.mock import patch, MagicMock # Add paths for imports sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..', 'src')) sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..', 'src', 'server')) class LiveFlaskAPITests(unittest.TestCase): """Tests that use actual Flask test client to test API endpoints.""" @classmethod def setUpClass(cls): """Set up Flask app for testing.""" try: # Mock all the complex dependencies before importing the app with patch('sys.modules') as mock_modules: # Mock modules that might not be available mock_modules['main'] = MagicMock() mock_modules['core.entities.series'] = MagicMock() mock_modules['core.entities'] = MagicMock() mock_modules['infrastructure.file_system'] = MagicMock() mock_modules['infrastructure.providers.provider_factory'] = MagicMock() mock_modules['web.controllers.auth_controller'] = MagicMock() mock_modules['config'] = MagicMock() mock_modules['application.services.queue_service'] = MagicMock() # Try to import the Flask app try: from app import app cls.app = app cls.app.config['TESTING'] = True cls.app.config['WTF_CSRF_ENABLED'] = False cls.client = app.test_client() cls.app_available = True except Exception as e: print(f"⚠️ Could not import Flask app: {e}") cls.app_available = False cls.app = None cls.client = None except Exception as e: print(f"⚠️ Could not set up Flask app: {e}") cls.app_available = False cls.app = None cls.client = None def setUp(self): """Set up for each test.""" if not self.app_available: self.skipTest("Flask app not available for testing") def test_static_routes_exist(self): """Test that static JavaScript and CSS routes exist.""" static_routes = [ '/static/js/keyboard-shortcuts.js', '/static/js/drag-drop.js', '/static/js/bulk-operations.js', '/static/js/user-preferences.js', '/static/js/advanced-search.js', '/static/css/ux-features.css' ] for route in static_routes: response = self.client.get(route) # Should return 200 (content) or 404 (route exists but no content) # Should NOT return 500 (server error) self.assertNotEqual(response.status_code, 500, f"Route {route} should not return server error") def test_main_page_routes(self): """Test that main page routes exist.""" routes = ['/', '/login', '/setup'] for route in routes: response = self.client.get(route) # Should return 200, 302 (redirect), or 404 # Should NOT return 500 (server error) self.assertIn(response.status_code, [200, 302, 404], f"Route {route} returned unexpected status: {response.status_code}") def test_api_auth_status_endpoint(self): """Test GET /api/auth/status endpoint.""" response = self.client.get('/api/auth/status') # Should return a valid HTTP status (not 500 error) self.assertNotEqual(response.status_code, 500, "Auth status endpoint should not return server error") # If it returns 200, should have JSON content if response.status_code == 200: try: data = json.loads(response.data) # Should have basic auth status fields expected_fields = ['authenticated', 'has_master_password', 'setup_required'] for field in expected_fields: self.assertIn(field, data, f"Auth status should include {field}") except json.JSONDecodeError: self.fail("Auth status should return valid JSON") def test_api_series_endpoint(self): """Test GET /api/series endpoint.""" response = self.client.get('/api/series') # Should return a valid HTTP status self.assertNotEqual(response.status_code, 500, "Series endpoint should not return server error") # If it returns 200, should have JSON content if response.status_code == 200: try: data = json.loads(response.data) # Should have basic series response structure expected_fields = ['status', 'series', 'total_series'] for field in expected_fields: self.assertIn(field, data, f"Series response should include {field}") except json.JSONDecodeError: self.fail("Series endpoint should return valid JSON") def test_api_status_endpoint(self): """Test GET /api/status endpoint.""" response = self.client.get('/api/status') # Should return a valid HTTP status self.assertNotEqual(response.status_code, 500, "Status endpoint should not return server error") # If it returns 200, should have JSON content if response.status_code == 200: try: data = json.loads(response.data) # Should have basic status fields expected_fields = ['success', 'directory', 'series_count'] for field in expected_fields: self.assertIn(field, data, f"Status response should include {field}") except json.JSONDecodeError: self.fail("Status endpoint should return valid JSON") def test_api_process_locks_endpoint(self): """Test GET /api/process/locks/status endpoint.""" response = self.client.get('/api/process/locks/status') # Should return a valid HTTP status self.assertNotEqual(response.status_code, 500, "Process locks endpoint should not return server error") # If it returns 200, should have JSON content if response.status_code == 200: try: data = json.loads(response.data) # Should have basic lock status fields expected_fields = ['success', 'locks'] for field in expected_fields: self.assertIn(field, data, f"Lock status should include {field}") if 'locks' in data: # Should have rescan and download lock info lock_types = ['rescan', 'download'] for lock_type in lock_types: self.assertIn(lock_type, data['locks'], f"Locks should include {lock_type}") except json.JSONDecodeError: self.fail("Process locks endpoint should return valid JSON") def test_api_search_endpoint_with_post(self): """Test POST /api/search endpoint with valid data.""" test_data = {'query': 'test anime'} response = self.client.post( '/api/search', data=json.dumps(test_data), content_type='application/json' ) # Should return a valid HTTP status self.assertNotEqual(response.status_code, 500, "Search endpoint should not return server error") # Should handle JSON input (200 success or 400 bad request) self.assertIn(response.status_code, [200, 400, 401, 403], f"Search endpoint returned unexpected status: {response.status_code}") def test_api_search_endpoint_empty_query(self): """Test POST /api/search endpoint with empty query.""" test_data = {'query': ''} response = self.client.post( '/api/search', data=json.dumps(test_data), content_type='application/json' ) # Should return 400 bad request for empty query if response.status_code == 200: try: data = json.loads(response.data) # If it processed the request, should indicate error if data.get('status') == 'error': self.assertIn('empty', data.get('message', '').lower(), "Should indicate query is empty") except json.JSONDecodeError: pass # OK if it's not JSON def test_api_scheduler_config_endpoint(self): """Test GET /api/scheduler/config endpoint.""" response = self.client.get('/api/scheduler/config') # Should return a valid HTTP status self.assertNotEqual(response.status_code, 500, "Scheduler config endpoint should not return server error") # If it returns 200, should have JSON content if response.status_code == 200: try: data = json.loads(response.data) # Should have basic config structure expected_fields = ['success', 'config'] for field in expected_fields: self.assertIn(field, data, f"Scheduler config should include {field}") except json.JSONDecodeError: self.fail("Scheduler config should return valid JSON") def test_api_logging_config_endpoint(self): """Test GET /api/logging/config endpoint.""" response = self.client.get('/api/logging/config') # Should return a valid HTTP status self.assertNotEqual(response.status_code, 500, "Logging config endpoint should not return server error") # If it returns 200, should have JSON content if response.status_code == 200: try: data = json.loads(response.data) # Should have basic config structure expected_fields = ['success', 'config'] for field in expected_fields: self.assertIn(field, data, f"Logging config should include {field}") except json.JSONDecodeError: self.fail("Logging config should return valid JSON") def test_api_advanced_config_endpoint(self): """Test GET /api/config/section/advanced endpoint.""" response = self.client.get('/api/config/section/advanced') # Should return a valid HTTP status self.assertNotEqual(response.status_code, 500, "Advanced config endpoint should not return server error") # If it returns 200, should have JSON content if response.status_code == 200: try: data = json.loads(response.data) # Should have basic config structure expected_fields = ['success', 'config'] for field in expected_fields: self.assertIn(field, data, f"Advanced config should include {field}") except json.JSONDecodeError: self.fail("Advanced config should return valid JSON") def test_api_logging_files_endpoint(self): """Test GET /api/logging/files endpoint.""" response = self.client.get('/api/logging/files') # Should return a valid HTTP status self.assertNotEqual(response.status_code, 500, "Logging files endpoint should not return server error") # If it returns 200, should have JSON content if response.status_code == 200: try: data = json.loads(response.data) # Should have basic response structure expected_fields = ['success', 'files'] for field in expected_fields: self.assertIn(field, data, f"Logging files should include {field}") # Files should be a list self.assertIsInstance(data['files'], list, "Files should be a list") except json.JSONDecodeError: self.fail("Logging files should return valid JSON") def test_nonexistent_api_endpoint(self): """Test that non-existent API endpoints return 404.""" response = self.client.get('/api/nonexistent/endpoint') # Should return 404 not found self.assertEqual(response.status_code, 404, "Non-existent endpoints should return 404") def test_api_endpoints_handle_invalid_methods(self): """Test that API endpoints handle invalid HTTP methods properly.""" # Test GET on POST-only endpoints post_only_endpoints = [ '/api/auth/login', '/api/auth/logout', '/api/rescan', '/api/download' ] for endpoint in post_only_endpoints: response = self.client.get(endpoint) # Should return 405 Method Not Allowed or 404 Not Found self.assertIn(response.status_code, [404, 405], f"GET on POST-only endpoint {endpoint} should return 404 or 405") def test_api_endpoints_content_type(self): """Test that API endpoints return proper content types.""" json_endpoints = [ '/api/auth/status', '/api/series', '/api/status', '/api/scheduler/config', '/api/logging/config' ] for endpoint in json_endpoints: response = self.client.get(endpoint) if response.status_code == 200: # Should have JSON content type or be valid JSON content_type = response.headers.get('Content-Type', '') if 'application/json' not in content_type: # If not explicitly JSON content type, should still be valid JSON try: json.loads(response.data) except json.JSONDecodeError: self.fail(f"Endpoint {endpoint} should return valid JSON") class APIEndpointDiscoveryTest(unittest.TestCase): """Test to discover and validate all available API endpoints.""" @classmethod def setUpClass(cls): """Set up Flask app for endpoint discovery.""" try: # Mock dependencies and import app with patch('sys.modules') as mock_modules: mock_modules['main'] = MagicMock() mock_modules['core.entities.series'] = MagicMock() mock_modules['core.entities'] = MagicMock() mock_modules['infrastructure.file_system'] = MagicMock() mock_modules['infrastructure.providers.provider_factory'] = MagicMock() mock_modules['web.controllers.auth_controller'] = MagicMock() mock_modules['config'] = MagicMock() mock_modules['application.services.queue_service'] = MagicMock() try: from app import app cls.app = app cls.app_available = True except Exception as e: print(f"⚠️ Could not import Flask app for discovery: {e}") cls.app_available = False cls.app = None except Exception as e: print(f"⚠️ Could not set up Flask app for discovery: {e}") cls.app_available = False cls.app = None def setUp(self): """Set up for each test.""" if not self.app_available: self.skipTest("Flask app not available for endpoint discovery") def test_discover_api_endpoints(self): """Discover all registered API endpoints in the Flask app.""" if not self.app: self.skipTest("Flask app not available") # Get all registered routes api_routes = [] other_routes = [] for rule in self.app.url_map.iter_rules(): if rule.rule.startswith('/api/'): methods = ', '.join(sorted(rule.methods - {'OPTIONS', 'HEAD'})) api_routes.append(f"{methods} {rule.rule}") else: other_routes.append(rule.rule) # Print discovered routes print(f"\n🔍 DISCOVERED API ROUTES ({len(api_routes)} total):") for route in sorted(api_routes): print(f" ✓ {route}") print(f"\n📋 DISCOVERED NON-API ROUTES ({len(other_routes)} total):") for route in sorted(other_routes)[:10]: # Show first 10 print(f" - {route}") if len(other_routes) > 10: print(f" ... and {len(other_routes) - 10} more") # Validate we found API routes self.assertGreater(len(api_routes), 0, "Should discover some API routes") # Validate common endpoints exist expected_patterns = [ '/api/auth/', '/api/series', '/api/status', '/api/config/' ] found_patterns = [] for pattern in expected_patterns: for route in api_routes: if pattern in route: found_patterns.append(pattern) break print(f"\n✅ Found {len(found_patterns)}/{len(expected_patterns)} expected API patterns:") for pattern in found_patterns: print(f" ✓ {pattern}") missing_patterns = set(expected_patterns) - set(found_patterns) if missing_patterns: print(f"\n⚠️ Missing expected patterns:") for pattern in missing_patterns: print(f" - {pattern}") if __name__ == '__main__': # Run the live Flask tests loader = unittest.TestLoader() # Load test classes suite = unittest.TestSuite() suite.addTests(loader.loadTestsFromTestCase(LiveFlaskAPITests)) suite.addTests(loader.loadTestsFromTestCase(APIEndpointDiscoveryTest)) # Run tests runner = unittest.TextTestRunner(verbosity=2) result = runner.run(suite) # Print summary print(f"\n{'='*60}") print(f"LIVE FLASK API TEST SUMMARY") print(f"{'='*60}") print(f"Tests run: {result.testsRun}") print(f"Failures: {len(result.failures)}") print(f"Errors: {len(result.errors)}") print(f"Skipped: {len(result.skipped) if hasattr(result, 'skipped') else 0}") if result.testsRun > 0: success_rate = ((result.testsRun - len(result.failures) - len(result.errors)) / result.testsRun * 100) print(f"Success rate: {success_rate:.1f}%") if result.failures: print(f"\n🔥 FAILURES:") for test, traceback in result.failures: print(f" - {test}") if result.errors: print(f"\n💥 ERRORS:") for test, traceback in result.errors: print(f" - {test}") # Summary message if result.wasSuccessful(): print(f"\n🎉 All live Flask API tests passed!") print(f"✅ API endpoints are responding correctly") print(f"✅ JSON responses are properly formatted") print(f"✅ HTTP methods are handled appropriately") print(f"✅ Error handling is working") else: print(f"\n⚠️ Some tests failed - check the Flask app setup") exit(0 if result.wasSuccessful() else 1)