# #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