DataOps-platform/tests/core/governance/test_responsibilities.py

from __future__ import annotations

import pytest


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


def test_device_scope_requires_one_accountable_asset_manager():
    from app.core.governance.responsibilities import (
        ResponsibilityValidationError,
        validate_matrix,
    )

    with pytest.raises(
        ResponsibilityValidationError,
        match="one accountable asset manager",
    ):
        validate_matrix(
            "device_asset",
            [
                {
                    "user_id": USER_A,
                    "responsibility_role": "data_steward",
                    "raci_role": "responsible",
                }
            ],
        )

    validated = validate_matrix(
        "device_asset",
        [
            {
                "user_id": USER_A,
                "responsibility_role": "asset_manager",
                "raci_role": "accountable",
            },
            {
                "user_id": USER_B,
                "responsibility_role": "data_steward",
                "raci_role": "responsible",
            },
        ],
    )

    assert [item.user_id for item in validated] == [USER_A, USER_B]


def test_responsibility_matrix_rejects_duplicates_and_unknown_values():
    from app.core.governance.responsibilities import (
        ResponsibilityValidationError,
        validate_matrix,
    )

    duplicate = {
        "user_id": USER_A,
        "responsibility_role": "asset_manager",
        "raci_role": "accountable",
    }
    with pytest.raises(ResponsibilityValidationError, match="duplicate"):
        validate_matrix("device_mapping", [duplicate, duplicate])

    with pytest.raises(ResponsibilityValidationError, match="resource type"):
        validate_matrix("dashboard", [duplicate])

    with pytest.raises(ResponsibilityValidationError, match="responsibility role"):
        validate_matrix(
            "business_domain",
            [
                {
                    "user_id": USER_A,
                    "responsibility_role": "supervisor",
                    "raci_role": "accountable",
                }
            ],
        )


def test_replace_is_revision_guarded_and_rejects_inactive_users():
    from app.core.governance.responsibilities import (
        ResponsibilityConflict,
        ResponsibilityService,
        ResponsibilityUserUnavailable,
    )

    class Repository:
        def replace(self, **kwargs):
            assert kwargs["expected_revision"] == 3
            if kwargs["assignments"][0].user_id == USER_B:
                raise ResponsibilityUserUnavailable("inactive user")
            raise ResponsibilityConflict("revision conflict")

    service = ResponsibilityService(Repository())

    with pytest.raises(ResponsibilityConflict):
        service.replace(
            resource_type="device_ontology",
            resource_uid="device-domain",
            assignments=[
                {
                    "user_id": USER_A,
                    "responsibility_role": "asset_manager",
                    "raci_role": "accountable",
                }
            ],
            expected_revision=3,
            actor_uid=USER_A,
        )

    with pytest.raises(ResponsibilityUserUnavailable):
        service.replace(
            resource_type="device_ontology",
            resource_uid="device-domain",
            assignments=[
                {
                    "user_id": USER_B,
                    "responsibility_role": "asset_manager",
                    "raci_role": "accountable",
                }
            ],
            expected_revision=3,
            actor_uid=USER_A,
        )