DataOps-platform/tests/test_active_metadata_api.py

from __future__ import annotations

ACTOR_UID = "01900000-0000-7000-8000-000000000101"
PLAN_UID = "01900000-0000-7000-8000-000000000201"
ASSET_UID = "01900000-0000-7000-8000-000000000301"
CORRECTION_UID = "01900000-0000-7000-8000-000000000401"


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

    def list_plans(self):
        return [{"uid": PLAN_UID, "name": "主动发现"}]

    def create_plan(self, payload, *, actor_uid):
        self.calls.append(("create_plan", payload, actor_uid))
        return {"uid": PLAN_UID, **payload}

    def execute(self, plan_uid, payload, *, batch_key, actor_uid):
        self.calls.append(("execute", plan_uid, batch_key, actor_uid))
        return {"uid": "run-1", "status": "completed"}

    def list_assets(self, source_uid):
        return [{"uid": ASSET_UID, "source_uid": source_uid}]

    def list_runs(self, plan_uid):
        return [{"uid": "run-1", "plan_uid": plan_uid}]

    def list_changes(self, run_uid):
        return [{"run_uid": run_uid, "change_type": "field_changed"}]

    def list_lineage(self, run_uid):
        return [{"run_uid": run_uid, "parse_status": "resolved"}]

    def list_health_signals(self, run_uid):
        return [{"run_uid": run_uid, "signal_type": "quality"}]

    def list_corrections(self, asset_uid=None):
        return [{"uid": CORRECTION_UID, "asset_uid": asset_uid}]

    def submit_correction(self, asset_uid, payload, *, actor_uid):
        return {"uid": CORRECTION_UID, "asset_uid": asset_uid, "status": "pending"}

    def resolve_correction(self, correction_uid, **kwargs):
        self.calls.append(("resolve", correction_uid, kwargs))
        return {"uid": correction_uid, "status": "resolved", "current_version": 2}


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


def test_active_metadata_read_operate_and_admin_permissions(monkeypatch):
    from app import create_app
    from app.api.meta_data import active_metadata

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

    assert client.get(
        "/api/meta/active-metadata/plans", headers=_headers("viewer")
    ).status_code == 200
    assert client.post(
        "/api/meta/active-metadata/plans",
        json={"name": "plan"},
        headers=_headers("editor"),
    ).status_code == 403
    assert client.post(
        "/api/meta/active-metadata/plans",
        json={"name": "plan"},
        headers=_headers("admin"),
    ).status_code == 201
    assert client.post(
        f"/api/meta/active-metadata/plans/{PLAN_UID}/runs",
        json={"snapshot": {"assets": []}},
        headers=_headers("viewer", **{"Idempotency-Key": "batch-1"}),
    ).status_code == 403
    executed = client.post(
        f"/api/meta/active-metadata/plans/{PLAN_UID}/runs",
        json={"snapshot": {"assets": []}},
        headers=_headers("editor", **{"Idempotency-Key": "batch-1"}),
    )
    assert executed.status_code == 200
    assert service.calls[-1] == ("execute", PLAN_UID, "batch-1", ACTOR_UID)

    resolved = client.post(
        f"/api/meta/active-metadata/corrections/{CORRECTION_UID}/resolve",
        json={
            "expected_version": 1,
            "decision": "accept",
            "resolution": {"comment": "accepted"},
        },
        headers=_headers("editor"),
    )
    assert resolved.status_code == 200


def test_active_metadata_paths_have_dedicated_policy():
    from app.core.system.permissions import (
        ACTIVE_METADATA_MANAGE,
        ACTIVE_METADATA_OPERATE,
        ACTIVE_METADATA_READ,
        permission_for_request,
    )

    base = "/api/meta/active-metadata"
    assert permission_for_request(f"{base}/plans", "GET") == (ACTIVE_METADATA_READ,)
    assert permission_for_request(f"{base}/plans", "POST") == (
        ACTIVE_METADATA_MANAGE,
    )
    assert permission_for_request(f"{base}/plans/{PLAN_UID}/runs", "POST") == (
        ACTIVE_METADATA_OPERATE,
    )
    assert permission_for_request(f"{base}/runs/run-1/lineage", "GET") == (
        ACTIVE_METADATA_READ,
    )