bvandeusen
7f8073c4c8
services/interpreter_client.py: sync (requests) client for the LibreTranslate- compatible /v1/translate — no new dep, mirrors the platform clients. translate() maps translatedText[]↔texts (order + length), returns detected_lang + engine_version (aggregate = first item, fine for a per-post [title, description] batch); passthrough items come back unchanged in their slot. InterpreterUnavailable on 503 / connection error (retry later), InterpreterBadRequest on 400. health() checks /v1/health engines.llm. 10 unit tests with mocked HTTP. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM
109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
"""Interpreter translation client (#143) — unit tests with mocked HTTP (no DB,
|
|
no real network)."""
|
|
import pytest
|
|
|
|
from backend.app.services import interpreter_client as ic
|
|
|
|
|
|
class _Resp:
|
|
def __init__(self, status_code=200, payload=None):
|
|
self.status_code = status_code
|
|
self._payload = payload or {}
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
def raise_for_status(self):
|
|
if self.status_code >= 400:
|
|
raise ic.requests.HTTPError(f"status {self.status_code}")
|
|
|
|
|
|
def test_translate_maps_batch_in_order(monkeypatch):
|
|
captured = {}
|
|
|
|
def fake_post(url, json, timeout):
|
|
captured["json"] = json
|
|
return _Resp(200, {
|
|
"translatedText": ["The cat is cute", "Blonde gal!"],
|
|
"detectedLanguage": {"language": "ja", "confidence": 0.98},
|
|
"interpreter": {"engine": "llm", "engine_version": "ollama:x:12b"},
|
|
})
|
|
|
|
monkeypatch.setattr(ic.requests, "post", fake_post)
|
|
out = ic.translate(["ねこが可愛い", "金髪ギャル！"], base_url="http://i.lan")
|
|
assert out["translations"] == ["The cat is cute", "Blonde gal!"]
|
|
assert out["detected_lang"] == "ja"
|
|
assert out["engine_version"] == "ollama:x:12b"
|
|
assert captured["json"]["q"] == ["ねこが可愛い", "金髪ギャル！"]
|
|
assert captured["json"]["target"] == "en"
|
|
|
|
|
|
def test_translate_passthrough_unchanged(monkeypatch):
|
|
# Already-English items come back unchanged in their slot (engine "none").
|
|
monkeypatch.setattr(ic.requests, "post", lambda *a, **k: _Resp(200, {
|
|
"translatedText": ["already english"],
|
|
"detectedLanguage": {"language": "en"},
|
|
"interpreter": {"engine": "none", "engine_version": None},
|
|
}))
|
|
out = ic.translate(["already english"], base_url="http://i.lan")
|
|
assert out["translations"] == ["already english"]
|
|
assert out["detected_lang"] == "en"
|
|
assert out["engine"] == "none"
|
|
|
|
|
|
def test_translate_503_raises_unavailable(monkeypatch):
|
|
monkeypatch.setattr(ic.requests, "post", lambda *a, **k: _Resp(503, {}))
|
|
with pytest.raises(ic.InterpreterUnavailable):
|
|
ic.translate(["x"], base_url="http://i.lan")
|
|
|
|
|
|
def test_translate_connection_error_raises_unavailable(monkeypatch):
|
|
def boom(*a, **k):
|
|
raise ic.requests.ConnectionError("refused")
|
|
|
|
monkeypatch.setattr(ic.requests, "post", boom)
|
|
with pytest.raises(ic.InterpreterUnavailable):
|
|
ic.translate(["x"], base_url="http://i.lan")
|
|
|
|
|
|
def test_translate_400_raises_bad_request(monkeypatch):
|
|
monkeypatch.setattr(
|
|
ic.requests, "post", lambda *a, **k: _Resp(400, {"error": "bad target"})
|
|
)
|
|
with pytest.raises(ic.InterpreterBadRequest):
|
|
ic.translate(["x"], base_url="http://i.lan")
|
|
|
|
|
|
def test_translate_empty_is_noop(monkeypatch):
|
|
def boom(*a, **k):
|
|
raise AssertionError("should not call the service for an empty batch")
|
|
|
|
monkeypatch.setattr(ic.requests, "post", boom)
|
|
assert ic.translate([], base_url="http://i.lan")["translations"] == []
|
|
|
|
|
|
def test_health_true_when_llm_up(monkeypatch):
|
|
monkeypatch.setattr(
|
|
ic.requests, "get", lambda *a, **k: _Resp(200, {"engines": {"llm": True}})
|
|
)
|
|
assert ic.health("http://i.lan") is True
|
|
|
|
|
|
def test_health_false_when_engine_down(monkeypatch):
|
|
monkeypatch.setattr(
|
|
ic.requests, "get", lambda *a, **k: _Resp(200, {"engines": {"llm": False}})
|
|
)
|
|
assert ic.health("http://i.lan") is False
|
|
|
|
|
|
def test_health_false_on_error(monkeypatch):
|
|
def boom(*a, **k):
|
|
raise ic.requests.ConnectionError("refused")
|
|
|
|
monkeypatch.setattr(ic.requests, "get", boom)
|
|
assert ic.health("http://i.lan") is False
|
|
|
|
|
|
def test_health_false_when_url_empty():
|
|
assert ic.health("") is False
|