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>
259 lines
No EOL
9.4 KiB
Python
259 lines
No EOL
9.4 KiB
Python
import asyncio
|
|
from datetime import datetime, timedelta
|
|
from io import BytesIO
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from fastapi import HTTPException, UploadFile
|
|
from google.cloud.exceptions import NotFound
|
|
|
|
from app.services.gcs import (
|
|
GCSService,
|
|
upload_file_to_gcs,
|
|
upload_vtt_to_gcs,
|
|
get_signed_download_url,
|
|
generate_signed_upload_url
|
|
)
|
|
|
|
|
|
class TestGCSService:
|
|
"""Test Google Cloud Storage service functionality"""
|
|
|
|
@pytest.fixture
|
|
def mock_storage_client(self):
|
|
"""Mock Google Cloud Storage client"""
|
|
with patch('app.services.gcs.storage.Client') as mock_client:
|
|
mock_bucket = MagicMock()
|
|
mock_client.return_value.bucket.return_value = mock_bucket
|
|
yield mock_client, mock_bucket
|
|
|
|
@pytest.fixture
|
|
def gcs_service(self, mock_storage_client):
|
|
"""Create GCS service instance with mocked client"""
|
|
mock_client, mock_bucket = mock_storage_client
|
|
service = GCSService()
|
|
service.bucket = mock_bucket
|
|
return service
|
|
|
|
@pytest.fixture
|
|
def sample_upload_file(self):
|
|
"""Create a sample UploadFile for testing"""
|
|
file_content = b"test video content"
|
|
file_obj = BytesIO(file_content)
|
|
from starlette.datastructures import Headers
|
|
upload_file = UploadFile(
|
|
file=file_obj,
|
|
filename="test.mp4",
|
|
headers=Headers({"content-type": "video/mp4"}),
|
|
)
|
|
return upload_file
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_file_to_gcs_success(self, gcs_service, sample_upload_file):
|
|
"""Test successful file upload to GCS"""
|
|
mock_blob = MagicMock()
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
result = await gcs_service.upload_file_to_gcs(
|
|
sample_upload_file,
|
|
"test/path.mp4"
|
|
)
|
|
|
|
assert result == "gs://test-bucket/test/path.mp4"
|
|
assert mock_blob.content_type == "video/mp4"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_file_with_custom_content_type(self, gcs_service, sample_upload_file):
|
|
"""Test file upload with custom content type"""
|
|
mock_blob = MagicMock()
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
await gcs_service.upload_file_to_gcs(
|
|
sample_upload_file,
|
|
"test/path.mp4",
|
|
content_type="application/octet-stream"
|
|
)
|
|
|
|
assert mock_blob.content_type == "application/octet-stream"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_file_failure(self, gcs_service, sample_upload_file):
|
|
"""Test file upload failure handling"""
|
|
mock_blob = MagicMock()
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
with patch.object(gcs_service.executor, 'submit') as mock_submit:
|
|
future = asyncio.Future()
|
|
future.set_exception(Exception("Upload failed"))
|
|
mock_submit.return_value = future
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await gcs_service.upload_file_to_gcs(sample_upload_file, "test/path.mp4")
|
|
|
|
assert exc_info.value.status_code == 500
|
|
assert "File upload failed" in str(exc_info.value.detail)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_text_to_gcs_success(self, gcs_service):
|
|
"""Test successful text upload to GCS"""
|
|
mock_blob = MagicMock()
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
result = await gcs_service.upload_text_to_gcs(
|
|
"test content",
|
|
"test/file.txt",
|
|
"text/plain"
|
|
)
|
|
|
|
assert result == "gs://test-bucket/test/file.txt"
|
|
assert mock_blob.content_type == "text/plain"
|
|
mock_blob.upload_from_string.assert_called_once_with("test content", content_type="text/plain")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_signed_url_success(self, gcs_service):
|
|
"""Test successful signed URL generation"""
|
|
mock_blob = MagicMock()
|
|
mock_blob.exists.return_value = True
|
|
mock_blob.generate_signed_url.return_value = "https://signed-url.example.com"
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
with patch.object(gcs_service.executor, 'submit') as mock_submit:
|
|
future = asyncio.Future()
|
|
future.set_result("https://signed-url.example.com")
|
|
mock_submit.return_value = future
|
|
|
|
result = await gcs_service.get_signed_url("test/file.mp4")
|
|
|
|
assert result == "https://signed-url.example.com"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_signed_url_file_not_found(self, gcs_service):
|
|
"""Test signed URL generation for non-existent file"""
|
|
mock_blob = MagicMock()
|
|
mock_blob.exists.return_value = False
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
with patch.object(gcs_service.executor, 'submit') as mock_submit:
|
|
future = asyncio.Future()
|
|
future.set_exception(NotFound("File not found"))
|
|
mock_submit.return_value = future
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await gcs_service.get_signed_url("test/nonexistent.mp4")
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert "File not found" in str(exc_info.value.detail)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_file_success(self, gcs_service):
|
|
"""Test successful file deletion"""
|
|
mock_blob = MagicMock()
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
result = await gcs_service.delete_file("test/file.mp4")
|
|
|
|
assert result is True
|
|
mock_blob.delete.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_file_not_found(self, gcs_service):
|
|
"""Test deleting non-existent file"""
|
|
mock_blob = MagicMock()
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
with patch.object(gcs_service.executor, 'submit') as mock_submit:
|
|
future = asyncio.Future()
|
|
future.set_exception(NotFound("File not found"))
|
|
mock_submit.return_value = future
|
|
|
|
result = await gcs_service.delete_file("test/nonexistent.mp4")
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_exists_true(self, gcs_service):
|
|
"""Test checking if file exists (true case)"""
|
|
mock_blob = MagicMock()
|
|
mock_blob.exists.return_value = True
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
with patch.object(gcs_service.executor, 'submit') as mock_submit:
|
|
future = asyncio.Future()
|
|
future.set_result(True)
|
|
mock_submit.return_value = future
|
|
|
|
result = await gcs_service.file_exists("test/file.mp4")
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_exists_false(self, gcs_service):
|
|
"""Test checking if file exists (false case)"""
|
|
mock_blob = MagicMock()
|
|
mock_blob.exists.return_value = False
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
with patch.object(gcs_service.executor, 'submit') as mock_submit:
|
|
future = asyncio.Future()
|
|
future.set_result(False)
|
|
mock_submit.return_value = future
|
|
|
|
result = await gcs_service.file_exists("test/file.mp4")
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestGCSConvenienceFunctions:
|
|
"""Test convenience functions for GCS operations"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_vtt_to_gcs(self):
|
|
"""Test VTT upload convenience function"""
|
|
vtt_content = """WEBVTT
|
|
|
|
00:00:01.000 --> 00:00:03.000
|
|
Hello world
|
|
"""
|
|
|
|
with patch('app.services.gcs.gcs_service.upload_text_to_gcs') as mock_upload:
|
|
mock_upload.return_value = "gs://bucket/test.vtt"
|
|
|
|
result = await upload_vtt_to_gcs(vtt_content, "test.vtt")
|
|
|
|
assert result == "gs://bucket/test.vtt"
|
|
mock_upload.assert_called_once_with(vtt_content, "test.vtt", "text/vtt; charset=utf-8")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_signed_download_url(self):
|
|
"""Test signed download URL convenience function"""
|
|
with patch('app.services.gcs.gcs_service.get_signed_url') as mock_get_url:
|
|
mock_get_url.return_value = "https://signed-url.example.com"
|
|
|
|
result = await get_signed_download_url("test/file.mp4", 12)
|
|
|
|
assert result == "https://signed-url.example.com"
|
|
mock_get_url.assert_called_once_with("test/file.mp4", 12)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_signed_upload_url(self):
|
|
"""Test signed upload URL generation"""
|
|
import app.services.gcs as gcs_module
|
|
|
|
expected_result = {
|
|
"url": "https://upload-url.example.com",
|
|
"fields": {"Content-Type": "video/mp4"}
|
|
}
|
|
|
|
mock_blob = MagicMock()
|
|
mock_blob.generate_signed_post_policy_v4.return_value = (
|
|
expected_result["url"],
|
|
expected_result["fields"]
|
|
)
|
|
mock_bucket = MagicMock()
|
|
mock_bucket.blob.return_value = mock_blob
|
|
|
|
# Patch _bucket directly to avoid triggering lazy GCS client initialization
|
|
with patch.object(gcs_module.gcs_service, '_bucket', mock_bucket):
|
|
result = await generate_signed_upload_url("test/file.mp4", "video/mp4")
|
|
|
|
assert result == expected_result |