"""Integration tests for backwards compatibility.""" import pytest from fastapi.testclient import TestClient from unittest.mock import Mock, patch, AsyncMock from main import app class TestBackwardsCompatibility: """Test that existing clients continue to work.""" def setup_method(self): """Set up test fixtures.""" self.client = TestClient(app) def test_info_endpoint_remains_public(self): """Test that info endpoint doesn't require authentication.""" response = self.client.get("/api/v1/info") assert response.status_code == 200 response_data = response.json() # Should return expected API info structure assert "version" in response_data assert "supported_video_formats" in response_data assert "supported_audio_formats" in response_data assert "quality_levels" in response_data assert "endpoints" in response_data # Verify expected formats are still supported assert ".mp4" in response_data["supported_video_formats"] assert "mp3" in response_data["supported_audio_formats"] assert "medium" in response_data["quality_levels"] def test_health_endpoint_if_exists(self): """Test health endpoint doesn't require authentication if it exists.""" response = self.client.get("/api/v1/health") # Health endpoint may or may not exist, but if it does, it should be public if response.status_code != 404: assert response.status_code in [200, 204] # Successful health check def test_root_endpoint_behavior(self): """Test that root endpoint behavior is preserved.""" response = self.client.get("/") # Root endpoint should either redirect to web interface or return info # Should not require authentication assert response.status_code in [200, 301, 302, 404] # Various acceptable responses assert response.status_code != 401 # Should not require auth def test_protected_endpoints_now_require_auth(self): """Test that previously unprotected endpoints now require authentication.""" protected_endpoints = [ ("GET", "/api/v1/jobs/test-job-id"), ("GET", "/api/v1/jobs/test-job-id/download"), ("POST", "/api/v1/extract"), ] for method, endpoint in protected_endpoints: if method == "GET": response = self.client.get(endpoint) elif method == "POST": # For POST extract, provide minimal valid data response = self.client.post( endpoint, files={"video": ("test.mp4", b"fake content", "video/mp4")}, data={"output_format": "mp3", "quality": "medium"} ) assert response.status_code == 401 assert "Authorization" in response.json()["error"] or \ "authentication" in response.json()["error"].lower() def test_error_response_format_compatibility(self): """Test that error response format is compatible with existing clients.""" # Test authentication error format response = self.client.get("/api/v1/jobs/test-job") assert response.status_code == 401 response_data = response.json() # Ensure error response has expected structure assert "error" in response_data assert isinstance(response_data["error"], str) # Optional fields that enhance but don't break compatibility if "code" in response_data: assert isinstance(response_data["code"], str) if "details" in response_data: assert isinstance(response_data["details"], str) def test_form_parameter_names_unchanged(self): """Test that form parameter names haven't changed.""" # Test that existing parameter names still work test_file_content = b"fake video content" response = self.client.post( "/api/v1/extract", files={"video": ("test.mp4", test_file_content, "video/mp4")}, data={ "output_format": "mp3", # Existing parameter "quality": "medium", # Existing parameter # job_id is new and optional } ) # Should fail on auth, not on parameter validation assert response.status_code == 401 assert "authorization" in response.json()["error"].lower() def test_existing_response_fields_preserved(self): """Test that existing response fields are preserved when using new auth.""" # This test would verify that authenticated requests return the same # response structure as before, just with added fields # For now, we'll test the expected response structure expected_job_response_fields = [ "job_id", "status", "message", "check_url", "file_size_mb" ] expected_status_response_fields = [ "job_id", "status", "created_at", "updated_at", "filename", "file_size_mb", "output_format", "quality" ] # These field lists represent the minimum compatibility requirements # New fields like external_job_id are additions, not replacements assert len(expected_job_response_fields) > 0 assert len(expected_status_response_fields) > 0 class TestLegacyClientCompatibility: """Test compatibility with legacy client patterns.""" def setup_method(self): """Set up test fixtures.""" self.client = TestClient(app) def test_content_type_handling_unchanged(self): """Test that content type handling for uploads hasn't changed.""" test_file_content = b"fake video content" # Test multipart/form-data upload (standard way) response = self.client.post( "/api/v1/extract", files={"video": ("test.mp4", test_file_content, "video/mp4")}, data={"output_format": "mp3", "quality": "medium"} ) # Should fail on auth, not on content type assert response.status_code == 401 assert "content-type" not in response.json()["error"].lower() def test_http_methods_unchanged(self): """Test that HTTP methods for endpoints haven't changed.""" endpoint_methods = [ ("/api/v1/info", "GET"), ("/api/v1/extract", "POST"), ("/api/v1/jobs/test-job", "GET"), ("/api/v1/jobs/test-job/download", "GET"), ] for endpoint, expected_method in endpoint_methods: if expected_method == "GET": response = self.client.get(endpoint) elif expected_method == "POST": response = self.client.post( endpoint, files={"video": ("test.mp4", b"fake", "video/mp4")}, data={"output_format": "mp3"} ) # Should not return 405 Method Not Allowed assert response.status_code != 405 def test_url_paths_unchanged(self): """Test that URL paths haven't changed.""" # Test that old URLs still work (may require auth now, but URLs are same) old_urls = [ "/api/v1/info", "/api/v1/extract", "/api/v1/jobs/test-job-id", "/api/v1/jobs/test-job-id/download", ] for url in old_urls: response = self.client.get(url) # Should not return 404 Not Found assert response.status_code != 404 def test_supported_formats_unchanged(self): """Test that supported video and audio formats haven't changed.""" response = self.client.get("/api/v1/info") assert response.status_code == 200 info = response.json() # Verify minimum expected formats are still supported expected_video_formats = [".mp4", ".avi", ".mov"] expected_audio_formats = ["mp3", "aac", "wav"] for fmt in expected_video_formats: assert fmt in info["supported_video_formats"] for fmt in expected_audio_formats: assert fmt in info["supported_audio_formats"] def test_quality_levels_unchanged(self): """Test that quality levels haven't changed.""" response = self.client.get("/api/v1/info") assert response.status_code == 200 info = response.json() expected_quality_levels = ["high", "medium", "low"] for quality in expected_quality_levels: assert quality in info["quality_levels"] class TestMigrationPath: """Test migration path for existing clients.""" def setup_method(self): """Set up test fixtures.""" self.client = TestClient(app) def test_clear_authentication_error_messages(self): """Test that authentication errors provide clear migration guidance.""" response = self.client.get("/api/v1/jobs/test-job") assert response.status_code == 401 error_msg = response.json()["error"] # Error message should be clear about what's needed assert "authorization" in error_msg.lower() or "bearer" in error_msg.lower() # Should include WWW-Authenticate header for proper HTTP compliance assert "WWW-Authenticate" in response.headers assert response.headers["WWW-Authenticate"] == "Bearer" def test_gradual_migration_support(self): """Test that clients can gradually migrate to new features.""" # New optional parameters should not break old clients test_file_content = b"fake video content" # Old-style request without new optional parameters response = self.client.post( "/api/v1/extract", files={"video": ("test.mp4", test_file_content, "video/mp4")}, data={ "output_format": "mp3", "quality": "medium" # No job_id parameter (new optional field) } ) # Should fail on auth, not on missing optional parameters assert response.status_code == 401 assert "job_id" not in response.json()["error"].lower() def test_response_structure_backwards_compatibility(self): """Test that response structures are backwards compatible.""" # When authentication is added, existing response fields should remain # This is important for client parsing logic response = self.client.get("/api/v1/info") info = response.json() # Core info structure should be preserved required_fields = ["version", "supported_video_formats", "supported_audio_formats"] for field in required_fields: assert field in info, f"Required field '{field}' missing from info response" def test_error_codes_are_new_additions(self): """Test that error codes are new additions, not changes to existing behavior.""" response = self.client.get("/api/v1/jobs/nonexistent") assert response.status_code == 401 # Auth error comes first # When error codes are present, they should be additions response_data = response.json() if "code" in response_data: # Code should be descriptive and not break existing error parsing assert isinstance(response_data["code"], str) assert len(response_data["code"]) > 0 class TestDocumentationCompatibility: """Test that API documentation remains compatible.""" def setup_method(self): """Set up test fixtures.""" self.client = TestClient(app) def test_openapi_schema_accessible(self): """Test that OpenAPI schema is still accessible.""" response = self.client.get("/openapi.json") assert response.status_code == 200 schema = response.json() # Should have standard OpenAPI structure assert "openapi" in schema assert "paths" in schema assert "info" in schema def test_api_endpoints_documented(self): """Test that API endpoints are documented.""" response = self.client.get("/openapi.json") schema = response.json() expected_paths = [ "/api/v1/info", "/api/v1/extract", "/api/v1/jobs/{job_id}", "/api/v1/jobs/{job_id}/download" ] for path in expected_paths: assert path in schema["paths"], f"Path {path} not documented in OpenAPI schema" def test_security_requirements_documented(self): """Test that security requirements are properly documented.""" response = self.client.get("/openapi.json") schema = response.json() # Check if security schemes are defined if "components" in schema and "securitySchemes" in schema["components"]: # Bearer token auth should be documented security_schemes = schema["components"]["securitySchemes"] # Look for Bearer token scheme bearer_scheme_found = False for scheme_name, scheme_def in security_schemes.items(): if scheme_def.get("type") == "http" and scheme_def.get("scheme") == "bearer": bearer_scheme_found = True break # If security is implemented, it should be documented # (This test may need adjustment based on actual OpenAPI generation) class TestPerformanceCompatibility: """Test that performance characteristics haven't degraded significantly.""" def setup_method(self): """Set up test fixtures.""" self.client = TestClient(app) def test_public_endpoint_performance_unchanged(self): """Test that public endpoints maintain good performance.""" import time # Test info endpoint performance start_time = time.time() response = self.client.get("/api/v1/info") end_time = time.time() assert response.status_code == 200 # Should respond quickly (less than 1 second for info endpoint) response_time = end_time - start_time assert response_time < 1.0, f"Info endpoint took {response_time:.2f}s, expected < 1.0s" def test_auth_validation_performance(self): """Test that authentication validation doesn't add significant overhead.""" import time # Test authentication error response time start_time = time.time() response = self.client.get("/api/v1/jobs/test-job") end_time = time.time() assert response.status_code == 401 # Auth validation should be fast (less than 0.5 seconds) response_time = end_time - start_time assert response_time < 0.5, f"Auth validation took {response_time:.2f}s, expected < 0.5s" def test_multiple_request_performance(self): """Test that authentication doesn't degrade under multiple requests.""" import time start_time = time.time() # Make multiple requests for i in range(10): response = self.client.get(f"/api/v1/jobs/test-job-{i}") assert response.status_code == 401 end_time = time.time() # Average should be reasonable total_time = end_time - start_time avg_time = total_time / 10 assert avg_time < 0.1, f"Average request time {avg_time:.3f}s, expected < 0.1s"