Merge branch 'refactor/TG-1122-td-export-import-domain-errors' into 'enterprise'

refactor(api): TD export/import domain exceptions and version marker

See merge request dkinternal/testgen/dataops-testgen!558

Author: ci bot

testgen/api/schemas.py

@@ -1,12 +1,12 @@
 """Pydantic request/response models for API v1 endpoints."""
 
 from datetime import datetime
-from enum import StrEnum
 from uuid import UUID
 
-from pydantic import BaseModel, field_validator
+from pydantic import BaseModel
 
 from testgen.common.enums import JobSource, JobStatus, PublicJobKey
+from testgen.common.test_definition_export_import_service import ImportConfig, ImportPayload, ImportResponse
 
 # --- Jobs ---
 
@@ -127,183 +127,14 @@ class ErrorResponse(BaseModel):
     errors: list[ErrorDetail]
 
 
-# --- Test Definition Export/Import ---
-
-
-class Origin(StrEnum):
-    manual = "manual"
-    auto = "auto"
-    both = "both"
-
-
-class ImportMode(StrEnum):
-    preview = "preview"
-    apply = "apply"
-    apply_strict = "apply_strict"
-
-
-class OnMatch(StrEnum):
-    overwrite_all = "overwrite_all"
-    overwrite_unlocked = "overwrite_unlocked"
-    skip = "skip"
-
-
-class OnNew(StrEnum):
-    skip = "skip"
-    create = "create"
-    create_and_lock = "create_and_lock"
-
-
-class OnAbsence(StrEnum):
-    do_nothing = "do_nothing"
-    delete_all = "delete_all"
-    delete_unlocked = "delete_unlocked"
-
-
-class ImportAction(StrEnum):
-    create = "create"
-    update = "update"
-    skip = "skip"
-    delete = "delete"
-
-
-class ImportReason(StrEnum):
-    matched = "matched"
-    no_match = "no_match"
-    policy = "policy"
-    locked = "locked"
-    invalid_test_type = "invalid_test_type"
-    invalid_table = "invalid_table"
-    missing_external_id = "missing_external_id"
-    absent = "absent"
-
-
-# Non-None defaults must match the ORM column defaults in TestDefinition:
-#   test_active=True (YNString default="Y"), lock_refresh=False (YNString default="N"),
-#   skip_errors=0 (ZeroIfEmptyInteger), window_days=0 (ZeroIfEmptyInteger),
-#   history_lookback=0 (Column default=0).
-# On export, the model_serializer omits fields matching these defaults to keep the file compact.
-# On import, model_fields_set distinguishes explicit from defaulted.
-class TestDefinitionExport(BaseModel):
-    """Test definition fields included in the export/import file."""
-
-    model_config = {"from_attributes": True}
-
-    # Matching / identity
-    test_type: str
-    external_id: UUID | None = None
-    last_auto_gen_date: datetime | None = None
-
-    # Definition fields
-    table_name: str | None = None
-    column_name: str | None = None
-    test_description: str | None = None
-    test_active: bool = True
-    severity: str | None = None
-    lock_refresh: bool = False
-    export_to_observability: bool | None = None
-    skip_errors: int = 0
-
-    # Calibration fields
-    baseline_ct: str | None = None
-    baseline_unique_ct: str | None = None
-    baseline_value: str | None = None
-    baseline_value_ct: str | None = None
-    threshold_value: str | None = None
-    baseline_sum: str | None = None
-    baseline_avg: str | None = None
-    baseline_sd: str | None = None
-    lower_tolerance: str | None = None
-    upper_tolerance: str | None = None
-
-    # Subset / grouping
-    subset_condition: str | None = None
-    groupby_names: str | None = None
-    having_condition: str | None = None
-    window_date_column: str | None = None
-    window_days: int = 0
-
-    # Referential
-    match_schema_name: str | None = None
-    match_table_name: str | None = None
-    match_column_names: str | None = None
-    match_subset_condition: str | None = None
-    match_groupby_names: str | None = None
-    match_having_condition: str | None = None
-
-    # Query / history
-    custom_query: str | None = None
-    history_calculation: str | None = None
-    history_calculation_upper: str | None = None
-    history_lookback: int = 0
-
-    @field_validator("skip_errors", "window_days", "history_lookback", mode="before")
-    @classmethod
-    def _coerce_none_to_zero(cls, v: int | None) -> int:
-        return v if v is not None else 0
-
-
-class ExportSource(BaseModel):
-    project_code: str
-    test_suite: str
-    table_group: str
-    table_group_schema: str
-    exported_at: datetime
-    testgen_version: str | None = None
-
-
-class ExportDocument(BaseModel):
-    version: int = 1
-    source: ExportSource
-    definitions: list[TestDefinitionExport]
-
-
-# --- Import ---
-
-
-class ImportConfig(BaseModel):
-    mode: ImportMode
-    on_match: OnMatch
-    on_new: OnNew
-    on_absence: OnAbsence
-
-
-class ImportPayload(BaseModel):
-    """Import payload — same structure as an export document, but definitions are typed."""
-
-    version: int = 1
-    source: ExportSource | None = None
-    definitions: list[TestDefinitionExport]
+# --- Test Definition Export/Import (wire types) ---
 
 
 class ImportRequest(BaseModel):
     config: ImportConfig
     payload: ImportPayload
 
 
-class ImportItemTD(BaseModel):
-    idx: int | None = None
-    target_id: UUID | None = None
-
-
-class ImportItem(BaseModel):
-    action: ImportAction
-    reason: ImportReason
-    tds: list[ImportItemTD]
-
-
-class ImportSummary(BaseModel):
-    created: int = 0
-    updated: int = 0
-    skipped: int = 0
-    deleted: int = 0
-
-
-class ImportResponse(BaseModel):
-    summary: ImportSummary
-    items: list[ImportItem]
-
-
 class ImportStrictError(ErrorResponse):
     """400 response for apply_strict when entries would be skipped."""
 

testgen/api/test_definitions.py

@@ -2,19 +2,23 @@
 
 from fastapi import APIRouter, Depends, HTTPException, Query
 
-from testgen.api import test_definition_service
-from testgen.api.deps import db_session, resolve_test_suite
+from testgen.api.deps import api_error, db_session, resolve_test_suite
 from testgen.api.schemas import (
     ErrorDetail,
     ErrorResponse,
-    ExportDocument,
-    ImportMode,
     ImportRequest,
-    ImportResponse,
     ImportStrictError,
-    Origin,
 )
 from testgen.common.models.test_suite import TestSuite
+from testgen.common.test_definition_export_import_service import (
+    ExportDocument,
+    ImportResponse,
+    ImportStrictViolation,
+    InvalidImportPayload,
+    Origin,
+    export_definitions,
+    import_definitions,
+)
 
 _error_responses = {
     404: {"model": ErrorResponse, "description": "Not found"},
@@ -39,33 +43,33 @@ def export_test_definitions(
     test_type: str | None = Query(default=None),
 ) -> ExportDocument:
     """Export test definitions from a test suite as a portable JSON document."""
-    return test_definition_service.export_definitions(test_suite, origin, table_name, test_type)
+    return export_definitions(test_suite, origin, table_name, test_type)
 
 
 @router.post(
     "/test-suites/{test_suite_id}/test-definition-import",
     response_model=ImportResponse,
     responses={
-        400: {"model": ImportStrictError, "description": "Invalid request or strict validation failed"},
+        400: {
+            "model": ImportStrictError | ErrorResponse,
+            "description": "Strict validation failed (includes the projected import result) or invalid payload",
+        },
     },
 )
 def import_test_definitions(
     body: ImportRequest,
     test_suite: TestSuite = resolve_test_suite("edit"),  # noqa: B008
 ) -> ImportResponse:
     """Import test definitions into a test suite from a portable JSON document."""
-    result = test_definition_service.import_definitions(test_suite, body.config, body.payload)
-
-    if body.config.mode == ImportMode.apply_strict and result.summary.skipped > 0:
+    try:
+        return import_definitions(test_suite, body.config, body.payload)
+    except InvalidImportPayload as err:
+        raise api_error(400, err.code, str(err)) from err
+    except ImportStrictViolation as err:
         raise HTTPException(
             status_code=400,
             detail=ImportStrictError(
-                errors=[ErrorDetail(
-                    code="strict_validation_failed",
-                    detail=f"{result.summary.skipped} test definition(s) would be skipped",
-                )],
-                import_result=result,
+                errors=[ErrorDetail(code="strict_validation_failed", detail=str(err))],
+                import_result=err.result,
             ).model_dump(mode="json"),
-        )
-
-    return result
+        ) from err