busya
488a8f349b
- ~60 new/extended test files across api, core, plugins, services, schemas: routes, superset clients, task_manager, lineage, git, translate, dashboard-testing, load-testing, migration, llm_analysis, scheduler, ssl - .coveragerc: enable branch coverage; exclude src/__tests__ (test files) and src/scripts (CLI/ops tools) from the denominator - bug fixes found while testing: * settings: PUT /settings/reports registered under duplicated prefix * schemas/lineage: FleetReportDTO missing run_status (route always 500) * dashboard_testing/baseline_inheritance: visual entry read wrong field * superset_client/_databases: logger extra name shadowed LogRecord attr * routes/datasets: _yaml_string_paths recursion without yield from * translate/sql_generator: restore explicit-type timestamp contract * baseline_catalog: remove unreachable dashboard_id fallback - conftest fixes: pytest_plugins to rootdir conftest (pytest 9), test filename collision, TMPDIR-safe integration fixtures
375 lines
15 KiB
Python
375 lines
15 KiB
Python
# #region Test.Schemas.Translate [C:3] [TYPE Module] [SEMANTICS test,schema,translate,validation,bcp47]
|
|
# @BRIEF Tests for schemas/translate.py — BCP-47 target-language validator and the
|
|
# model-level validators on TranslateJobCreate / TranslateJobUpdate /
|
|
# BulkFindReplaceRequest / TargetSchemaValidationRequest.
|
|
# @RELATION BINDS_TO -> [Schemas.Translate.TranslateSchemas]
|
|
# @TEST_EDGE: missing_field -> optional fields default correctly
|
|
# @TEST_EDGE: invalid_type -> BCP-47 / enum / length violations raise ValidationError
|
|
# @TEST_EDGE: conditional_required -> direct_db requires connection_id; sqllab requires target_database_id
|
|
# @TEST_EDGE: cross_field -> source_key_cols/target_key_cols length mismatch rejected
|
|
# @TEST_EDGE: constraint_floor -> numeric performance knobs reject out-of-range values
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from src.schemas.translate import (
|
|
BulkFindReplaceRequest,
|
|
TargetSchemaValidationRequest,
|
|
TranslateJobCreate,
|
|
TranslateJobUpdate,
|
|
_validate_bcp47_list,
|
|
)
|
|
|
|
|
|
class TestBcp47List:
|
|
"""_validate_bcp47_list — raw BCP-47 tag list validator."""
|
|
|
|
def test_none_returns_none(self):
|
|
assert _validate_bcp47_list(None) is None
|
|
|
|
def test_empty_list_returns_empty(self):
|
|
assert _validate_bcp47_list([]) == []
|
|
|
|
def test_valid_tags_passthrough(self):
|
|
tags = ["en", "ru", "zh-CN", "pt-BR"]
|
|
assert _validate_bcp47_list(tags) == tags
|
|
|
|
def test_empty_string_tag_raises(self):
|
|
with pytest.raises(ValueError, match="must be a non-empty string"):
|
|
_validate_bcp47_list(["en", ""])
|
|
|
|
def test_whitespace_only_tag_raises(self):
|
|
with pytest.raises(ValueError, match="must be a non-empty string"):
|
|
_validate_bcp47_list(["en", " "])
|
|
|
|
def test_invalid_format_raises(self):
|
|
with pytest.raises(ValueError, match="Invalid BCP-47 tag"):
|
|
_validate_bcp47_list(["en", "not a tag"])
|
|
|
|
def test_single_char_tag_raises(self):
|
|
with pytest.raises(ValueError, match="Invalid BCP-47 tag"):
|
|
_validate_bcp47_list(["e"])
|
|
|
|
|
|
class TestTranslateJobCreate:
|
|
"""TranslateJobCreate — create payload with conditional model validator."""
|
|
|
|
def test_minimal_defaults(self):
|
|
job = TranslateJobCreate(name="job-1")
|
|
assert job.name == "job-1"
|
|
assert job.description is None
|
|
assert job.source_dialect == ""
|
|
assert job.target_dialect == ""
|
|
assert job.database_dialect is None
|
|
assert job.source_key_cols == []
|
|
assert job.target_key_cols == []
|
|
assert job.context_columns == []
|
|
assert job.target_languages == []
|
|
assert job.dictionary_ids == []
|
|
assert job.include_source_reference is True
|
|
assert job.batch_size == 50
|
|
assert job.disable_reasoning is False
|
|
assert job.upsert_strategy == "MERGE"
|
|
assert job.insert_method == "sqllab"
|
|
assert job.connection_id is None
|
|
assert job.llm_batch_max_rows is None
|
|
assert job.llm_concurrency is None
|
|
assert job.insert_concurrency is None
|
|
assert job.max_in_flight_batches is None
|
|
|
|
def test_full_valid(self):
|
|
job = TranslateJobCreate(
|
|
name="job-1",
|
|
description="desc",
|
|
source_dialect="postgresql",
|
|
target_dialect="clickhouse",
|
|
database_dialect="postgresql",
|
|
source_key_cols=["id"],
|
|
target_key_cols=["id"],
|
|
target_languages=["en", "zh-CN"],
|
|
multi_lang_mode="per_language",
|
|
batch_aggressiveness="fast",
|
|
insert_method="direct_db",
|
|
connection_id="conn-1",
|
|
llm_batch_max_rows=25,
|
|
llm_concurrency=8,
|
|
insert_concurrency=2,
|
|
max_in_flight_batches=16,
|
|
include_source_reference=False,
|
|
)
|
|
assert job.connection_id == "conn-1"
|
|
assert job.llm_batch_max_rows == 25
|
|
assert job.include_source_reference is False
|
|
|
|
def test_direct_db_requires_connection_id(self):
|
|
with pytest.raises(ValidationError, match="connection_id is required"):
|
|
TranslateJobCreate(name="job-1", insert_method="direct_db")
|
|
|
|
def test_direct_db_with_connection_ok(self):
|
|
job = TranslateJobCreate(name="job-1", insert_method="direct_db", connection_id="conn-1")
|
|
assert job.insert_method == "direct_db"
|
|
|
|
def test_sqllab_without_connection_ok(self):
|
|
job = TranslateJobCreate(name="job-1", insert_method="sqllab")
|
|
assert job.connection_id is None
|
|
|
|
def test_key_cols_length_mismatch_raises(self):
|
|
with pytest.raises(ValidationError, match="must have equal lengths"):
|
|
TranslateJobCreate(
|
|
name="job-1",
|
|
source_key_cols=["id"],
|
|
target_key_cols=["id", "name"],
|
|
)
|
|
|
|
def test_key_cols_equal_length_ok(self):
|
|
job = TranslateJobCreate(
|
|
name="job-1",
|
|
source_key_cols=["id", "ts"],
|
|
target_key_cols=["id", "ts"],
|
|
)
|
|
assert job.source_key_cols == ["id", "ts"]
|
|
|
|
def test_invalid_multi_lang_mode_raises(self):
|
|
with pytest.raises(ValidationError, match="single_call"):
|
|
TranslateJobCreate(name="job-1", multi_lang_mode="parallel")
|
|
|
|
def test_valid_multi_lang_modes_ok(self):
|
|
for mode in ("single_call", "per_language"):
|
|
job = TranslateJobCreate(name="job-1", multi_lang_mode=mode)
|
|
assert job.multi_lang_mode == mode
|
|
|
|
def test_invalid_batch_aggressiveness_raises(self):
|
|
with pytest.raises(ValidationError, match="safe"):
|
|
TranslateJobCreate(name="job-1", batch_aggressiveness="max")
|
|
|
|
def test_valid_batch_aggressiveness_ok(self):
|
|
for level in ("safe", "balanced", "fast"):
|
|
job = TranslateJobCreate(name="job-1", batch_aggressiveness=level)
|
|
assert job.batch_aggressiveness == level
|
|
|
|
def test_invalid_target_languages_raises(self):
|
|
with pytest.raises(ValidationError, match="Invalid BCP-47 tag"):
|
|
TranslateJobCreate(name="job-1", target_languages=["en", "bad tag"])
|
|
|
|
def test_valid_target_languages_ok(self):
|
|
job = TranslateJobCreate(name="job-1", target_languages=["en", "pt-BR"])
|
|
assert job.target_languages == ["en", "pt-BR"]
|
|
|
|
def test_llm_batch_max_rows_bounds(self):
|
|
with pytest.raises(ValidationError):
|
|
TranslateJobCreate(name="job-1", llm_batch_max_rows=0)
|
|
with pytest.raises(ValidationError):
|
|
TranslateJobCreate(name="job-1", llm_batch_max_rows=51)
|
|
assert TranslateJobCreate(name="job-1", llm_batch_max_rows=50).llm_batch_max_rows == 50
|
|
|
|
def test_llm_concurrency_bounds(self):
|
|
with pytest.raises(ValidationError):
|
|
TranslateJobCreate(name="job-1", llm_concurrency=17)
|
|
assert TranslateJobCreate(name="job-1", llm_concurrency=16).llm_concurrency == 16
|
|
|
|
def test_insert_concurrency_bounds(self):
|
|
with pytest.raises(ValidationError):
|
|
TranslateJobCreate(name="job-1", insert_concurrency=5)
|
|
assert TranslateJobCreate(name="job-1", insert_concurrency=4).insert_concurrency == 4
|
|
|
|
def test_max_in_flight_batches_bounds(self):
|
|
with pytest.raises(ValidationError):
|
|
TranslateJobCreate(name="job-1", max_in_flight_batches=3)
|
|
assert TranslateJobCreate(name="job-1", max_in_flight_batches=128).max_in_flight_batches == 128
|
|
|
|
def test_serialize_roundtrip(self):
|
|
job = TranslateJobCreate(name="job-1", target_languages=["en"])
|
|
data = job.model_dump()
|
|
restored = TranslateJobCreate.model_validate(data)
|
|
assert restored.name == "job-1"
|
|
assert restored.target_languages == ["en"]
|
|
|
|
|
|
class TestTranslateJobUpdate:
|
|
"""TranslateJobUpdate — partial update with enum + cross-field validation."""
|
|
|
|
def test_empty_all_optional(self):
|
|
update = TranslateJobUpdate()
|
|
assert update.name is None
|
|
assert update.source_key_cols is None
|
|
assert update.target_key_cols is None
|
|
assert update.multi_lang_mode is None
|
|
assert update.batch_aggressiveness is None
|
|
assert update.insert_method is None
|
|
assert update.status is None
|
|
|
|
def test_partial_fields(self):
|
|
update = TranslateJobUpdate(name="renamed", status="RUNNING", upsert_strategy="INSERT")
|
|
assert update.name == "renamed"
|
|
assert update.status == "RUNNING"
|
|
assert update.upsert_strategy == "INSERT"
|
|
|
|
def test_key_cols_length_mismatch_raises(self):
|
|
with pytest.raises(ValidationError, match="must have equal lengths"):
|
|
TranslateJobUpdate(source_key_cols=["id"], target_key_cols=["id", "name"])
|
|
|
|
def test_key_cols_equal_length_ok(self):
|
|
update = TranslateJobUpdate(source_key_cols=["id"], target_key_cols=["id"])
|
|
assert update.source_key_cols == ["id"]
|
|
|
|
def test_invalid_multi_lang_mode_raises(self):
|
|
with pytest.raises(ValidationError, match="single_call"):
|
|
TranslateJobUpdate(multi_lang_mode="parallel")
|
|
|
|
def test_valid_multi_lang_mode_ok(self):
|
|
update = TranslateJobUpdate(multi_lang_mode="per_language")
|
|
assert update.multi_lang_mode == "per_language"
|
|
|
|
def test_invalid_batch_aggressiveness_raises(self):
|
|
with pytest.raises(ValidationError, match="safe"):
|
|
TranslateJobUpdate(batch_aggressiveness="max")
|
|
|
|
def test_valid_batch_aggressiveness_ok(self):
|
|
update = TranslateJobUpdate(batch_aggressiveness="balanced")
|
|
assert update.batch_aggressiveness == "balanced"
|
|
|
|
def test_invalid_target_languages_raises(self):
|
|
with pytest.raises(ValidationError, match="Invalid BCP-47 tag"):
|
|
TranslateJobUpdate(target_languages=["en", "bad tag"])
|
|
|
|
def test_valid_target_languages_ok(self):
|
|
update = TranslateJobUpdate(target_languages=["ru"])
|
|
assert update.target_languages == ["ru"]
|
|
|
|
def test_knob_bounds(self):
|
|
with pytest.raises(ValidationError):
|
|
TranslateJobUpdate(llm_batch_max_rows=0)
|
|
with pytest.raises(ValidationError):
|
|
TranslateJobUpdate(llm_concurrency=17)
|
|
with pytest.raises(ValidationError):
|
|
TranslateJobUpdate(max_in_flight_batches=3)
|
|
update = TranslateJobUpdate(llm_batch_max_rows=10, llm_concurrency=4)
|
|
assert update.llm_batch_max_rows == 10
|
|
|
|
def test_serialize_roundtrip(self):
|
|
update = TranslateJobUpdate(name="new", batch_aggressiveness="fast")
|
|
data = update.model_dump()
|
|
restored = TranslateJobUpdate.model_validate(data)
|
|
assert restored.name == "new"
|
|
assert restored.batch_aggressiveness == "fast"
|
|
|
|
|
|
class TestBulkFindReplaceRequest:
|
|
"""BulkFindReplaceRequest — find-pattern length guard."""
|
|
|
|
def test_minimal_valid(self):
|
|
req = BulkFindReplaceRequest(find_pattern="old", replacement_text="new", target_language="en")
|
|
assert req.is_regex is False
|
|
assert req.preview is True
|
|
assert req.submit_to_dictionary is False
|
|
assert req.submit_to_dictionary_with_context is False
|
|
assert req.dictionary_id is None
|
|
assert req.usage_notes is None
|
|
|
|
def test_exactly_500_chars_ok(self):
|
|
req = BulkFindReplaceRequest(find_pattern="x" * 500, replacement_text="new", target_language="en")
|
|
assert len(req.find_pattern) == 500
|
|
|
|
def test_over_500_chars_rejected_by_constraint(self):
|
|
# pydantic max_length=500 fires before the custom validator, so the
|
|
# rejection surfaces as the field constraint error.
|
|
with pytest.raises(ValidationError, match="at most 500"):
|
|
BulkFindReplaceRequest(find_pattern="x" * 501, replacement_text="new", target_language="en")
|
|
|
|
def test_flags_and_optional_fields(self):
|
|
req = BulkFindReplaceRequest(
|
|
find_pattern=r"^old$",
|
|
is_regex=True,
|
|
replacement_text="new",
|
|
target_language="en",
|
|
preview=False,
|
|
submit_to_dictionary=True,
|
|
dictionary_id="dict-1",
|
|
usage_notes="note",
|
|
submit_to_dictionary_with_context=True,
|
|
)
|
|
assert req.is_regex is True
|
|
assert req.preview is False
|
|
assert req.submit_to_dictionary is True
|
|
assert req.dictionary_id == "dict-1"
|
|
assert req.submit_to_dictionary_with_context is True
|
|
|
|
def test_serialize_roundtrip(self):
|
|
req = BulkFindReplaceRequest(find_pattern="old", replacement_text="new", target_language="en")
|
|
data = req.model_dump()
|
|
restored = BulkFindReplaceRequest.model_validate(data)
|
|
assert restored.find_pattern == "old"
|
|
assert restored.target_language == "en"
|
|
|
|
|
|
class TestTargetSchemaValidationRequest:
|
|
"""TargetSchemaValidationRequest — conditional backend/connection requirement."""
|
|
|
|
def test_sqllab_with_database_ok(self):
|
|
req = TargetSchemaValidationRequest(
|
|
environment_id="env-1",
|
|
target_database_id="db-1",
|
|
target_table="my_table",
|
|
)
|
|
assert req.target_schema == "public"
|
|
assert req.target_key_cols == []
|
|
assert req.insert_method is None
|
|
|
|
def test_sqllab_requires_target_database(self):
|
|
with pytest.raises(ValidationError, match="target_database_id is required"):
|
|
TargetSchemaValidationRequest(
|
|
environment_id="env-1",
|
|
target_table="my_table",
|
|
insert_method="sqllab",
|
|
)
|
|
|
|
def test_default_insert_method_requires_target_database(self):
|
|
with pytest.raises(ValidationError, match="target_database_id is required"):
|
|
TargetSchemaValidationRequest(environment_id="env-1", target_table="my_table")
|
|
|
|
def test_direct_db_requires_connection_id(self):
|
|
with pytest.raises(ValidationError, match="connection_id is required"):
|
|
TargetSchemaValidationRequest(
|
|
environment_id="env-1",
|
|
target_table="my_table",
|
|
insert_method="direct_db",
|
|
)
|
|
|
|
def test_direct_db_with_connection_ok(self):
|
|
req = TargetSchemaValidationRequest(
|
|
environment_id="env-1",
|
|
target_table="my_table",
|
|
insert_method="direct_db",
|
|
connection_id="conn-1",
|
|
)
|
|
assert req.connection_id == "conn-1"
|
|
assert req.target_database_id is None
|
|
|
|
def test_column_mapping_fields(self):
|
|
req = TargetSchemaValidationRequest(
|
|
environment_id="env-1",
|
|
target_database_id="db-1",
|
|
target_table="t",
|
|
target_key_cols=["id"],
|
|
target_column="translated",
|
|
translation_column="source",
|
|
target_language_column="lang",
|
|
target_source_column="src_text",
|
|
target_source_language_column="src_lang",
|
|
)
|
|
assert req.target_key_cols == ["id"]
|
|
assert req.target_column == "translated"
|
|
assert req.target_source_language_column == "src_lang"
|
|
|
|
def test_serialize_roundtrip(self):
|
|
req = TargetSchemaValidationRequest(
|
|
environment_id="env-1",
|
|
target_database_id="db-1",
|
|
target_table="t",
|
|
)
|
|
data = req.model_dump()
|
|
restored = TargetSchemaValidationRequest.model_validate(data)
|
|
assert restored.target_table == "t"
|
|
assert restored.target_schema == "public"
|
|
# #endregion Test.Schemas.Translate
|