DataOps-platform/tests/data_research/test_data_element_api.py

from __future__ import annotations

from dataclasses import replace

import pytest

from app.core.data_research.data_elements import DataElementRecord


class FakeElementService:
    def __init__(self):
        self.record = DataElementRecord(
            uid="element-1",
            code="ORDER_ID",
            status="draft",
            current_version=1,
            snapshot={"definition": "订单唯一标识"},
            created_by="editor-1",
            updated_by="editor-1",
        )
        self.actions = []

    def create_draft(self, payload, actor_uid):
        self.actions.append(("create", payload, actor_uid))
        return self.record

    def list(self, *, status=None, business_domain_uid=None):
        self.actions.append(("list", status, business_domain_uid))
        return [replace(self.record, status="published")]

    def transition(self, uid, target_status, *, expected_version, actor_uid):
        self.actions.append(("transition", uid, target_status, expected_version, actor_uid))
        return replace(
            self.record,
            uid=uid,
            status=target_status,
            current_version=expected_version + 1,
            updated_by=actor_uid,
        )


class FakeDecisionService:
    def __init__(self):
        self.calls = []

    def decide(self, items, *, actor_uid):
        self.calls.append((items, actor_uid))
        return [
            type(
                "Decision",
                (),
                {
                    "uid": "decision-1",
                    "candidate_uid": "candidate-1",
                    "action": "ignore",
                    "data_element_uid": None,
                    "evidence_uids": ("evidence-1",),
                    "actor_uid": actor_uid,
                    "reason": "重复字段",
                },
            )()
        ]


@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

    elements = FakeElementService()
    decisions = FakeDecisionService()

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

    monkeypatch.setattr(permissions, "authenticate_request", identity)
    monkeypatch.setattr(routes, "get_data_element_service", lambda: elements)
    monkeypatch.setattr(routes, "get_candidate_decision_service", lambda: decisions)
    app = create_app()
    app.config.update(TESTING=True)
    return app.test_client(), elements, decisions


def test_editor_creates_draft_and_submits_review(client):
    http, elements, _decisions = client
    headers = {"Authorization": "Bearer editor"}
    created = http.post(
        "/api/development/v1/data-elements",
        headers=headers,
        json={"code": "ORDER_ID", "definition": "订单唯一标识"},
    )
    reviewed = http.post(
        "/api/development/v1/data-elements/element-1/transitions",
        headers=headers,
        json={"target_status": "in_review", "expected_version": 1},
    )

    assert created.status_code == 201
    assert created.get_json()["data"]["status"] == "draft"
    assert reviewed.status_code == 200
    assert reviewed.get_json()["data"]["status"] == "in_review"
    assert elements.actions[-1] == (
        "transition",
        "element-1",
        "in_review",
        1,
        "editor-1",
    )


def test_viewer_lists_published_elements_for_business_domain(client):
    http, elements, _decisions = client
    listed = http.get(
        "/api/development/v1/data-elements"
        "?status=published&business_domain_uid=domain-1",
        headers={"Authorization": "Bearer editor"},
    )

    assert listed.status_code == 200
    assert listed.get_json()["data"]["records"][0]["uid"] == "element-1"
    assert elements.actions[-1] == ("list", "published", "domain-1")


def test_only_publisher_can_publish(client):
    http, _elements, _decisions = client
    payload = {"target_status": "published", "expected_version": 2}

    denied = http.post(
        "/api/development/v1/data-elements/element-1/transitions",
        headers={"Authorization": "Bearer editor"},
        json=payload,
    )
    allowed = http.post(
        "/api/development/v1/data-elements/element-1/transitions",
        headers={"Authorization": "Bearer admin"},
        json=payload,
    )

    assert denied.status_code == 403
    assert allowed.status_code == 200
    assert allowed.get_json()["data"]["current_version"] == 3


def test_candidate_decisions_are_editor_gated(client):
    http, _elements, decisions = client
    response = http.post(
        "/api/development/v1/candidate-decisions",
        headers={"Authorization": "Bearer editor"},
        json={"decisions": [{"candidate_uid": "candidate-1", "action": "ignore"}]},
    )

    assert response.status_code == 200
    assert response.get_json()["data"][0]["evidence_uids"] == ["evidence-1"]
    assert decisions.calls[0][1] == "editor-1"