Spaces:
Sleeping
Sleeping
| """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" |