sepehr
d40d7f3e86
All checks were successful
Deploy to Production / Build and Deploy (push) Successful in 3m22s
333 lines
13 KiB
Python
333 lines
13 KiB
Python
"""
|
|
Tests for Track C1.1 — NewProviderAsLegacyAdapter and LegacyProviderAsNewAdapter.
|
|
|
|
Covers:
|
|
- NewProviderAsLegacyAdapter:
|
|
* translate() delegates to translate_text()
|
|
* translate_batch() delegates to translate_batch()
|
|
* get_name() and is_available() pass through
|
|
* errors propagate as exceptions (legacy code expects this)
|
|
- LegacyProviderAsNewAdapter:
|
|
* translate_text() wraps legacy.translate()
|
|
* translate_batch() groups by language pair
|
|
* get_name() and is_available() pass through
|
|
- coerce_to_legacy / coerce_to_new:
|
|
* no-op when already in target shape
|
|
* wraps when in the other shape
|
|
"""
|
|
import pytest
|
|
from typing import List
|
|
from unittest.mock import MagicMock
|
|
|
|
from services.providers.base import TranslationProvider as NewTranslationProvider
|
|
from services.providers.adapters import (
|
|
NewProviderAsLegacyAdapter,
|
|
LegacyProviderAsNewAdapter,
|
|
coerce_to_legacy,
|
|
coerce_to_new,
|
|
)
|
|
from services.providers.schemas import (
|
|
TranslationRequest,
|
|
TranslationResponse,
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# Test helpers — new-style mock
|
|
# ============================================================================
|
|
|
|
class MockNewProvider(NewTranslationProvider):
|
|
"""Minimal new-style provider for testing the adapter."""
|
|
|
|
def __init__(self, translations: dict = None, name: str = "mock-new"):
|
|
self._t = translations or {}
|
|
self._name = name
|
|
self.translate_text_calls = []
|
|
self.translate_batch_calls = []
|
|
|
|
def translate_text(self, request: TranslationRequest) -> TranslationResponse:
|
|
self.translate_text_calls.append(request)
|
|
translated = self._t.get(request.text, f"NEW_{request.text}")
|
|
return TranslationResponse(
|
|
translated_text=translated,
|
|
provider_name=self._name,
|
|
source_language=request.source_language or "auto",
|
|
target_language=request.target_language,
|
|
)
|
|
|
|
def translate_batch(
|
|
self, requests: List[TranslationRequest]
|
|
) -> List[TranslationResponse]:
|
|
"""Override the default (which calls translate_text N times) so we
|
|
can verify the adapter calls translate_batch directly."""
|
|
self.translate_batch_calls.append(requests)
|
|
return [self.translate_text(req) for req in requests]
|
|
|
|
def get_name(self) -> str:
|
|
return self._name
|
|
|
|
def is_available(self) -> bool:
|
|
return True
|
|
|
|
|
|
# ============================================================================
|
|
# Test helpers — legacy mock
|
|
# ============================================================================
|
|
|
|
class MockLegacyProvider:
|
|
"""Minimal legacy provider for testing the adapter."""
|
|
|
|
def __init__(self, translations: dict = None, name: str = "mock-legacy"):
|
|
self._t = translations or {}
|
|
self._name = name
|
|
self.calls = []
|
|
|
|
def translate(self, text, target_language, source_language="auto"):
|
|
self.calls.append((text, target_language, source_language))
|
|
return self._t.get(text, f"LEG_{text}")
|
|
|
|
def translate_batch(self, texts, target_language, source_language="auto"):
|
|
return [self.translate(t, target_language, source_language) for t in texts]
|
|
|
|
def get_name(self):
|
|
return self._name
|
|
|
|
def is_available(self):
|
|
return True
|
|
|
|
|
|
# ============================================================================
|
|
# NewProviderAsLegacyAdapter
|
|
# ============================================================================
|
|
|
|
class TestNewProviderAsLegacyAdapter:
|
|
def test_translate_delegates(self):
|
|
new = MockNewProvider({"Hello": "Bonjour"})
|
|
legacy = NewProviderAsLegacyAdapter(new)
|
|
|
|
result = legacy.translate("Hello", "fr", "en")
|
|
assert result == "Bonjour"
|
|
assert len(new.translate_text_calls) == 1
|
|
assert new.translate_text_calls[0].text == "Hello"
|
|
assert new.translate_text_calls[0].target_language == "fr"
|
|
|
|
def test_translate_default_source(self):
|
|
new = MockNewProvider()
|
|
legacy = NewProviderAsLegacyAdapter(new)
|
|
legacy.translate("Hello", "fr")
|
|
assert new.translate_text_calls[0].source_language == "auto"
|
|
|
|
def test_translate_batch_delegates(self):
|
|
new = MockNewProvider({"A": "TR_A", "B": "TR_B"})
|
|
legacy = NewProviderAsLegacyAdapter(new)
|
|
result = legacy.translate_batch(["A", "B"], "fr")
|
|
assert result == ["TR_A", "TR_B"]
|
|
assert len(new.translate_batch_calls) == 1
|
|
|
|
def test_translate_batch_empty(self):
|
|
new = MockNewProvider()
|
|
legacy = NewProviderAsLegacyAdapter(new)
|
|
result = legacy.translate_batch([], "fr")
|
|
assert result == []
|
|
|
|
def test_get_name_passthrough(self):
|
|
new = MockNewProvider(name="my-new")
|
|
legacy = NewProviderAsLegacyAdapter(new)
|
|
assert legacy.get_name() == "my-new"
|
|
|
|
def test_is_available_passthrough(self):
|
|
new = MockNewProvider()
|
|
legacy = NewProviderAsLegacyAdapter(new)
|
|
assert legacy.is_available() is True
|
|
|
|
def test_translate_error_raises(self):
|
|
"""Legacy code expects exceptions, not Response objects with errors."""
|
|
|
|
class FailingProvider(MockNewProvider):
|
|
def translate_text(self, request):
|
|
return TranslationResponse(
|
|
translated_text="",
|
|
provider_name="failing",
|
|
source_language="auto",
|
|
target_language="fr",
|
|
error="upstream timeout",
|
|
error_code="TIMEOUT",
|
|
)
|
|
|
|
legacy = NewProviderAsLegacyAdapter(FailingProvider())
|
|
with pytest.raises(Exception, match="TIMEOUT"):
|
|
legacy.translate("Hello", "fr")
|
|
|
|
def test_health_check(self):
|
|
new = MockNewProvider()
|
|
legacy = NewProviderAsLegacyAdapter(new)
|
|
status = legacy.health_check()
|
|
assert status["name"] == "mock-new"
|
|
assert status["available"] is True
|
|
|
|
|
|
# ============================================================================
|
|
# LegacyProviderAsNewAdapter
|
|
# ============================================================================
|
|
|
|
class TestLegacyProviderAsNewAdapter:
|
|
def test_translate_text_wraps(self):
|
|
legacy = MockLegacyProvider({"Hello": "Bonjour"})
|
|
adapter = LegacyProviderAsNewAdapter(legacy)
|
|
|
|
result = adapter.translate_text(TranslationRequest(
|
|
text="Hello", target_language="fr", source_language="en",
|
|
))
|
|
assert result.translated_text == "Bonjour"
|
|
assert result.provider_name == "mock-legacy"
|
|
assert not result.error
|
|
assert len(legacy.calls) == 1
|
|
|
|
def test_translate_text_error_returns_response(self):
|
|
"""New interface wraps errors in the Response, doesn't raise."""
|
|
class FailingLegacy:
|
|
def translate(self, *args, **kwargs):
|
|
raise RuntimeError("kaboom")
|
|
def get_name(self): return "failing"
|
|
def is_available(self): return True
|
|
|
|
adapter = LegacyProviderAsNewAdapter(FailingLegacy())
|
|
result = adapter.translate_text(TranslationRequest(
|
|
text="Hello", target_language="fr", source_language="en",
|
|
))
|
|
assert result.translated_text == ""
|
|
assert "kaboom" in result.error
|
|
assert result.error_code == "LEGACY_PROVIDER_ERROR"
|
|
|
|
def test_translate_batch_single_group(self):
|
|
legacy = MockLegacyProvider({"A": "TR_A", "B": "TR_B"})
|
|
adapter = LegacyProviderAsNewAdapter(legacy)
|
|
|
|
results = adapter.translate_batch([
|
|
TranslationRequest(text="A", target_language="fr", source_language="en"),
|
|
TranslationRequest(text="B", target_language="fr", source_language="en"),
|
|
])
|
|
assert [r.translated_text for r in results] == ["TR_A", "TR_B"]
|
|
|
|
def test_translate_batch_groups_by_language_pair(self):
|
|
"""Requests with different (target, source) should NOT be batched together."""
|
|
legacy = MockLegacyProvider()
|
|
adapter = LegacyProviderAsNewAdapter(legacy)
|
|
|
|
adapter.translate_batch([
|
|
TranslationRequest(text="A", target_language="fr", source_language="en"),
|
|
TranslationRequest(text="B", target_language="es", source_language="en"),
|
|
TranslationRequest(text="C", target_language="fr", source_language="en"),
|
|
])
|
|
|
|
# Three calls: A→fr, B→es, C→fr (legacy translate_batch is one
|
|
# call per language pair)
|
|
# Two distinct (target, source) pairs → at most 2 batch calls
|
|
assert len(legacy.calls) >= 2
|
|
|
|
def test_translate_batch_empty(self):
|
|
legacy = MockLegacyProvider()
|
|
adapter = LegacyProviderAsNewAdapter(legacy)
|
|
results = adapter.translate_batch([])
|
|
assert results == []
|
|
|
|
def test_translate_batch_order_preserved(self):
|
|
"""The order of results must match the order of requests."""
|
|
legacy = MockLegacyProvider({"X": "TR_X", "Y": "TR_Y", "Z": "TR_Z"})
|
|
adapter = LegacyProviderAsNewAdapter(legacy)
|
|
|
|
results = adapter.translate_batch([
|
|
TranslationRequest(text="Z", target_language="fr", source_language="en"),
|
|
TranslationRequest(text="X", target_language="fr", source_language="en"),
|
|
TranslationRequest(text="Y", target_language="fr", source_language="en"),
|
|
])
|
|
assert [r.translated_text for r in results] == ["TR_Z", "TR_X", "TR_Y"]
|
|
|
|
def test_get_name_passthrough(self):
|
|
legacy = MockLegacyProvider(name="my-legacy")
|
|
adapter = LegacyProviderAsNewAdapter(legacy)
|
|
assert adapter.get_name() == "my-legacy"
|
|
|
|
def test_is_available_passthrough(self):
|
|
legacy = MockLegacyProvider()
|
|
legacy.is_available = lambda: False
|
|
adapter = LegacyProviderAsNewAdapter(legacy)
|
|
assert adapter.is_available() is False
|
|
|
|
def test_custom_provider_name_overrides(self):
|
|
legacy = MockLegacyProvider(name="original-name")
|
|
adapter = LegacyProviderAsNewAdapter(legacy, provider_name="override-name")
|
|
assert adapter.get_name() == "override-name"
|
|
|
|
|
|
# ============================================================================
|
|
# Coercion helpers
|
|
# ============================================================================
|
|
|
|
class TestCoercion:
|
|
def test_new_already_legacy_noop(self):
|
|
new = MockNewProvider()
|
|
adapter = NewProviderAsLegacyAdapter(new)
|
|
result = coerce_to_legacy(adapter)
|
|
# Should return the existing adapter, not wrap it again
|
|
assert result is adapter
|
|
|
|
def test_new_provider_to_legacy_wraps(self):
|
|
new = MockNewProvider()
|
|
result = coerce_to_legacy(new)
|
|
assert isinstance(result, NewProviderAsLegacyAdapter)
|
|
assert result._new is new
|
|
|
|
def test_legacy_provider_unchanged(self):
|
|
legacy = MockLegacyProvider()
|
|
result = coerce_to_legacy(legacy)
|
|
assert result is legacy
|
|
|
|
def test_legacy_already_new_noop(self):
|
|
legacy = MockLegacyProvider()
|
|
adapter = LegacyProviderAsNewAdapter(legacy)
|
|
result = coerce_to_new(adapter)
|
|
assert result is adapter
|
|
|
|
def test_legacy_provider_to_new_wraps(self):
|
|
legacy = MockLegacyProvider()
|
|
result = coerce_to_new(legacy)
|
|
assert isinstance(result, NewTranslationProvider)
|
|
assert isinstance(result, LegacyProviderAsNewAdapter)
|
|
|
|
def test_new_provider_unchanged(self):
|
|
new = MockNewProvider()
|
|
result = coerce_to_new(new)
|
|
assert result is new
|
|
|
|
|
|
# ============================================================================
|
|
# Round-trip
|
|
# ============================================================================
|
|
|
|
class TestRoundTrip:
|
|
"""An adapter should be reversible: a string that comes out of one
|
|
side should match what the other side would have returned."""
|
|
|
|
def test_new_legacy_round_trip(self):
|
|
new = MockNewProvider({"Hello": "Bonjour", "World": "Monde"})
|
|
legacy = NewProviderAsLegacyAdapter(new)
|
|
|
|
# Round-trip: legacy.translate → new.translate_text
|
|
via_legacy = legacy.translate("Hello", "fr", "en")
|
|
via_new = new.translate_text(TranslationRequest(
|
|
text="Hello", target_language="fr", source_language="en",
|
|
)).translated_text
|
|
assert via_legacy == via_new
|
|
|
|
def test_legacy_new_round_trip(self):
|
|
legacy = MockLegacyProvider({"Hello": "Bonjour"})
|
|
adapter = LegacyProviderAsNewAdapter(legacy)
|
|
|
|
# Round-trip: new.translate_text → legacy.translate
|
|
via_new = adapter.translate_text(TranslationRequest(
|
|
text="Hello", target_language="fr", source_language="en",
|
|
)).translated_text
|
|
via_legacy = legacy.translate("Hello", "fr", "en")
|
|
assert via_new == via_legacy
|