ss-tools/backend/tests/plugins/translate/test_batch_insert_coverage.py

# #region Test.BatchInsert.Coverage [C:3] [TYPE Module] [SEMANTICS test,translate,batch,insert,coverage]
# @BRIEF Coverage extension for _batch_insert.py — edge cases not covered in test_batch_insert.py.
# @RELATION BINDS_TO -> [BatchInsertService]
# @TEST_EDGE: missing_key_in_source_data -> base_row[k] = None
# @TEST_EDGE: empty_columns_fallback -> Uses effective_target as single column
# @TEST_EDGE: sql_chunk_is_none -> Skipped in chunked insert loop

import json
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src"))

import pytest
from unittest.mock import AsyncMock, MagicMock, patch

from src.models.translate import TranslationJob, TranslationRecord, TranslationLanguage
from src.plugins.translate._batch_insert import (
    _build_insert_rows,
    insert_batch_to_target,
)
from .conftest import JOB_ID


class TestBuildInsertRowsEdgeCases:
    """_build_insert_rows — edge cases for coverage."""

    def _make_record(self, source_sql="hello", target_sql="bonjour",
                     source_data=None, languages=None):
        rec = MagicMock(spec=TranslationRecord)
        rec.source_sql = source_sql
        rec.target_sql = target_sql
        rec.source_data = source_data or {}
        rec.languages = languages or []
        rec.status = "SUCCESS"
        return rec

    def test_missing_key_in_source_data(self):
        """When target_key_cols key is missing from source_data -> base_row[k] = None (line 158)."""
        lang = MagicMock(spec=TranslationLanguage)
        lang.language_code = "fr"
        lang.final_value = "bonjour"
        lang.translated_value = "bonjour"
        lang.source_language_detected = "en"

        # source_data has 'id' but NOT 'created_at'
        rec = self._make_record(
            source_sql="hello", target_sql="bonjour",
            source_data={"id": 1}, languages=[lang],
        )
        job = MagicMock(spec=TranslationJob)
        job.target_key_cols = ["id", "created_at"]
        job.target_source_column = None
        job.target_source_language_column = None
        job.target_language_column = "lang"
        job.context_columns = []

        rows = _build_insert_rows([rec], job, "translated_text", "en", [])
        assert len(rows) == 2
        # 'id' present in source_data -> value should be 1
        assert rows[0].get("id") == 1
        # 'created_at' missing -> should be None
        assert rows[0].get("created_at") is None
        assert rows[1].get("created_at") is None

    def test_primary_language_column_with_target_language(self):
        """When target_language_column is set and languages exist, translation row gets correct lang."""
        lang = MagicMock(spec=TranslationLanguage)
        lang.language_code = "de"
        lang.final_value = "hallo"
        lang.translated_value = "hallo"
        lang.source_language_detected = "en"

        rec = self._make_record(
            source_sql="hello", target_sql="hallo",
            source_data={"id": 1}, languages=[lang],
        )
        job = MagicMock(spec=TranslationJob)
        job.target_key_cols = ["id"]
        job.target_source_column = "source"
        job.target_source_language_column = "src_lang"
        job.target_language_column = "lang"
        job.context_columns = []

        rows = _build_insert_rows([rec], job, "translated_text", "en", [])
        assert len(rows) == 2
        assert rows[0]["is_original"] == 1
        assert rows[0]["lang"] == "en"  # detected source lang
        assert rows[1]["is_original"] == 0
        assert rows[1]["lang"] == "de"  # translation lang

    def test_multiple_languages_with_skip(self):
        """Multiple languages: skip language matching detected source."""
        en_lang = MagicMock(spec=TranslationLanguage)
        en_lang.language_code = "en"
        en_lang.final_value = "hello"
        en_lang.translated_value = "hello"
        en_lang.source_language_detected = "en"

        fr_lang = MagicMock(spec=TranslationLanguage)
        fr_lang.language_code = "fr"
        fr_lang.final_value = "bonjour"
        fr_lang.translated_value = "bonjour"
        fr_lang.source_language_detected = "en"

        de_lang = MagicMock(spec=TranslationLanguage)
        de_lang.language_code = "de"
        de_lang.final_value = "hallo"
        de_lang.translated_value = "hallo"
        de_lang.source_language_detected = "en"

        rec = self._make_record(
            source_sql="hello", target_sql="bonjour",
            source_data={"id": 1}, languages=[en_lang, fr_lang, de_lang],
        )
        job = MagicMock(spec=TranslationJob)
        job.target_key_cols = []
        job.target_source_column = None
        job.target_source_language_column = None
        job.target_language_column = None
        job.context_columns = []

        rows = _build_insert_rows([rec], job, "translated_text", "en", [])
        # Original + 2 translations (en skipped, fr + de kept)
        assert len(rows) == 3
        assert rows[0]["is_original"] == 1
        assert rows[1]["is_original"] == 0
        assert rows[2]["is_original"] == 0
        # The first non-original should be fr (first in order after en skip)
        assert rows[1]["translated_text"] == "bonjour"
        assert rows[2]["translated_text"] == "hallo"


class TestInsertBatchToTargetEdgeCases:
    """insert_batch_to_target — edge case coverage for lines 47-48, 68."""

    @pytest.mark.asyncio
    async def test_empty_columns_fallback(self, db_session):
        """When _build_target_columns returns empty, falls back to single column (lines 47-48)."""
        config_manager = MagicMock()
        job = MagicMock(spec=TranslationJob)
        job.target_column = None
        job.translation_column = None
        job.target_languages = None
        job.target_schema = None
        job.target_table = "translated_data"
        job.target_key_cols = []
        job.upsert_strategy = "MERGE"
        job.environment_id = None
        job.source_dialect = None
        job.target_database_id = None
        job.database_dialect = "postgresql"
        job.target_dialect = None

        rec = MagicMock(spec=TranslationRecord)
        rec.target_sql = "bonjour"

        # Force _build_target_columns to return empty, _build_insert_rows to return non-empty
        from src.plugins.translate._batch_insert import _build_target_columns
        original_build_columns = _build_target_columns

        with patch("src.plugins.translate._batch_insert._build_target_columns", return_value=[]):
            with patch("src.plugins.translate._batch_insert._build_context_keys", return_value=[]):
                with patch("src.plugins.translate._batch_insert._build_insert_rows",
                           return_value=[{"text": "bonjour"}]):
                    with patch("src.plugins.translate._batch_insert._resolve_insert_backend",
                               return_value=("postgresql", AsyncMock(), None)):
                        with patch("src.plugins.translate._batch_insert.SQLGenerator.generate_batch",
                                   return_value=[("INSERT INTO ...", 1)]):
                            with patch("src.plugins.translate._batch_insert._execute_insert_sql",
                                       AsyncMock()):
                                with patch("src.plugins.translate._batch_insert._fetch_batch_records",
                                           return_value=[rec]):
                                    result = await insert_batch_to_target(
                                        db_session, config_manager, job, "batch-1", "run-1"
                                    )
                                    assert result is None

    @pytest.mark.asyncio
    async def test_sql_chunk_is_none_skipped(self, db_session):
        """When a SQL chunk has sql=None, it is skipped (line 68)."""
        config_manager = MagicMock()
        job = MagicMock(spec=TranslationJob)
        job.target_column = "text"
        job.translation_column = None
        job.target_languages = None
        job.target_schema = None
        job.target_table = "translated_data"
        job.target_key_cols = []
        job.upsert_strategy = "MERGE"
        job.environment_id = None
        job.source_dialect = None
        job.target_database_id = None
        job.database_dialect = "postgresql"
        job.target_dialect = None

        rec = MagicMock(spec=TranslationRecord)
        rec.target_sql = "bonjour"

        # generate_batch returns [ (None, 0), ("INSERT ...", 1) ]
        # The None chunk should be skipped via continue, the second should execute
        mock_executor = AsyncMock()
        mock_executor.execute_and_poll.return_value = {"status": "success"}

        with patch("src.plugins.translate._batch_insert._fetch_batch_records", return_value=[rec]):
            with patch("src.plugins.translate._batch_insert._build_target_columns",
                       return_value=["text", "is_original"]):
                with patch("src.plugins.translate._batch_insert._build_context_keys", return_value=[]):
                    with patch("src.plugins.translate._batch_insert._build_insert_rows",
                               return_value=[{"text": "bonjour", "is_original": 1}]):
                        with patch("src.plugins.translate._batch_insert._resolve_insert_backend",
                                   return_value=("postgresql", mock_executor, None)):
                            with patch("src.plugins.translate._batch_insert.SQLGenerator.generate_batch",
                                       return_value=[(None, 0), ("INSERT INTO target ...", 1)]):
                                result = await insert_batch_to_target(
                                    db_session, config_manager, job, "batch-1", "run-1"
                                )
                                assert result is None
                                # execute_and_poll should be called only once (second chunk)
                                mock_executor.execute_and_poll.assert_called_once()
# #endregion Test.BatchInsert.Coverage