DataOps-platform/tests/data_research/test_ingestion_models.py

from __future__ import annotations

from sqlalchemy import CheckConstraint, UniqueConstraint


def test_ingestion_models_expose_stable_control_plane_tables():
    from app.models.data_research import (
        EvidenceFragment,
        ExtractionCandidate,
        IngestionJob,
        IngestionSource,
        SourceArtifact,
    )

    assert IngestionSource.__tablename__ == "ingestion_sources"
    assert IngestionJob.__tablename__ == "ingestion_jobs"
    assert SourceArtifact.__tablename__ == "source_artifacts"
    assert EvidenceFragment.__tablename__ == "evidence_fragments"
    assert ExtractionCandidate.__tablename__ == "extraction_candidates"

    assert IngestionJob.__table__.c.source_uid.foreign_keys
    assert IngestionJob.__table__.c.artifact_uid.foreign_keys
    assert EvidenceFragment.__table__.c.job_uid.foreign_keys
    assert ExtractionCandidate.__table__.c.job_uid.foreign_keys


def test_ingestion_job_has_idempotency_and_status_constraints():
    from app.models.data_research import IngestionJob

    constraints = list(IngestionJob.__table__.constraints)
    unique_columns = {
        tuple(column.name for column in constraint.columns)
        for constraint in constraints
        if isinstance(constraint, UniqueConstraint)
    }
    check_sql = " ".join(
        str(constraint.sqltext)
        for constraint in constraints
        if isinstance(constraint, CheckConstraint)
    )

    assert ("idempotency_key",) in unique_columns
    assert "attempt_count" in IngestionJob.__table__.c
    assert "failure_stage" in IngestionJob.__table__.c
    assert "attempt_count >= 0" in check_sql
    for status in (
        "created",
        "queued",
        "extracting",
        "normalizing",
        "matching",
        "awaiting_review",
        "published",
        "partial",
        "failed",
        "cancelled",
    ):
        assert status in check_sql


def test_catalog_snapshot_has_one_immutable_record_per_job_attempt():
    from app.models.data_research import CatalogSnapshot

    constraints = list(CatalogSnapshot.__table__.constraints)
    unique_columns = {
        tuple(column.name for column in constraint.columns)
        for constraint in constraints
        if isinstance(constraint, UniqueConstraint)
    }

    assert CatalogSnapshot.__tablename__ == "catalog_snapshots"
    assert CatalogSnapshot.__table__.c.job_uid.foreign_keys
    assert ("job_uid", "attempt") in unique_columns
    assert "content_hash" in CatalogSnapshot.__table__.c


def test_model_serialization_redacts_control_plane_secrets():
    from app.models.data_research import IngestionSource, SourceArtifact

    source = IngestionSource(
        uid="00000000-0000-0000-0000-000000000001",
        source_type="database",
        name="orders",
        config={"secret_ref": "vault://orders", "schema": "public"},
        permission_scope={"roles": ["editor"]},
        status="active",
    )
    artifact = SourceArtifact(
        uid="00000000-0000-0000-0000-000000000002",
        source_uid=source.uid,
        filename="orders.csv",
        media_type="text/csv",
        size_bytes=12,
        content_hash="a" * 64,
        storage_ref="minio://private/orders.csv?token=secret",
        parser_version="csv-v1",
    )

    assert "config" not in source.to_dict()
    assert "vault://orders" not in str(source.to_dict())
    assert "storage_ref" not in artifact.to_dict()
    assert "token=secret" not in str(artifact.to_dict())
    assert artifact.to_dict()["content_hash"] == "a" * 64