26bd096a06
Major changes across backend, frontend, infrastructure: - Provider system with model selection (Google, DeepL, OpenAI, Ollama, Google Cloud) - Admin panel: user management, pricing, settings - Glossary system with CSV import/export - Subscription and tier quota management - Security hardening (rate limiting, API key auth, path traversal fixes) - Docker compose for dev, prod, and IONOS deployment - Alembic migrations for new tables - Frontend: dashboard, pricing page, landing page, i18n (en/fr) - Test suite and verification scripts Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
423 lines
16 KiB
Python
423 lines
16 KiB
Python
"""
|
|
Integration tests for GoogleTranslationProvider.
|
|
|
|
Tests for error handling, retry logic, and health checks.
|
|
Uses mocking to simulate various API error scenarios.
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
import time
|
|
|
|
from services.providers.google_provider import (
|
|
GoogleTranslationProvider,
|
|
GoogleProviderError,
|
|
GOOGLE_QUOTA_EXCEEDED,
|
|
GOOGLE_INVALID_KEY,
|
|
GOOGLE_NETWORK_ERROR,
|
|
GOOGLE_UNSUPPORTED_LANGUAGE,
|
|
GOOGLE_TEXT_TOO_LONG,
|
|
)
|
|
from services.providers.schemas import TranslationRequest
|
|
|
|
|
|
class TestGoogleProviderHealthCheck:
|
|
"""Tests for health check functionality."""
|
|
|
|
@pytest.fixture
|
|
def provider(self):
|
|
return GoogleTranslationProvider(use_cache=False)
|
|
|
|
def test_health_check_returns_status(self, provider):
|
|
"""Test health check returns ProviderHealthStatus."""
|
|
status = provider.health_check()
|
|
|
|
assert status.name == "google"
|
|
assert isinstance(status.available, bool)
|
|
assert status.latency_ms is not None
|
|
|
|
def test_health_check_includes_last_check_timestamp(self, provider):
|
|
"""Test health check includes last_check timestamp."""
|
|
status = provider.health_check()
|
|
|
|
assert status.last_check is not None
|
|
assert "T" in status.last_check # ISO format
|
|
|
|
def test_health_check_caches_result(self, provider):
|
|
"""Test health check result is cached for 60 seconds."""
|
|
status1 = provider.health_check()
|
|
status2 = provider.health_check()
|
|
|
|
assert status1.last_check == status2.last_check
|
|
|
|
def test_health_check_cache_ttl(self, provider):
|
|
"""Test health check cache expires after TTL."""
|
|
provider._health_cache_ttl = 0.1 # 100ms TTL for testing
|
|
|
|
status1 = provider.health_check()
|
|
time.sleep(0.15)
|
|
status2 = provider.health_check()
|
|
|
|
assert status1.last_check != status2.last_check
|
|
|
|
|
|
class TestGoogleProviderErrorCodes:
|
|
"""Tests for specific Google error codes."""
|
|
|
|
@pytest.fixture
|
|
def provider(self):
|
|
return GoogleTranslationProvider(use_cache=False)
|
|
|
|
def test_quota_exceeded_error(self, provider):
|
|
"""Test GOOGLE_QUOTA_EXCEEDED error on 429 response."""
|
|
request = TranslationRequest(text="Hello", target_language="fr")
|
|
|
|
with patch.object(provider, "_make_api_request") as mock_api:
|
|
mock_api.side_effect = GoogleProviderError(
|
|
code=GOOGLE_QUOTA_EXCEEDED,
|
|
message="Quota Google Translate dépassé. Réessayez demain.",
|
|
details={"reset_at": "2024-01-16T00:00:00Z"},
|
|
)
|
|
|
|
response = provider.translate_text(request)
|
|
|
|
assert response.error is not None
|
|
assert response.error_code == GOOGLE_QUOTA_EXCEEDED
|
|
assert (
|
|
"quota" in response.error.lower() or "dépassé" in response.error.lower()
|
|
)
|
|
|
|
def test_invalid_key_error(self, provider):
|
|
"""Test GOOGLE_INVALID_KEY error on 401 response."""
|
|
request = TranslationRequest(text="Hello", target_language="fr")
|
|
|
|
with patch.object(provider, "_make_api_request") as mock_api:
|
|
mock_api.side_effect = GoogleProviderError(
|
|
code=GOOGLE_INVALID_KEY,
|
|
message="Clé API Google invalide. Contactez l'administrateur.",
|
|
)
|
|
|
|
response = provider.translate_text(request)
|
|
|
|
assert response.error is not None
|
|
assert response.error_code == GOOGLE_INVALID_KEY
|
|
|
|
def test_network_error(self, provider):
|
|
"""Test GOOGLE_NETWORK_ERROR on timeout/connection error."""
|
|
request = TranslationRequest(text="Hello", target_language="fr")
|
|
|
|
with patch.object(provider, "_make_api_request") as mock_api:
|
|
mock_api.side_effect = GoogleProviderError(
|
|
code=GOOGLE_NETWORK_ERROR,
|
|
message="Service Google Translate indisponible. Réessayez.",
|
|
)
|
|
|
|
response = provider.translate_text(request)
|
|
|
|
assert response.error is not None
|
|
assert response.error_code == GOOGLE_NETWORK_ERROR
|
|
|
|
def test_unsupported_language_error(self, provider):
|
|
"""Test GOOGLE_UNSUPPORTED_LANGUAGE for invalid language."""
|
|
request = TranslationRequest(text="Hello", target_language="xx")
|
|
|
|
with patch.object(provider, "_make_api_request") as mock_api:
|
|
mock_api.side_effect = GoogleProviderError(
|
|
code=GOOGLE_UNSUPPORTED_LANGUAGE,
|
|
message="Langue 'xx' non supportée par Google.",
|
|
details={"unsupported_language": "xx"},
|
|
)
|
|
|
|
response = provider.translate_text(request)
|
|
|
|
assert response.error is not None
|
|
assert response.error_code == GOOGLE_UNSUPPORTED_LANGUAGE
|
|
|
|
def test_text_too_long_error(self, provider):
|
|
"""Test GOOGLE_TEXT_TOO_LONG for text exceeding limit."""
|
|
long_text = "x" * 5001
|
|
request = TranslationRequest(text=long_text, target_language="fr")
|
|
|
|
with patch.object(provider, "_make_api_request") as mock_api:
|
|
mock_api.side_effect = GoogleProviderError(
|
|
code=GOOGLE_TEXT_TOO_LONG,
|
|
message="Texte trop long (max 5000 caractères par requête).",
|
|
details={"text_length": 5001, "max_length": 5000},
|
|
)
|
|
|
|
response = provider.translate_text(request)
|
|
|
|
assert response.error is not None
|
|
assert response.error_code == GOOGLE_TEXT_TOO_LONG
|
|
|
|
|
|
class TestGoogleProviderRetryLogic:
|
|
"""Tests for retry logic with exponential backoff."""
|
|
|
|
@pytest.fixture
|
|
def provider(self):
|
|
return GoogleTranslationProvider(use_cache=False)
|
|
|
|
def test_retry_on_transient_error(self, provider):
|
|
"""Test that transient errors trigger retry."""
|
|
request = TranslationRequest(text="Hello", target_language="fr")
|
|
|
|
call_count = 0
|
|
|
|
def mock_api_call(*args, **kwargs):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
if call_count < 3:
|
|
raise GoogleProviderError(
|
|
code=GOOGLE_NETWORK_ERROR, message="Temporary network error"
|
|
)
|
|
return "Bonjour"
|
|
|
|
with patch.object(provider, "_make_api_request", side_effect=mock_api_call):
|
|
response = provider.translate_text(request)
|
|
|
|
assert call_count == 3
|
|
assert response.translated_text == "Bonjour"
|
|
assert response.error is None
|
|
|
|
def test_max_retries_exceeded(self, provider):
|
|
"""Test that max retries is respected."""
|
|
request = TranslationRequest(text="Hello", target_language="fr")
|
|
|
|
with patch.object(provider, "_make_api_request") as mock_api:
|
|
mock_api.side_effect = GoogleProviderError(
|
|
code=GOOGLE_NETWORK_ERROR, message="Network error"
|
|
)
|
|
|
|
response = provider.translate_text(request)
|
|
|
|
assert response.error is not None
|
|
assert mock_api.call_count == 4 # Initial + 3 retries
|
|
|
|
def test_no_retry_on_invalid_key(self, provider):
|
|
"""Test that invalid key errors don't retry."""
|
|
request = TranslationRequest(text="Hello", target_language="fr")
|
|
|
|
with patch.object(provider, "_make_api_request") as mock_api:
|
|
mock_api.side_effect = GoogleProviderError(
|
|
code=GOOGLE_INVALID_KEY, message="Invalid key"
|
|
)
|
|
|
|
response = provider.translate_text(request)
|
|
|
|
assert response.error is not None
|
|
assert mock_api.call_count == 1 # No retries for auth errors
|
|
|
|
|
|
class TestGoogleProviderTimeout:
|
|
"""Tests for timeout configuration."""
|
|
|
|
@pytest.fixture
|
|
def provider(self):
|
|
return GoogleTranslationProvider(use_cache=False)
|
|
|
|
def test_default_timeout(self, provider):
|
|
"""Test default timeout is 30 seconds."""
|
|
assert provider.timeout == 30
|
|
|
|
def test_custom_timeout(self):
|
|
"""Test custom timeout configuration."""
|
|
provider = GoogleTranslationProvider(use_cache=False, timeout=60)
|
|
assert provider.timeout == 60
|
|
|
|
def test_timeout_raises_network_error(self, provider):
|
|
"""Test that timeout raises GOOGLE_NETWORK_ERROR."""
|
|
request = TranslationRequest(text="Hello", target_language="fr")
|
|
|
|
import socket
|
|
|
|
with patch.object(provider, "_make_api_request") as mock_api:
|
|
mock_api.side_effect = socket.timeout("Request timed out")
|
|
|
|
response = provider.translate_text(request)
|
|
|
|
assert response.error is not None
|
|
assert response.error_code == GOOGLE_NETWORK_ERROR
|
|
|
|
|
|
class TestGoogleProviderErrorFormat:
|
|
"""Tests for JSON error format compliance."""
|
|
|
|
@pytest.fixture
|
|
def provider(self):
|
|
return GoogleTranslationProvider(use_cache=False)
|
|
|
|
def test_error_response_format(self, provider):
|
|
"""Test that errors return JSON: {error, message, details?} format."""
|
|
request = TranslationRequest(text="Hello", target_language="fr")
|
|
|
|
with patch.object(provider, "_make_api_request") as mock_api:
|
|
mock_api.side_effect = GoogleProviderError(
|
|
code=GOOGLE_QUOTA_EXCEEDED,
|
|
message="Quota exceeded",
|
|
details={"reset_at": "2024-01-16T00:00:00Z"},
|
|
)
|
|
|
|
response = provider.translate_text(request)
|
|
|
|
assert response.error is not None
|
|
assert response.error_code is not None
|
|
error_dict = response.to_error_dict()
|
|
assert "error" in error_dict
|
|
assert "message" in error_dict
|
|
|
|
def test_error_no_document_content_in_response(self, provider):
|
|
"""Test that error response never contains document content."""
|
|
sensitive_text = "SENSITIVE_DATA_12345"
|
|
request = TranslationRequest(text=sensitive_text, target_language="fr")
|
|
|
|
with patch.object(provider, "_make_api_request") as mock_api:
|
|
mock_api.side_effect = GoogleProviderError(
|
|
code=GOOGLE_NETWORK_ERROR, message="Network error"
|
|
)
|
|
|
|
response = provider.translate_text(request)
|
|
|
|
assert sensitive_text not in str(response.error)
|
|
assert sensitive_text not in str(response.to_error_dict())
|
|
|
|
|
|
class TestGoogleProviderLogging:
|
|
"""Tests for structlog logging."""
|
|
|
|
@pytest.fixture
|
|
def provider(self):
|
|
return GoogleTranslationProvider(use_cache=False)
|
|
|
|
def test_error_logged_with_structlog(self, provider):
|
|
"""Test errors are logged with structlog (no document content)."""
|
|
request = TranslationRequest(text="Hello", target_language="fr")
|
|
|
|
with patch.object(provider, "_make_api_request") as mock_api:
|
|
mock_api.side_effect = GoogleProviderError(
|
|
code=GOOGLE_QUOTA_EXCEEDED, message="Quota exceeded"
|
|
)
|
|
|
|
with patch("services.providers.google_provider.logger") as mock_logger:
|
|
response = provider.translate_text(request)
|
|
|
|
assert mock_logger.error.called or mock_logger.warning.called
|
|
|
|
def test_log_contains_metadata_not_content(self, provider):
|
|
"""Test logs contain metadata (text_length) not content."""
|
|
request = TranslationRequest(text="Secret content", target_language="fr")
|
|
|
|
with patch.object(provider, "_make_api_request") as mock_api:
|
|
mock_api.side_effect = GoogleProviderError(
|
|
code=GOOGLE_NETWORK_ERROR, message="Network error"
|
|
)
|
|
|
|
with patch("services.providers.google_provider.logger") as mock_logger:
|
|
response = provider.translate_text(request)
|
|
|
|
if mock_logger.error.called:
|
|
call_args = str(mock_logger.error.call_args)
|
|
assert "Secret content" not in call_args
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestGoogleProviderRealAPI:
|
|
"""Integration tests with real Google Translate API (via deep_translator)."""
|
|
|
|
@pytest.fixture
|
|
def provider(self):
|
|
return GoogleTranslationProvider(use_cache=False)
|
|
|
|
def test_real_translation_en_to_fr(self, provider):
|
|
"""Test real translation from English to French."""
|
|
request = TranslationRequest(text="Hello", target_language="fr")
|
|
response = provider.translate_text(request)
|
|
|
|
assert response.error is None
|
|
assert response.translated_text.lower() in ["bonjour", "salut", "hello"]
|
|
assert response.provider_name == "google"
|
|
|
|
def test_real_translation_with_auto_detect(self, provider):
|
|
"""Test translation with automatic language detection."""
|
|
request = TranslationRequest(
|
|
text="Bonjour le monde", target_language="en", source_language="auto"
|
|
)
|
|
response = provider.translate_text(request)
|
|
|
|
assert response.error is None
|
|
assert (
|
|
"world" in response.translated_text.lower()
|
|
or "hello" in response.translated_text.lower()
|
|
)
|
|
|
|
def test_real_health_check(self, provider):
|
|
"""Test real health check."""
|
|
status = provider.health_check()
|
|
|
|
assert status.name == "google"
|
|
assert status.available is True
|
|
assert status.latency_ms is not None
|
|
assert status.last_check is not None
|
|
|
|
def test_real_batch_translation(self, provider):
|
|
"""Test real batch translation."""
|
|
requests = [
|
|
TranslationRequest(text="Hello", target_language="es"),
|
|
TranslationRequest(text="World", target_language="es"),
|
|
]
|
|
responses = provider.translate_batch(requests)
|
|
|
|
assert len(responses) == 2
|
|
assert all(r.error is None for r in responses)
|
|
assert "hola" in responses[0].translated_text.lower()
|
|
assert "mundo" in responses[1].translated_text.lower()
|
|
|
|
|
|
class TestGoogleProviderOptimization:
|
|
"""Tests for API usage optimization."""
|
|
|
|
@pytest.fixture
|
|
def provider(self):
|
|
return GoogleTranslationProvider(use_cache=False)
|
|
|
|
def test_skip_translation_same_language(self, provider):
|
|
"""Test translation is skipped when source == target."""
|
|
request = TranslationRequest(
|
|
text="Hello World", target_language="en", source_language="en"
|
|
)
|
|
response = provider.translate_text(request)
|
|
|
|
assert response.translated_text == "Hello World"
|
|
assert response.from_cache is False
|
|
|
|
def test_translation_not_skipped_auto_detect(self, provider):
|
|
"""Test translation is not skipped with auto-detect."""
|
|
request = TranslationRequest(
|
|
text="Bonjour", target_language="en", source_language="auto"
|
|
)
|
|
|
|
with patch.object(provider, "_make_api_request", return_value="Hello"):
|
|
response = provider.translate_text(request)
|
|
|
|
assert response.translated_text == "Hello"
|
|
provider._make_api_request.assert_called_once()
|
|
|
|
def test_usage_metrics_logged(self, provider):
|
|
"""Test that usage metrics are logged on success."""
|
|
request = TranslationRequest(text="Hello", target_language="fr")
|
|
|
|
with patch.object(provider, "_make_api_request", return_value="Bonjour"):
|
|
with patch("services.providers.google_provider.logger") as mock_logger:
|
|
response = provider.translate_text(request)
|
|
|
|
# Check that success was logged with metrics
|
|
success_calls = [
|
|
call
|
|
for call in mock_logger.info.call_args_list
|
|
if "google_translation_success" in str(call)
|
|
]
|
|
assert len(success_calls) > 0
|
|
log_msg = str(success_calls[0])
|
|
assert "chars=" in log_msg
|
|
assert "source_lang=" in log_msg
|