DataOps-platform/tests/integration/test_kestra_mcp_stdio.py

import json
import os
import subprocess
from pathlib import Path

import pytest


ROOT = Path(__file__).resolve().parents[2]
COMPOSE = ROOT / "deploy/docker/docker-compose.yml"

pytestmark = pytest.mark.skipif(
    os.getenv("KESTRA_INTEGRATION") != "1",
    reason="set KESTRA_INTEGRATION=1 for the local Kestra MCP contract",
)


def test_kestra_mcp_stdio_initializes_and_exposes_only_governed_groups():
    messages = [
        {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-06-18",
                "capabilities": {},
                "clientInfo": {"name": "dataops-v51", "version": "1.0"},
            },
        },
        {
            "jsonrpc": "2.0",
            "method": "notifications/initialized",
            "params": {},
        },
        {"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}},
    ]
    payload = "\n".join(
        json.dumps(message, separators=(",", ":")) for message in messages
    ) + "\n"
    result = subprocess.run(
        [
            "docker",
            "compose",
            "-f",
            str(COMPOSE),
            "--profile",
            "kestra-mcp",
            "run",
            "--rm",
            "-T",
            "--no-deps",
            "kestra-mcp",
        ],
        cwd=ROOT,
        input=payload,
        check=True,
        capture_output=True,
        text=True,
        timeout=60,
    )
    responses = {
        item["id"]: item
        for line in result.stdout.splitlines()
        if line.strip()
        for item in [json.loads(line)]
        if "id" in item
    }

    assert responses[1]["result"]["serverInfo"]["name"]
    tools = responses[2]["result"]["tools"]
    assert tools
    names = {tool["name"] for tool in tools}
    assert any("flow" in name for name in names)
    assert any("execution" in name for name in names)
    for forbidden in ("namespace_file", "kv_", "tenant", "auditlog"):
        assert not any(forbidden in name for name in names)