296 lines
No EOL
12 KiB
Python
296 lines
No EOL
12 KiB
Python
import json
|
|
from unittest.mock import AsyncMock, MagicMock, patch, mock_open
|
|
|
|
import pytest
|
|
|
|
from app.services.gemini import GeminiService
|
|
|
|
|
|
class TestGeminiService:
|
|
"""Test Gemini AI service functionality"""
|
|
|
|
@pytest.fixture
|
|
def gemini_service(self):
|
|
"""Create Gemini service instance with mocked dependencies"""
|
|
with patch('app.services.gemini.genai'):
|
|
service = GeminiService()
|
|
service.model = MagicMock()
|
|
return service
|
|
|
|
@pytest.fixture
|
|
def valid_gemini_response(self):
|
|
"""Sample valid Gemini response"""
|
|
return {
|
|
"language": "en",
|
|
"confidence": 0.92,
|
|
"summary": "A short video about accessibility features in web development.",
|
|
"transcript_plaintext": "Hello everyone, today we'll learn about accessibility features.",
|
|
"captions_vtt": """WEBVTT
|
|
|
|
00:00:01.000 --> 00:00:03.000
|
|
Hello everyone, today we'll
|
|
|
|
00:00:03.000 --> 00:00:05.000
|
|
learn about accessibility features.
|
|
""",
|
|
"audio_description_vtt": """WEBVTT
|
|
|
|
00:00:00.500 --> 00:00:01.000
|
|
[Upbeat intro music plays]
|
|
|
|
00:00:05.500 --> 00:00:07.000
|
|
[Speaker gestures toward screen]
|
|
"""
|
|
}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_accessibility_success(self, gemini_service, valid_gemini_response):
|
|
"""Test successful accessibility extraction"""
|
|
# Mock file upload and model response
|
|
mock_response = MagicMock()
|
|
mock_response.text = json.dumps(valid_gemini_response)
|
|
gemini_service.model.generate_content.return_value = mock_response
|
|
|
|
with patch('app.services.gemini.genai.upload_file') as mock_upload:
|
|
mock_upload.return_value = MagicMock()
|
|
|
|
with patch.object(gemini_service, '_load_prompt') as mock_load_prompt:
|
|
mock_load_prompt.return_value = "Test prompt"
|
|
|
|
result = await gemini_service.extract_accessibility("/tmp/test.mp4")
|
|
|
|
assert result == valid_gemini_response
|
|
assert result["confidence"] == 0.92
|
|
assert result["language"] == "en"
|
|
assert "WEBVTT" in result["captions_vtt"]
|
|
assert "WEBVTT" in result["audio_description_vtt"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_accessibility_with_markdown_formatting(self, gemini_service, valid_gemini_response):
|
|
"""Test handling Gemini response with markdown formatting"""
|
|
# Mock response with markdown formatting
|
|
mock_response = MagicMock()
|
|
mock_response.text = f"```json\n{json.dumps(valid_gemini_response)}\n```"
|
|
gemini_service.model.generate_content.return_value = mock_response
|
|
|
|
with patch('app.services.gemini.genai.upload_file'):
|
|
with patch.object(gemini_service, '_load_prompt', return_value="Test prompt"):
|
|
result = await gemini_service.extract_accessibility("/tmp/test.mp4")
|
|
|
|
assert result == valid_gemini_response
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_accessibility_invalid_json(self, gemini_service):
|
|
"""Test handling of invalid JSON response"""
|
|
# Mock invalid JSON response
|
|
mock_response = MagicMock()
|
|
mock_response.text = "invalid json content"
|
|
gemini_service.model.generate_content.return_value = mock_response
|
|
|
|
with patch('app.services.gemini.genai.upload_file'):
|
|
with patch.object(gemini_service, '_load_prompt', return_value="Test prompt"):
|
|
with patch.object(gemini_service, '_self_heal_response') as mock_self_heal:
|
|
mock_self_heal.return_value = {"language": "en", "confidence": 0.8}
|
|
|
|
result = await gemini_service.extract_accessibility("/tmp/test.mp4")
|
|
|
|
assert result == {"language": "en", "confidence": 0.8}
|
|
mock_self_heal.assert_called_once_with("/tmp/test.mp4", "invalid json content")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_accessibility_missing_fields(self, gemini_service):
|
|
"""Test error handling for missing required fields"""
|
|
incomplete_response = {
|
|
"language": "en",
|
|
"confidence": 0.92
|
|
# Missing required fields
|
|
}
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.text = json.dumps(incomplete_response)
|
|
gemini_service.model.generate_content.return_value = mock_response
|
|
|
|
with patch('app.services.gemini.genai.upload_file'):
|
|
with patch.object(gemini_service, '_load_prompt', return_value="Test prompt"):
|
|
with pytest.raises(ValueError, match="Missing required field"):
|
|
await gemini_service.extract_accessibility("/tmp/test.mp4")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_accessibility_invalid_vtt_format(self, gemini_service):
|
|
"""Test error handling for invalid VTT format"""
|
|
invalid_response = {
|
|
"language": "en",
|
|
"confidence": 0.92,
|
|
"summary": "Test summary",
|
|
"transcript_plaintext": "Test transcript",
|
|
"captions_vtt": "Invalid VTT content", # Missing WEBVTT header
|
|
"audio_description_vtt": "WEBVTT\n\n00:00:01.000 --> 00:00:02.000\nValid AD"
|
|
}
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.text = json.dumps(invalid_response)
|
|
gemini_service.model.generate_content.return_value = mock_response
|
|
|
|
with patch('app.services.gemini.genai.upload_file'):
|
|
with patch.object(gemini_service, '_load_prompt', return_value="Test prompt"):
|
|
with pytest.raises(ValueError, match="Invalid captions VTT format"):
|
|
await gemini_service.extract_accessibility("/tmp/test.mp4")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_self_heal_response_success(self, gemini_service, valid_gemini_response):
|
|
"""Test successful self-healing of invalid response"""
|
|
mock_response = MagicMock()
|
|
mock_response.text = json.dumps(valid_gemini_response)
|
|
gemini_service.model.generate_content.return_value = mock_response
|
|
|
|
with patch('app.services.gemini.genai.upload_file'):
|
|
result = await gemini_service._self_heal_response("/tmp/test.mp4", "invalid json")
|
|
|
|
assert result == valid_gemini_response
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_self_heal_response_reask(self, gemini_service):
|
|
"""Test self-healing when Gemini returns REASK"""
|
|
mock_response = MagicMock()
|
|
mock_response.text = "REASK"
|
|
gemini_service.model.generate_content.return_value = mock_response
|
|
|
|
with patch('app.services.gemini.genai.upload_file'):
|
|
with pytest.raises(ValueError, match="Gemini unable to self-heal response"):
|
|
await gemini_service._self_heal_response("/tmp/test.mp4", "invalid json")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_transcreate_content_success(self, gemini_service):
|
|
"""Test successful content transcreation"""
|
|
transcreate_response = {
|
|
"captions_vtt": """WEBVTT
|
|
|
|
00:00:01.000 --> 00:00:03.000
|
|
Hola a todos, hoy vamos a
|
|
|
|
00:00:03.000 --> 00:00:05.000
|
|
aprender sobre características de accesibilidad.
|
|
""",
|
|
"audio_description_vtt": """WEBVTT
|
|
|
|
00:00:00.500 --> 00:00:01.000
|
|
[Música de introducción alegre]
|
|
|
|
00:00:05.500 --> 00:00:07.000
|
|
[El presentador gesticula hacia la pantalla]
|
|
"""
|
|
}
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.text = json.dumps(transcreate_response)
|
|
gemini_service.model.generate_content.return_value = mock_response
|
|
|
|
with patch.object(gemini_service, '_load_prompt', return_value="Test prompt {TARGET_LANGUAGE}"):
|
|
result = await gemini_service.transcreate_content(
|
|
"English captions VTT",
|
|
"English AD VTT",
|
|
"es",
|
|
"Brand guidelines"
|
|
)
|
|
|
|
assert result == transcreate_response
|
|
assert "WEBVTT" in result["captions_vtt"]
|
|
assert "WEBVTT" in result["audio_description_vtt"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_transcreate_content_missing_fields(self, gemini_service):
|
|
"""Test transcreation with missing required fields"""
|
|
incomplete_response = {
|
|
"captions_vtt": "Some content"
|
|
# Missing audio_description_vtt
|
|
}
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.text = json.dumps(incomplete_response)
|
|
gemini_service.model.generate_content.return_value = mock_response
|
|
|
|
with patch.object(gemini_service, '_load_prompt', return_value="Test prompt {TARGET_LANGUAGE}"):
|
|
with pytest.raises(ValueError, match="Missing required VTT fields"):
|
|
await gemini_service.transcreate_content(
|
|
"English captions VTT",
|
|
"English AD VTT",
|
|
"es"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_transcreate_content_invalid_json(self, gemini_service):
|
|
"""Test transcreation with invalid JSON response"""
|
|
mock_response = MagicMock()
|
|
mock_response.text = "invalid json"
|
|
gemini_service.model.generate_content.return_value = mock_response
|
|
|
|
with patch.object(gemini_service, '_load_prompt', return_value="Test prompt {TARGET_LANGUAGE}"):
|
|
with pytest.raises(ValueError, match="Invalid JSON response from transcreation"):
|
|
await gemini_service.transcreate_content(
|
|
"English captions VTT",
|
|
"English AD VTT",
|
|
"es"
|
|
)
|
|
|
|
def test_load_prompt_success(self, gemini_service):
|
|
"""Test successful prompt loading"""
|
|
prompt_content = "Test prompt content with {TARGET_LANGUAGE} placeholder"
|
|
|
|
with patch('pathlib.Path.read_text', return_value=prompt_content):
|
|
result = gemini_service._load_prompt("test_prompt.md")
|
|
assert result == prompt_content
|
|
|
|
def test_load_prompt_file_not_found(self, gemini_service):
|
|
"""Test prompt loading with missing file"""
|
|
with patch('pathlib.Path.read_text', side_effect=FileNotFoundError):
|
|
with pytest.raises(FileNotFoundError):
|
|
gemini_service._load_prompt("nonexistent.md")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_transcreate_with_markdown_response(self, gemini_service):
|
|
"""Test transcreation handling markdown-formatted response"""
|
|
transcreate_response = {
|
|
"captions_vtt": "Test VTT",
|
|
"audio_description_vtt": "Test AD VTT"
|
|
}
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.text = f"```json\n{json.dumps(transcreate_response)}\n```"
|
|
gemini_service.model.generate_content.return_value = mock_response
|
|
|
|
with patch.object(gemini_service, '_load_prompt', return_value="Test prompt {TARGET_LANGUAGE}"):
|
|
result = await gemini_service.transcreate_content(
|
|
"English captions VTT",
|
|
"English AD VTT",
|
|
"es"
|
|
)
|
|
|
|
assert result == transcreate_response
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestGeminiServiceIntegration:
|
|
"""Integration tests for Gemini service (requires actual API key)"""
|
|
|
|
@pytest.mark.skip(reason="Requires actual Gemini API key and video file")
|
|
@pytest.mark.asyncio
|
|
async def test_real_gemini_extraction(self):
|
|
"""Test real Gemini extraction (requires setup)"""
|
|
# This test should be enabled when running with real credentials
|
|
service = GeminiService()
|
|
|
|
# Would require a real test video file
|
|
# result = await service.extract_accessibility("/path/to/test/video.mp4")
|
|
# assert "captions_vtt" in result
|
|
# assert "audio_description_vtt" in result
|
|
pass
|
|
|
|
@pytest.mark.skip(reason="Requires actual Gemini API key")
|
|
@pytest.mark.asyncio
|
|
async def test_real_transcreation(self):
|
|
"""Test real transcreation (requires setup)"""
|
|
# This test should be enabled when running with real credentials
|
|
service = GeminiService()
|
|
|
|
# Would test actual transcreation
|
|
pass |