From 5fd370c0932c30ea0121f992f711b50ee53d4438 Mon Sep 17 00:00:00 2001 From: Vadym Samoilenko Date: Thu, 30 Apr 2026 14:02:04 +0100 Subject: [PATCH] =?UTF-8?q?test:=20fix=20all=20unit=20tests=20=E2=80=94=20?= =?UTF-8?q?168=20passing,=200=20failures?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- backend/app/core/security.py | 10 +- backend/app/lib/vtt.py | 17 +- backend/app/services/emailer.py | 4 +- backend/app/services/gcs.py | 26 +- backend/app/services/tts.py | 59 +-- backend/tests/conftest.py | 18 + .../tests/unit/test_cross_tenant_isolation.py | 2 +- backend/tests/unit/test_emailer.py | 341 +++++++--------- backend/tests/unit/test_gcs.py | 130 +++---- backend/tests/unit/test_gemini.py | 366 ++++++++---------- backend/tests/unit/test_models.py | 6 +- backend/tests/unit/test_security.py | 3 +- backend/tests/unit/test_tts.py | 43 +- 13 files changed, 462 insertions(+), 563 deletions(-) create mode 100644 backend/tests/conftest.py diff --git a/backend/app/core/security.py b/backend/app/core/security.py index 63d5d1b..cf44b0b 100644 --- a/backend/app/core/security.py +++ b/backend/app/core/security.py @@ -1,5 +1,5 @@ from datetime import datetime, timedelta -from typing import Any, Optional, Union +from typing import Any from fastapi import HTTPException, status from jose import JWTError, jwt @@ -11,8 +11,8 @@ pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def create_access_token( - subject: Union[str, Any], - expires_delta: Optional[timedelta] = None, + subject: str | Any, + expires_delta: timedelta | None = None, org_ids: list[str] | None = None, ) -> str: if expires_delta: @@ -28,7 +28,7 @@ def create_access_token( def create_refresh_token( - subject: Union[str, Any], expires_delta: Optional[timedelta] = None + subject: str | Any, expires_delta: timedelta | None = None ) -> str: if expires_delta: expire = datetime.utcnow() + expires_delta @@ -41,6 +41,8 @@ def create_refresh_token( def verify_password(plain_password: str, hashed_password: str) -> bool: + if not hashed_password: + return False return pwd_context.verify(plain_password, hashed_password) diff --git a/backend/app/lib/vtt.py b/backend/app/lib/vtt.py index 994b2b9..89f58bd 100644 --- a/backend/app/lib/vtt.py +++ b/backend/app/lib/vtt.py @@ -49,13 +49,12 @@ class VTTParser: text_lines.append(lines[i].strip()) i += 1 - if text_lines: - cues.append(VTTCue( - start_time=start_time, - end_time=end_time, - text="\n".join(text_lines), - identifier=identifier - )) + cues.append(VTTCue( + start_time=start_time, + end_time=end_time, + text="\n".join(text_lines), + identifier=identifier + )) else: i += 1 @@ -80,7 +79,7 @@ class VTTParser: lines.append(cue.text) lines.append("") # Empty line between cues - return "\n".join(lines) + return "\n".join(lines) + "\n" @staticmethod def _parse_timestamp(timestamp: str) -> float: @@ -121,7 +120,7 @@ class VTTParser: secs = seconds % 60 whole_secs = int(secs) - milliseconds = int((secs - whole_secs) * 1000) + milliseconds = round((secs - whole_secs) * 1000) return f"{hours:02d}:{minutes:02d}:{whole_secs:02d}.{milliseconds:03d}" diff --git a/backend/app/services/emailer.py b/backend/app/services/emailer.py index ce3e2c8..9ba23de 100644 --- a/backend/app/services/emailer.py +++ b/backend/app/services/emailer.py @@ -1,5 +1,7 @@ +import html as _html from datetime import datetime + from jinja2 import Template from ..core.config import settings @@ -385,7 +387,7 @@ class EmailService: template = Template(template_str) return template.render( - job_title=job_title, + job_title=_html.escape(job_title), download_links=download_links ) diff --git a/backend/app/services/gcs.py b/backend/app/services/gcs.py index 5c5f597..7594d80 100644 --- a/backend/app/services/gcs.py +++ b/backend/app/services/gcs.py @@ -1,7 +1,6 @@ import asyncio from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta -from typing import Optional from fastapi import HTTPException, UploadFile from google.cloud import storage @@ -13,16 +12,27 @@ from ..core.logging import get_logger logger = get_logger(__name__) class GCSService: - def __init__(self): - self.client = storage.Client(project=settings.gcp_project_id) - self.bucket = self.client.bucket(settings.gcs_bucket) + def __init__(self) -> None: + self._client: storage.Client | None = None + self._bucket = None self.executor = ThreadPoolExecutor(max_workers=4) + @property + def bucket(self): + if self._bucket is None: + self._client = storage.Client(project=settings.gcp_project_id) + self._bucket = self._client.bucket(settings.gcs_bucket) + return self._bucket + + @bucket.setter + def bucket(self, value) -> None: + self._bucket = value + async def upload_file_to_gcs( self, file: UploadFile, destination_path: str, - content_type: Optional[str] = None + content_type: str | None = None ) -> str: """Upload file to GCS and return the GCS URI""" def _upload(): @@ -184,7 +194,7 @@ async def generate_signed_upload_url( """Generate a signed URL for direct browser-to-GCS upload""" def _generate(): blob = gcs_service.bucket.blob(blob_path) - + # Generate signed POST URL url, fields = blob.generate_signed_post_policy_v4( expiration=timedelta(hours=1), @@ -196,8 +206,8 @@ async def generate_signed_upload_url( "Content-Type": content_type } ) - + return {"url": url, "fields": fields} - + loop = asyncio.get_event_loop() return await loop.run_in_executor(gcs_service.executor, _generate) diff --git a/backend/app/services/tts.py b/backend/app/services/tts.py index 861d9e6..200a878 100644 --- a/backend/app/services/tts.py +++ b/backend/app/services/tts.py @@ -1,6 +1,5 @@ import io from dataclasses import dataclass -from typing import Optional import aiohttp from google.cloud import texttospeech @@ -47,8 +46,8 @@ class TTSService: self, ad_vtt_content: str, language_code: str = "en-US", - voice_name: Optional[str] = None, - provider: Optional[str] = None, + voice_name: str | None = None, + provider: str | None = None, model: str = "flash", speed: float = 1.0, style_prompt: str = "", @@ -114,8 +113,8 @@ class TTSService: self, ad_vtt_content: str, language_code: str = "en-US", - voice_name: Optional[str] = None, - provider: Optional[str] = None, + voice_name: str | None = None, + provider: str | None = None, model: str = "flash", speed: float = 1.0, style_prompt: str = "", @@ -219,7 +218,7 @@ class TTSService: self, ad_vtt_content: str, language_code: str = "en-US", - voice_name: Optional[str] = None + voice_name: str | None = None ) -> bytes: """Generate MP3 using Google TTS with 2-second pauses between passages""" @@ -281,7 +280,7 @@ class TTSService: self, ad_vtt_content: str, language_code: str = "en-US", - voice_name: Optional[str] = None, + voice_name: str | None = None, stability: float = 0.5, similarity_boost: float = 0.5, ) -> bytes: @@ -339,7 +338,7 @@ class TTSService: self, text: str, language_code: str, - voice_name: Optional[str] = None + voice_name: str | None = None ) -> bytes: """Synthesize a single text string to audio using Google TTS""" # Configure voice @@ -404,7 +403,7 @@ class TTSService: error_text = await response.text() raise ValueError(f"ElevenLabs TTS failed: {response.status} - {error_text}") - def _get_elevenlabs_voice(self, language_code: str, voice_name: Optional[str] = None) -> str: + def _get_elevenlabs_voice(self, language_code: str, voice_name: str | None = None) -> str: """Get ElevenLabs voice ID for language""" if voice_name: return voice_name @@ -452,28 +451,32 @@ class TTSService: def _parse_timestamp(self, timestamp: str) -> float: """Convert VTT timestamp to seconds""" # Format: HH:MM:SS.mmm or MM:SS.mmm - parts = timestamp.split(":") + try: + parts = timestamp.split(":") - if len(parts) == 3: # HH:MM:SS.mmm - hours, minutes, seconds = parts - elif len(parts) == 2: # MM:SS.mmm - hours, minutes, seconds = "0", parts[0], parts[1] - else: - raise ValueError(f"Invalid timestamp format: {timestamp}") + if len(parts) == 3: # HH:MM:SS.mmm + hours, minutes, seconds = parts + elif len(parts) == 2: # MM:SS.mmm + hours, minutes, seconds = "0", parts[0], parts[1] + else: + raise ValueError(f"Invalid timestamp format: {timestamp}") - # Parse seconds and milliseconds - sec_parts = seconds.split(".") - seconds = int(sec_parts[0]) - milliseconds = int(sec_parts[1]) if len(sec_parts) > 1 else 0 + # Parse seconds and milliseconds + sec_parts = seconds.split(".") + seconds_int = int(sec_parts[0]) + milliseconds = int(sec_parts[1]) if len(sec_parts) > 1 else 0 - total_seconds = ( - int(hours) * 3600 + - int(minutes) * 60 + - seconds + - milliseconds / 1000.0 - ) - - return total_seconds + total_seconds = ( + int(hours) * 3600 + + int(minutes) * 60 + + seconds_int + + milliseconds / 1000.0 + ) + return total_seconds + except (ValueError, IndexError) as e: + if "Invalid timestamp format" in str(e): + raise + raise ValueError(f"Invalid timestamp format: {timestamp}") from e # Global service instance diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py new file mode 100644 index 0000000..ec7d79c --- /dev/null +++ b/backend/tests/conftest.py @@ -0,0 +1,18 @@ +import os + +# Set required env vars before any app module is imported. +# Settings() is instantiated at module level in config.py, so these must +# be present before collection starts. +_TEST_DEFAULTS = { + "APP_ENV": "test", + "JWT_SECRET": "test-jwt-secret-at-least-32-chars-long", + "MONGODB_URI": "mongodb://localhost:27017/test_accessible_video", + "REDIS_URL": "redis://localhost:6379/1", + "GCP_PROJECT_ID": "test-gcp-project", + "GCS_BUCKET": "test-bucket", + "GEMINI_API_KEY": "test-gemini-key", + "CLIENT_BASE_URL": "http://localhost:3000", +} + +for key, value in _TEST_DEFAULTS.items(): + os.environ.setdefault(key, value) diff --git a/backend/tests/unit/test_cross_tenant_isolation.py b/backend/tests/unit/test_cross_tenant_isolation.py index 18c7b37..57adc69 100644 --- a/backend/tests/unit/test_cross_tenant_isolation.py +++ b/backend/tests/unit/test_cross_tenant_isolation.py @@ -10,7 +10,7 @@ MT-18: list_for_reviewer must only return jobs from the reviewer's orgs. """ import pytest -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock, MagicMock, patch from fastapi import HTTPException from app.core.dependencies import assert_job_in_user_org, get_user_org_ids diff --git a/backend/tests/unit/test_emailer.py b/backend/tests/unit/test_emailer.py index 579afba..4ec76cc 100644 --- a/backend/tests/unit/test_emailer.py +++ b/backend/tests/unit/test_emailer.py @@ -1,4 +1,4 @@ -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -6,236 +6,177 @@ from app.services.emailer import EmailService class TestEmailService: - """Test email service functionality""" + """Test email service (Mailgun-based)""" @pytest.fixture - def email_service(self): - """Create email service with mocked SendGrid client""" - with patch('app.services.emailer.settings') as mock_settings: - mock_settings.sendgrid_api_key = "test_api_key" - mock_settings.email_from = "support@example.com" - - with patch('app.services.emailer.SendGridAPIClient') as mock_client: - service = EmailService() - service.client = MagicMock() - return service + def service(self): + return EmailService() @pytest.fixture def sample_download_links(self): - """Sample download links for testing""" 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" + "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" - } + "audio_description_mp3": "https://signed-url.example.com/es/ad.mp3", + }, } - @pytest.mark.asyncio - async def test_send_completion_email_success(self, email_service, sample_download_links): - """Test successful completion email sending""" - # Mock successful SendGrid response - mock_response = MagicMock() - mock_response.status_code = 202 - email_service.client.send.return_value = mock_response - - result = await email_service.send_completion_email( - recipient_email="client@example.com", - job_title="Test Video Project", - download_links=sample_download_links - ) - - assert result is True - email_service.client.send.assert_called_once() + # ── _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_completion_email_no_client(self): - """Test email sending when client is not configured""" - service = EmailService() - service.client = None - - result = await service.send_completion_email( - recipient_email="client@example.com", - job_title="Test Video", - download_links={} - ) - + 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", "

HTML

") assert result is False @pytest.mark.asyncio - async def test_send_completion_email_api_failure(self, email_service, sample_download_links): - """Test email sending with SendGrid API failure""" - # Mock failed SendGrid response + 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", "

Hi

") + + assert result is True + + @pytest.mark.asyncio + async def test_send_api_failure(self, service): mock_response = MagicMock() mock_response.status_code = 400 - email_service.client.send.return_value = mock_response - - result = await email_service.send_completion_email( - recipient_email="client@example.com", - job_title="Test Video", - download_links=sample_download_links - ) - + 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", "

Hi

") + assert result is False @pytest.mark.asyncio - async def test_send_completion_email_exception(self, email_service, sample_download_links): - """Test email sending with exception""" - # Mock SendGrid client raising exception - email_service.client.send.side_effect = Exception("SendGrid error") - - result = await email_service.send_completion_email( - recipient_email="client@example.com", - job_title="Test Video", - download_links=sample_download_links - ) - + 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", "

Hi

") + assert result is False - def test_render_completion_template_basic(self, email_service, sample_download_links): - """Test rendering completion email template""" - html_content = email_service._render_completion_template( - job_title="Test Video Project", - download_links=sample_download_links - ) - - # Check that key elements are present - assert "Test Video Project" in html_content - assert "EN Assets" in html_content - assert "ES Assets" in html_content - assert "captions.vtt" in html_content - assert "audio_description_vtt" in html_content - assert "audio_description_mp3" in html_content - assert "24 hours" in html_content # Expiry warning - - def test_render_completion_template_single_language(self, email_service): - """Test rendering template with single language""" - download_links = { - "en": { - "captions_vtt": "https://example.com/captions.vtt" - } - } - - html_content = email_service._render_completion_template( - job_title="English Only Video", - download_links=download_links - ) - - assert "English Only Video" in html_content - assert "EN Assets" in html_content - assert "ES Assets" not in html_content - - def test_render_completion_template_no_downloads(self, email_service): - """Test rendering template with no download links""" - html_content = email_service._render_completion_template( - job_title="Empty Job", - download_links={} - ) - - assert "Empty Job" in html_content - assert "" in html_content - assert "24 hours" in html_content - - def test_render_completion_template_html_structure(self, email_service, sample_download_links): - """Test that rendered template has proper HTML structure""" - html_content = email_service._render_completion_template( - job_title="Test Video", - download_links=sample_download_links - ) - - # Check HTML structure - assert html_content.startswith("") - assert "" in html_content - assert "" in html_content - assert "" in html_content - assert "" in html_content - assert "font-family: Arial" in html_content # CSS present - - def test_render_completion_template_download_link_formatting(self, email_service): - """Test that download links are properly formatted in template""" - download_links = { - "en": { - "captions_vtt": "https://example.com/captions.vtt", - "audio_description_mp3": "https://example.com/ad.mp3" - } - } - - html_content = email_service._render_completion_template( - job_title="Test Video", - download_links=download_links - ) - - # Check that file types are properly formatted - assert "Download Captions Vtt" in html_content - assert "Download Audio Description Mp3" in html_content - assert 'href="https://example.com/captions.vtt"' in html_content - assert 'href="https://example.com/ad.mp3"' in html_content - - def test_service_initialization_with_api_key(self): - """Test service initialization with SendGrid API key""" - with patch('app.services.emailer.settings') as mock_settings: - mock_settings.sendgrid_api_key = "test_api_key" - - with patch('app.services.emailer.SendGridAPIClient') as mock_client: - service = EmailService() - - mock_client.assert_called_once_with(api_key="test_api_key") - assert service.client is not None - - def test_service_initialization_without_api_key(self): - """Test service initialization without SendGrid API key""" - with patch('app.services.emailer.settings') as mock_settings: - mock_settings.sendgrid_api_key = "" - - service = EmailService() - - assert service.client is None + # ── send_completion_email ───────────────────────────────────────────────── @pytest.mark.asyncio - async def test_send_completion_email_mail_object_creation(self, email_service, sample_download_links): - """Test that Mail object is created correctly""" - mock_response = MagicMock() - mock_response.status_code = 202 - email_service.client.send.return_value = mock_response - - with patch('app.services.emailer.Mail') as mock_mail: - mock_mail_instance = MagicMock() - mock_mail.return_value = mock_mail_instance - - await email_service.send_completion_email( + 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 + download_links=sample_download_links, ) - - # Verify Mail object was created with correct parameters - mock_mail.assert_called_once() - call_args = mock_mail.call_args - - # Check that from_email, to_emails, subject, and html_content are set - assert call_args is not None - email_service.client.send.assert_called_once_with(mock_mail_instance) + assert result is True + mock_send.assert_called_once() + # Subject should contain job title + assert "Test Video" in mock_send.call_args[0][1] - def test_template_injection_safety(self, email_service): - """Test that template is safe from injection attacks""" - malicious_title = "Malicious Title" - malicious_links = { - "en": { - "captions_vtt": "javascript:alert('xss')" - } - } - - html_content = email_service._render_completion_template( - job_title=malicious_title, - download_links=malicious_links + # ── _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, ) - - # Jinja2 should escape HTML by default - assert "Title", + download_links={}, + ) + assert "