DataOps-platform/tests/data_research/test_ontology_api.py

from __future__ import annotations

from dataclasses import replace

import pytest

from app.core.data_research.ontology.models import DomainLink, GraphDocument, Ontology, OntologyVersion
from app.core.data_research.ontology.validation import ValidationIssue


class FakeOntologyService:
    def __init__(self):
        self.ontology = Ontology(
            uid="ontology-1",
            code="CUSTOMER",
            name="客户本体",
            owner_uid="owner-1",
            domain_links=(DomainLink("domain-1", "owner"), DomainLink("domain-2", "contributor")),
        )
        self.version = OntologyVersion(
            uid="version-1",
            ontology_uid="ontology-1",
            version=1,
            graph_document=GraphDocument.from_dict({"domain_links": [{"domain_uid": "domain-1", "role": "owner"}]}),
            content_hash="a" * 64,
        )
        self.calls = []

    def list(self):
        return [self.ontology]

    def get(self, uid):
        assert uid == self.ontology.uid
        return self.ontology

    def list_versions(self, uid):
        assert uid == self.ontology.uid
        return [self.version]

    def latest_version(self, uid):
        assert uid == self.ontology.uid
        return self.version

    def create(self, payload, actor_uid):
        self.calls.append(("create", payload, actor_uid))
        return self.ontology

    def save_draft(self, uid, graph, expected_revision, actor_uid):
        self.calls.append(("draft", uid, expected_revision, actor_uid))
        return self.version

    def validate(self, uid):
        self.calls.append(("validate", uid))
        return [ValidationIssue("ONTOLOGY_OWNER_REQUIRED", "owner required", "domain_links")]

    def publish(self, uid, idempotency_key, actor_uid):
        self.calls.append(("publish", uid, idempotency_key, actor_uid))
        return replace(self.version, status="published")

    def diff(self, uid, left, right):
        return {"classes": {"added": [], "removed": [], "changed": []}}

    def rollback(self, uid, target_version_uid, expected_revision, actor_uid):
        self.calls.append(("rollback", uid, target_version_uid, expected_revision, actor_uid))
        return replace(self.version, uid="version-2", version=2, parent_version_uid="version-1")


@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

    service = FakeOntologyService()

    def identity():
        token = request.headers.get("Authorization", "")
        role = token.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_service", lambda: service)
    app = create_app()
    app.config.update(TESTING=True)
    return app.test_client(), service


def test_viewer_is_read_only_and_multi_domain_roles_are_returned(client):
    http, _service = client
    headers = {"Authorization": "Bearer viewer"}
    listed = http.get("/api/development/v1/ontologies", headers=headers)
    denied = http.post(
        "/api/development/v1/ontologies",
        headers=headers,
        json={"code": "CUSTOMER", "name": "客户本体"},
    )
    assert listed.status_code == 200
    assert listed.get_json()["data"][0]["domain_links"][1]["role"] == "contributor"
    assert denied.status_code == 403


def test_viewer_loads_ontology_detail_versions_and_latest_graph(client):
    http, _service = client
    headers = {"Authorization": "Bearer viewer"}

    detail = http.get(
        "/api/development/v1/ontologies/ontology-1", headers=headers
    )
    versions = http.get(
        "/api/development/v1/ontologies/ontology-1/versions", headers=headers
    )
    graph = http.get(
        "/api/development/v1/ontologies/ontology-1/graph", headers=headers
    )

    assert detail.status_code == 200
    assert detail.get_json()["data"]["code"] == "CUSTOMER"
    assert versions.status_code == 200
    assert versions.get_json()["data"][0]["uid"] == "version-1"
    assert graph.status_code == 200
    assert graph.headers["ETag"] == '"0"'
    assert graph.get_json()["data"]["graph_document"]["domain_links"][0] == {
        "domain_uid": "domain-1",
        "role": "owner",
    }


def test_editor_saves_etag_draft_but_cannot_publish(client):
    http, service = client
    headers = {"Authorization": "Bearer editor", "If-Match": '"0"'}
    saved = http.patch(
        "/api/development/v1/ontologies/ontology-1/graph",
        headers=headers,
        json={"classes": [], "domain_links": []},
    )
    denied = http.post(
        "/api/development/v1/ontologies/ontology-1/publish",
        headers={"Authorization": "Bearer editor", "Idempotency-Key": "publish-1"},
    )
    assert saved.status_code == 200
    assert service.calls[-1][2] == 0
    assert denied.status_code == 403


def test_admin_validates_publishes_diffs_and_rolls_back(client):
    http, service = client
    headers = {"Authorization": "Bearer admin"}
    validation = http.post("/api/development/v1/ontologies/ontology-1/validate", headers=headers)
    published = http.post(
        "/api/development/v1/ontologies/ontology-1/publish",
        headers={**headers, "Idempotency-Key": "publish-1"},
    )
    diffed = http.get(
        "/api/development/v1/ontologies/ontology-1/diff?left=version-0&right=version-1",
        headers=headers,
    )
    rolled_back = http.post(
        "/api/development/v1/ontologies/ontology-1/rollback",
        headers=headers,
        json={"target_version_uid": "version-1", "expected_revision": 1},
    )
    assert validation.status_code == 200
    assert validation.get_json()["data"][0]["code"] == "ONTOLOGY_OWNER_REQUIRED"
    assert published.get_json()["data"]["status"] == "published"
    assert diffed.status_code == 200
    assert rolled_back.get_json()["data"]["parent_version_uid"] == "version-1"