busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: massive coverage expansion — 15 new test modules + assistant tool fixes + orthogonal testing

Browse Source 
- 10 translate plugin test files (100% coverage on 12 modules)
- assistant/handler tools: 85+ tests covering dispatch, registry, resolvers, routes, llm_planner, 13 tool handlers
- clean release: artifact_catalog_loader, mappers, approval, publication tests
- API routes: translate_helpers, validation_service extensions, datasets to 100%
- notifications: providers/service tests
- services: profile_preference_service
- docs/orthogonal-test-report.md — full speckit.tests audit
- Fixes: 3 git_base async mock failures, 4 assistant handler permission-check patches
- .gitignore: coverage artifacts
This commit is contained in:
busya
2026-06-15 15:38:59 +03:00
parent 6ba381676a
commit f75c15dbc6
67 changed files with 10769 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files

166
backend/tests/plugins/translate/test_orchestrator_config.py Normal file
View File

@@ -0,0 +1,166 @@
# #region Test.OrchestratorConfig [C:3] [TYPE Module] [SEMANTICS test, translate, config, hash, snapshot]
# @BRIEF Tests for orchestrator_config.py — compute_config_hash, compute_dict_snapshot_hash.
# @RELATION BINDS_TO -> [orchestrator_config]
# @TEST_CONTRACT: compute_config_hash -> str | deterministic 16-char hash from job config
# @TEST_CONTRACT: compute_dict_snapshot_hash -> str | hash from dictionary entries
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, patch
import pytest
from src.plugins.translate.orchestrator_config import (
compute_config_hash,
compute_dict_snapshot_hash,
)
class TestComputeConfigHash:
"""compute_config_hash — deterministic hash from job config."""
def _make_job(self, overrides: dict | None = None) -> MagicMock:
"""Create a mock job with default config."""
job = MagicMock()
job.source_dialect = "postgresql"
job.target_dialect = "clickhouse"
job.source_datasource_id = "ds-1"
job.translation_column = "name"
job.context_columns = ["desc"]
job.target_languages = ["ru", "de"]
job.provider_id = "provider-1"
job.batch_size = 50
job.upsert_strategy = "MERGE"
if overrides:
for k, v in overrides.items():
setattr(job, k, v)
return job
def test_hash_length(self):
"""Hash is 16 characters."""
job = self._make_job()
h = compute_config_hash(job)
assert isinstance(h, str)
assert len(h) == 16
def test_deterministic(self):
"""Same config produces same hash."""
job1 = self._make_job()
job2 = self._make_job()
assert compute_config_hash(job1) == compute_config_hash(job2)
def test_different_config_different_hash(self):
"""Different config produces different hash."""
job1 = self._make_job({"batch_size": 50})
job2 = self._make_job({"batch_size": 100})
assert compute_config_hash(job1) != compute_config_hash(job2)
def test_source_dialect_affects_hash(self):
"""Changing source_dialect changes hash."""
job1 = self._make_job({"source_dialect": "postgresql"})
job2 = self._make_job({"source_dialect": "mysql"})
assert compute_config_hash(job1) != compute_config_hash(job2)
def test_target_languages_sorted(self):
"""target_languages are sorted before hashing."""
job1 = self._make_job({"target_languages": ["de", "ru"]})
job2 = self._make_job({"target_languages": ["ru", "de"]})
assert compute_config_hash(job1) == compute_config_hash(job2)
def test_target_languages_empty(self):
"""Empty target_languages handled."""
job = self._make_job({"target_languages": None})
h = compute_config_hash(job)
assert isinstance(h, str)
assert len(h) == 16
class TestComputeDictSnapshotHash:
"""compute_dict_snapshot_hash — dictionary state hash."""
def test_no_dictionary_links(self):
"""No dictionary links returns hash of empty bytes."""
db = MagicMock()
db.query.return_value.filter.return_value.all.return_value = []
h = compute_dict_snapshot_hash(db, "job-1")
assert isinstance(h, str)
assert len(h) == 16
def test_with_dictionary_links(self):
"""Dictionary links with entries produce a hash."""
db = MagicMock()
mock_link = MagicMock()
mock_link.dictionary_id = "dict-1"
db.query.return_value.filter.return_value.all.return_value = [mock_link]
db.query.return_value.filter.return_value.scalar.return_value = 5 # entry_count
db.query.return_value.filter.return_value.scalar.side_effect = [
5, # entry_count for dict-1
datetime(2026, 1, 15, 12, 0, 0, tzinfo=timezone.utc), # max_updated
]
h = compute_dict_snapshot_hash(db, "job-1")
assert isinstance(h, str)
assert len(h) == 16
def test_deterministic_with_same_data(self):
"""Same data produces same hash."""
db1 = MagicMock()
db2 = MagicMock()
mock_link = MagicMock()
mock_link.dictionary_id = "dict-1"
db1.query.return_value.filter.return_value.all.return_value = [mock_link]
db2.query.return_value.filter.return_value.all.return_value = [mock_link]
db1.query.return_value.filter.return_value.scalar.side_effect = [5, datetime(2026, 1, 15, 12, 0, 0, tzinfo=timezone.utc)]
db2.query.return_value.filter.return_value.scalar.side_effect = [5, datetime(2026, 1, 15, 12, 0, 0, tzinfo=timezone.utc)]
h1 = compute_dict_snapshot_hash(db1, "job-1")
h2 = compute_dict_snapshot_hash(db2, "job-1")
assert h1 == h2
def test_different_entry_count_changes_hash(self):
"""Different entry count produces different hash."""
db1 = MagicMock()
db2 = MagicMock()
mock_link = MagicMock()
mock_link.dictionary_id = "dict-1"
db1.query.return_value.filter.return_value.all.return_value = [mock_link]
db2.query.return_value.filter.return_value.all.return_value = [mock_link]
db1.query.return_value.filter.return_value.scalar.side_effect = [5, datetime(2026, 1, 15, 12, 0, 0, tzinfo=timezone.utc)]
db2.query.return_value.filter.return_value.scalar.side_effect = [10, datetime(2026, 1, 15, 12, 0, 0, tzinfo=timezone.utc)]
h1 = compute_dict_snapshot_hash(db1, "job-1")
h2 = compute_dict_snapshot_hash(db2, "job-1")
assert h1 != h2
def test_max_updated_none_handled(self):
"""When max_updated is None, '0' is used."""
db = MagicMock()
mock_link = MagicMock()
mock_link.dictionary_id = "dict-1"
db.query.return_value.filter.return_value.all.return_value = [mock_link]
db.query.return_value.filter.return_value.scalar.side_effect = [5, None]
h = compute_dict_snapshot_hash(db, "job-1")
assert isinstance(h, str)
assert len(h) == 16
def test_multiple_dictionaries(self):
"""Multiple dictionary links handled."""
db = MagicMock()
link1 = MagicMock()
link1.dictionary_id = "dict-a"
link2 = MagicMock()
link2.dictionary_id = "dict-b"
db.query.return_value.filter.return_value.all.return_value = [link2, link1] # unsorted
db.query.return_value.filter.return_value.scalar.side_effect = [
3, datetime(2026, 1, 15, 12, 0, 0, tzinfo=timezone.utc), # dict-a
7, datetime(2026, 2, 1, 8, 0, 0, tzinfo=timezone.utc), # dict-b
]
h = compute_dict_snapshot_hash(db, "job-1")
assert isinstance(h, str)
assert len(h) == 16
# #endregion Test.OrchestratorConfig

160
backend/tests/plugins/translate/test_orchestrator_lang_stats.py Normal file
View File

@@ -0,0 +1,160 @@
# #region Test.OrchestratorLangStats [C:3] [TYPE Module] [SEMANTICS test, translate, language, stats, aggregation]
# @BRIEF Tests for orchestrator_lang_stats.py — update_language_stats.
# @RELATION BINDS_TO -> [update_language_stats]
# @TEST_CONTRACT: update_language_stats -> None | aggregates TranslationLanguage entries into stats
# @TEST_EDGE: empty_record_ids -> early return
# @TEST_EDGE: various_statuses -> correct counting
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 unittest.mock import MagicMock, patch
import pytest
from src.plugins.translate.orchestrator_lang_stats import update_language_stats
class TestUpdateLanguageStats:
"""update_language_stats — aggregate language entries into stats objects."""
def _make_lang_entry(self, code: str, status: str, translated_value: str | None = None):
entry = MagicMock()
entry.language_code = code
entry.status = status
entry.translated_value = translated_value
return entry
def test_no_records_returns_early(self):
"""No records for run_id -> early return."""
db = MagicMock()
db.query.return_value.filter.return_value.all.return_value = []
event_log = MagicMock()
stats_map = {}
with patch("src.plugins.translate.orchestrator_lang_stats.logger") as mock_logger:
update_language_stats(db, event_log, "run-1", stats_map)
mock_logger.reason.assert_called_once()
def test_single_language_all_translated(self):
"""Single language with all translated entries."""
db = MagicMock()
# Return one record
db.query.return_value.filter.return_value.all.side_effect = [
[MagicMock(id="rec-1")], # TranslationRecords
[self._make_lang_entry("ru", "translated", "привет")], # TranslationLanguages
]
ru_stat = MagicMock()
stats_map = {"ru": ru_stat}
update_language_stats(db, None, "run-1", stats_map)
assert ru_stat.total_rows == 1
assert ru_stat.translated_rows == 1
assert ru_stat.failed_rows == 0
assert ru_stat.skipped_rows == 0
assert ru_stat.token_count >= 1
assert ru_stat.estimated_cost >= 0
db.flush.assert_called_once()
def test_multiple_languages_multiple_statuses(self):
"""Multiple languages with various statuses counted correctly."""
db = MagicMock()
db.query.return_value.filter.return_value.all.side_effect = [
[MagicMock(id="rec-1"), MagicMock(id="rec-2")], # Records
[ # Language entries
self._make_lang_entry("ru", "translated", "привет"),
self._make_lang_entry("ru", "failed", None),
self._make_lang_entry("de", "translated", "hallo"),
self._make_lang_entry("de", "skipped", None),
self._make_lang_entry("de", "approved", "OK"),
self._make_lang_entry("fr", "edited", "salut"),
],
]
ru_stat = MagicMock()
de_stat = MagicMock()
fr_stat = MagicMock()
stats_map = {"ru": ru_stat, "de": de_stat, "fr": fr_stat}
update_language_stats(db, None, "run-1", stats_map)
assert ru_stat.total_rows == 2
assert ru_stat.translated_rows == 1
assert ru_stat.failed_rows == 1
assert ru_stat.skipped_rows == 0
assert de_stat.total_rows == 3
assert de_stat.translated_rows == 2 # translated + approved
assert de_stat.failed_rows == 0
assert de_stat.skipped_rows == 1
assert fr_stat.total_rows == 1
assert fr_stat.translated_rows == 1 # edited
assert fr_stat.failed_rows == 0
assert fr_stat.skipped_rows == 0
def test_missing_language_in_stats_map(self):
"""Language in DB but not in stats_map -> ignored."""
db = MagicMock()
db.query.return_value.filter.return_value.all.side_effect = [
[MagicMock(id="rec-1")],
[self._make_lang_entry("ru", "translated", "привет")],
]
ru_stat = MagicMock()
stats_map = {"ru": ru_stat}
update_language_stats(db, None, "run-1", stats_map)
assert ru_stat.total_rows == 1
def test_language_not_in_agg_uses_defaults(self):
"""Language in stats_map but no entries -> all zeros."""
db = MagicMock()
db.query.return_value.filter.return_value.all.side_effect = [
[MagicMock(id="rec-1")],
[self._make_lang_entry("ru", "translated", "привет")],
]
ru_stat = MagicMock()
de_stat = MagicMock()
stats_map = {"ru": ru_stat, "de": de_stat}
update_language_stats(db, None, "run-1", stats_map)
assert ru_stat.total_rows == 1
assert de_stat.total_rows == 0
assert de_stat.translated_rows == 0
assert de_stat.failed_rows == 0
assert de_stat.skipped_rows == 0
def test_token_count_distribution(self):
"""Token count is distributed equally among languages."""
db = MagicMock()
db.query.return_value.filter.return_value.all.side_effect = [
[MagicMock(id="rec-1")],
[self._make_lang_entry("ru", "translated", "hello world")],
]
ru_stat = MagicMock()
de_stat = MagicMock()
stats_map = {"ru": ru_stat, "de": de_stat}
update_language_stats(db, None, "run-1", stats_map)
# total_chars = 11, total_tokens = 11 // 4 = 2, each gets 2 // 2 = 1
assert ru_stat.token_count == 1
assert de_stat.token_count == 1
def test_zero_token_falls_back_to_one(self):
"""When no text translated, total_tokens defaults to 1."""
db = MagicMock()
db.query.return_value.filter.return_value.all.side_effect = [
[MagicMock(id="rec-1")],
[self._make_lang_entry("ru", "failed", None)],
]
ru_stat = MagicMock()
stats_map = {"ru": ru_stat}
update_language_stats(db, None, "run-1", stats_map)
assert ru_stat.token_count >= 1
# #endregion Test.OrchestratorLangStats

111
backend/tests/plugins/translate/test_orchestrator_validation.py Normal file
View File

@@ -0,0 +1,111 @@
# #region Test.OrchestratorValidation [C:3] [TYPE Module] [SEMANTICS test, translate, validation, preconditions]
# @BRIEF Tests for orchestrator_validation.py — validate_job_preconditions.
# @RELATION BINDS_TO -> [validate_job_preconditions]
# @TEST_CONTRACT: validate_job_preconditions -> None | raises ValueError on precondition failure
# @TEST_EDGE: draft_status_fails
# @TEST_EDGE: missing_target_table
# @TEST_EDGE: missing_provider_id
# @TEST_EDGE: missing_translation_column
# @TEST_EDGE: no_accepted_preview_session
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 unittest.mock import MagicMock, patch
import pytest
from src.plugins.translate.orchestrator_validation import validate_job_preconditions
class TestValidateJobPreconditions:
"""validate_job_preconditions — precondition checks."""
def _make_job(self, overrides: dict | None = None) -> MagicMock:
"""Create a mock job with valid defaults."""
job = MagicMock()
job.id = "job-1"
job.status = "READY"
job.target_table = "target_table"
job.provider_id = "provider-1"
job.translation_column = "name"
job.source_datasource_id = None
if overrides:
for k, v in overrides.items():
setattr(job, k, v)
return job
def test_ready_passes_with_source_datasource(self):
"""READY job with source datasource passes without preview session."""
job = self._make_job({"source_datasource_id": "ds-1"})
db = MagicMock()
# Should not query preview sessions
validate_job_preconditions(job, db)
db.query.assert_not_called()
def test_ready_no_datasource_needs_preview(self):
"""READY job without datasource and no accepted preview fails."""
job = self._make_job({"source_datasource_id": None})
db = MagicMock()
db.query.return_value.filter.return_value.order_by.return_value.first.return_value = None
with pytest.raises(ValueError, match="no accepted preview session"):
validate_job_preconditions(job, db)
def test_ready_no_datasource_with_preview_passes(self):
"""READY job without datasource but with accepted preview passes."""
job = self._make_job({"source_datasource_id": None})
db = MagicMock()
mock_session = MagicMock()
mock_session.status = "APPLIED"
db.query.return_value.filter.return_value.order_by.return_value.first.return_value = mock_session
validate_job_preconditions(job, db)
def test_draft_status_fails(self):
"""DRAFT job raises ValueError."""
job = self._make_job({"status": "DRAFT"})
db = MagicMock()
with pytest.raises(ValueError, match="Cannot run job.*DRAFT"):
validate_job_preconditions(job, db)
def test_missing_target_table(self):
"""Job with no target_table raises ValueError."""
job = self._make_job({"target_table": None})
db = MagicMock()
with pytest.raises(ValueError, match="no target table"):
validate_job_preconditions(job, db)
def test_missing_provider_id(self):
"""Job with no provider_id raises ValueError."""
job = self._make_job({"provider_id": None})
db = MagicMock()
with pytest.raises(ValueError, match="no LLM provider"):
validate_job_preconditions(job, db)
def test_missing_translation_column(self):
"""Job with no translation_column raises ValueError."""
job = self._make_job({"translation_column": None})
db = MagicMock()
with pytest.raises(ValueError, match="no translation column"):
validate_job_preconditions(job, db)
def test_scheduled_run_skips_preview_check(self):
"""Scheduled run (is_scheduled=True) skips preview session check."""
job = self._make_job({"source_datasource_id": None})
db = MagicMock()
# Would fail without is_scheduled=True, but should pass with it
validate_job_preconditions(job, db, is_scheduled=True)
db.query.assert_not_called()
def test_validation_order_draft_checked_first(self):
"""DRAFT is checked before target_table."""
job = self._make_job({"status": "DRAFT", "target_table": None})
db = MagicMock()
with pytest.raises(ValueError, match="DRAFT"):
validate_job_preconditions(job, db)
# #endregion Test.OrchestratorValidation

82
backend/tests/plugins/translate/test_plugin.py Normal file
View File

@@ -0,0 +1,82 @@
# #region Test.TranslatePlugin [C:3] [TYPE Module] [SEMANTICS test,translate,plugin,interface]
# @BRIEF Tests for plugin.py — TranslatePlugin class.
# @RELATION BINDS_TO -> [TranslatePlugin]
# @TEST_CONTRACT: properties -> str | id, name, description, version
# @TEST_CONTRACT: get_schema -> dict | JSON schema with source_dialect, target_dialect, job_id
# @TEST_CONTRACT: execute -> NotImplementedError | deferred implementation
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:")
import pytest
from src.plugins.translate.plugin import TranslatePlugin
class TestTranslatePlugin:
"""TranslatePlugin — metadata properties and schema."""
def setup_method(self):
self.plugin = TranslatePlugin()
def test_id(self):
"""Plugin ID is 'translate'."""
assert self.plugin.id == "translate"
def test_name(self):
"""Plugin name is 'LLM Table Translation'."""
assert self.plugin.name == "LLM Table Translation"
def test_description(self):
"""Plugin description is about cross-dialect SQL translation."""
desc = self.plugin.description
assert "Cross-dialect" in desc
assert "SQL" in desc
def test_version(self):
"""Plugin version is 1.0.0."""
assert self.plugin.version == "1.0.0"
def test_get_schema_structure(self):
"""Schema is a dict with type, properties, required."""
schema = self.plugin.get_schema()
assert isinstance(schema, dict)
assert schema["type"] == "object"
assert "properties" in schema
assert "required" in schema
def test_get_schema_contains_source_dialect(self):
"""Schema includes source_dialect string field."""
props = self.plugin.get_schema()["properties"]
assert "source_dialect" in props
assert props["source_dialect"]["type"] == "string"
def test_get_schema_contains_target_dialect(self):
"""Schema includes target_dialect string field."""
props = self.plugin.get_schema()["properties"]
assert "target_dialect" in props
assert props["target_dialect"]["type"] == "string"
def test_get_schema_contains_job_id(self):
"""Schema includes job_id string field."""
props = self.plugin.get_schema()["properties"]
assert "job_id" in props
assert props["job_id"]["type"] == "string"
def test_get_schema_required_job_id(self):
"""job_id is required."""
schema = self.plugin.get_schema()
assert "job_id" in schema["required"]
@pytest.mark.asyncio
async def test_execute_not_implemented(self):
"""execute raises NotImplementedError."""
with pytest.raises(NotImplementedError, match="not yet implemented"):
await self.plugin.execute({"job_id": "test"})
# #endregion Test.TranslatePlugin

218
backend/tests/plugins/translate/test_preview_prompt_helpers.py Normal file
View File

@@ -0,0 +1,218 @@
# #region Test.PreviewPromptHelpers [C:3] [TYPE Module] [SEMANTICS test, translate, preview, token, budget, helpers]
# @BRIEF Tests for preview_prompt_helpers.py — estimate_token_budget_for_rows, compute_build_token_metadata.
# @RELATION BINDS_TO -> [preview_prompt_helpers]
# @TEST_CONTRACT: estimate_token_budget_for_rows -> dict | adjusts rows based on token budget
# @TEST_CONTRACT: compute_build_token_metadata -> dict | returns token estimates with cost warning
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 unittest.mock import MagicMock, patch
import pytest
from src.plugins.translate.preview_prompt_helpers import (
estimate_token_budget_for_rows,
compute_build_token_metadata,
)
class TestEstimateTokenBudgetForRows:
"""estimate_token_budget_for_rows — adjust row count based on budget."""
def test_no_adjustment_needed(self):
"""When budget fits, rows unchanged."""
job = MagicMock()
job.translation_column = "name"
job.context_columns = []
source_rows = [{"name": "hello"}, {"name": "world"}]
with patch("src.plugins.translate.preview_prompt_helpers.estimate_token_budget") as mock_budget:
mock_budget.return_value = {
"batch_size_adjusted": 5,
"estimated_input_tokens": 100,
"estimated_output_tokens": 200,
"batch_size": 2,
}
result = estimate_token_budget_for_rows(source_rows, ["ru"], job, None)
assert result["actual_row_count"] == 2
assert result["source_rows"] == source_rows
assert result["token_budget"]["batch_size_adjusted"] == 5
def test_adjustment_reduces_rows(self):
"""When budget is tight, rows are truncated to fit."""
job = MagicMock()
job.translation_column = "name"
job.context_columns = []
source_rows = [{"name": f"row{i}"} for i in range(10)]
with patch("src.plugins.translate.preview_prompt_helpers.estimate_token_budget") as mock_budget:
# First call: batch_size_adjusted smaller than actual
# Second call: after truncation
mock_budget.side_effect = [
{
"batch_size_adjusted": 3,
"estimated_input_tokens": 1000,
"estimated_output_tokens": 2000,
"batch_size": 10,
},
{
"batch_size_adjusted": 3,
"estimated_input_tokens": 300,
"estimated_output_tokens": 600,
"batch_size": 3,
},
]
result = estimate_token_budget_for_rows(source_rows, ["ru", "de"], job, None)
assert result["actual_row_count"] == 3
assert len(result["source_rows"]) == 3
assert result["token_budget"]["batch_size"] == 3
def test_empty_rows(self):
"""Empty source rows handled."""
job = MagicMock()
job.translation_column = "name"
job.context_columns = []
with patch("src.plugins.translate.preview_prompt_helpers.estimate_token_budget") as mock_budget:
mock_budget.return_value = {
"batch_size_adjusted": 0,
"estimated_input_tokens": 0,
"estimated_output_tokens": 0,
"batch_size": 0,
}
result = estimate_token_budget_for_rows([], ["ru"], job, None)
assert result["actual_row_count"] == 0
assert result["source_rows"] == []
def test_provider_model_passed(self):
"""Provider model string passed to estimate_token_budget."""
job = MagicMock()
job.translation_column = "name"
job.context_columns = []
with patch("src.plugins.translate.preview_prompt_helpers.estimate_token_budget") as mock_budget:
mock_budget.return_value = {
"batch_size_adjusted": 2,
"estimated_input_tokens": 100,
"estimated_output_tokens": 200,
"batch_size": 2,
}
result = estimate_token_budget_for_rows(
[{"name": "hello"}], ["ru"], job, "gpt-4o-mini"
)
assert result["actual_row_count"] == 1
# verify provider_model was passed (checked via call args)
call_kwargs = mock_budget.call_args[1]
assert call_kwargs["provider_info"] == "gpt-4o-mini"
class TestComputeBuildTokenMetadata:
"""compute_build_token_metadata — token estimation metadata."""
def test_basic_metadata(self):
"""Returns all expected keys with estimates."""
with patch("src.plugins.translate.preview_prompt_helpers.TokenEstimator") as MockTE:
MockTE.estimate_prompt_tokens.return_value = 100
MockTE.estimate_output_tokens.return_value = 200
MockTE.estimate_cost.return_value = 0.0006
MockTE.check_cost_warning.return_value = None
result = compute_build_token_metadata(
prompt="translate this",
row_meta=[{"id": 1}],
target_languages=["ru"],
num_languages=1,
dictionary_section="## Dictionary\nterm -> word",
actual_row_count=5,
sample_size=5,
)
assert result["prompt"] == "translate this"
assert result["row_meta"] == [{"id": 1}]
assert result["target_languages"] == ["ru"]
assert result["num_languages"] == 1
assert result["dictionary_section"] == "## Dictionary\nterm -> word"
assert result["sample_prompt_tokens"] == 100
assert result["sample_output_tokens"] == 200
assert result["sample_total_tokens"] == 300
assert result["sample_cost"] == 0.0006
assert result["total_est_rows"] == 50 # sample_size * 10
assert result["total_est_tokens"] > 0
assert result["total_est_cost"] > 0
assert result["cost_warning"] is None
assert result["actual_row_count"] == 5
def test_with_cost_warning(self):
"""Cost warning returned when sample exceeds threshold."""
with patch("src.plugins.translate.preview_prompt_helpers.TokenEstimator") as MockTE:
MockTE.estimate_prompt_tokens.return_value = 500
MockTE.estimate_output_tokens.return_value = 2000
MockTE.estimate_cost.return_value = 0.005
MockTE.check_cost_warning.return_value = (
"Large preview — estimated 2500 tokens, ~$0.0050 cost (across 2 languages)"
)
result = compute_build_token_metadata(
prompt="translate all this text",
row_meta=[{"id": 1}, {"id": 2}],
target_languages=["ru", "de"],
num_languages=2,
dictionary_section="",
actual_row_count=2,
sample_size=100,
)
assert result["cost_warning"] is not None
assert "Large preview" in result["cost_warning"]
assert "2 languages" in result["cost_warning"]
def test_empty_prompt(self):
"""Empty prompt handled by TokenEstimator."""
with patch("src.plugins.translate.preview_prompt_helpers.TokenEstimator") as MockTE:
MockTE.estimate_prompt_tokens.return_value = 0
MockTE.estimate_output_tokens.return_value = 0
MockTE.estimate_cost.return_value = 0.0
MockTE.check_cost_warning.return_value = None
result = compute_build_token_metadata(
prompt="", row_meta=[], target_languages=[], num_languages=1,
dictionary_section="", actual_row_count=0, sample_size=0,
)
assert result["sample_prompt_tokens"] == 0
assert result["sample_output_tokens"] == 0
assert result["sample_total_tokens"] == 0
assert result["sample_cost"] == 0.0
def test_total_est_replaces_row_count_in_prompt(self):
"""total_est_tokens substitutes actual_row_count with total_est_rows in prompt."""
with patch("src.plugins.translate.preview_prompt_helpers.TokenEstimator") as MockTE:
MockTE.estimate_prompt_tokens.side_effect = lambda p: len(p)
MockTE.estimate_output_tokens.return_value = 200
MockTE.estimate_cost.return_value = 0.001
MockTE.check_cost_warning.return_value = None
result = compute_build_token_metadata(
prompt="Translate 5 rows",
row_meta=[{"id": 1}],
target_languages=["ru"],
num_languages=1,
dictionary_section="",
actual_row_count=5,
sample_size=5,
)
# Second call: prompt.replace("5", "50")
assert MockTE.estimate_prompt_tokens.call_count == 2
# #endregion Test.PreviewPromptHelpers

103
backend/tests/plugins/translate/test_preview_resolve_provider.py Normal file
View File

@@ -0,0 +1,103 @@
# #region Test.PreviewResolveProvider [C:3] [TYPE Module] [SEMANTICS test, translate, preview, provider, model, resolve]
# @BRIEF Tests for preview_resolve_provider.py — resolve_provider_model.
# @RELATION BINDS_TO -> [resolve_provider_model]
# @TEST_CONTRACT: resolve_provider_model -> str|None | returns default model or None
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 unittest.mock import MagicMock, patch
import pytest
from src.plugins.translate.preview_resolve_provider import resolve_provider_model
class TestResolveProviderModel:
"""resolve_provider_model — resolve LLM model name for job."""
def _make_job(self, provider_id: str | None = "provider-1") -> MagicMock:
job = MagicMock()
job.provider_id = provider_id
return job
def test_with_provider_and_default_model(self):
"""Provider with default_model returns it."""
db = MagicMock()
job = self._make_job("provider-1")
mock_provider = MagicMock()
mock_provider.default_model = "gpt-4o"
with patch("src.services.llm_provider.LLMProviderService") as MockCls:
mock_service = MagicMock()
mock_service.get_provider.return_value = mock_provider
MockCls.return_value = mock_service
result = resolve_provider_model(db, job)
assert result == "gpt-4o"
def test_with_provider_no_default_model_returns_fallback(self):
"""Provider with None default_model returns 'gpt-4o-mini'."""
db = MagicMock()
job = self._make_job("provider-1")
mock_provider = MagicMock()
mock_provider.default_model = None
with patch("src.services.llm_provider.LLMProviderService") as MockCls:
mock_service = MagicMock()
mock_service.get_provider.return_value = mock_provider
MockCls.return_value = mock_service
result = resolve_provider_model(db, job)
assert result == "gpt-4o-mini"
def test_no_provider_id(self):
"""No provider_id returns None."""
job = self._make_job(None)
result = resolve_provider_model(MagicMock(), job)
assert result is None
def test_exception_during_resolve_returns_none(self):
"""Exception during provider resolve returns None."""
db = MagicMock()
job = self._make_job("provider-1")
with patch("src.services.llm_provider.LLMProviderService", side_effect=Exception("DB error")):
result = resolve_provider_model(db, job)
assert result is None
def test_provider_not_found_returns_none(self):
"""Provider not found (get_provider returns None) -> returns None."""
db = MagicMock()
job = self._make_job("provider-1")
with patch("src.services.llm_provider.LLMProviderService") as MockCls:
mock_service = MagicMock()
mock_service.get_provider.return_value = None
MockCls.return_value = mock_service
result = resolve_provider_model(db, job)
assert result is None
def test_provider_service_initialized_with_db(self):
"""LLMProviderService initialized with the given db session."""
db = MagicMock()
job = self._make_job("provider-1")
with patch("src.services.llm_provider.LLMProviderService") as MockCls:
mock_service = MagicMock()
mock_service.get_provider.return_value = MagicMock(default_model="claude-3-5-sonnet")
MockCls.return_value = mock_service
resolve_provider_model(db, job)
MockCls.assert_called_once_with(db)
# #endregion Test.PreviewResolveProvider

162
backend/tests/plugins/translate/test_preview_session_ops.py Normal file
View File

@@ -0,0 +1,162 @@
# #region Test.PreviewSessionOps [C:3] [TYPE Module] [SEMANTICS test, translate, preview, session, query]
# @BRIEF Tests for preview_session_ops.py — get_preview_session.
# @RELATION BINDS_TO -> [preview_session_ops]
# @TEST_CONTRACT: get_preview_session -> dict | returns latest session with records
# @TEST_EDGE: no_session_found -> raises ValueError
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, patch
import pytest
from src.plugins.translate.preview_session_ops import get_preview_session
class TestGetPreviewSession:
"""get_preview_session — fetch latest preview session for a job."""
def _make_session(self, overrides: dict | None = None) -> MagicMock:
"""Create a mock preview session."""
session = MagicMock()
session.id = "session-1"
session.job_id = "job-1"
session.status = "APPLIED"
session.created_by = "user-1"
session.created_at = datetime(2026, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
session.expires_at = datetime(2026, 2, 15, 12, 0, 0, tzinfo=timezone.utc)
if overrides:
for k, v in overrides.items():
setattr(session, k, v)
return session
def _make_record(self) -> MagicMock:
"""Create a mock preview record."""
r = MagicMock()
r.id = "rec-1"
r.source_sql = "SELECT * FROM foo"
r.target_sql = "SELECT * FROM bar"
r.source_object_type = "table"
r.source_object_id = "obj-1"
r.source_object_name = "foo"
r.status = "done"
r.feedback = None
r.languages = []
return r
def _make_job(self, target_languages=None, target_dialect=None) -> MagicMock:
"""Create a mock translation job."""
job = MagicMock()
job.id = "job-1"
job.target_languages = target_languages if target_languages is not None else ["ru", "de"]
job.target_dialect = target_dialect if target_dialect else "clickhouse"
return job
def _setup_db_mock(self, session, records, job):
"""Configure db mock chain for the three queries in get_preview_session."""
db = MagicMock()
# Query 1: TranslationPreviewSession (filter + order_by + first)
db.query.return_value.filter.return_value.order_by.return_value.first.return_value = session
# Query 2: TranslationPreviewRecord (filter + all)
db.query.return_value.filter.return_value.all.return_value = records
# Query 3: TranslationJob (filter + first — no order_by)
db.query.return_value.filter.return_value.first.return_value = job
return db
def test_returns_session_with_records(self):
"""Returns session dict with serialized records."""
session = self._make_session()
record = self._make_record()
job = self._make_job()
db = self._setup_db_mock(session, [record], job)
with patch("src.plugins.translate.preview_session_ops._serialize_preview_record") as mock_ser:
mock_ser.return_value = {"id": "rec-1", "source_sql": "SELECT * FROM foo"}
result = get_preview_session(db, "job-1")
assert result["id"] == "session-1"
assert result["job_id"] == "job-1"
assert result["status"] == "APPLIED"
assert result["created_by"] == "user-1"
assert "created_at" in result
assert "expires_at" in result
assert result["target_languages"] == ["ru", "de"]
assert len(result["records"]) == 1
assert result["records"][0]["id"] == "rec-1"
def test_no_session_found_raises(self):
"""No session found raises ValueError."""
db = MagicMock()
db.query.return_value.filter.return_value.order_by.return_value.first.return_value = None
with pytest.raises(ValueError, match="No preview session found for job 'job-1'"):
get_preview_session(db, "job-1")
def test_job_not_found_defaults_to_en(self):
"""Job not found falls back to ['en'] (last fallback)."""
session = self._make_session()
db = MagicMock()
db.query.return_value.filter.return_value.order_by.return_value.first.return_value = session
db.query.return_value.filter.return_value.all.return_value = []
# No job found -> first (no order_by) returns None
db.query.return_value.filter.return_value.first.return_value = None
with patch("src.plugins.translate.preview_session_ops._serialize_preview_record") as mock_ser:
result = get_preview_session(db, "job-1")
# When job is None -> the fallback is ["en"]
assert result["target_languages"] == ["en"]
def test_job_no_target_languages_uses_dialect(self):
"""Job with None target_languages uses target_dialect."""
session = self._make_session()
job = self._make_job(target_dialect="postgresql")
job.target_languages = None # explicitly override to None
db = self._setup_db_mock(session, [], job)
with patch("src.plugins.translate.preview_session_ops._serialize_preview_record") as mock_ser:
result = get_preview_session(db, "job-1")
assert result["target_languages"] == ["postgresql"]
def test_target_languages_not_list_normalized(self):
"""target_languages that is not a list gets wrapped in list."""
session = self._make_session()
job = self._make_job(target_languages="ru") # string instead of list
db = self._setup_db_mock(session, [], job)
with patch("src.plugins.translate.preview_session_ops._serialize_preview_record") as mock_ser:
result = get_preview_session(db, "job-1")
assert result["target_languages"] == ["ru"]
def test_session_expires_at_none(self):
"""Session without expires_at returns None for that field."""
session = self._make_session({"expires_at": None})
job = self._make_job()
db = self._setup_db_mock(session, [], job)
with patch("src.plugins.translate.preview_session_ops._serialize_preview_record") as mock_ser:
result = get_preview_session(db, "job-1")
assert result["expires_at"] is None
def test_empty_records(self):
"""No records returns empty records list."""
session = self._make_session()
job = self._make_job()
db = self._setup_db_mock(session, [], job)
with patch("src.plugins.translate.preview_session_ops._serialize_preview_record"):
result = get_preview_session(db, "job-1")
assert result["records"] == []
# #endregion Test.PreviewSessionOps

650
backend/tests/plugins/translate/test_service_target_schema.py Normal file
View File

@@ -0,0 +1,650 @@
# #region Test.TargetSchemaValidation [C:3] [TYPE Module] [SEMANTICS test, translate, schema, validation]
# @BRIEF Tests for service_target_schema.py — _build_expected_columns, _extract_columns_from_rows,
# _parse_sqllab_result, validate_target_table_schema.
# @RELATION BINDS_TO -> [TargetSchemaValidation]
# @TEST_CONTRACT: _build_expected_columns -> list[TargetSchemaColumnInfo] | deduplicated column list
# @TEST_CONTRACT: _extract_columns_from_rows -> list[dict] | normalised column info
# @TEST_CONTRACT: _parse_sqllab_result -> tuple | data_rows + table_exists
# @TEST_CONTRACT: validate_target_table_schema -> TargetSchemaValidationResponse | full schema diff
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 unittest.mock import AsyncMock, MagicMock, patch
import pytest
from src.schemas.translate import (
TargetSchemaColumnInfo,
TargetSchemaValidationRequest,
TargetSchemaValidationResponse,
)
from src.plugins.translate.service_target_schema import (
_build_expected_columns,
_extract_columns_from_rows,
_parse_sqllab_result,
validate_target_table_schema,
)
class TestBuildExpectedColumns:
"""_build_expected_columns — build deduplicated column list from request."""
def _make_req(self, overrides: dict | None = None) -> TargetSchemaValidationRequest:
defaults = {
"environment_id": "env-1",
"target_database_id": "db-1",
"target_table": "my_table",
"target_schema": "public",
"target_key_cols": ["id"],
"target_column": None,
"translation_column": "name",
"target_language_column": "lang",
"target_source_column": "src",
"target_source_language_column": "src_lang",
}
if overrides:
defaults.update(overrides)
return TargetSchemaValidationRequest(**defaults)
def test_with_all_columns(self):
"""All column types included in order."""
req = self._make_req()
cols = _build_expected_columns(req)
names = [c.name for c in cols]
assert names == ["id", "name", "lang", "src", "src_lang", "context", "is_original"]
def test_basic_minimal(self):
"""Minimal config: only key cols and translation column."""
req = self._make_req({
"target_key_cols": ["id"],
"translation_column": "name",
"target_column": None,
"target_language_column": None,
"target_source_column": None,
"target_source_language_column": None,
})
cols = _build_expected_columns(req)
names = [c.name for c in cols]
assert names == ["id", "name", "context", "is_original"]
def test_empty_key_cols(self):
"""Empty key cols list handled."""
req = self._make_req({
"target_key_cols": [],
"translation_column": "name",
"target_column": None,
"target_language_column": None,
"target_source_column": None,
"target_source_language_column": None,
})
cols = _build_expected_columns(req)
names = [c.name for c in cols]
assert names == ["name", "context", "is_original"]
def test_target_column_overrides_translation(self):
"""target_column used instead of translation_column when set."""
req = self._make_req({
"target_key_cols": [],
"translation_column": "source_col",
"target_column": "target_col",
"target_language_column": None,
"target_source_column": None,
"target_source_language_column": None,
})
cols = _build_expected_columns(req)
names = [c.name for c in cols]
assert "target_col" in names
assert "source_col" not in names
def test_deduplication(self):
"""Duplicate column names are removed."""
req = self._make_req({
"target_key_cols": ["id", "id"],
"target_column": "name",
"translation_column": "name",
"target_language_column": None,
"target_source_column": None,
"target_source_language_column": None,
})
cols = _build_expected_columns(req)
names = [c.name for c in cols]
assert names == ["id", "name", "context", "is_original"]
def test_none_key_cols(self):
"""None key cols handled."""
req = self._make_req({
"target_key_cols": None,
"translation_column": "name",
"target_column": None,
"target_language_column": None,
"target_source_column": None,
"target_source_language_column": None,
})
cols = _build_expected_columns(req)
names = [c.name for c in cols]
assert names == ["name", "context", "is_original"]
def test_all_nullable_fields(self):
"""All optional column fields are None."""
req = self._make_req({
"target_key_cols": None,
"translation_column": None,
"target_column": None,
"target_language_column": None,
"target_source_column": None,
"target_source_language_column": None,
})
cols = _build_expected_columns(req)
names = [c.name for c in cols]
# Only the always-included columns
assert names == ["context", "is_original"]
def test_return_type(self):
"""Returns list of TargetSchemaColumnInfo."""
req = self._make_req()
cols = _build_expected_columns(req)
for c in cols:
assert isinstance(c, TargetSchemaColumnInfo)
assert c.name
assert c.data_type is None
assert c.is_nullable is True
class TestExtractColumnsFromRows:
"""_extract_columns_from_rows — parse column info from data rows."""
def test_postgres_format(self):
"""PostgreSQL information_schema.columns format."""
rows = [
{"column_name": "id", "data_type": "integer", "is_nullable": "YES"},
{"column_name": "name", "data_type": "varchar", "is_nullable": "NO"},
]
cols = _extract_columns_from_rows(rows)
assert len(cols) == 2
assert cols[0]["name"] == "id"
assert cols[0]["type"] == "integer"
assert cols[0]["is_nullable"] is True
assert cols[1]["name"] == "name"
assert cols[1]["type"] == "varchar"
assert cols[1]["is_nullable"] is False
def test_clickhouse_format(self):
"""ClickHouse DESCRIBE TABLE format."""
rows = [
{"name": "id", "type": "Int32"},
{"name": "name", "type": "String"},
]
cols = _extract_columns_from_rows(rows)
assert len(cols) == 2
assert cols[0]["name"] == "id"
assert cols[0]["type"] == "Int32"
assert cols[0]["is_nullable"] is True # default for CH
def test_mysql_format(self):
"""MySQL SHOW COLUMNS format."""
rows = [
{"Field": "id", "Type": "int(11)"},
{"Field": "name", "Type": "varchar(255)"},
]
cols = _extract_columns_from_rows(rows)
assert len(cols) == 2
assert cols[0]["name"] == "id"
assert cols[0]["type"] == "int(11)"
def test_non_dict_row_skipped(self):
"""Non-dict rows are skipped."""
rows = [{"column_name": "id"}, "not a dict", 42]
cols = _extract_columns_from_rows(rows)
assert len(cols) == 1
def test_missing_column_name_skipped(self):
"""Rows without any known column key are skipped."""
rows = [{"unknown_key": "value"}]
cols = _extract_columns_from_rows(rows)
assert len(cols) == 0
def test_empty_rows(self):
"""Empty list returns empty list."""
assert _extract_columns_from_rows([]) == []
def test_is_nullable_string_handling(self):
"""is_nullable string 'YES'/'NO' converted to bool."""
rows = [
{"column_name": "a", "is_nullable": "YES"},
{"column_name": "b", "is_nullable": "NO"},
{"column_name": "c", "is_nullable": "yes"},
{"column_name": "d", "is_nullable": None},
]
cols = _extract_columns_from_rows(rows)
assert cols[0]["is_nullable"] is True
assert cols[1]["is_nullable"] is False
assert cols[2]["is_nullable"] is True
assert cols[3]["is_nullable"] is True # None -> default True
def test_type_none_handled(self):
"""type is None when not found in row."""
rows = [{"column_name": "id"}]
cols = _extract_columns_from_rows(rows)
assert cols[0]["type"] is None
class TestParseSqllabResult:
"""_parse_sqllab_result — extract data rows from SQL Lab response."""
def test_format_direct_data(self):
"""Format 1: direct 'data' key."""
result = {"data": [{"col": 1}, {"col": 2}]}
rows, exists = _parse_sqllab_result(result)
assert len(rows) == 2
assert exists is True
def test_format_async_result(self):
"""Format 2: nested 'result' dict."""
result = {"result": {"data": [{"col": 1}]}}
rows, exists = _parse_sqllab_result(result)
assert len(rows) == 1
def test_format_get_query_results(self):
"""Format 3: 'results' dict with 'data'."""
result = {"results": {"data": [{"col": 1}]}}
rows, exists = _parse_sqllab_result(result)
assert len(rows) == 1
def test_format_get_query_results_nested(self):
"""Format 3: 'results' -> 'result' -> 'data'."""
result = {"results": {"result": {"data": [{"col": 1}]}}}
rows, exists = _parse_sqllab_result(result)
assert len(rows) == 1
def test_format_results_list(self):
"""Format 4: 'results' as list."""
result = {"results": [{"col": 1}]}
rows, exists = _parse_sqllab_result(result)
assert len(rows) == 1
def test_empty_result(self):
"""None/empty result returns empty."""
rows, exists = _parse_sqllab_result({})
assert rows == []
assert exists is False
def test_none_result(self):
"""None result returns empty."""
rows, exists = _parse_sqllab_result(None)
assert rows == []
assert exists is False
def test_no_data_key(self):
"""Result without data/result/results returns empty."""
result = {"status": "success"}
rows, exists = _parse_sqllab_result(result)
assert rows == []
assert exists is False
def test_raw_response_is_falsy(self):
"""raw_response is a falsy value -> returns empty."""
result = {"raw_response": None}
rows, exists = _parse_sqllab_result(result)
assert rows == []
assert exists is False
def test_raw_response_is_empty_string(self):
"""raw_response is empty string -> returns empty."""
result = {"raw_response": ""}
rows, exists = _parse_sqllab_result(result)
assert rows == []
assert exists is False
def test_with_raw_response_wrapper(self):
"""Result with raw_response wrapper unwrapped."""
result = {"raw_response": {"data": [{"col": 1}]}}
rows, exists = _parse_sqllab_result(result)
assert len(rows) == 1
def test_normalize_list_rows_to_dicts(self):
"""List-of-lists with columns metadata normalized to dicts."""
result = {
"data": [["id1", "name1"], ["id2", "name2"]],
"columns": [{"name": "id"}, {"name": "name"}],
}
rows, exists = _parse_sqllab_result(result)
assert len(rows) == 2
assert rows[0] == {"id": "id1", "name": "name1"}
assert rows[1] == {"id": "id2", "name": "name2"}
def test_table_not_exists_empty_data(self):
"""Empty data means table does not exist."""
result = {"data": []}
rows, exists = _parse_sqllab_result(result)
assert rows == []
assert exists is False
def test_data_not_list(self):
"""data key exists but is not a list."""
result = {"data": "not a list"}
rows, exists = _parse_sqllab_result(result)
assert rows == []
assert exists is False
class TestValidateTargetTableSchema:
"""validate_target_table_schema — full schema validation flow."""
@pytest.mark.asyncio
async def test_invalid_table_name_sql_injection(self):
"""Invalid table name (SQL injection) returns error response."""
req = TargetSchemaValidationRequest(
environment_id="env-1",
target_database_id="db-1",
target_table="users; DROP TABLE",
target_schema="public",
)
config_manager = MagicMock()
result = await validate_target_table_schema(req, config_manager)
assert result.table_exists is False
assert "Invalid target table name" in result.error
assert len(result.missing_columns) > 0
assert result.all_present is False
@pytest.mark.asyncio
async def test_invalid_schema_name(self):
"""Invalid schema name returns error response."""
req = TargetSchemaValidationRequest(
environment_id="env-1",
target_database_id="db-1",
target_table="my_table",
target_schema="schema; DROP",
)
config_manager = MagicMock()
result = await validate_target_table_schema(req, config_manager)
assert result.table_exists is False
assert "Invalid target schema name" in result.error
assert result.all_present is False
@pytest.mark.asyncio
async def test_direct_db_path(self):
"""Direct DB path uses DbExecutor instead of SQL Lab."""
req = TargetSchemaValidationRequest(
environment_id="env-1",
target_database_id="db-1",
target_table="my_table",
target_schema="public",
insert_method="direct_db",
connection_id="conn-1",
)
config_manager = MagicMock()
mock_column = MagicMock()
mock_column.name = "id"
mock_column.data_type = "integer"
mock_executor = AsyncMock()
mock_executor.fetch_schema.return_value = [mock_column]
with patch("src.plugins.translate.service_target_schema.ConnectionService") as MockCS, \
patch("src.plugins.translate.service_target_schema.DbExecutor", return_value=mock_executor):
result = await validate_target_table_schema(req, config_manager)
assert result.table_exists is True
assert len(result.actual_columns) == 1
assert result.actual_columns[0].name == "id"
assert result.actual_columns[0].data_type == "integer"
@pytest.mark.asyncio
async def test_direct_db_no_columns(self):
"""Direct DB path with None fetch returns table_exists=False."""
req = TargetSchemaValidationRequest(
environment_id="env-1",
target_database_id="db-1",
target_table="my_table",
target_schema="public",
insert_method="direct_db",
connection_id="conn-1",
)
config_manager = MagicMock()
mock_executor = AsyncMock()
mock_executor.fetch_schema.return_value = None
with patch("src.plugins.translate.service_target_schema.ConnectionService"), \
patch("src.plugins.translate.service_target_schema.DbExecutor", return_value=mock_executor):
result = await validate_target_table_schema(req, config_manager)
assert result.table_exists is False
assert "Failed to fetch schema" in (result.error or "")
def _make_sqllab_executor(self, backend="postgresql", db_name="test_db"):
"""Create a mock executor with proper sync/async methods."""
executor = MagicMock()
executor.resolve_database_id = AsyncMock()
executor.get_database_backend.return_value = backend
executor.get_database_name.return_value = db_name
executor.execute_and_poll = AsyncMock()
return executor
@pytest.mark.asyncio
async def test_sqllab_path_success(self):
"""SQL Lab path: query succeeds, columns match."""
req = TargetSchemaValidationRequest(
environment_id="env-1",
target_database_id="db-1",
target_table="my_table",
target_schema="public",
target_key_cols=["id"],
translation_column="name",
target_column=None,
target_language_column=None,
target_source_column=None,
target_source_language_column=None,
)
config_manager = MagicMock()
executor = self._make_sqllab_executor()
executor.execute_and_poll.return_value = {
"status": "success",
"data": [
{"column_name": "id", "data_type": "integer", "is_nullable": "NO"},
{"column_name": "name", "data_type": "varchar", "is_nullable": "YES"},
{"column_name": "context", "data_type": "text", "is_nullable": "YES"},
{"column_name": "is_original", "data_type": "boolean", "is_nullable": "YES"},
],
}
with patch("src.plugins.translate.service_target_schema.SupersetSqlLabExecutor", return_value=executor):
result = await validate_target_table_schema(req, config_manager)
assert result.table_exists is True
assert result.all_present is True
assert len(result.expected_columns) == 4 # id, name, context, is_original
assert len(result.missing_columns) == 0
@pytest.mark.asyncio
async def test_sqllab_path_missing_columns(self):
"""SQL Lab path: some expected columns missing."""
req = TargetSchemaValidationRequest(
environment_id="env-1",
target_database_id="db-1",
target_table="my_table",
target_schema="public",
target_key_cols=["id"],
translation_column="name",
)
config_manager = MagicMock()
executor = self._make_sqllab_executor()
executor.execute_and_poll.return_value = {
"status": "success",
"data": [
{"column_name": "id", "data_type": "integer", "is_nullable": "NO"},
],
}
with patch("src.plugins.translate.service_target_schema.SupersetSqlLabExecutor", return_value=executor):
result = await validate_target_table_schema(req, config_manager)
assert result.table_exists is True
assert result.all_present is False
assert len(result.missing_columns) > 0
missing_names = [c.name for c in result.missing_columns]
assert "name" in missing_names
@pytest.mark.asyncio
async def test_sqllab_path_clickhouse(self):
"""ClickHouse backend uses system.columns."""
req = TargetSchemaValidationRequest(
environment_id="env-1",
target_database_id="db-1",
target_table="my_table",
target_schema="public",
target_key_cols=["id"],
translation_column="name",
)
config_manager = MagicMock()
executor = self._make_sqllab_executor(backend="clickhouse", db_name="test_ch")
executor.execute_and_poll.return_value = {
"status": "success",
"data": [
{"name": "id", "type": "Int32"},
{"name": "name", "type": "String"},
{"name": "context", "type": "String"},
{"name": "is_original", "type": "UInt8"},
],
}
with patch("src.plugins.translate.service_target_schema.SupersetSqlLabExecutor", return_value=executor):
result = await validate_target_table_schema(req, config_manager)
assert result.table_exists is True
assert result.all_present is True
# Verify clickhouse SQL was used
call_sql = executor.execute_and_poll.call_args[1]["sql"]
assert "system.columns" in call_sql
assert "SELECT name, type" in call_sql
@pytest.mark.asyncio
async def test_sqllab_path_failed_status(self):
"""SQL Lab query fails status -> error response."""
req = TargetSchemaValidationRequest(
environment_id="env-1",
target_database_id="db-1",
target_table="my_table",
)
config_manager = MagicMock()
executor = self._make_sqllab_executor()
executor.execute_and_poll.return_value = {
"status": "failed",
"error_message": "Table does not exist",
}
with patch("src.plugins.translate.service_target_schema.SupersetSqlLabExecutor", return_value=executor):
result = await validate_target_table_schema(req, config_manager)
assert result.table_exists is False
assert "Superset SQL Lab error" in (result.error or "")
assert result.database_name == "test_db"
assert result.database_backend == "postgresql"
@pytest.mark.asyncio
async def test_sqllab_path_exception(self):
"""Exception during SQL Lab execution -> error response."""
req = TargetSchemaValidationRequest(
environment_id="env-1",
target_database_id="db-1",
target_table="my_table",
)
config_manager = MagicMock()
executor = MagicMock()
executor.resolve_database_id = AsyncMock(side_effect=Exception("Connection refused"))
with patch("src.plugins.translate.service_target_schema.SupersetSqlLabExecutor", return_value=executor):
result = await validate_target_table_schema(req, config_manager)
assert result.table_exists is False
assert "Failed to validate" in (result.error or "")
@pytest.mark.asyncio
async def test_sqllab_with_extra_columns(self):
"""Extra columns in target table are reported."""
req = TargetSchemaValidationRequest(
environment_id="env-1",
target_database_id="db-1",
target_table="my_table",
target_key_cols=["id"],
translation_column="name",
)
config_manager = MagicMock()
executor = self._make_sqllab_executor()
executor.execute_and_poll.return_value = {
"status": "success",
"data": [
{"column_name": "id", "data_type": "integer"},
{"column_name": "name", "data_type": "varchar"},
{"column_name": "context", "data_type": "text"},
{"column_name": "is_original", "data_type": "boolean"},
{"column_name": "extra_col", "data_type": "text"},
],
}
with patch("src.plugins.translate.service_target_schema.SupersetSqlLabExecutor", return_value=executor):
result = await validate_target_table_schema(req, config_manager)
assert result.all_present is True
extra_names = [c.name for c in result.extra_columns]
assert "extra_col" in extra_names
@pytest.mark.asyncio
async def test_clickhousedb_normalized_to_clickhouse(self):
"""Backend 'clickhousedb' normalized to 'clickhouse' for SQL generation."""
req = TargetSchemaValidationRequest(
environment_id="env-1",
target_database_id="db-1",
target_table="my_table",
target_schema="default",
)
config_manager = MagicMock()
executor = self._make_sqllab_executor(backend="clickhousedb", db_name="test_ch")
executor.execute_and_poll.return_value = {
"status": "success",
"data": [],
}
with patch("src.plugins.translate.service_target_schema.SupersetSqlLabExecutor", return_value=executor):
result = await validate_target_table_schema(req, config_manager)
call_sql = executor.execute_and_poll.call_args[1]["sql"]
assert "system.columns" in call_sql
@pytest.mark.asyncio
async def test_greenplum_normalized_to_postgresql(self):
"""Backend 'greenplum' normalized to 'postgresql' for SQL generation."""
req = TargetSchemaValidationRequest(
environment_id="env-1",
target_database_id="db-1",
target_table="my_table",
target_schema="public",
)
config_manager = MagicMock()
executor = self._make_sqllab_executor(backend="greenplum", db_name="test_gp")
executor.execute_and_poll.return_value = {
"status": "success",
"data": [],
}
with patch("src.plugins.translate.service_target_schema.SupersetSqlLabExecutor", return_value=executor):
result = await validate_target_table_schema(req, config_manager)
call_sql = executor.execute_and_poll.call_args[1]["sql"]
assert "information_schema.columns" in call_sql
# #endregion Test.TargetSchemaValidation

217
backend/tests/plugins/translate/test_service_utils.py Normal file
View File

@@ -0,0 +1,217 @@
# #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

125
backend/tests/plugins/translate/test_text_cleaner.py Normal file
View File

@@ -0,0 +1,125 @@
# #region Test.TextCleaner [C:3] [TYPE Module] [SEMANTICS test,translate,text,cleaner,whitespace,truncation]
# @BRIEF Tests for _text_cleaner.py — normalize_whitespace, truncate_text, clean_text.
# @RELATION BINDS_TO -> [TextCleaner]
# @TEST_CONTRACT: normalize_whitespace -> str | whitespace collapsed + trimmed
# @TEST_CONTRACT: truncate_text -> str | truncated with "..." or unchanged
# @TEST_CONTRACT: clean_text -> str | normalized then truncated
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:")
import pytest
from src.plugins.translate._text_cleaner import (
normalize_whitespace,
truncate_text,
clean_text,
)
class TestNormalizeWhitespace:
"""normalize_whitespace — collapse whitespace, trim."""
def test_basic_trim(self):
"""Leading/trailing spaces are removed."""
assert normalize_whitespace(" hello world ") == "hello world"
def test_collapse_multi_spaces(self):
"""Multiple internal spaces collapsed to single."""
assert normalize_whitespace("hello world") == "hello world"
def test_tabs_newlines(self):
"""Tabs and newlines collapse to single space."""
assert normalize_whitespace("hello\t\tworld\n\nfoo") == "hello world foo"
def test_empty_string(self):
"""Empty string returns empty string."""
assert normalize_whitespace("") == ""
def test_whitespace_only(self):
"""Whitespace-only string returns empty string."""
assert normalize_whitespace(" \t\n ") == ""
def test_no_change(self):
"""Already clean text unchanged."""
assert normalize_whitespace("hello world") == "hello world"
def test_single_word(self):
"""Single word is trimmed."""
assert normalize_whitespace(" hello ") == "hello"
class TestTruncateText:
"""truncate_text — truncate with '...' suffix."""
def test_no_truncation_needed(self):
"""Short text returned unchanged."""
assert truncate_text("hello", 10) == "hello"
def test_exact_length(self):
"""Text equal to max_length returned unchanged."""
assert truncate_text("hello", 5) == "hello"
def test_truncation_applied(self):
"""Text longer than max_length is truncated with '...'."""
assert truncate_text("hello world", 5) == "hello..."
def test_empty_string(self):
"""Empty string with any max_length returns empty."""
assert truncate_text("", 100) == ""
def test_default_max_length(self):
"""Default max_length is 500."""
short = "a" * 400
long_str = "a" * 600
assert truncate_text(short) == short
assert truncate_text(long_str) == "a" * 500 + "..."
def test_zero_max_length(self):
"""max_length=0 truncates everything."""
assert truncate_text("hello", 0) == "..."
def test_unicode_preserved(self):
"""Unicode characters preserved through truncation."""
text = "héllo wörld"
assert truncate_text(text, 5) == "héllo..."
def test_cjk_characters(self):
"""CJK characters counted as single chars."""
text = "你好世界"
assert truncate_text(text, 2) == "你好..."
class TestCleanText:
"""clean_text — normalize then truncate."""
def test_normalize_and_truncate(self):
"""Input is normalized then truncated."""
result = clean_text(" hello world ", 5)
assert result == "hello..."
def test_no_truncation_needed(self):
"""Normalized text within max_length unchanged."""
result = clean_text(" hello world ", 50)
assert result == "hello world"
def test_empty_string(self):
"""Empty string stays empty."""
assert clean_text("") == ""
def test_default_max_length(self):
"""Default max_length of 500 used."""
text = " " + "a" * 600 + " "
result = clean_text(text)
assert result == "a" * 500 + "..."
def test_whitespace_only(self):
"""Whitespace-only input returns empty string."""
assert clean_text(" \t\n ") == ""
# #endregion Test.TextCleaner