DataOps-platform/tests/data_research/test_ontology_dynamic_api.py

from __future__ import annotations

import pytest


class DynamicService:
    def generate(self, ontology_uid, payload, actor_uid):
        return [{"uid": "suggestion-1", "kind": "class", "confidence": 0.95, "evidence_uids": ["domain-1"]}]

    def decide(self, ontology_uid, change_set_uid, decisions, actor_uid):
        return {"uid": change_set_uid, "ontology_uid": ontology_uid, "decisions": decisions, "actor_uid": actor_uid}


class ExchangeService:
    def export(self, ontology_uid, format_name):
        assert (ontology_uid, format_name) == ("ontology-1", "json")
        return b'{"ontology_uid":"ontology-1"}', "application/json", "ontology-1.json"

    def import_document(self, content, format_name, actor_uid):
        return {"ontology_uid": "ontology-2", "version": 1, "format": format_name, "actor_uid": actor_uid}


class SemanticService:
    def trace_property(self, uid, *, allowed_domains, limit, after_uid):
        return {"items": [{"property_uid": uid, "business_domain_uid": next(iter(allowed_domains))}], "count": 1, "truncated": False, "next_cursor": None}


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

    from app import create_app
    from app.api.data_development import routes
    from app.core.system import permissions

    def identity():
        role = request.headers.get("Authorization", "").removeprefix("Bearer ")
        return {"id": f"{role}-1", "roles": [role]} if role in {"viewer", "editor", "admin"} else None

    monkeypatch.setattr(permissions, "authenticate_request", identity)
    monkeypatch.setattr(routes, "get_ontology_dynamic_service", lambda: DynamicService())
    monkeypatch.setattr(routes, "get_ontology_exchange_service", lambda: ExchangeService())
    monkeypatch.setattr(routes, "get_semantic_query_service", lambda: SemanticService())
    app = create_app()
    app.config.update(TESTING=True)
    return app.test_client()


def test_dynamic_suggestion_and_audited_decision_endpoints(client):
    headers = {"Authorization": "Bearer editor"}
    generated = client.post(
        "/api/development/v1/ontologies/ontology-1/suggestions",
        headers=headers,
        json={"domains": [], "data_elements": [], "foreign_keys": []},
    )
    decided = client.post(
        "/api/development/v1/ontologies/ontology-1/change-sets/change-1/decisions",
        headers=headers,
        json={"decisions": [{"suggestion_uid": "suggestion-1", "decision": "accept"}]},
    )
    assert generated.get_json()["data"][0]["evidence_uids"] == ["domain-1"]
    assert decided.get_json()["data"]["actor_uid"] == "editor-1"


def test_exchange_and_read_only_semantic_query_endpoints(client):
    exported = client.get(
        "/api/development/v1/ontologies/ontology-1/export?format=json",
        headers={"Authorization": "Bearer viewer"},
    )
    imported = client.post(
        "/api/development/v1/ontologies/import?format=json",
        headers={"Authorization": "Bearer editor"},
        data=b'{"ontology_uid":"ontology-2"}',
        content_type="application/json",
    )
    queried = client.get(
        "/api/development/v1/semantic/properties/property-1?business_domain_uid=sales&limit=20",
        headers={"Authorization": "Bearer viewer"},
    )
    assert exported.status_code == 200 and exported.mimetype == "application/json"
    assert imported.status_code == 201
    assert queried.get_json()["data"]["items"][0]["business_domain_uid"] == "sales"