DataOps-platform/tests/integration/test_data_research_docker_api.py

"""Opt-in end-to-end acceptance for the containerized data-research API."""

from __future__ import annotations

import os
import uuid

import pytest
import requests
from minio import Minio
from sqlalchemy import create_engine, text


pytestmark = pytest.mark.skipif(
    os.getenv("RUN_DATA_RESEARCH_DOCKER_API") != "1",
    reason="set RUN_DATA_RESEARCH_DOCKER_API=1 for Docker API acceptance",
)


def _data(response, expected_status):
    assert response.status_code == expected_status, response.text
    payload = response.json()
    assert payload["code"] == 200, payload
    return payload["data"]


def test_file_ingestion_and_dynamic_ontology_publish_through_live_api():
    base_url = os.getenv("TEST_BACKEND_URL", "http://127.0.0.1:15500/api")
    database_url = os.environ["TEST_DATABASE_URL"]
    engine = create_engine(database_url, pool_pre_ping=True)
    minio = Minio(
        os.getenv("TEST_MINIO_HOST", "127.0.0.1:19000"),
        access_key="dataops-test",
        secret_key="dataops-test-password",
        secure=False,
    )
    source_uid = str(uuid.uuid4())
    owner_domain_uid = str(uuid.uuid4())
    contributor_domain_uid = str(uuid.uuid4())
    ontology_uid = None
    artifact_uid = None
    artifact_object_key = None

    login = requests.post(
        f"{base_url}/system/auth/login",
        json={"username": "admin", "password": os.environ["TEST_ADMIN_PASSWORD"]},
        timeout=20,
    )
    token = _data(login, 200)["token"]
    headers = {"Authorization": f"Bearer {token}"}

    try:
        with engine.begin() as connection:
            connection.execute(
                text(
                    "INSERT INTO public.ingestion_sources "
                    "(uid, source_type, name, created_by) "
                    "VALUES (CAST(:uid AS uuid), 'file', :name, 'docker-api-test')"
                ),
                {"uid": source_uid, "name": f"docker-api-{source_uid}"},
            )

        csv_content = b"code,name\nCUSTOMER_ID,Customer identifier\n"
        uploaded = requests.post(
            f"{base_url}/development/v1/sources/files",
            headers=headers,
            data={"source_uid": source_uid, "parser_version": "csv-v1"},
            files={"file": ("elements.csv", csv_content, "text/csv")},
            timeout=20,
        )
        artifact = _data(uploaded, 201)
        artifact_uid = artifact["uid"]

        duplicate = requests.post(
            f"{base_url}/development/v1/sources/files",
            headers=headers,
            data={"source_uid": source_uid, "parser_version": "csv-v1"},
            files={"file": ("elements.csv", csv_content, "text/csv")},
            timeout=20,
        )
        assert _data(duplicate, 200)["uid"] == artifact_uid

        job_payload = {
            "source_uid": source_uid,
            "artifact_uid": artifact_uid,
            "job_type": "file_extract",
            "parser_version": "csv-v1",
            "parameters": {"delimiter": ","},
        }
        created_job = _data(
            requests.post(
                f"{base_url}/development/v1/ingestion-jobs",
                headers=headers,
                json=job_payload,
                timeout=20,
            ),
            201,
        )
        duplicate_job = _data(
            requests.post(
                f"{base_url}/development/v1/ingestion-jobs",
                headers=headers,
                json=job_payload,
                timeout=20,
            ),
            200,
        )
        assert duplicate_job["uid"] == created_job["uid"]
        cancelled = _data(
            requests.post(
                f"{base_url}/development/v1/ingestion-jobs/"
                f"{created_job['uid']}/cancel",
                headers=headers,
                timeout=20,
            ),
            200,
        )
        assert cancelled["status"] == "cancelled"

        ontology = _data(
            requests.post(
                f"{base_url}/development/v1/ontologies",
                headers=headers,
                json={
                    "code": f"DOCKER_{uuid.uuid4().hex[:12].upper()}",
                    "name": "Docker API customer ontology",
                    "domain_links": [
                        {"domain_uid": owner_domain_uid, "role": "owner"},
                        {
                            "domain_uid": contributor_domain_uid,
                            "role": "contributor",
                        },
                    ],
                },
                timeout=20,
            ),
            201,
        )
        ontology_uid = ontology["uid"]
        graph = {
            "classes": [{"uid": "customer", "name": "Customer"}],
            "properties": [
                {
                    "uid": "customer-id",
                    "name": "customerId",
                    "class_uid": "customer",
                    "required": True,
                    "cardinality": "1",
                    "data_element_uid": "customer-id-element",
                }
            ],
            "relations": [],
            "constraints": [],
            "domain_links": [
                {"domain_uid": owner_domain_uid, "role": "owner"},
                {
                    "domain_uid": contributor_domain_uid,
                    "role": "contributor",
                },
            ],
            "element_mappings": [
                {
                    "property_uid": "customer-id",
                    "data_element_uid": "customer-id-element",
                }
            ],
        }
        draft = _data(
            requests.patch(
                f"{base_url}/development/v1/ontologies/{ontology_uid}/graph",
                headers={**headers, "If-Match": '"0"'},
                json=graph,
                timeout=20,
            ),
            200,
        )
        assert draft["version"] == 1
        assert _data(
            requests.post(
                f"{base_url}/development/v1/ontologies/{ontology_uid}/validate",
                headers=headers,
                timeout=20,
            ),
            200,
        ) == []
        published = _data(
            requests.post(
                f"{base_url}/development/v1/ontologies/{ontology_uid}/publish",
                headers={**headers, "Idempotency-Key": f"docker-{ontology_uid}"},
                timeout=20,
            ),
            200,
        )
        assert published["status"] == "published"
        exported = requests.get(
            f"{base_url}/development/v1/ontologies/{ontology_uid}/export?format=json",
            headers=headers,
            timeout=20,
        )
        assert exported.status_code == 200
        assert exported.json()["graph_document"]["classes"][0]["uid"] == "customer"

        with engine.connect() as connection:
            artifact_object_key = connection.execute(
                text(
                    "SELECT storage_ref FROM public.source_artifacts "
                    "WHERE uid = CAST(:uid AS uuid)"
                ),
                {"uid": artifact_uid},
            ).scalar_one().removeprefix("minio://dataops-bucket/")
            outbox_payload = connection.execute(
                text(
                    "SELECT payload FROM public.outbox_events "
                    "WHERE aggregate_id = :uid "
                    "AND event_type = 'ontology.version_published'"
                ),
                {"uid": ontology_uid},
            ).scalar_one()
        assert outbox_payload["ontology_uid"] == ontology_uid
    finally:
        with engine.begin() as connection:
            if ontology_uid:
                connection.execute(
                    text(
                        "DELETE FROM public.outbox_consumptions WHERE event_id IN "
                        "(SELECT event_id FROM public.outbox_events "
                        "WHERE aggregate_id = :uid)"
                    ),
                    {"uid": ontology_uid},
                )
                connection.execute(
                    text("DELETE FROM public.outbox_events WHERE aggregate_id = :uid"),
                    {"uid": ontology_uid},
                )
                connection.execute(
                    text(
                        "UPDATE public.ontologies SET active_version_uid = NULL "
                        "WHERE uid = CAST(:uid AS uuid)"
                    ),
                    {"uid": ontology_uid},
                )
                for table in (
                    "ontology_publish_runs",
                    "ontology_change_sets",
                    "ontology_domain_links",
                    "ontology_versions",
                    "ontologies",
                ):
                    connection.execute(
                        text(
                            f"DELETE FROM public.{table} "
                            "WHERE ontology_uid = CAST(:uid AS uuid)"
                            if table != "ontologies"
                            else "DELETE FROM public.ontologies "
                            "WHERE uid = CAST(:uid AS uuid)"
                        ),
                        {"uid": ontology_uid},
                    )
            connection.execute(
                text(
                    "DELETE FROM public.ingestion_jobs "
                    "WHERE source_uid = CAST(:uid AS uuid)"
                ),
                {"uid": source_uid},
            )
            connection.execute(
                text(
                    "DELETE FROM public.source_artifacts "
                    "WHERE source_uid = CAST(:uid AS uuid)"
                ),
                {"uid": source_uid},
            )
            connection.execute(
                text(
                    "DELETE FROM public.ingestion_sources "
                    "WHERE uid = CAST(:uid AS uuid)"
                ),
                {"uid": source_uid},
            )
        if artifact_object_key:
            minio.remove_object("dataops-bucket", artifact_object_key)
        engine.dispose()