67 lines
2.9 KiB
Python
67 lines
2.9 KiB
Python
# #region Test.Translate.RunPreflight [C:3] [TYPE Module] [SEMANTICS test,translate,preflight,lingua]
|
|
# @BRIEF Verify run preflight returns aggregate scope only and applies the Lingua SLA fallback.
|
|
# @RELATION BINDS_TO -> [TranslationRunPreflight]
|
|
# @TEST_EDGE: missing_job -> ValueError
|
|
# @TEST_EDGE: slow_lingua -> distribution omitted and skip recommended
|
|
# @TEST_EDGE: incremental_scope -> existing rows excluded before cost calculation
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from src.plugins.translate.run_preflight import TranslationRunPreflight
|
|
|
|
|
|
def _job():
|
|
return SimpleNamespace(
|
|
id="job-1", source_dialect="postgres", target_dialect="en",
|
|
source_datasource_id="ds-1", translation_column="text", context_columns=[],
|
|
target_languages=["en", "ru"], provider_id="provider-1", batch_size=50,
|
|
upsert_strategy="MERGE", source_key_cols=["id"], target_key_cols=["id"],
|
|
)
|
|
|
|
|
|
def _service(job=None):
|
|
db = MagicMock()
|
|
db.query.return_value.filter.return_value.first.return_value = job
|
|
return TranslationRunPreflight(db, MagicMock())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_preflight_returns_aggregate_scope_and_lingua_distribution():
|
|
service = _service(_job())
|
|
rows = [
|
|
{"source_text": "Hello", "source_data": {"id": 1}},
|
|
{"source_text": "Привет", "source_data": {"id": 2}},
|
|
]
|
|
with patch("src.plugins.translate.run_preflight.fetch_source_rows", new=AsyncMock(return_value=rows)), \
|
|
patch.object(service, "_detect_distribution", new=AsyncMock(return_value=({"en": 1, "ru": 1}, 10))), \
|
|
patch("src.plugins.translate.run_preflight.compute_config_hash", return_value="cfg-1"):
|
|
result = await service.calculate("job-1", full_translation=True)
|
|
|
|
assert result["total_source_rows"] == 2
|
|
assert result["eligible_rows"] == 2
|
|
assert result["language_distribution"] == {"en": 1, "ru": 1}
|
|
assert result["lingua_accepted"] is True
|
|
assert result["recommended_language_detection"] == "auto"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_preflight_hides_distribution_when_lingua_misses_sla():
|
|
service = _service(_job())
|
|
rows = [{"source_text": "Hello", "source_data": {"id": i}} for i in range(1000)]
|
|
with patch("src.plugins.translate.run_preflight.fetch_source_rows", new=AsyncMock(return_value=rows)), \
|
|
patch.object(service, "_detect_distribution", new=AsyncMock(return_value=({"en": 1}, 3_000))):
|
|
result = await service.calculate("job-1", full_translation=True)
|
|
|
|
assert result["language_distribution"] is None
|
|
assert result["lingua_accepted"] is False
|
|
assert result["recommended_language_detection"] == "skip"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_preflight_rejects_unknown_job():
|
|
with pytest.raises(ValueError, match="job-404"):
|
|
await _service(None).calculate("job-404", full_translation=False)
|