backend/tests/test_deepl_backend.py
Rama master · réplica de lectura del repositorio original.
"""Tests for translation.deepl_backend — DeepL translation backend."""
from unittest.mock import MagicMock, patch
import pytest
from translation.base import TranslationResult
def _make_backend(**config_overrides):
"""Create a DeepLBackend instance without triggering __init__ side effects."""
from translation.concurrency import get_concurrency
from translation.deepl_backend import DeepLBackend
get_concurrency().register("deepl", 3)
backend = DeepLBackend.__new__(DeepLBackend)
backend.config = {"api_key": "test-key:fx", **config_overrides}
backend._client = None
backend._glossary_cache = {}
return backend
class TestToDeeplLang:
"""Tests for the _to_deepl_lang helper function."""
def test_known_language_code(self):
from translation.deepl_backend import _to_deepl_lang
assert _to_deepl_lang("en") == "EN"
assert _to_deepl_lang("de") == "DE"
def test_portuguese_source_stays_bare(self):
from translation.deepl_backend import _to_deepl_lang
assert _to_deepl_lang("pt") == "PT"
def test_english_target_uses_regional_variant(self):
"""DeepL rejects bare EN as a target: 'deprecated, use EN-GB or EN-US'."""
from translation.deepl_backend import _to_deepl_lang
assert _to_deepl_lang("en", target=True) == "EN-US"
def test_portuguese_target_uses_regional_variant(self):
from translation.deepl_backend import _to_deepl_lang
assert _to_deepl_lang("pt", target=True) == "PT-BR"
def test_target_flag_does_not_affect_other_languages(self):
from translation.deepl_backend import _to_deepl_lang
assert _to_deepl_lang("de", target=True) == "DE"
assert _to_deepl_lang("ja", target=True) == "JA"
def test_norwegian_maps_to_nb(self):
from translation.deepl_backend import _to_deepl_lang
assert _to_deepl_lang("nb") == "NB"
def test_unknown_language_falls_back_to_uppercase(self):
from translation.deepl_backend import _to_deepl_lang
assert _to_deepl_lang("xx") == "XX"
def test_case_insensitive_input(self):
from translation.deepl_backend import _to_deepl_lang
assert _to_deepl_lang("EN") == "EN"
assert _to_deepl_lang("De") == "DE"
class TestDeepLBackendAttributes:
"""Tests for class-level attributes and config."""
def test_name_and_display_name(self):
backend = _make_backend()
assert backend.name == "deepl"
assert backend.display_name == "DeepL"
def test_supports_glossary(self):
backend = _make_backend()
assert backend.supports_glossary is True
def test_supports_batch(self):
backend = _make_backend()
assert backend.supports_batch is True
assert backend.max_batch_size == 50
def test_config_fields_returned(self):
backend = _make_backend()
fields = backend.get_config_fields()
assert len(fields) == 1
assert fields[0]["key"] == "api_key"
class TestGetClient:
"""Tests for lazy client initialization."""
@patch("translation.deepl_backend._DEEPL_AVAILABLE", False)
def test_raises_when_deepl_not_installed(self):
backend = _make_backend()
with pytest.raises(RuntimeError, match="deepl package not installed"):
backend._get_client()
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
def test_raises_when_no_api_key(self):
backend = _make_backend(api_key="")
with pytest.raises(RuntimeError, match="API key not configured"):
backend._get_client()
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
@patch("translation.deepl_backend.deepl")
def test_creates_client_on_first_call(self, mock_deepl):
backend = _make_backend()
mock_deepl.DeepLClient.return_value = MagicMock()
client = backend._get_client()
mock_deepl.DeepLClient.assert_called_once_with(auth_key="test-key:fx")
assert client is not None
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
def test_returns_cached_client_on_second_call(self):
backend = _make_backend()
mock_client = MagicMock()
backend._client = mock_client
assert backend._get_client() is mock_client
class TestTranslateBatch:
"""Tests for translate_batch method."""
@patch("translation.deepl_backend._DEEPL_AVAILABLE", False)
def test_raises_when_package_missing(self):
backend = _make_backend()
with pytest.raises(RuntimeError, match="deepl package not installed"):
backend.translate_batch(["Hello"], "en", "de")
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
def test_successful_translation(self):
backend = _make_backend()
mock_client = MagicMock()
mock_result = MagicMock()
mock_result.text = "Hallo"
mock_client.translate_text.return_value = [mock_result]
backend._client = mock_client
result = backend.translate_batch(["Hello"], "en", "de")
assert result.success is True
assert result.translated_lines == ["Hallo"]
assert result.backend_name == "deepl"
assert result.characters_used == 5
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
def test_multiple_lines(self):
backend = _make_backend()
mock_client = MagicMock()
r1, r2 = MagicMock(), MagicMock()
r1.text = "Hallo"
r2.text = "Welt"
mock_client.translate_text.return_value = [r1, r2]
backend._client = mock_client
result = backend.translate_batch(["Hello", "World"], "en", "de")
assert result.translated_lines == ["Hallo", "Welt"]
assert result.characters_used == 10
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
def test_api_error_returns_failure_result(self):
backend = _make_backend()
mock_client = MagicMock()
mock_client.translate_text.side_effect = Exception("API quota exceeded")
backend._client = mock_client
result = backend.translate_batch(["Hello"], "en", "de")
assert result.success is False
assert result.translated_lines == []
assert "API quota exceeded" in result.error
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
def test_glossary_entries_used(self):
backend = _make_backend()
mock_client = MagicMock()
mock_result = MagicMock()
mock_result.text = "Hallo"
mock_client.translate_text.return_value = [mock_result]
mock_glossary = MagicMock()
mock_client.create_glossary.return_value = mock_glossary
backend._client = mock_client
glossary = [{"source_term": "hello", "target_term": "hallo"}]
result = backend.translate_batch(["Hello"], "en", "de", glossary_entries=glossary)
assert result.success is True
# Verify glossary was passed to translate_text
call_kwargs = mock_client.translate_text.call_args[1]
assert "glossary" in call_kwargs
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
def test_english_target_sent_as_regional_variant(self):
"""Prod incident: DeepL 400s on target_lang="EN" and the breaker latched OPEN."""
backend = _make_backend()
mock_client = MagicMock()
mock_result = MagicMock()
mock_result.text = "Hello"
mock_client.translate_text.return_value = [mock_result]
backend._client = mock_client
result = backend.translate_batch(["Hallo"], "de", "en")
assert result.success is True
call_kwargs = mock_client.translate_text.call_args[1]
assert call_kwargs["target_lang"] == "EN-US"
assert call_kwargs["source_lang"] == "DE"
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
def test_glossary_uses_base_target_code(self):
"""DeepL's glossary API rejects the regional variant translate_text requires."""
backend = _make_backend()
mock_client = MagicMock()
mock_result = MagicMock()
mock_result.text = "Hello"
mock_client.translate_text.return_value = [mock_result]
mock_client.create_glossary.return_value = MagicMock()
backend._client = mock_client
glossary = [{"source_term": "hallo", "target_term": "hello"}]
backend.translate_batch(["Hallo"], "de", "en", glossary_entries=glossary)
glossary_kwargs = mock_client.create_glossary.call_args[1]
assert glossary_kwargs["target_lang"] == "EN"
assert glossary_kwargs["source_lang"] == "DE"
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
def test_hung_call_is_abandoned_after_timeout(self):
"""A translate_text call that never returns must not hang the job forever.
Reproduces the prod incident: the deepl SDK's own retry/backoff didn't
surface an exception even after 12+ minutes (TCP CLOSE_WAIT with
unread bytes). _run_with_timeout must bound the wait regardless.
"""
import threading
import time
backend = _make_backend()
backend.timeout_s = 0.1
mock_client = MagicMock()
release_event = threading.Event()
def _never_returns_in_time(*args, **kwargs):
release_event.wait(timeout=5) # released by the test at teardown
return [MagicMock(text="unused")]
mock_client.translate_text.side_effect = _never_returns_in_time
backend._client = mock_client
started = time.monotonic()
result = backend.translate_batch(["Hello"], "en", "de")
elapsed = time.monotonic() - started
assert result.success is False
assert "0.1s" in result.error
assert elapsed < 1.0 # bounded by timeout_s, not the 5s side_effect wait
release_event.set() # let the abandoned thread finish so it doesn't leak past the test
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
def test_translate_batch_uses_concurrency_slot(self):
"""translate_batch must acquire the shared per-backend concurrency slot."""
from translation.concurrency import get_concurrency
backend = _make_backend()
mock_client = MagicMock()
mock_result = MagicMock()
mock_result.text = "Hallo"
mock_client.translate_text.return_value = [mock_result]
backend._client = mock_client
with patch.object(get_concurrency(), "slot", wraps=get_concurrency().slot) as mock_slot:
result = backend.translate_batch(["Hello"], "en", "de")
assert result.success is True
mock_slot.assert_called_once_with("deepl", timeout_s=backend.timeout_s)
class TestGetOrCreateGlossary:
"""Tests for glossary caching and creation."""
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
def test_glossary_cached_on_second_call(self):
backend = _make_backend()
mock_client = MagicMock()
mock_glossary = MagicMock()
mock_client.create_glossary.return_value = mock_glossary
backend._client = mock_client
entries = [{"source_term": "hello", "target_term": "hallo"}]
g1 = backend._get_or_create_glossary("EN", "DE", entries)
g2 = backend._get_or_create_glossary("EN", "DE", entries)
assert g1 is g2
mock_client.create_glossary.assert_called_once()
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
def test_empty_entries_returns_none(self):
backend = _make_backend()
mock_client = MagicMock()
backend._client = mock_client
result = backend._get_or_create_glossary(
"EN", "DE", [{"source_term": "", "target_term": ""}]
)
assert result is None
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
def test_glossary_creation_failure_returns_none(self):
backend = _make_backend()
mock_client = MagicMock()
mock_client.create_glossary.side_effect = Exception("quota exceeded")
backend._client = mock_client
entries = [{"source_term": "hello", "target_term": "hallo"}]
result = backend._get_or_create_glossary("EN", "DE", entries)
assert result is None
class TestHealthCheck:
"""Tests for health_check method."""
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
def test_healthy_free_plan(self):
backend = _make_backend(api_key="test-key:fx")
mock_client = MagicMock()
mock_usage = MagicMock()
mock_usage.character.count = 1000
mock_usage.character.limit = 500000
mock_client.get_usage.return_value = mock_usage
backend._client = mock_client
healthy, msg = backend.health_check()
assert healthy is True
assert "Free" in msg
assert "1000/500000" in msg
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
def test_healthy_pro_plan(self):
backend = _make_backend(api_key="test-key-pro")
mock_client = MagicMock()
mock_usage = MagicMock()
mock_usage.character.count = 500
mock_usage.character.limit = 10000000
mock_client.get_usage.return_value = mock_usage
backend._client = mock_client
healthy, msg = backend.health_check()
assert healthy is True
assert "Pro" in msg
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
def test_unhealthy_on_exception(self):
backend = _make_backend()
mock_client = MagicMock()
mock_client.get_usage.side_effect = Exception("connection refused")
backend._client = mock_client
healthy, msg = backend.health_check()
assert healthy is False
assert "connection refused" in msg
class TestGetUsage:
"""Tests for get_usage method."""
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
def test_returns_usage_dict(self):
backend = _make_backend(api_key="test-key:fx")
mock_client = MagicMock()
mock_usage = MagicMock()
mock_usage.character.count = 2500
mock_usage.character.limit = 500000
mock_client.get_usage.return_value = mock_usage
backend._client = mock_client
usage = backend.get_usage()
assert usage["characters_used"] == 2500
assert usage["characters_limit"] == 500000
assert usage["plan"] == "Free"
@patch("translation.deepl_backend._DEEPL_AVAILABLE", True)
def test_returns_empty_dict_on_error(self):
backend = _make_backend()
mock_client = MagicMock()
mock_client.get_usage.side_effect = Exception("failed")
backend._client = mock_client
usage = backend.get_usage()
assert usage == {}