# #region Test.ServiceUtils [C:3] [TYPE Module] [SEMANTICS test,translate,utils,dialect,response,job] # @BRIEF Tests for service_utils.py — _extract_dialect, job_to_response. # @RELATION BINDS_TO -> [ServiceUtils] # @TEST_CONTRACT: _extract_dialect -> str | normalised dialect name # @TEST_CONTRACT: job_to_response -> TranslateJobResponse | field mapping import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src")) import os os.environ.setdefault("SECRET_KEY", "test-secret-key-for-testing") os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:") os.environ.setdefault("AUTH_DATABASE_URL", "sqlite:///:memory:") from datetime import datetime, timezone from unittest.mock import MagicMock import pytest from src.plugins.translate.service_utils import _extract_dialect, job_to_response from src.schemas.translate import TranslateJobResponse class TestExtractDialect: """_extract_dialect — URI parsing and dialect normalisation.""" def test_plain_engine_name(self): """Plain engine name returned as-is.""" assert _extract_dialect("postgresql") == "postgresql" def test_uri_format(self): """URI scheme extracted.""" assert _extract_dialect("postgresql://user:pass@host:5432/db") == "postgresql" def test_clickhousedb_normalised(self): """clickhousedb variant normalised to clickhouse.""" assert _extract_dialect("clickhousedb") == "clickhouse" def test_clickhousedb_uri(self): """clickhousedb:// URI normalised to clickhouse.""" assert _extract_dialect("clickhousedb://localhost:8123") == "clickhouse" def test_greenplum_normalised(self): """greenplum normalised to postgresql.""" assert _extract_dialect("greenplum") == "postgresql" def test_engine_plus_scheme(self): """Engine+driver scheme stripped.""" assert _extract_dialect("postgresql+psycopg2://host/db") == "postgresql" def test_empty_string(self): """Empty string returns unknown.""" assert _extract_dialect("") == "unknown" def test_whitespace_only(self): """Whitespace-only returns unknown.""" assert _extract_dialect(" ") == "unknown" def test_none_backend(self): """None returns unknown.""" assert _extract_dialect(None) == "unknown" def test_mysql(self): """MySQL engine name pass through.""" assert _extract_dialect("mysql") == "mysql" assert _extract_dialect("mysql://user@host/db") == "mysql" def test_case_insensitive(self): """Case is normalised to lower.""" assert _extract_dialect("PostgreSQL") == "postgresql" assert _extract_dialect("ClickHouseDB://host/db") == "clickhouse" def test_unknown_dialect(self): """Unknown dialect passes through as lowercased.""" assert _extract_dialect("sqlite") == "sqlite" def test_exception_fallback(self): """Exception during parsing falls back to 'unknown'.""" # Pass an object that has strip() but raises on split class _StrLike: def strip(self): return "valid" def split(self, sep=None, maxsplit=-1): raise TypeError("split failed") assert _extract_dialect(_StrLike()) == "unknown" class TestJobToResponse: """job_to_response — TranslationJob ORM -> TranslateJobResponse schema.""" def _make_job(self, overrides: dict | None = None) -> MagicMock: """Create a mock TranslationJob with required fields.""" now = datetime.now(timezone.utc) job = MagicMock() job.id = "job-1" job.name = "Test Job" job.description = "A test job" job.source_dialect = "postgresql" job.target_dialect = "clickhouse" job.database_dialect = "postgresql" job.source_datasource_id = "ds-1" job.source_table = "source_table" job.target_schema = "public" job.target_table = "target_table" job.source_key_cols = ["id"] job.target_key_cols = ["id"] job.translation_column = "name" job.target_column = "name_translated" job.target_language_column = "lang" job.target_source_column = "source_text" job.target_source_language_column = "source_lang" job.context_columns = ["desc"] job.target_languages = ["ru", "de"] job.provider_id = "provider-1" job.batch_size = 50 job.upsert_strategy = "MERGE" job.status = "READY" job.created_by = "user-1" job.created_at = now job.updated_at = now job.environment_id = "env-1" job.target_database_id = "db-1" job.insert_method = "sqllab" job.connection_id = None job.disable_reasoning = False if overrides: for k, v in overrides.items(): setattr(job, k, v) return job def test_basic_conversion(self): """All fields mapped correctly from job to response.""" job = self._make_job() response = job_to_response(job) assert isinstance(response, TranslateJobResponse) assert response.id == "job-1" assert response.name == "Test Job" assert response.description == "A test job" assert response.source_dialect == "postgresql" assert response.target_dialect == "clickhouse" assert response.database_dialect == "postgresql" assert response.source_datasource_id == "ds-1" assert response.source_table == "source_table" assert response.target_schema == "public" assert response.target_table == "target_table" assert response.source_key_cols == ["id"] assert response.target_key_cols == ["id"] assert response.translation_column == "name" assert response.target_column == "name_translated" assert response.target_language_column == "lang" assert response.target_source_column == "source_text" assert response.target_source_language_column == "source_lang" assert response.context_columns == ["desc"] assert response.target_languages == ["ru", "de"] assert response.provider_id == "provider-1" assert response.batch_size == 50 assert response.upsert_strategy == "MERGE" assert response.status == "READY" assert response.created_by == "user-1" assert response.dictionary_ids == [] assert response.environment_id == "env-1" assert response.target_database_id == "db-1" assert response.insert_method == "sqllab" assert response.connection_id is None assert response.disable_reasoning is False def test_with_dictionary_ids(self): """Dictionary IDs passed through.""" job = self._make_job() response = job_to_response(job, dict_ids=["dict-1", "dict-2"]) assert response.dictionary_ids == ["dict-1", "dict-2"] def test_none_key_cols_default_to_empty(self): """None source_key_cols / target_key_cols default to [].""" job = self._make_job({"source_key_cols": None, "target_key_cols": None}) response = job_to_response(job) assert response.source_key_cols == [] assert response.target_key_cols == [] def test_none_context_columns_default_to_empty(self): """None context_columns defaults to [].""" job = self._make_job({"context_columns": None}) response = job_to_response(job) assert response.context_columns == [] def test_none_batch_size_defaults_to_50(self): """None batch_size defaults to 50.""" job = self._make_job({"batch_size": None}) response = job_to_response(job) assert response.batch_size == 50 def test_none_upsert_strategy_defaults_to_merge(self): """None upsert_strategy defaults to MERGE.""" job = self._make_job({"upsert_strategy": None}) response = job_to_response(job) assert response.upsert_strategy == "MERGE" def test_none_insert_method_defaults_to_sqllab(self): """None insert_method defaults to sqllab.""" job = self._make_job({"insert_method": None}) response = job_to_response(job) assert response.insert_method == "sqllab" def test_direct_db_insert_method(self): """direct_db insert_method preserved.""" job = self._make_job({"insert_method": "direct_db", "connection_id": "conn-1"}) response = job_to_response(job) assert response.insert_method == "direct_db" assert response.connection_id == "conn-1" def test_disable_reasoning_true(self): """disable_reasoning=True preserved.""" job = self._make_job({"disable_reasoning": True}) response = job_to_response(job) assert response.disable_reasoning is True # #endregion Test.ServiceUtils