Spaces:
Sleeping
Sleeping
| """Unit tests for Job entity and repository with external job ID support.""" | |
| import pytest | |
| import asyncio | |
| from datetime import datetime, timedelta | |
| from domain.entities.job import Job | |
| from domain.exceptions.domain_exceptions import DuplicateExternalJobIdError, ValidationError | |
| from domain.services.validation_service import ValidationService | |
| from infrastructure.repositories.job_repository import InMemoryJobRepository, JobRecord | |
| class TestJobEntityWithExternalId: | |
| """Test Job entity with external job ID and bearer token support.""" | |
| def test_job_creation_with_external_id(self): | |
| """Test creating a job with external job ID.""" | |
| job = Job.create_new( | |
| video_filename="test.mp4", | |
| file_size_bytes=1000000, | |
| output_format="mp3", | |
| quality="medium", | |
| external_job_id="ext-job-123", | |
| bearer_token="bearer-token-xyz" | |
| ) | |
| assert job.external_job_id == "ext-job-123" | |
| assert job.bearer_token == "bearer-token-xyz" | |
| assert job.has_external_job_id is True | |
| assert job.id is not None # Internal ID still generated | |
| def test_job_creation_without_external_id(self): | |
| """Test creating a job without external job ID (backwards compatibility).""" | |
| job = Job.create_new( | |
| video_filename="test.mp4", | |
| file_size_bytes=1000000, | |
| output_format="mp3", | |
| quality="medium" | |
| ) | |
| assert job.external_job_id is None | |
| assert job.bearer_token is None | |
| assert job.has_external_job_id is False | |
| assert job.id is not None # Internal ID still generated | |
| def test_job_creation_with_empty_external_id(self): | |
| """Test creating a job with empty external job ID.""" | |
| job = Job.create_new( | |
| video_filename="test.mp4", | |
| file_size_bytes=1000000, | |
| output_format="mp3", | |
| quality="medium", | |
| external_job_id="", | |
| bearer_token=None | |
| ) | |
| assert job.external_job_id == "" | |
| assert job.has_external_job_id is False | |
| def test_clear_bearer_token(self): | |
| """Test clearing bearer token for security.""" | |
| job = Job.create_new( | |
| video_filename="test.mp4", | |
| file_size_bytes=1000000, | |
| output_format="mp3", | |
| quality="medium", | |
| bearer_token="secret-token" | |
| ) | |
| assert job.bearer_token == "secret-token" | |
| original_updated_at = job.updated_at | |
| job.clear_bearer_token() | |
| assert job.bearer_token is None | |
| assert job.updated_at > original_updated_at | |
| def test_has_external_job_id_property(self): | |
| """Test the has_external_job_id property.""" | |
| # With valid external ID | |
| job1 = Job.create_new("test.mp4", 1000, "mp3", "medium", external_job_id="ext-123") | |
| assert job1.has_external_job_id is True | |
| # With None external ID | |
| job2 = Job.create_new("test.mp4", 1000, "mp3", "medium", external_job_id=None) | |
| assert job2.has_external_job_id is False | |
| # With empty string external ID | |
| job3 = Job.create_new("test.mp4", 1000, "mp3", "medium", external_job_id="") | |
| assert job3.has_external_job_id is False | |
| class TestValidationServiceExternalJobId: | |
| """Test ValidationService external job ID validation.""" | |
| def setup_method(self): | |
| """Set up test fixtures.""" | |
| self.validation_service = ValidationService( | |
| max_file_size_mb=100.0, | |
| supported_video_formats=['.mp4', '.avi'], | |
| supported_audio_formats=['mp3', 'aac'] | |
| ) | |
| def test_validate_external_job_id_valid(self): | |
| """Test validation of valid external job IDs.""" | |
| # Valid alphanumeric | |
| self.validation_service.validate_external_job_id("job123") | |
| # Valid with underscores and hyphens | |
| self.validation_service.validate_external_job_id("job_123-abc") | |
| # Valid single character | |
| self.validation_service.validate_external_job_id("a") | |
| # Valid 50 characters (max length) | |
| long_id = "a" * 50 | |
| self.validation_service.validate_external_job_id(long_id) | |
| # Valid None (optional field) | |
| self.validation_service.validate_external_job_id(None) | |
| # Valid empty string (optional field) | |
| self.validation_service.validate_external_job_id("") | |
| def test_validate_external_job_id_invalid(self): | |
| """Test validation of invalid external job IDs.""" | |
| # Too long (51 characters) | |
| with pytest.raises(ValidationError, match="must be 50 characters or less"): | |
| long_id = "a" * 51 | |
| self.validation_service.validate_external_job_id(long_id) | |
| # Contains invalid characters | |
| with pytest.raises(ValidationError, match="must contain only alphanumeric"): | |
| self.validation_service.validate_external_job_id("job@123") | |
| with pytest.raises(ValidationError, match="must contain only alphanumeric"): | |
| self.validation_service.validate_external_job_id("job 123") # space | |
| with pytest.raises(ValidationError, match="must contain only alphanumeric"): | |
| self.validation_service.validate_external_job_id("job.123") # dot | |
| with pytest.raises(ValidationError, match="must contain only alphanumeric"): | |
| self.validation_service.validate_external_job_id("job/123") # slash | |
| class TestJobRepositoryWithExternalId: | |
| """Test InMemoryJobRepository with external job ID support.""" | |
| def setup_method(self): | |
| """Set up test fixtures.""" | |
| self.repo = InMemoryJobRepository() | |
| async def test_create_job_with_external_id(self): | |
| """Test creating a job with external job ID.""" | |
| job = await self.repo.create( | |
| job_id="internal-123", | |
| filename="test.mp4", | |
| file_size_mb=10.0, | |
| output_format="mp3", | |
| quality="medium", | |
| external_job_id="ext-job-456", | |
| bearer_token="bearer-xyz" | |
| ) | |
| assert job.id == "internal-123" | |
| assert job.external_job_id == "ext-job-456" | |
| assert job.bearer_token == "bearer-xyz" | |
| assert job.filename == "test.mp4" | |
| async def test_create_job_without_external_id(self): | |
| """Test creating a job without external job ID (backwards compatibility).""" | |
| job = await self.repo.create( | |
| job_id="internal-123", | |
| filename="test.mp4", | |
| file_size_mb=10.0, | |
| output_format="mp3", | |
| quality="medium" | |
| ) | |
| assert job.id == "internal-123" | |
| assert job.external_job_id is None | |
| assert job.bearer_token is None | |
| async def test_duplicate_external_job_id_raises_error(self): | |
| """Test that duplicate external job IDs raise an error.""" | |
| # Create first job | |
| await self.repo.create( | |
| job_id="internal-1", | |
| filename="test1.mp4", | |
| file_size_mb=10.0, | |
| output_format="mp3", | |
| quality="medium", | |
| external_job_id="duplicate-id" | |
| ) | |
| # Try to create second job with same external ID | |
| with pytest.raises(DuplicateExternalJobIdError) as exc_info: | |
| await self.repo.create( | |
| job_id="internal-2", | |
| filename="test2.mp4", | |
| file_size_mb=20.0, | |
| output_format="aac", | |
| quality="high", | |
| external_job_id="duplicate-id" | |
| ) | |
| assert exc_info.value.external_job_id == "duplicate-id" | |
| assert "already exists" in str(exc_info.value) | |
| async def test_get_by_external_id_success(self): | |
| """Test retrieving a job by external job ID.""" | |
| # Create job with external ID | |
| await self.repo.create( | |
| job_id="internal-123", | |
| filename="test.mp4", | |
| file_size_mb=10.0, | |
| output_format="mp3", | |
| quality="medium", | |
| external_job_id="ext-job-456" | |
| ) | |
| # Retrieve by external ID | |
| job = await self.repo.get_by_external_id("ext-job-456") | |
| assert job is not None | |
| assert job.id == "internal-123" | |
| assert job.external_job_id == "ext-job-456" | |
| async def test_get_by_external_id_not_found(self): | |
| """Test retrieving a job by non-existent external job ID.""" | |
| job = await self.repo.get_by_external_id("non-existent-id") | |
| assert job is None | |
| async def test_get_by_internal_id_still_works(self): | |
| """Test that retrieving by internal ID still works.""" | |
| # Create job with external ID | |
| await self.repo.create( | |
| job_id="internal-123", | |
| filename="test.mp4", | |
| file_size_mb=10.0, | |
| output_format="mp3", | |
| quality="medium", | |
| external_job_id="ext-job-456" | |
| ) | |
| # Retrieve by internal ID | |
| job = await self.repo.get("internal-123") | |
| assert job is not None | |
| assert job.id == "internal-123" | |
| assert job.external_job_id == "ext-job-456" | |
| async def test_clear_bearer_token(self): | |
| """Test clearing bearer token from repository.""" | |
| # Create job with bearer token | |
| await self.repo.create( | |
| job_id="internal-123", | |
| filename="test.mp4", | |
| file_size_mb=10.0, | |
| output_format="mp3", | |
| quality="medium", | |
| bearer_token="secret-token" | |
| ) | |
| # Verify token exists | |
| job = await self.repo.get("internal-123") | |
| assert job.bearer_token == "secret-token" | |
| # Clear token | |
| result = await self.repo.clear_bearer_token("internal-123") | |
| assert result is True | |
| # Verify token cleared | |
| job = await self.repo.get("internal-123") | |
| assert job.bearer_token is None | |
| async def test_clear_bearer_token_nonexistent_job(self): | |
| """Test clearing bearer token for non-existent job.""" | |
| result = await self.repo.clear_bearer_token("non-existent-id") | |
| assert result is False | |
| async def test_delete_job_with_external_id(self): | |
| """Test deleting a job removes it from external ID index.""" | |
| # Create job with external ID | |
| await self.repo.create( | |
| job_id="internal-123", | |
| filename="test.mp4", | |
| file_size_mb=10.0, | |
| output_format="mp3", | |
| quality="medium", | |
| external_job_id="ext-job-456" | |
| ) | |
| # Verify job exists | |
| job = await self.repo.get_by_external_id("ext-job-456") | |
| assert job is not None | |
| # Delete job | |
| result = await self.repo.delete("internal-123") | |
| assert result is True | |
| # Verify job no longer accessible by external ID | |
| job = await self.repo.get_by_external_id("ext-job-456") | |
| assert job is None | |
| # Verify job no longer accessible by internal ID | |
| job = await self.repo.get("internal-123") | |
| assert job is None | |
| async def test_multiple_jobs_with_external_ids(self): | |
| """Test managing multiple jobs with different external IDs.""" | |
| # Create multiple jobs | |
| await self.repo.create("internal-1", "test1.mp4", 10.0, "mp3", "medium", "ext-1") | |
| await self.repo.create("internal-2", "test2.mp4", 20.0, "aac", "high", "ext-2") | |
| await self.repo.create("internal-3", "test3.mp4", 30.0, "wav", "low") # No external ID | |
| # Verify all can be retrieved correctly | |
| job1 = await self.repo.get_by_external_id("ext-1") | |
| assert job1.id == "internal-1" | |
| job2 = await self.repo.get_by_external_id("ext-2") | |
| assert job2.id == "internal-2" | |
| job3 = await self.repo.get("internal-3") | |
| assert job3.external_job_id is None | |
| # Verify external ID lookup for job without external ID returns None | |
| job3_by_ext = await self.repo.get_by_external_id("internal-3") | |
| assert job3_by_ext is None |