480 lines
20 KiB
Python
480 lines
20 KiB
Python
"""
|
|
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) |