DataOps-platform/tests/test_permission_matrix.py

def test_fixed_role_permission_matrix_is_monotonic():
    from app.core.system.permissions import ROLE_PERMISSIONS

    viewer = ROLE_PERMISSIONS["viewer"]
    editor = ROLE_PERMISSIONS["editor"]
    admin = ROLE_PERMISSIONS["admin"]

    assert viewer < editor < admin
    assert "governance:read" in viewer
    assert "governance:edit" in editor
    assert "users:manage" in admin
    assert "workflow:activate" in admin
    assert "ingestion:run" in editor
    assert "data-elements:edit" in editor
    assert "ontologies:edit" in editor
    assert "governance:responsibilities:read" in viewer
    assert "ingestion:admin" in admin
    assert "evidence:download" in admin
    assert "data-elements:publish" in admin
    assert "ontologies:publish" in admin
    assert "governance:responsibilities:manage" in admin
    assert "device-assets:edit" in editor
    assert "device-semantics:edit" in editor
    assert "device-semantics:review" not in editor
    assert "device-semantics:review" in admin
    assert "device-entities:edit" in editor
    assert "device-entities:review" not in editor
    assert "device-entities:review" in admin
    assert "device-quality:edit" in editor
    assert "device-quality:execute" in editor
    assert "device-quality:publish" not in editor
    assert "device-quality:publish" in admin
    assert "quality-issues:edit" in editor
    assert "quality-issues:review" not in editor
    assert "quality-issues:review" in admin
    assert "device-observability:edit" in editor
    assert "data-observability:read" in viewer
    assert "data-observability:operate" in editor
    assert "data-observability:manage" not in editor
    assert "data-observability:manage" in admin
    assert "governance-audit:read" not in editor
    assert "governance-audit:read" in admin
    assert "governance-audit:seal" in admin
    assert "edge-gateways:read" in viewer
    assert "edge-gateways:operate" in editor
    assert "edge-gateways:manage" not in editor
    assert "edge-gateways:manage" in admin


def test_edge_gateway_machine_routes_are_public_but_human_routes_are_governed():
    from app.core.system.permissions import PUBLIC, permission_for_request

    gateway = "11111111-1111-1111-1111-111111111111"
    assert permission_for_request("/api/datasource/edge/register", "POST") == (PUBLIC,)
    for suffix in ("heartbeat", "tasks/pull", "events", "releases/ack", "reconcile", "tasks/task-1/outcome"):
        assert permission_for_request(f"/api/datasource/edge/gateways/{gateway}/{suffix}", "POST") == (PUBLIC,)
    assert permission_for_request("/api/datasource/edge/enrollments", "POST") == ("edge-gateways:manage",)
    assert permission_for_request("/api/datasource/edge/gateways", "GET") == ("edge-gateways:read",)
    assert permission_for_request("/api/datasource/edge/tasks", "POST") == ("edge-gateways:operate",)
    assert permission_for_request("/api/datasource/edge/tasks/task-1/cancel", "POST") == ("edge-gateways:operate",)
    assert permission_for_request(f"/api/datasource/edge/gateways/{gateway}/rotate", "POST") == ("edge-gateways:manage",)


def test_data_development_paths_have_specific_write_policies():
    from app.core.system.permissions import permission_for_request

    assert permission_for_request(
        "/api/development/v1/ingestion-jobs", "POST"
    ) == ("ingestion:run",)
    assert permission_for_request(
        "/api/development/v1/ingestion-jobs/job-1/retry", "POST"
    ) == ("ingestion:admin",)
    assert permission_for_request(
        "/api/development/v1/ingestion-jobs/job-1/cancel", "POST"
    ) == ("ingestion:run",)
    assert permission_for_request(
        "/api/development/v1/ingestion-jobs/job-1/execute", "POST"
    ) == ("ingestion:run",)
    assert permission_for_request(
        "/api/development/v1/ingestion-jobs/job-1/catalog-snapshots", "GET"
    ) == ("governance:read",)
    assert permission_for_request(
        "/api/development/v1/ingestion-jobs/job-1/evidence", "GET"
    ) == ("governance:read",)
    assert permission_for_request(
        "/api/development/v1/data-elements", "POST"
    ) == ("data-elements:edit",)
    assert permission_for_request(
        "/api/development/v1/data-elements/element-1/transitions", "POST"
    ) == ("data-elements:edit",)
    assert permission_for_request(
        "/api/development/v1/candidate-decisions", "POST"
    ) == ("data-elements:edit",)
    assert permission_for_request(
        "/api/development/v1/device-assets", "GET"
    ) == ("governance:read",)
    assert permission_for_request(
        "/api/development/v1/device-assets/asset-1", "GET"
    ) == ("governance:read",)
    assert permission_for_request(
        "/api/development/v1/device-assets/import", "POST"
    ) == ("device-assets:edit",)
    assert permission_for_request(
        "/api/development/v1/device-semantics/profile", "GET"
    ) == ("governance:read",)
    assert permission_for_request(
        "/api/development/v1/device-semantics/bootstrap", "POST"
    ) == ("device-semantics:edit",)
    assert permission_for_request(
        "/api/development/v1/device-semantics/codes/code-1/submit",
        "POST",
    ) == ("device-semantics:edit",)
    assert permission_for_request(
        "/api/development/v1/device-semantics/codes/code-1/review",
        "POST",
    ) == ("device-semantics:review",)
    assert permission_for_request(
        "/api/development/v1/device-entities/candidates", "GET"
    ) == ("governance:read",)
    assert permission_for_request(
        "/api/development/v1/device-entities/candidates/generate", "POST"
    ) == ("device-entities:edit",)
    assert permission_for_request(
        "/api/development/v1/device-entities/candidates/candidate-1/review",
        "POST",
    ) == ("device-entities:review",)
    assert permission_for_request(
        "/api/development/v1/device-entities/merges/merge-1/rollback",
        "POST",
    ) == ("device-entities:review",)
    assert permission_for_request(
        "/api/development/v1/device-quality/profile", "GET"
    ) == ("governance:read",)
    assert permission_for_request(
        "/api/development/v1/device-quality/bootstrap", "POST"
    ) == ("device-quality:edit",)
    assert permission_for_request(
        "/api/development/v1/device-quality/profile/versions/version-1/publish",
        "POST",
    ) == ("device-quality:publish",)
    assert permission_for_request(
        "/api/development/v1/device-quality/runs", "POST"
    ) == ("device-quality:execute",)
    assert permission_for_request(
        "/api/development/v1/device-quality/issues", "GET"
    ) == ("governance:read",)
    assert permission_for_request(
        "/api/development/v1/device-quality/issues/import", "POST"
    ) == ("quality-issues:edit",)
    assert permission_for_request(
        "/api/development/v1/device-quality/issues/issue-1/review",
        "POST",
    ) == ("quality-issues:review",)
    assert permission_for_request(
        "/api/development/v1/device-observability/events",
        "GET",
    ) == ("governance:read",)
    assert permission_for_request(
        "/api/development/v1/device-observability/graph",
        "GET",
    ) == ("governance:read",)
    assert permission_for_request(
        "/api/development/v1/device-observability/root-cause",
        "GET",
    ) == ("governance:read",)
    assert permission_for_request(
        "/api/development/v1/device-observability/import",
        "POST",
    ) == ("device-observability:edit",)
    assert permission_for_request(
        "/api/development/v1/governance-metrics/summary",
        "GET",
    ) == ("governance:read",)
    assert permission_for_request(
        "/api/development/v1/governance-metrics/details",
        "GET",
    ) == ("governance:read",)
    assert permission_for_request(
        "/api/datafactory/observability/overview",
        "GET",
    ) == ("data-observability:read",)
    assert permission_for_request(
        "/api/datafactory/observability/collect",
        "POST",
    ) == ("data-observability:operate",)
    assert permission_for_request(
        "/api/datafactory/observability/slos",
        "POST",
    ) == ("data-observability:manage",)
    assert permission_for_request(
        "/api/system/governance-audit/events",
        "GET",
    ) == ("governance-audit:read",)
    assert permission_for_request(
        "/api/system/governance-audit/seals",
        "POST",
    ) == ("governance-audit:seal",)


def test_business_domain_read_endpoints_are_available_to_viewers():
    from app.core.system.permissions import permission_for_request

    assert permission_for_request("/api/bd/list", "POST") == (
        "governance:read",
    )
    assert permission_for_request("/api/bd/detail", "POST") == (
        "governance:read",
    )
    assert permission_for_request("/api/bd/save", "POST") == (
        "governance:edit",
    )


def test_permissions_are_derived_server_side_from_roles():
    from app.core.system.permissions import permissions_for_roles

    permissions = permissions_for_roles(["viewer", "editor"])

    assert "governance:read" in permissions
    assert "governance:edit" in permissions
    assert "users:manage" not in permissions


def test_every_api_path_has_an_explicit_policy():
    from app import create_app
    from app.core.system.permissions import permission_for_request

    app = create_app()
    for rule in app.url_map.iter_rules():
        if not rule.rule.startswith("/api/"):
            continue
        for method in rule.methods - {"HEAD", "OPTIONS"}:
            assert permission_for_request(rule.rule, method) is not None, (
                method,
                rule.rule,
            )


def test_knowledge_search_and_ask_posts_are_read_only():
    from app.core.system.permissions import READ_GOVERNANCE, permission_for_request

    assert permission_for_request("/api/knowledge/search", "POST") == (
        READ_GOVERNANCE,
    )
    assert permission_for_request("/api/knowledge/ask", "POST") == (READ_GOVERNANCE,)


def test_knowledge_admin_routes_require_dedicated_manage_permission():
    from app.core.system.permissions import KNOWLEDGE_MANAGE, permission_for_request

    assert permission_for_request("/api/knowledge/admin/sync", "GET") == (
        KNOWLEDGE_MANAGE,
    )
    assert permission_for_request(
        "/api/knowledge/admin/change-sets/id/retry", "POST"
    ) == (KNOWLEDGE_MANAGE,)
    assert permission_for_request(
        "/api/knowledge/admin/device-sources", "GET"
    ) == (KNOWLEDGE_MANAGE,)
    assert permission_for_request(
        "/api/knowledge/admin/device-sources/id/scope", "PUT"
    ) == (KNOWLEDGE_MANAGE,)
    assert permission_for_request(
        "/api/knowledge/admin/query-audits", "GET"
    ) == (KNOWLEDGE_MANAGE,)


def test_data_rule_routes_separate_read_authoring_and_release_permissions():
    from app.core.system.permissions import (
        DATAFLOW_RELEASE,
        RULES_EDIT,
        RULES_PUBLISH,
        RULES_READ,
        permission_for_request,
    )

    assert permission_for_request("/api/rules/capabilities", "GET") == (
        RULES_READ,
    )
    assert permission_for_request("/api/rules/interpret", "POST") == (
        RULES_EDIT,
    )
    assert permission_for_request("/api/rules/validate", "POST") == (
        RULES_EDIT,
    )
    assert permission_for_request(
        "/api/rules/versions/id/publish", "POST"
    ) == (RULES_PUBLISH,)
    assert permission_for_request(
        "/api/rules/production-lines/id/release", "POST"
    ) == (DATAFLOW_RELEASE,)


def test_data_factory_actions_have_independent_permissions():
    from app.core.system.permissions import (
        DATAFLOW_ACTIVATE,
        DATAFLOW_CANARY,
        DATAFLOW_DEPLOY,
        DATAFLOW_ROLLBACK,
        RULES_EXECUTE,
        permission_for_request,
    )

    assert permission_for_request("/api/rules/deployments", "POST") == (
        DATAFLOW_DEPLOY,
    )
    assert permission_for_request(
        "/api/rules/deployments/id/deploy-disabled", "POST"
    ) == (DATAFLOW_DEPLOY,)
    assert permission_for_request(
        "/api/rules/deployments/id/canary", "POST"
    ) == (DATAFLOW_CANARY, RULES_EXECUTE)
    assert permission_for_request(
        "/api/rules/deployments/id/execute", "POST"
    ) == (RULES_EXECUTE,)
    assert permission_for_request(
        "/api/rules/deployments/id/activate", "POST"
    ) == (DATAFLOW_ACTIVATE,)
    assert permission_for_request(
        "/api/rules/deployments/id/rollback", "POST"
    ) == (DATAFLOW_ROLLBACK,)
    assert permission_for_request(
        "/api/rules/deployments/id/reconcile-deploy", "POST"
    ) == (DATAFLOW_DEPLOY,)
    assert permission_for_request(
        "/api/rules/deployments/id/reconcile-activate", "POST"
    ) == (DATAFLOW_ACTIVATE,)
    assert permission_for_request(
        "/api/rules/deployments/id/reconcile-rollback", "POST"
    ) == (DATAFLOW_ROLLBACK,)
    assert permission_for_request(
        "/api/rules/deployments/id/reconcile-execute", "POST"
    ) == (RULES_EXECUTE,)