187 lines
8.2 KiB
Python
187 lines
8.2 KiB
Python
# #region Test.LLMTranslationService.Extra [C:3] [TYPE Module] [SEMANTICS test, llm, translation, coverage]
|
|
# @BRIEF Additional coverage for LLMTranslationService edge cases — remaining uncovered paths.
|
|
# @RELATION BINDS_TO -> [LLMTranslationService]
|
|
# @TEST_EDGE: undetected_language_triggers_review -> Review flag set on TranslationLanguage
|
|
# @TEST_EDGE: recovery_with_und_detected_lang -> Fallback to LLM response field
|
|
# @TEST_EDGE: empty_source_text_cache_check -> continue on empty source in _check_cache
|
|
# @TEST_EDGE: provider_token_config_error -> Graceful fallback on config error
|
|
# @TEST_EDGE: backward_compat_delegation -> call_openai_compatible / _parse_llm_response wrappers
|
|
import json
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from src.models.translate import TranslationLanguage, TranslationRecord
|
|
from src.plugins.translate._llm_call import LLMTranslationService, MAX_RETRIES_PER_BATCH
|
|
|
|
from .conftest import JOB_ID, RUN_ID
|
|
|
|
|
|
def _make_job(session, provider_id="test-provider", target_langs=None):
|
|
from src.models.translate import TranslationJob
|
|
job = session.query(TranslationJob).filter(TranslationJob.id == JOB_ID).first()
|
|
if job is None:
|
|
job = TranslationJob(
|
|
id=JOB_ID, name="LLM Extra Test", source_dialect="en",
|
|
target_dialect=target_langs[0] if target_langs else "fr",
|
|
target_languages=target_langs or ["fr"],
|
|
status="ACTIVE", provider_id=provider_id,
|
|
)
|
|
session.add(job)
|
|
else:
|
|
if provider_id:
|
|
job.provider_id = provider_id
|
|
if target_langs:
|
|
job.target_languages = target_langs
|
|
job.target_dialect = target_langs[0]
|
|
session.commit()
|
|
return job
|
|
|
|
|
|
def _batch_rows(count=2, detected_lang="en"):
|
|
return [
|
|
{
|
|
"row_index": str(i),
|
|
"source_text": f"Hello world {i}",
|
|
"source_data": {"table": "test"},
|
|
"_detected_lang": detected_lang,
|
|
}
|
|
for i in range(count)
|
|
]
|
|
|
|
|
|
class TestCreateRecordsUndetectedLang:
|
|
"""Cover _create_records_from_translations with undetected language (needs_review=True)."""
|
|
|
|
def test_undetected_lang_triggers_needs_review(self, db_with_run):
|
|
session, run_id = db_with_run
|
|
svc = LLMTranslationService(session)
|
|
rows = _batch_rows(1, detected_lang="und")
|
|
translations = {"0": {"fr": "Bonjour", "translation": "Bonjour"}}
|
|
result = svc._create_records_from_translations(
|
|
rows, run_id, "batch-1", ["fr"], translations, [], 0,
|
|
)
|
|
assert result["successful"] == 1
|
|
langs = session.query(TranslationLanguage).all()
|
|
assert len(langs) == 1
|
|
assert langs[0].needs_review is True
|
|
|
|
def test_detected_lang_same_as_target_lang_skipped(self, db_with_run):
|
|
"""When detected_lang matches target language code, skip that language row."""
|
|
session, run_id = db_with_run
|
|
svc = LLMTranslationService(session)
|
|
rows = _batch_rows(1, detected_lang="fr")
|
|
translations = {"0": {"fr": "Bonjour", "translation": "Bonjour", "de": "Hallo"}}
|
|
result = svc._create_records_from_translations(
|
|
rows, run_id, "batch-1", ["fr", "de"], translations, [], 0,
|
|
)
|
|
assert result["successful"] == 1
|
|
langs = session.query(TranslationLanguage).all()
|
|
lang_codes = [l.language_code for l in langs]
|
|
assert "fr" not in lang_codes # skipped because detected_lang == "fr"
|
|
assert "de" in lang_codes
|
|
|
|
def test_detected_lang_fallback_to_llm_response(self, db_with_run):
|
|
"""When _detected_lang is 'und', fall back to LLM response field."""
|
|
session, run_id = db_with_run
|
|
svc = LLMTranslationService(session)
|
|
rows = _batch_rows(1, detected_lang="und")
|
|
translations = {
|
|
"0": {
|
|
"fr": "Bonjour",
|
|
"translation": "Bonjour",
|
|
"detected_source_language": "en",
|
|
}
|
|
}
|
|
result = svc._create_records_from_translations(
|
|
rows, run_id, "batch-1", ["fr"], translations, [], 0,
|
|
)
|
|
assert result["successful"] == 1
|
|
langs = session.query(TranslationLanguage).all()
|
|
assert langs[0].source_language_detected == "en"
|
|
|
|
|
|
class TestRecoveryUndetectedLang:
|
|
"""Cover _try_recover_partial with undetected language paths."""
|
|
|
|
def test_recovery_with_und_detected_lang(self, db_with_run):
|
|
"""Recovery with _detected_lang='und' falls back to LLM field."""
|
|
session, run_id = db_with_run
|
|
svc = LLMTranslationService(session)
|
|
rows = _batch_rows(1, detected_lang="und")
|
|
translations_response = json.dumps([{"row_id": "0", "fr": "Bonjour"}])
|
|
|
|
with patch('src.plugins.translate._llm_call.parse_llm_response',
|
|
return_value={"0": {"fr": "Bonjour", "detected_source_language": "fr"}}):
|
|
recovered = svc._try_recover_partial(
|
|
translations_response, rows, run_id, "batch-1", ["fr"],
|
|
)
|
|
assert recovered == {"0"}
|
|
# When detected_lang="und" and response has detected_source_language="fr",
|
|
# detected_lang is set to "fr" — but target_languages=["fr"] so no
|
|
# TranslationLanguage is created (same-lang skip). Only TranslationRecord exists.
|
|
records = session.query(TranslationRecord).all()
|
|
assert len(records) == 1
|
|
|
|
def test_recovery_needs_review(self, db_with_run):
|
|
"""Recovery with undetected lang sets needs_review=True."""
|
|
session, run_id = db_with_run
|
|
svc = LLMTranslationService(session)
|
|
rows = _batch_rows(1, detected_lang="und")
|
|
translations_response = json.dumps([{"row_id": "0", "fr": "Bonjour"}])
|
|
|
|
with patch('src.plugins.translate._llm_call.parse_llm_response',
|
|
return_value={"0": {"fr": "Bonjour", "detected_source_language": "und"}}):
|
|
recovered = svc._try_recover_partial(
|
|
translations_response, rows, run_id, "batch-1", ["fr"],
|
|
)
|
|
assert recovered == {"0"}
|
|
langs = session.query(TranslationLanguage).all()
|
|
assert langs[0].needs_review is True
|
|
|
|
|
|
class TestBackwardCompatDelegation:
|
|
"""Cover backward-compatibility static wrapper methods."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_call_openai_compatible_delegates(self):
|
|
with patch('src.plugins.translate._llm_call.call_openai_compatible',
|
|
new=AsyncMock(return_value=("response", "stop"))):
|
|
result = await LLMTranslationService.call_openai_compatible(
|
|
"url", "key", "model", "prompt"
|
|
)
|
|
assert result == ("response", "stop")
|
|
|
|
def test_parse_llm_response_delegates(self):
|
|
with patch('src.plugins.translate._llm_call.parse_llm_response',
|
|
return_value={"0": {"fr": "test"}}):
|
|
result = LLMTranslationService._parse_llm_response(
|
|
'{"fr": "test"}', 1, ["fr"]
|
|
)
|
|
assert result == {"0": {"fr": "test"}}
|
|
|
|
|
|
class TestCreateRecordsEnforceDict:
|
|
"""Cover dictionary enforcement with dict_matches."""
|
|
|
|
def test_dict_enforcement_called(self, db_with_run):
|
|
"""_enforce_dictionary is called when dict_matches and source_text exist."""
|
|
session, run_id = db_with_run
|
|
svc = LLMTranslationService(session)
|
|
rows = _batch_rows(1, detected_lang="und")
|
|
translations = {"0": {"fr": "Bonjour", "translation": "Bonjour"}}
|
|
dict_matches = [{"entry_id": "e1", "source_term": "hello", "target_term": "hola"}]
|
|
|
|
with patch('src.plugins.translate._llm_call._enforce_dictionary') as mock_enf:
|
|
result = svc._create_records_from_translations(
|
|
rows, run_id, "batch-1", ["fr"], translations, dict_matches, 0,
|
|
)
|
|
assert result["successful"] == 1
|
|
mock_enf.assert_called_once()
|
|
|
|
def test_plv_fallback_to_translation_key(self, db_session):
|
|
"""When per-language values are empty but translation key exists."""
|
|
td = {"translation": "Bonjour le monde"}
|
|
result = LLMTranslationService._extract_per_lang_values(td, ["fr"])
|
|
assert result == {"fr": "Bonjour le monde"}
|
|
# #endregion Test.LLMTranslationService.Extra
|