"""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 imported_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 detail = _data( requests.get( f"{base_url}/development/v1/ontologies/{ontology_uid}", headers=headers, timeout=20, ), 200, ) assert detail["uid"] == ontology_uid assert detail["draft_revision"] == 1 graph_response = requests.get( f"{base_url}/development/v1/ontologies/{ontology_uid}/graph", headers=headers, timeout=20, ) graph_snapshot = _data(graph_response, 200) assert graph_response.headers["ETag"] == '"1"' assert graph_snapshot["graph_document"]["classes"][0]["uid"] == "customer" versions = _data( requests.get( f"{base_url}/development/v1/ontologies/{ontology_uid}/versions", headers=headers, timeout=20, ), 200, ) assert versions[0]["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" imported = _data( requests.post( f"{base_url}/development/v1/ontologies/import?format=json", headers=headers, files={ "file": ( "ontology.json", exported.content, "application/json", ) }, timeout=20, ), 201, ) imported_ontology_uid = imported["ontology_uid"] imported_graph = _data( requests.get( f"{base_url}/development/v1/ontologies/" f"{imported_ontology_uid}/graph", headers=headers, timeout=20, ), 200, ) assert imported_graph["graph_document"] == graph 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: for cleanup_ontology_uid in ( imported_ontology_uid, ontology_uid, ): if not cleanup_ontology_uid: continue connection.execute( text( "DELETE FROM public.outbox_consumptions WHERE event_id IN " "(SELECT event_id FROM public.outbox_events " "WHERE aggregate_id = :uid)" ), {"uid": cleanup_ontology_uid}, ) connection.execute( text("DELETE FROM public.outbox_events WHERE aggregate_id = :uid"), {"uid": cleanup_ontology_uid}, ) connection.execute( text( "UPDATE public.ontologies SET active_version_uid = NULL " "WHERE uid = CAST(:uid AS uuid)" ), {"uid": cleanup_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": cleanup_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()