DataOps-platform/tests/test_responsibility_api.py

from __future__ import annotations


USER_A = "01900000-0000-7000-8000-000000000101"


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

    def get(self, resource_type, resource_uid):
        self.calls.append(("get", resource_type, resource_uid))
        return {
            "resource_type": resource_type,
            "resource_uid": resource_uid,
            "revision": 2,
            "assignments": [],
        }

    def replace(self, **kwargs):
        self.calls.append(("replace", kwargs))
        return {
            "resource_type": kwargs["resource_type"],
            "resource_uid": kwargs["resource_uid"],
            "revision": kwargs["expected_revision"] + 1,
            "assignments": [
                {
                    **item,
                    "username": "asset_admin",
                    "display_name": "设备资产管理员",
                }
                for item in kwargs["assignments"]
            ],
        }


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


def test_responsibility_matrix_is_readable_but_only_admin_can_replace(monkeypatch):
    from app import create_app
    from app.api.system import responsibilities

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

    readable = client.get(
        "/api/system/responsibilities/device_asset/device-1",
        headers=_headers("viewer"),
    )
    assert readable.status_code == 200
    assert readable.headers["ETag"] == '"2"'

    payload = {
        "assignments": [
            {
                "user_id": USER_A,
                "responsibility_role": "asset_manager",
                "raci_role": "accountable",
            }
        ]
    }
    forbidden = client.put(
        "/api/system/responsibilities/device_asset/device-1",
        json=payload,
        headers=_headers("editor", **{"If-Match": '"2"'}),
    )
    assert forbidden.status_code == 403

    missing_revision = client.put(
        "/api/system/responsibilities/device_asset/device-1",
        json=payload,
        headers=_headers("admin"),
    )
    assert missing_revision.status_code == 428

    updated = client.put(
        "/api/system/responsibilities/device_asset/device-1",
        json=payload,
        headers=_headers("admin", **{"If-Match": '"2"'}),
    )
    assert updated.status_code == 200
    assert updated.headers["ETag"] == '"3"'
    assert service.calls[-1][1]["actor_uid"] == USER_A


def test_responsibility_paths_have_dedicated_permissions():
    from app.core.system.permissions import (
        RESPONSIBILITIES_MANAGE,
        RESPONSIBILITIES_READ,
        permission_for_request,
    )

    path = "/api/system/responsibilities/device_asset/device-1"
    assert permission_for_request(path, "GET") == (RESPONSIBILITIES_READ,)
    assert permission_for_request(path, "PUT") == (RESPONSIBILITIES_MANAGE,)