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>
182 lines
7.6 KiB
Python
182 lines
7.6 KiB
Python
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from app.services.emailer import EmailService
|
|
|
|
|
|
class TestEmailService:
|
|
"""Test email service (Mailgun-based)"""
|
|
|
|
@pytest.fixture
|
|
def service(self):
|
|
return EmailService()
|
|
|
|
@pytest.fixture
|
|
def sample_download_links(self):
|
|
return {
|
|
"en": {
|
|
"captions_vtt": "https://signed-url.example.com/en/captions.vtt",
|
|
"audio_description_vtt": "https://signed-url.example.com/en/ad.vtt",
|
|
"audio_description_mp3": "https://signed-url.example.com/en/ad.mp3",
|
|
},
|
|
"es": {
|
|
"captions_vtt": "https://signed-url.example.com/es/captions.vtt",
|
|
"audio_description_vtt": "https://signed-url.example.com/es/ad.vtt",
|
|
"audio_description_mp3": "https://signed-url.example.com/es/ad.mp3",
|
|
},
|
|
}
|
|
|
|
# ── _configured property ───────────────────────────────────────────────────
|
|
|
|
def test_configured_when_keys_present(self, service):
|
|
with patch("app.services.emailer.settings") as s:
|
|
s.mailgun_api_key = "key"
|
|
s.mailgun_domain = "mg.example.com"
|
|
assert service._configured is True
|
|
|
|
def test_not_configured_when_key_missing(self, service):
|
|
with patch("app.services.emailer.settings") as s:
|
|
s.mailgun_api_key = ""
|
|
s.mailgun_domain = "mg.example.com"
|
|
assert service._configured is False
|
|
|
|
def test_not_configured_when_domain_missing(self, service):
|
|
with patch("app.services.emailer.settings") as s:
|
|
s.mailgun_api_key = "key"
|
|
s.mailgun_domain = ""
|
|
assert service._configured is False
|
|
|
|
# ── _send ─────────────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_returns_false_when_not_configured(self, service):
|
|
with patch("app.services.emailer.settings") as s:
|
|
s.mailgun_api_key = ""
|
|
s.mailgun_domain = ""
|
|
result = await service._send("to@example.com", "Subject", "<p>HTML</p>")
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_success(self, service):
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=False)
|
|
mock_client.post = AsyncMock(return_value=mock_response)
|
|
|
|
with patch("app.services.emailer.settings") as s:
|
|
s.mailgun_api_key = "key"
|
|
s.mailgun_domain = "mg.example.com"
|
|
s.mailgun_from = "noreply@example.com"
|
|
with patch("httpx.AsyncClient", return_value=mock_client):
|
|
result = await service._send("to@example.com", "Subject", "<p>Hi</p>")
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_api_failure(self, service):
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 400
|
|
mock_response.text = "Bad request"
|
|
|
|
mock_client = AsyncMock()
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=False)
|
|
mock_client.post = AsyncMock(return_value=mock_response)
|
|
|
|
with patch("app.services.emailer.settings") as s:
|
|
s.mailgun_api_key = "key"
|
|
s.mailgun_domain = "mg.example.com"
|
|
s.mailgun_from = "noreply@example.com"
|
|
with patch("httpx.AsyncClient", return_value=mock_client):
|
|
result = await service._send("to@example.com", "Subject", "<p>Hi</p>")
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_exception_returns_false(self, service):
|
|
with patch("app.services.emailer.settings") as s:
|
|
s.mailgun_api_key = "key"
|
|
s.mailgun_domain = "mg.example.com"
|
|
s.mailgun_from = "noreply@example.com"
|
|
with patch("httpx.AsyncClient", side_effect=Exception("network error")):
|
|
result = await service._send("to@example.com", "Subject", "<p>Hi</p>")
|
|
|
|
assert result is False
|
|
|
|
# ── send_completion_email ─────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_completion_email_calls_send(self, service, sample_download_links):
|
|
with patch.object(service, "_send", new_callable=AsyncMock) as mock_send:
|
|
mock_send.return_value = True
|
|
result = await service.send_completion_email(
|
|
recipient_email="client@example.com",
|
|
job_title="Test Video",
|
|
download_links=sample_download_links,
|
|
)
|
|
assert result is True
|
|
mock_send.assert_called_once()
|
|
# Subject should contain job title
|
|
assert "Test Video" in mock_send.call_args[0][1]
|
|
|
|
# ── _render_completion_template ───────────────────────────────────────────
|
|
|
|
def test_render_template_basic(self, service, sample_download_links):
|
|
html = service._render_completion_template(
|
|
job_title="Test Video Project",
|
|
download_links=sample_download_links,
|
|
)
|
|
assert "Test Video Project" in html
|
|
assert "EN Assets" in html
|
|
assert "ES Assets" in html
|
|
assert "24 hours" in html
|
|
|
|
def test_render_template_single_language(self, service):
|
|
html = service._render_completion_template(
|
|
job_title="English Only",
|
|
download_links={"en": {"captions_vtt": "https://example.com/cap.vtt"}},
|
|
)
|
|
assert "English Only" in html
|
|
assert "EN Assets" in html
|
|
assert "ES Assets" not in html
|
|
|
|
def test_render_template_no_downloads(self, service):
|
|
html = service._render_completion_template(job_title="Empty Job", download_links={})
|
|
assert "Empty Job" in html
|
|
assert "<!DOCTYPE html>" in html
|
|
assert "24 hours" in html
|
|
|
|
def test_render_template_html_structure(self, service, sample_download_links):
|
|
html = service._render_completion_template(
|
|
job_title="Test", download_links=sample_download_links
|
|
)
|
|
assert html.startswith("<!DOCTYPE html>") or "<!DOCTYPE html>" in html
|
|
assert "<html>" in html
|
|
assert "</html>" in html
|
|
assert "<body>" in html
|
|
assert "font-family: Arial" in html
|
|
|
|
def test_render_template_download_links(self, service):
|
|
html = service._render_completion_template(
|
|
job_title="Test",
|
|
download_links={"en": {
|
|
"captions_vtt": "https://example.com/captions.vtt",
|
|
"audio_description_mp3": "https://example.com/ad.mp3",
|
|
}},
|
|
)
|
|
assert "Download Captions Vtt" in html
|
|
assert "Download Audio Description Mp3" in html
|
|
assert 'href="https://example.com/captions.vtt"' in html
|
|
assert 'href="https://example.com/ad.mp3"' in html
|
|
|
|
def test_render_template_xss_escaping(self, service):
|
|
html = service._render_completion_template(
|
|
job_title="<script>alert('xss')</script>Title",
|
|
download_links={},
|
|
)
|
|
assert "<script>" not in html
|
|
assert "Title" in html
|