DataOps-platform/tests/test_unified_responsibility_api.py

from __future__ import annotations

USER_A = "01900000-0000-7000-8000-000000006101"
POLICY_UID = "01900000-0000-7000-8000-000000006102"
DELEGATION_UID = "01900000-0000-7000-8000-000000006103"


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

    def resolve(self, resource_type, resource_uid, at=None):
        self.calls.append(("resolve", resource_type, resource_uid, at))
        return {
            "resource_type": resource_type,
            "resource_uid": resource_uid,
            "status": "resolved",
            "final_owners": [{"effective_user_id": USER_A}],
            "grants_data_access": False,
        }

    def set_parent(self, payload, expected_revision, actor_uid):
        self.calls.append(("set_parent", payload, expected_revision, actor_uid))
        return {**payload, "revision": expected_revision + 1}

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

    def list_delegations(self, status=None):
        self.calls.append(("list_delegations", status))
        return []

    def revoke_delegation(self, uid, expected_version, actor_uid):
        self.calls.append(("revoke", uid, expected_version, actor_uid))
        return {"uid": uid, "status": "revoked", "current_version": 2}

    def expire_delegations(self, at=None, actor_uid=None):
        self.calls.append(("expire", at, actor_uid))
        return []

    def transfer_departing_user(self, payload, actor_uid):
        self.calls.append(("transfer", payload, actor_uid))
        return {"uid": DELEGATION_UID, "delegation_type": "departure_transfer"}

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

    def list_policies(self):
        self.calls.append(("list_policies",))
        return []

    def revise_policy(self, uid, definition, expected_version, actor_uid):
        self.calls.append(("revise_policy", uid, definition, expected_version, actor_uid))
        return {"uid": uid, "current_version": expected_version + 1}

    def publish_policy(self, uid, expected_version, actor_uid):
        self.calls.append(("publish_policy", uid, expected_version, actor_uid))
        return {"uid": uid, "status": "published", "current_version": expected_version}

    def evaluate_joint_review(self, resource_type, resource_uid, decisions):
        self.calls.append(("evaluate", resource_type, resource_uid, decisions))
        return {"status": "approved", "deterministic": True}

    def operations(self, owner_uid):
        self.calls.append(("operations", owner_uid))
        return {"owner_uid": owner_uid, "tasks": [], "metrics": [], "grants_data_access": False}


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


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

    service = FakeUnifiedService()
    monkeypatch.setattr(responsibilities, "_unified_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)
    return app.test_client(), service


def test_effective_owner_is_readable_without_granting_access(monkeypatch):
    client, service = _client(monkeypatch)

    response = client.get(
        "/api/system/responsibilities/data_asset/asset-1/resolved",
        headers=_headers("viewer"),
    )

    assert response.status_code == 200
    assert response.get_json()["data"]["status"] == "resolved"
    assert response.get_json()["data"]["grants_data_access"] is False
    assert service.calls[-1][:3] == ("resolve", "data_asset", "asset-1")


def test_hierarchy_delegation_and_policy_mutations_are_admin_only_and_versioned(monkeypatch):
    client, service = _client(monkeypatch)
    hierarchy = {"parent_type": "business_domain", "parent_uid": "domain-1"}

    forbidden = client.put(
        "/api/system/responsibilities/hierarchy/data_asset/asset-1",
        json=hierarchy,
        headers=_headers("editor", **{"If-Match": '"0"'}),
    )
    assert forbidden.status_code == 403

    missing = client.put(
        "/api/system/responsibilities/hierarchy/data_asset/asset-1",
        json=hierarchy,
        headers=_headers("admin"),
    )
    assert missing.status_code == 428

    updated = client.put(
        "/api/system/responsibilities/hierarchy/data_asset/asset-1",
        json=hierarchy,
        headers=_headers("admin", **{"If-Match": '"0"'}),
    )
    assert updated.status_code == 200
    assert updated.headers["ETag"] == '"1"'

    created = client.post(
        "/api/system/responsibilities/delegations",
        json={"delegation_type": "temporary"},
        headers=_headers("admin"),
    )
    assert created.status_code == 201

    revised = client.post(
        f"/api/system/responsibilities/policies/{POLICY_UID}/revisions",
        json={"definition": {"require_all_domains": True}},
        headers=_headers("admin", **{"If-Match": '"1"'}),
    )
    assert revised.status_code == 200
    assert revised.headers["ETag"] == '"2"'
    assert service.calls[-1][3] == 1


def test_read_models_cover_delegations_policies_operations_and_joint_review(monkeypatch):
    client, service = _client(monkeypatch)

    assert client.get(
        "/api/system/responsibilities/delegations?status=active",
        headers=_headers("viewer"),
    ).status_code == 200
    assert client.get(
        "/api/system/responsibilities/policies",
        headers=_headers("viewer"),
    ).status_code == 200
    operations = client.get(
        f"/api/system/responsibilities/operations?owner_uid={USER_A}",
        headers=_headers("viewer"),
    )
    assert operations.status_code == 200
    assert operations.get_json()["data"]["grants_data_access"] is False
    evaluated = client.post(
        "/api/system/responsibilities/data_product/product-1/joint-review/evaluate",
        json={"decisions": []},
        headers=_headers("admin"),
    )
    assert evaluated.status_code == 200
    assert evaluated.get_json()["data"]["deterministic"] is True