DataOps-platform/tests/data_research/test_development_api.py

from __future__ import annotations

from dataclasses import replace

import pytest


class FakeDevelopmentService:
    def __init__(self, record):
        self.record = record
        self.actions = []

    def create_job(self, payload, actor_uid):
        self.actions.append(("create", payload, actor_uid))
        return replace(self.record, actor_uid=actor_uid), True

    def list_jobs(self, filters=None):
        self.actions.append(("list", filters or {}))
        return [self.record]

    def get_job(self, uid):
        self.actions.append(("get", uid))
        return self.record

    def retry(self, uid):
        self.actions.append(("retry", uid))
        return replace(self.record, status="queued", last_error=None)

    def cancel(self, uid):
        self.actions.append(("cancel", uid))
        return replace(self.record, status="cancelled")


class FakeSourceRegistrar:
    def __init__(self):
        self.actions = []

    def ensure(self, source_uid, actor_uid):
        self.actions.append((source_uid, actor_uid))
        return object(), True


class FakeCatalogExecutor:
    def __init__(self, record):
        self.record = record
        self.actions = []

    def execute(self, uid):
        self.actions.append(uid)
        return replace(
            self.record,
            status="awaiting_review",
            attempt_count=1,
            statistics={
                "snapshot_uid": "snapshot-1",
                "asset_count": 2,
                "field_count": 8,
                "evidence_count": 8,
            },
        )


class FakeCatalogSnapshotRepository:
    def list(self, job_uid):
        from app.core.data_research.catalog.models import CatalogSnapshotRecord

        return [
            CatalogSnapshotRecord(
                uid="snapshot-1",
                job_uid=job_uid,
                source_uid="00000000-0000-0000-0000-000000000001",
                attempt=1,
                database_type="postgresql",
                content_hash="b" * 64,
                snapshot={
                    "data_source_uid": "00000000-0000-0000-0000-000000000001",
                    "database_type": "postgresql",
                    "assets": [],
                },
                evidence_count=8,
            )
        ]


class FakeEvidenceService:
    def list_for_job(self, job_uid):
        return [
            {
                "uid": "evidence-1",
                "job_uid": job_uid,
                "locator": {
                    "kind": "database.column",
                    "schema": "asset",
                    "table": "equipment",
                    "column": "equipment_code",
                },
                "excerpt": "password=[redacted]",
                "confidence": 1.0,
            }
        ]


@pytest.fixture()
def development_client(monkeypatch):
    from flask import request

    from app import create_app
    from app.api.data_development import routes
    from app.core.data_research.models import IngestionJobRecord
    from app.core.system import permissions

    record = IngestionJobRecord(
        uid="00000000-0000-0000-0000-000000000010",
        source_uid="00000000-0000-0000-0000-000000000001",
        artifact_uid="00000000-0000-0000-0000-000000000002",
        job_type="file_extract",
        parser_version="csv-v1",
        idempotency_key="a" * 64,
        actor_uid="editor-1",
        parameters={"schema": "public"},
    )
    service = FakeDevelopmentService(record)
    registrar = FakeSourceRegistrar()
    executor = FakeCatalogExecutor(record)

    def identity():
        header = request.headers.get("Authorization", "")
        if header == "Bearer viewer":
            return {"id": "viewer-1", "roles": ["viewer"]}
        if header == "Bearer editor":
            return {"id": "editor-1", "roles": ["editor"]}
        if header == "Bearer other-editor":
            return {"id": "editor-2", "roles": ["editor"]}
        if header == "Bearer admin":
            return {"id": "admin-1", "roles": ["admin"]}
        return None

    monkeypatch.setattr(permissions, "authenticate_request", identity)
    monkeypatch.setattr(routes, "get_ingestion_service", lambda: service)
    monkeypatch.setattr(
        routes,
        "get_database_source_registration_service",
        lambda: registrar,
    )
    monkeypatch.setattr(
        routes,
        "get_catalog_ingestion_executor",
        lambda: executor,
    )
    monkeypatch.setattr(
        routes,
        "get_catalog_snapshot_repository",
        lambda: FakeCatalogSnapshotRepository(),
    )
    monkeypatch.setattr(
        routes,
        "get_evidence_service",
        lambda: FakeEvidenceService(),
    )
    app = create_app()
    app.config.update(TESTING=True)
    service.registrar = registrar
    service.executor = executor
    return app.test_client(), service


def job_payload():
    return {
        "source_uid": "00000000-0000-0000-0000-000000000001",
        "artifact_uid": "00000000-0000-0000-0000-000000000002",
        "job_type": "file_extract",
        "parser_version": "csv-v1",
        "parameters": {"schema": "public"},
        "password": "must-not-return",
    }


def test_ingestion_api_requires_authentication_and_run_permission(development_client):
    client, _service = development_client

    assert client.post("/api/development/v1/ingestion-jobs", json=job_payload()).status_code == 401
    assert client.post(
        "/api/development/v1/ingestion-jobs",
        json=job_payload(),
        headers={"Authorization": "Bearer viewer"},
    ).status_code == 403


def test_editor_creates_and_lists_secret_free_jobs(development_client):
    client, service = development_client
    created = client.post(
        "/api/development/v1/ingestion-jobs",
        json=job_payload(),
        headers={"Authorization": "Bearer editor"},
    )
    listed = client.get(
        "/api/development/v1/ingestion-jobs?status=created",
        headers={"Authorization": "Bearer editor"},
    )

    assert created.status_code == 201
    assert created.get_json()["data"]["uid"].endswith("0010")
    assert "must-not-return" not in created.get_data(as_text=True)
    assert listed.status_code == 200
    assert listed.get_json()["data"]["total"] == 1
    assert service.actions[-1] == ("list", {"status": "created"})


def test_retry_is_admin_only_and_cancel_is_owner_or_admin(development_client):
    client, _service = development_client
    uid = "00000000-0000-0000-0000-000000000010"

    editor_retry = client.post(
        f"/api/development/v1/ingestion-jobs/{uid}/retry",
        headers={"Authorization": "Bearer editor"},
    )
    admin_retry = client.post(
        f"/api/development/v1/ingestion-jobs/{uid}/retry",
        headers={"Authorization": "Bearer admin"},
    )
    other_cancel = client.post(
        f"/api/development/v1/ingestion-jobs/{uid}/cancel",
        headers={"Authorization": "Bearer other-editor"},
    )
    owner_cancel = client.post(
        f"/api/development/v1/ingestion-jobs/{uid}/cancel",
        headers={"Authorization": "Bearer editor"},
    )

    assert editor_retry.status_code == 403
    assert admin_retry.status_code == 200
    assert other_cancel.status_code == 403
    assert owner_cancel.status_code == 200


def test_catalog_job_registers_source_then_executes_with_report(
    development_client,
):
    client, service = development_client
    payload = {
        **job_payload(),
        "artifact_uid": None,
        "job_type": "catalog_collect",
        "parser_version": "catalog-v1",
    }

    created = client.post(
        "/api/development/v1/ingestion-jobs",
        json=payload,
        headers={"Authorization": "Bearer editor"},
    )
    executed = client.post(
        "/api/development/v1/ingestion-jobs/"
        "00000000-0000-0000-0000-000000000010/execute",
        headers={"Authorization": "Bearer editor"},
    )

    assert created.status_code == 201
    assert service.registrar.actions == [
        ("00000000-0000-0000-0000-000000000001", "editor-1")
    ]
    assert executed.status_code == 200
    assert executed.get_json()["data"]["status"] == "awaiting_review"
    assert executed.get_json()["data"]["attempt_count"] == 1
    assert service.executor.actions == [
        "00000000-0000-0000-0000-000000000010"
    ]


def test_catalog_snapshots_and_evidence_are_readable_without_secrets(
    development_client,
):
    client, _service = development_client
    uid = "00000000-0000-0000-0000-000000000010"

    snapshots = client.get(
        f"/api/development/v1/ingestion-jobs/{uid}/catalog-snapshots",
        headers={"Authorization": "Bearer viewer"},
    )
    evidence = client.get(
        f"/api/development/v1/ingestion-jobs/{uid}/evidence",
        headers={"Authorization": "Bearer viewer"},
    )

    assert snapshots.status_code == 200
    assert snapshots.get_json()["data"]["records"][0]["attempt"] == 1
    assert evidence.status_code == 200
    text = evidence.get_data(as_text=True)
    assert "clear-secret" not in text
    assert "equipment_code" in text