DataOps-platform/tests/data_research/test_semantic_governance_api.py

from __future__ import annotations

import uuid

import pytest


class FakeSemanticService:
    def __init__(self):
        self.calls = []
        self.asset = {
            "uid": str(uuid.uuid4()),
            "asset_kind": "business_term",
            "code": "CUSTOMER_LEVEL",
            "name": "客户等级",
            "owner_uid": str(uuid.uuid4()),
            "business_domain_uid": str(uuid.uuid4()),
            "status": "draft",
            "current_version": 1,
            "definition": {"definition": "客户运营分层等级", "aliases": []},
        }

    def list(self, **filters):
        self.calls.append(("list", filters))
        return [self.asset]

    def create_draft(self, kind, payload, *, actor_uid):
        self.calls.append(("create", kind, payload, actor_uid))
        return self.asset

    def submit(self, uid, *, expected_version, actor_uid):
        self.calls.append(("submit", uid, expected_version, actor_uid))
        return {**self.asset, "status": "in_review"}

    def review(self, uid, **payload):
        self.calls.append(("review", uid, payload))
        return {**self.asset, "status": "approved"}

    def publish(self, uid, *, expected_version, actor_uid):
        self.calls.append(("publish", uid, expected_version, actor_uid))
        return {**self.asset, "status": "published"}

    def rollback(self, uid, **payload):
        self.calls.append(("rollback", uid, payload))
        return {
            **self.asset,
            "status": "published",
            "current_version": 2,
        }

    def list_field_mappings(self, **filters):
        self.calls.append(("mappings", filters))
        return []

    def map_physical_field(self, payload, *, actor_uid):
        self.calls.append(("map", payload, actor_uid))
        return {"uid": str(uuid.uuid4()), **payload, "status": "published"}


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

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

    service = FakeSemanticService()

    def identity():
        role = request.headers.get("Authorization", "").removeprefix("Bearer ")
        if role not in {"viewer", "editor", "admin"}:
            return None
        return {"id": f"00000000-0000-7000-8000-0000000000{role[0]}", "roles": [role]}

    monkeypatch.setattr(permissions, "authenticate_request", identity)
    monkeypatch.setattr(
        semantic_routes,
        "get_semantic_governance_service",
        lambda: service,
    )
    app = create_app()
    app.config.update(TESTING=True)
    return app.test_client(), service


def test_viewer_lists_but_cannot_create(client):
    http, service = client
    listed = http.get(
        "/api/development/v1/semantic-assets?asset_kind=business_term",
        headers={"Authorization": "Bearer viewer"},
    )
    denied = http.post(
        "/api/development/v1/semantic-assets",
        headers={"Authorization": "Bearer viewer"},
        json={"asset_kind": "business_term"},
    )

    assert listed.status_code == 200
    assert listed.get_json()["data"][0]["code"] == "CUSTOMER_LEVEL"
    assert service.calls[0] == (
        "list",
        {"asset_kind": "business_term", "status": None},
    )
    assert denied.status_code == 403


def test_editor_creates_submits_and_maps_but_cannot_approve(client):
    http, service = client
    headers = {"Authorization": "Bearer editor"}
    created = http.post(
        "/api/development/v1/semantic-assets",
        headers=headers,
        json={"asset_kind": "business_term", "code": "CUSTOMER_LEVEL"},
    )
    submitted = http.post(
        f"/api/development/v1/semantic-assets/{service.asset['uid']}/submit",
        headers=headers,
        json={"expected_version": 1},
    )
    mapped = http.post(
        "/api/development/v1/semantic-field-mappings",
        headers=headers,
        json={
            "asset_uid": str(uuid.uuid4()),
            "field_name": "customer_level",
            "data_element_uid": str(uuid.uuid4()),
            "owner_uid": str(uuid.uuid4()),
        },
    )
    denied = http.post(
        f"/api/development/v1/semantic-assets/{service.asset['uid']}/review",
        headers=headers,
        json={
            "expected_version": 1,
            "decision": "approve",
            "reason": "approved",
        },
    )

    assert created.status_code == 201
    assert submitted.get_json()["data"]["status"] == "in_review"
    assert mapped.status_code == 201
    assert denied.status_code == 403


def test_admin_approves_publishes_and_rolls_back(client):
    http, service = client
    headers = {"Authorization": "Bearer admin"}
    reviewed = http.post(
        f"/api/development/v1/semantic-assets/{service.asset['uid']}/review",
        headers=headers,
        json={
            "expected_version": 1,
            "decision": "approve",
            "reason": "企业术语审核通过",
        },
    )
    published = http.post(
        f"/api/development/v1/semantic-assets/{service.asset['uid']}/publish",
        headers=headers,
        json={"expected_version": 1},
    )
    rolled_back = http.post(
        f"/api/development/v1/semantic-assets/{service.asset['uid']}/rollback",
        headers=headers,
        json={
            "expected_version": 1,
            "target_version": 1,
            "reason": "恢复验收版本",
        },
    )

    assert reviewed.get_json()["data"]["status"] == "approved"
    assert published.get_json()["data"]["status"] == "published"
    assert rolled_back.get_json()["data"]["current_version"] == 2