Spaces:
Sleeping
Sleeping
File size: 12,563 Bytes
dbe78dd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 | """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()
@pytest.mark.asyncio
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"
@pytest.mark.asyncio
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
@pytest.mark.asyncio
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)
@pytest.mark.asyncio
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"
@pytest.mark.asyncio
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
@pytest.mark.asyncio
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"
@pytest.mark.asyncio
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
@pytest.mark.asyncio
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
@pytest.mark.asyncio
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
@pytest.mark.asyncio
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 |