5fd370c093
- conftest.py: set required env vars before app import to prevent Settings() crash
- gcs.py: lazy bucket init checks _bucket instead of _client; add @bucket.setter
- vtt.py: fix float precision in _format_timestamp; include empty-text cues in parser
- security.py: guard verify_password against empty hash (passlib UnknownHashError)
- tts.py: _parse_timestamp raises ValueError("Invalid timestamp format: …")
- emailer.py: HTML-escape job_title in _render_completion_template (XSS fix)
- test_emailer.py: rewrite for Mailgun-based service (replaced SendGrid)
- test_gcs.py: fix UploadFile constructor, MIME type, remove executor.submit mock
- test_gemini.py: patch module-level client instead of non-existent genai.upload_file;
translate_vtt tests use numbered-list mock responses matching new implementation
- test_tts.py: fix aiohttp async CM mock pattern; fix error message match
- test_models.py: update JobCreate to use source_is_english instead of language
- test_security.py: set jwt_access_ttl_min in token test
- test_cross_tenant_isolation.py: add patch to imports
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
352 lines
No EOL
11 KiB
Python
352 lines
No EOL
11 KiB
Python
from datetime import datetime
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from app.models.job import (
|
|
Job,
|
|
JobCreate,
|
|
JobStatus,
|
|
JobUpdate,
|
|
LangOutput,
|
|
RequestedOutputs,
|
|
Source,
|
|
)
|
|
from app.models.user import User, UserCreate, UserRole
|
|
|
|
|
|
class TestJobModel:
|
|
"""Test Job model validation and behavior"""
|
|
|
|
def test_job_creation_minimal(self):
|
|
"""Test creating job with minimal required fields"""
|
|
job_data = {
|
|
"client_id": "user123",
|
|
"title": "Test Video",
|
|
"source": {
|
|
"filename": "test.mp4",
|
|
"gcs_uri": "gs://bucket/test.mp4"
|
|
},
|
|
"requested_outputs": {
|
|
"captions_vtt": True,
|
|
"audio_description_vtt": True,
|
|
"audio_description_mp3": True
|
|
}
|
|
}
|
|
|
|
job = Job(**job_data)
|
|
|
|
assert job.client_id == "user123"
|
|
assert job.title == "Test Video"
|
|
assert job.status == JobStatus.CREATED
|
|
assert job.source.language == "en" # Default value
|
|
assert job.requested_outputs.captions_vtt is True
|
|
|
|
def test_job_creation_full(self):
|
|
"""Test creating job with all fields"""
|
|
job_data = {
|
|
"_id": "job123",
|
|
"client_id": "user123",
|
|
"title": "Test Video",
|
|
"source": {
|
|
"filename": "test.mp4",
|
|
"original_filename": "original_test.mp4",
|
|
"gcs_uri": "gs://bucket/test.mp4",
|
|
"duration_s": 120.5,
|
|
"language": "en"
|
|
},
|
|
"requested_outputs": {
|
|
"captions_vtt": True,
|
|
"audio_description_vtt": True,
|
|
"audio_description_mp3": True,
|
|
"languages": ["es", "fr"],
|
|
"transcreation": ["es"]
|
|
},
|
|
"status": "pending_qc",
|
|
"outputs": {
|
|
"en": {
|
|
"captions_vtt_gcs": "gs://bucket/en/captions.vtt",
|
|
"ad_vtt_gcs": "gs://bucket/en/ad.vtt"
|
|
}
|
|
},
|
|
"ai": {
|
|
"confidence": 0.95,
|
|
"ingestion_json": {"test": "data"}
|
|
},
|
|
"created_at": datetime.utcnow(),
|
|
"updated_at": datetime.utcnow()
|
|
}
|
|
|
|
job = Job(**job_data)
|
|
|
|
assert job.id == "job123"
|
|
assert job.source.duration_s == 120.5
|
|
assert job.requested_outputs.languages == ["es", "fr"]
|
|
assert job.requested_outputs.transcreation == ["es"]
|
|
assert job.ai.confidence == 0.95
|
|
|
|
def test_job_status_enum(self):
|
|
"""Test job status enum values"""
|
|
assert JobStatus.CREATED == "created"
|
|
assert JobStatus.PENDING_QC == "pending_qc"
|
|
assert JobStatus.APPROVED_ENGLISH == "approved_english"
|
|
assert JobStatus.COMPLETED == "completed"
|
|
|
|
def test_source_validation(self):
|
|
"""Test source model validation"""
|
|
# Valid source
|
|
source = Source(
|
|
filename="test.mp4",
|
|
gcs_uri="gs://bucket/test.mp4"
|
|
)
|
|
assert source.language == "en" # Default
|
|
|
|
# Invalid language (too short)
|
|
with pytest.raises(ValidationError):
|
|
Source(
|
|
filename="test.mp4",
|
|
gcs_uri="gs://bucket/test.mp4",
|
|
language="x" # Too short
|
|
)
|
|
|
|
def test_requested_outputs_defaults(self):
|
|
"""Test RequestedOutputs default values"""
|
|
outputs = RequestedOutputs()
|
|
|
|
assert outputs.captions_vtt is True
|
|
assert outputs.audio_description_vtt is True
|
|
assert outputs.audio_description_mp3 is True
|
|
assert outputs.languages == []
|
|
assert outputs.transcreation == []
|
|
|
|
def test_lang_output_model(self):
|
|
"""Test language output model"""
|
|
lang_output = LangOutput(
|
|
captions_vtt_gcs="gs://bucket/es/captions.vtt",
|
|
ad_vtt_gcs="gs://bucket/es/ad.vtt",
|
|
ad_mp3_gcs="gs://bucket/es/ad.mp3",
|
|
origin="translate",
|
|
qa_notes="Looks good"
|
|
)
|
|
|
|
assert lang_output.origin == "translate"
|
|
assert lang_output.qa_notes == "Looks good"
|
|
|
|
def test_lang_output_invalid_origin(self):
|
|
"""Test LangOutput with invalid origin value"""
|
|
with pytest.raises(ValidationError):
|
|
LangOutput(
|
|
captions_vtt_gcs="gs://bucket/es/captions.vtt",
|
|
origin="invalid_origin" # Not in allowed values
|
|
)
|
|
|
|
def test_job_create_schema(self):
|
|
"""Test JobCreate schema"""
|
|
job_create = JobCreate(
|
|
title="Test Video",
|
|
source_is_english=True,
|
|
requested_outputs=RequestedOutputs(
|
|
captions_vtt=True,
|
|
languages=["es", "fr"]
|
|
)
|
|
)
|
|
|
|
assert job_create.title == "Test Video"
|
|
assert job_create.source_is_english is True
|
|
assert job_create.requested_outputs.languages == ["es", "fr"]
|
|
|
|
def test_job_update_schema(self):
|
|
"""Test JobUpdate schema"""
|
|
job_update = JobUpdate(
|
|
title="Updated Title",
|
|
status=JobStatus.PENDING_QC
|
|
)
|
|
|
|
assert job_update.title == "Updated Title"
|
|
assert job_update.status == JobStatus.PENDING_QC
|
|
assert job_update.review is None # Optional field
|
|
|
|
|
|
class TestUserModel:
|
|
"""Test User model validation and behavior"""
|
|
|
|
def test_user_creation_minimal(self):
|
|
"""Test creating user with minimal required fields"""
|
|
user_data = {
|
|
"email": "test@example.com",
|
|
"hashed_password": "hashed_password_here",
|
|
"full_name": "Test User"
|
|
}
|
|
|
|
user = User(**user_data)
|
|
|
|
assert user.email == "test@example.com"
|
|
assert user.role == UserRole.CLIENT # Default
|
|
assert user.is_active is True # Default
|
|
|
|
def test_user_creation_full(self):
|
|
"""Test creating user with all fields"""
|
|
user_data = {
|
|
"_id": "user123",
|
|
"email": "admin@example.com",
|
|
"hashed_password": "hashed_password_here",
|
|
"full_name": "Admin User",
|
|
"role": "admin",
|
|
"is_active": True,
|
|
"created_at": datetime.utcnow(),
|
|
"updated_at": datetime.utcnow()
|
|
}
|
|
|
|
user = User(**user_data)
|
|
|
|
assert user.id == "user123"
|
|
assert user.role == UserRole.ADMIN
|
|
assert user.is_active is True
|
|
|
|
def test_user_invalid_email(self):
|
|
"""Test user creation with invalid email"""
|
|
with pytest.raises(ValidationError):
|
|
User(
|
|
email="invalid_email", # Not a valid email format
|
|
hashed_password="hashed_password",
|
|
full_name="Test User"
|
|
)
|
|
|
|
def test_user_role_enum(self):
|
|
"""Test user role enum values"""
|
|
assert UserRole.CLIENT == "client"
|
|
assert UserRole.REVIEWER == "reviewer"
|
|
assert UserRole.ADMIN == "admin"
|
|
|
|
def test_user_create_schema(self):
|
|
"""Test UserCreate schema"""
|
|
user_create = UserCreate(
|
|
email="newuser@example.com",
|
|
password="plain_password",
|
|
full_name="New User",
|
|
role=UserRole.REVIEWER
|
|
)
|
|
|
|
assert user_create.email == "newuser@example.com"
|
|
assert user_create.password == "plain_password"
|
|
assert user_create.role == UserRole.REVIEWER
|
|
|
|
def test_user_create_default_role(self):
|
|
"""Test UserCreate with default role"""
|
|
user_create = UserCreate(
|
|
email="newuser@example.com",
|
|
password="plain_password",
|
|
full_name="New User"
|
|
)
|
|
|
|
assert user_create.role == UserRole.CLIENT
|
|
|
|
|
|
class TestJobStatusTransitions:
|
|
"""Test job status state machine transitions"""
|
|
|
|
def test_valid_status_transitions(self):
|
|
"""Test that job can be created with valid statuses"""
|
|
valid_statuses = [
|
|
JobStatus.CREATED,
|
|
JobStatus.INGESTING,
|
|
JobStatus.AI_PROCESSING,
|
|
JobStatus.PENDING_QC,
|
|
JobStatus.APPROVED_ENGLISH,
|
|
JobStatus.REJECTED,
|
|
JobStatus.TRANSLATING,
|
|
JobStatus.TTS_GENERATING,
|
|
JobStatus.PENDING_FINAL_REVIEW,
|
|
JobStatus.COMPLETED
|
|
]
|
|
|
|
job_base = {
|
|
"client_id": "user123",
|
|
"title": "Test Video",
|
|
"source": {
|
|
"filename": "test.mp4",
|
|
"gcs_uri": "gs://bucket/test.mp4"
|
|
},
|
|
"requested_outputs": {
|
|
"captions_vtt": True
|
|
}
|
|
}
|
|
|
|
for status in valid_statuses:
|
|
job = Job(**{**job_base, "status": status})
|
|
assert job.status == status
|
|
|
|
def test_job_status_string_values(self):
|
|
"""Test that job status enum has correct string values"""
|
|
# This ensures the state machine values match the plan
|
|
expected_statuses = [
|
|
"created", "ingesting", "ai_processing", "pending_qc",
|
|
"approved_english", "rejected", "translating",
|
|
"tts_generating", "pending_final_review", "completed"
|
|
]
|
|
|
|
actual_statuses = [status.value for status in JobStatus]
|
|
|
|
for expected in expected_statuses:
|
|
assert expected in actual_statuses
|
|
|
|
|
|
class TestModelFieldValidation:
|
|
"""Test field-level validation for models"""
|
|
|
|
def test_source_language_constraint(self):
|
|
"""Test source language field constraints"""
|
|
# Valid languages
|
|
valid_languages = ["en", "es", "fr", "de", "pt-BR", "zh-CN"]
|
|
|
|
for lang in valid_languages:
|
|
source = Source(
|
|
filename="test.mp4",
|
|
gcs_uri="gs://bucket/test.mp4",
|
|
language=lang
|
|
)
|
|
assert source.language == lang
|
|
|
|
def test_source_language_too_short(self):
|
|
"""Test source language validation for too short values"""
|
|
with pytest.raises(ValidationError):
|
|
Source(
|
|
filename="test.mp4",
|
|
gcs_uri="gs://bucket/test.mp4",
|
|
language="x" # Too short
|
|
)
|
|
|
|
def test_source_language_too_long(self):
|
|
"""Test source language validation for too long values"""
|
|
with pytest.raises(ValidationError):
|
|
Source(
|
|
filename="test.mp4",
|
|
gcs_uri="gs://bucket/test.mp4",
|
|
language="this_is_too_long" # Too long
|
|
)
|
|
|
|
def test_job_title_required(self):
|
|
"""Test that job title is required"""
|
|
with pytest.raises(ValidationError):
|
|
Job(
|
|
client_id="user123",
|
|
# title missing
|
|
source={
|
|
"filename": "test.mp4",
|
|
"gcs_uri": "gs://bucket/test.mp4"
|
|
},
|
|
requested_outputs={}
|
|
)
|
|
|
|
def test_client_id_required(self):
|
|
"""Test that client_id is required"""
|
|
with pytest.raises(ValidationError):
|
|
Job(
|
|
# client_id missing
|
|
title="Test Video",
|
|
source={
|
|
"filename": "test.mp4",
|
|
"gcs_uri": "gs://bucket/test.mp4"
|
|
},
|
|
requested_outputs={}
|
|
) |