DataOps-platform/tests/test_product_governance_api.py

from __future__ import annotations

USER_UID = "01900000-0000-7000-8000-000000008801"
PRODUCT_UID = "01900000-0000-7000-8000-000000008802"
APPLICATION_UID = "01900000-0000-7000-8000-000000008803"
CONTRACT_UID = "01900000-0000-7000-8000-000000008804"


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

    def list_products(self, **filters):
        self.calls.append(("list_products", filters))
        return [{"uid": PRODUCT_UID, "name": "设备健康产品", "status": "draft"}]

    def register_product(self, payload, actor_uid):
        self.calls.append(("register_product", payload, actor_uid))
        return {"uid": PRODUCT_UID, "current_version": 1, **payload}

    def transition_product(self, uid, payload, expected_version, actor_uid):
        self.calls.append(("transition_product", uid, payload, expected_version, actor_uid))
        return {"uid": uid, "status": "in_review", "current_version": 2}

    def create_application(self, payload, actor_uid):
        self.calls.append(("create_application", payload, actor_uid))
        return {"uid": APPLICATION_UID, "current_version": 1, **payload}

    def submit_application(self, uid, payload, expected_version, actor_uid):
        self.calls.append(("submit_application", uid, payload, expected_version, actor_uid))
        return {"uid": uid, "status": "pending_approval", "current_version": 2}

    def create_contract(self, product_uid, payload, actor_uid):
        self.calls.append(("create_contract", product_uid, payload, actor_uid))
        return {"uid": CONTRACT_UID, "product_uid": product_uid, "current_version": 1}

    def generate_certificate(self, product_uid, payload, actor_uid):
        self.calls.append(("generate_certificate", product_uid, payload, actor_uid))
        return {"uid": "certificate-1", "product_uid": product_uid, "status": "qualified"}

    def dashboard(self):
        return {"product_count": 1, "qualified_count": 1}


def _headers(role, **extra):
    return {"Authorization": f"Bearer {role}", **extra}


def _client(monkeypatch):
    from app import create_app
    from app.api.data_service import product_governance_routes

    service = FakeProductGovernanceService()
    monkeypatch.setattr(product_governance_routes, "_service", lambda: service)
    monkeypatch.setattr(
        "app.core.system.auth.load_identity_from_token",
        lambda token, secret: (
            {"id": USER_UID, "username": token, "roles": [token]}
            if token in {"viewer", "editor", "admin"}
            else None
        ),
    )
    app = create_app()
    app.config.update(TESTING=True)
    return app.test_client(), service


def test_viewer_can_read_governed_products_and_dashboard(monkeypatch):
    client, service = _client(monkeypatch)
    products = client.get(
        "/api/dataservice/governance/products?status=draft",
        headers=_headers("viewer"),
    )
    assert products.status_code == 200
    assert products.get_json()["data"][0]["uid"] == PRODUCT_UID
    assert service.calls[-1] == ("list_products", {"status": "draft"})

    dashboard = client.get(
        "/api/dataservice/governance/dashboard", headers=_headers("viewer")
    )
    assert dashboard.status_code == 200
    assert dashboard.get_json()["data"]["qualified_count"] == 1


def test_editor_creates_and_submits_application_but_cannot_register_product(monkeypatch):
    client, service = _client(monkeypatch)
    payload = {
        "title": "设备健康数据申请",
        "source_type": "api",
        "source_ref": {"endpoint": "/health"},
    }
    created = client.post(
        "/api/dataservice/governance/applications",
        json=payload,
        headers=_headers("editor"),
    )
    assert created.status_code == 201
    assert created.headers["ETag"] == '"1"'

    missing_version = client.post(
        f"/api/dataservice/governance/applications/{APPLICATION_UID}/submit",
        json={"workflow_uid": "workflow-1"},
        headers=_headers("editor"),
    )
    assert missing_version.status_code == 428
    submitted = client.post(
        f"/api/dataservice/governance/applications/{APPLICATION_UID}/submit",
        json={"workflow_uid": "workflow-1"},
        headers=_headers("editor", **{"If-Match": '"1"'}),
    )
    assert submitted.status_code == 200
    assert submitted.headers["ETag"] == '"2"'
    assert service.calls[-1][3] == 1

    forbidden = client.post(
        "/api/dataservice/governance/products",
        json={"name": "不得登记"},
        headers=_headers("editor"),
    )
    assert forbidden.status_code == 403
    owner_transition = client.post(
        f"/api/dataservice/governance/products/{PRODUCT_UID}/transition",
        json={"action": "submit_review", "reason": "责任人提交"},
        headers=_headers("editor", **{"If-Match": '"1"'}),
    )
    assert owner_transition.status_code == 200


def test_admin_manages_lifecycle_contract_and_certificate(monkeypatch):
    client, service = _client(monkeypatch)
    product = client.post(
        "/api/dataservice/governance/products",
        json={"name": "设备健康产品"},
        headers=_headers("admin"),
    )
    assert product.status_code == 201
    assert product.headers["ETag"] == '"1"'

    transitioned = client.post(
        f"/api/dataservice/governance/products/{PRODUCT_UID}/transition",
        json={"action": "submit_review", "reason": "治理评审"},
        headers=_headers("admin", **{"If-Match": '"1"'}),
    )
    assert transitioned.status_code == 200
    assert transitioned.headers["ETag"] == '"2"'

    contract = client.post(
        f"/api/dataservice/governance/products/{PRODUCT_UID}/contracts",
        json={"definition": {"schema": {"fields": []}}},
        headers=_headers("admin"),
    )
    assert contract.status_code == 201
    certificate = client.post(
        f"/api/dataservice/governance/products/{PRODUCT_UID}/certificates",
        json={"approval_task_uid": "task-1", "evidence_refs": {}},
        headers=_headers("admin"),
    )
    assert certificate.status_code == 201
    assert certificate.get_json()["data"]["status"] == "qualified"