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")


@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)

    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)
    app = create_app()
    app.config.update(TESTING=True)
    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