DataOps-platform/tests/security/test_scheduling_tool_boundaries.py

import pytest

from app.core.mcp.gateway import ALLOWED_SCHEDULING_TOOLS, SchedulingPolicy
from app.core.mcp.identity import AgentIdentity


def identity(role):
    return AgentIdentity(
        subject=f"agent-{role}",
        roles=frozenset({role}),
        business_domains=frozenset({"sales"}),
        environments=frozenset({"test"}),
        correlation_id="correlation-a",
    )


def test_dangerous_raw_kestra_tools_are_not_exposed():
    assert {
        "create_candidate_plan",
        "deploy_disabled_version",
        "run_canary",
        "promote_candidate",
        "pause_schedule",
        "retry_failed_execution",
        "backfill_bounded_window",
        "rollback_to_previous_version",
    } == ALLOWED_SCHEDULING_TOOLS
    for forbidden in (
        "delete_flow",
        "delete_execution",
        "change_status",
        "put_kv",
        "write_file",
        "create_yaml",
        "configure_connection",
    ):
        assert forbidden not in ALLOWED_SCHEDULING_TOOLS


def test_viewer_cannot_mutate_and_editor_cannot_promote():
    policy = SchedulingPolicy()

    with pytest.raises(PermissionError, match="role"):
        policy.authorize_tool(
            identity("viewer"),
            "create_candidate_plan",
            domain="sales",
            environment="test",
        )
    with pytest.raises(PermissionError, match="role"):
        policy.authorize_tool(
            identity("editor"),
            "promote_candidate",
            domain="sales",
            environment="test",
        )