unveilAiDotOrg/audio-processor
4
1"""Unit tests for API endpoints with authentication and external job ID support."""2import pytest3import base644import json5from fastapi.testclient import TestClient6from unittest.mock import Mock, AsyncMock, patch7from fastapi import HTTPException8 9# Test the authentication logic separately to avoid complex setup10class TestAPIEndpointAuthentication:11 """Test authentication logic for API endpoints."""12 13 def setup_method(self):14 """Set up test fixtures."""15 # Create a valid JWT token for testing16 header = {"alg": "HS256", "typ": "JWT"}17 payload = {"sub": "1234567890", "name": "John Doe", "iat": 1516239022}18 19 header_b64 = base64.urlsafe_b64encode(20 json.dumps(header).encode()21 ).decode().rstrip('=')22 payload_b64 = base64.urlsafe_b64encode(23 json.dumps(payload).encode()24 ).decode().rstrip('=')25 26 self.valid_token = f"{header_b64}.{payload_b64}.test_signature"27 self.valid_auth_header = f"Bearer {self.valid_token}"28 29 def test_extract_endpoint_requires_authentication(self):30 """Test that extract endpoint requires authentication."""31 # Simulate the authentication logic that would happen in the endpoint32 def simulate_extract_auth_check(authorization_header):33 """Simulate the authentication check for extract endpoint."""34 if not authorization_header:35 raise HTTPException(status_code=401, detail="Missing Authorization header")36 37 if not authorization_header.startswith("Bearer "):38 raise HTTPException(status_code=401, detail="Invalid Authorization header format")39 40 return True41 42 # Test successful authentication43 result = simulate_extract_auth_check(self.valid_auth_header)44 assert result is True45 46 # Test missing authentication47 with pytest.raises(HTTPException) as exc_info:48 simulate_extract_auth_check(None)49 assert exc_info.value.status_code == 40150 assert "Missing Authorization header" in str(exc_info.value.detail)51 52 # Test invalid format53 with pytest.raises(HTTPException) as exc_info:54 simulate_extract_auth_check("Basic dXNlcjpwYXNz")55 assert exc_info.value.status_code == 40156 assert "Invalid Authorization header format" in str(exc_info.value.detail)57 58 def test_job_status_endpoint_requires_authentication(self):59 """Test that job status endpoint requires authentication."""60 def simulate_job_status_auth_check(authorization_header):61 """Simulate the authentication check for job status endpoint."""62 if not authorization_header:63 raise HTTPException(status_code=401, detail="Missing Authorization header")64 return True65 66 # Test successful authentication67 result = simulate_job_status_auth_check(self.valid_auth_header)68 assert result is True69 70 # Test missing authentication71 with pytest.raises(HTTPException) as exc_info:72 simulate_job_status_auth_check(None)73 assert exc_info.value.status_code == 40174 75 def test_download_endpoint_requires_authentication(self):76 """Test that download endpoint requires authentication."""77 def simulate_download_auth_check(authorization_header):78 """Simulate the authentication check for download endpoint."""79 if not authorization_header:80 raise HTTPException(status_code=401, detail="Missing Authorization header")81 return True82 83 # Test successful authentication84 result = simulate_download_auth_check(self.valid_auth_header)85 assert result is True86 87 # Test missing authentication88 with pytest.raises(HTTPException) as exc_info:89 simulate_download_auth_check(None)90 assert exc_info.value.status_code == 40191 92 def test_info_endpoint_public(self):93 """Test that info endpoint doesn't require authentication."""94 def simulate_info_endpoint():95 """Simulate the info endpoint (no auth required)."""96 return {97 "version": "1.0.0",98 "supported_video_formats": ['.mp4', '.avi'],99 "supported_audio_formats": ['mp3', 'aac'],100 "quality_levels": ['high', 'medium', 'low']101 }102 103 # Should work without any authentication104 result = simulate_info_endpoint()105 assert "version" in result106 assert result["version"] == "1.0.0"107 108 def test_health_endpoint_public(self):109 """Test that health endpoint doesn't require authentication."""110 def simulate_health_endpoint():111 """Simulate the health endpoint (no auth required)."""112 return {"status": "healthy", "service": "audio-extractor-api"}113 114 # Should work without any authentication115 result = simulate_health_endpoint()116 assert result["status"] == "healthy"117 assert result["service"] == "audio-extractor-api"118 119 120class TestExternalJobIdValidation:121 """Test external job ID validation in API endpoints."""122 123 def test_valid_external_job_ids(self):124 """Test validation of valid external job IDs."""125 def validate_external_job_id(job_id):126 """Simulate external job ID validation."""127 if job_id is None or job_id == "":128 return True # Optional field129 130 if len(job_id) > 50:131 raise HTTPException(status_code=400, detail="External job ID must be 50 characters or less")132 133 import re134 if not re.match(r'^[a-zA-Z0-9_-]+$', job_id):135 raise HTTPException(status_code=400, detail="External job ID must contain only alphanumeric characters, underscores, and hyphens")136 137 return True138 139 # Valid cases140 assert validate_external_job_id(None) is True141 assert validate_external_job_id("") is True142 assert validate_external_job_id("job123") is True143 assert validate_external_job_id("job_123-abc") is True144 assert validate_external_job_id("a" * 50) is True # Max length145 146 # Invalid cases147 with pytest.raises(HTTPException) as exc_info:148 validate_external_job_id("a" * 51) # Too long149 assert exc_info.value.status_code == 400150 assert "50 characters or less" in str(exc_info.value.detail)151 152 with pytest.raises(HTTPException) as exc_info:153 validate_external_job_id("job@123") # Invalid character154 assert exc_info.value.status_code == 400155 assert "alphanumeric characters" in str(exc_info.value.detail)156 157 with pytest.raises(HTTPException) as exc_info:158 validate_external_job_id("job 123") # Space159 assert exc_info.value.status_code == 400160 161 162class TestAPIResponseUpdates:163 """Test that API responses include external job IDs when provided."""164 165 def test_job_creation_response_includes_external_job_id(self):166 """Test that job creation response includes external job ID."""167 # Simulate JobCreationDTO168 from dataclasses import dataclass169 from typing import Optional170 171 @dataclass172 class MockJobCreationDTO:173 job_id: str174 external_job_id: Optional[str] = None175 status: str = "processing"176 message: str = "Job created"177 check_url: str = "/api/v1/jobs/123"178 file_size_mb: float = 10.0179 180 # With external job ID181 dto_with_external = MockJobCreationDTO(182 job_id="internal-123",183 external_job_id="ext-job-456"184 )185 186 assert dto_with_external.job_id == "internal-123"187 assert dto_with_external.external_job_id == "ext-job-456"188 189 # Without external job ID190 dto_without_external = MockJobCreationDTO(191 job_id="internal-789"192 )193 194 assert dto_without_external.job_id == "internal-789"195 assert dto_without_external.external_job_id is None196 197 def test_job_status_response_includes_external_job_id(self):198 """Test that job status response includes external job ID."""199 from dataclasses import dataclass200 from typing import Optional201 from datetime import datetime202 203 @dataclass204 class MockJobStatusDTO:205 job_id: str206 external_job_id: Optional[str] = None207 status: str = "processing"208 created_at: datetime = None209 updated_at: datetime = None210 211 # With external job ID212 now = datetime.utcnow()213 dto_with_external = MockJobStatusDTO(214 job_id="internal-123",215 external_job_id="ext-job-456",216 status="completed",217 created_at=now,218 updated_at=now219 )220 221 assert dto_with_external.job_id == "internal-123"222 assert dto_with_external.external_job_id == "ext-job-456"223 assert dto_with_external.status == "completed"224 225 def test_extract_endpoint_form_parameter_handling(self):226 """Test that extract endpoint handles job_id form parameter correctly."""227 def simulate_extract_form_handling(form_data):228 """Simulate form data handling for extract endpoint."""229 # Extract job_id from form data230 job_id = form_data.get("job_id")231 232 # Validate if provided233 if job_id and job_id != "":234 # Simulate validation235 if len(job_id) > 50:236 raise HTTPException(status_code=400, detail="External job ID too long")237 238 return {"extracted_job_id": job_id, "valid": True}239 else:240 return {"extracted_job_id": None, "valid": True}241 242 # Test with job_id provided243 form_with_job_id = {"video": "test.mp4", "job_id": "ext-job-123"}244 result = simulate_extract_form_handling(form_with_job_id)245 assert result["extracted_job_id"] == "ext-job-123"246 assert result["valid"] is True247 248 # Test without job_id249 form_without_job_id = {"video": "test.mp4"}250 result = simulate_extract_form_handling(form_without_job_id)251 assert result["extracted_job_id"] is None252 assert result["valid"] is True253 254 # Test with empty job_id255 form_with_empty_job_id = {"video": "test.mp4", "job_id": ""}256 result = simulate_extract_form_handling(form_with_empty_job_id)257 assert result["extracted_job_id"] is None # Empty string should be treated as None258 assert result["valid"] is True259 260 261class TestDuplicateExternalJobIdHandling:262 """Test handling of duplicate external job IDs in API endpoints."""263 264 def test_duplicate_external_job_id_error_handling(self):265 """Test that duplicate external job ID errors are handled properly."""266 from domain.exceptions.domain_exceptions import DuplicateExternalJobIdError267 268 def simulate_job_creation_with_duplicate_check(external_job_id):269 """Simulate job creation with duplicate external ID check."""270 # Simulate existing external job IDs271 existing_external_ids = ["existing-job-1", "existing-job-2"]272 273 if external_job_id in existing_external_ids:274 raise DuplicateExternalJobIdError(external_job_id)275 276 return {"job_id": "new-internal-id", "external_job_id": external_job_id}277 278 # Test successful creation with unique external ID279 result = simulate_job_creation_with_duplicate_check("unique-job-123")280 assert result["external_job_id"] == "unique-job-123"281 282 # Test duplicate external ID error283 with pytest.raises(DuplicateExternalJobIdError) as exc_info:284 simulate_job_creation_with_duplicate_check("existing-job-1")285 286 assert exc_info.value.external_job_id == "existing-job-1"287 assert "already exists" in str(exc_info.value)288 289 290class TestEndpointIntegrationLogic:291 """Test the integration logic of endpoints with authentication and external job IDs."""292 293 def test_extract_endpoint_complete_flow(self):294 """Test the complete flow of the extract endpoint."""295 def simulate_complete_extract_flow(auth_header, form_data):296 """Simulate the complete extract endpoint flow."""297 # 1. Authentication check298 if not auth_header or not auth_header.startswith("Bearer "):299 raise HTTPException(status_code=401, detail="Authentication required")300 301 # 2. Extract external job ID302 external_job_id = form_data.get("job_id")303 304 # 3. Validate external job ID format305 if external_job_id and len(external_job_id) > 50:306 raise HTTPException(status_code=400, detail="Invalid external job ID")307 308 # 4. Check for duplicates309 existing_ids = ["existing-1", "existing-2"]310 if external_job_id in existing_ids:311 raise HTTPException(status_code=400, detail="Duplicate external job ID")312 313 # 5. Create job314 internal_job_id = "internal-12345"315 316 # 6. Return response317 return {318 "job_id": internal_job_id,319 "external_job_id": external_job_id,320 "status": "processing",321 "message": "Job created successfully"322 }323 324 # Test successful flow325 auth_header = "Bearer valid.jwt.token"326 form_data = {"video": "test.mp4", "job_id": "my-job-123"}327 328 result = simulate_complete_extract_flow(auth_header, form_data)329 assert result["job_id"] == "internal-12345"330 assert result["external_job_id"] == "my-job-123"331 assert result["status"] == "processing"332 333 # Test without external job ID334 form_data_no_job_id = {"video": "test.mp4"}335 result = simulate_complete_extract_flow(auth_header, form_data_no_job_id)336 assert result["job_id"] == "internal-12345"337 assert result["external_job_id"] is None338 339 def test_job_status_endpoint_complete_flow(self):340 """Test the complete flow of the job status endpoint."""341 def simulate_complete_job_status_flow(auth_header, job_id):342 """Simulate the complete job status endpoint flow."""343 # 1. Authentication check344 if not auth_header or not auth_header.startswith("Bearer "):345 raise HTTPException(status_code=401, detail="Authentication required")346 347 # 2. Look up job348 mock_jobs = {349 "internal-123": {350 "job_id": "internal-123",351 "external_job_id": "ext-job-456",352 "status": "completed",353 "filename": "test.mp4"354 },355 "internal-789": {356 "job_id": "internal-789",357 "external_job_id": None,358 "status": "processing",359 "filename": "test2.mp4"360 }361 }362 363 if job_id not in mock_jobs:364 raise HTTPException(status_code=404, detail="Job not found")365 366 # 3. Return job status367 return mock_jobs[job_id]368 369 # Test successful lookup with external job ID370 auth_header = "Bearer valid.jwt.token"371 result = simulate_complete_job_status_flow(auth_header, "internal-123")372 assert result["job_id"] == "internal-123"373 assert result["external_job_id"] == "ext-job-456"374 assert result["status"] == "completed"375 376 # Test successful lookup without external job ID377 result = simulate_complete_job_status_flow(auth_header, "internal-789")378 assert result["job_id"] == "internal-789"379 assert result["external_job_id"] is None380 assert result["status"] == "processing"381 382 # Test job not found383 with pytest.raises(HTTPException) as exc_info:384 simulate_complete_job_status_flow(auth_header, "non-existent")385 assert exc_info.value.status_code == 404386 assert "Job not found" in str(exc_info.value.detail)