DataOps-platform/tests/mcp/test_stdio_runtime.py

import json
import os
import subprocess
from pathlib import Path

import pytest

from app.core.mcp.runtime import identity_from_env
from app.core.mcp.servers import CONTEXT_TOOL_NAMES, SCHEDULING_TOOL_NAMES

ROOT = Path(__file__).resolve().parents[2]


def runtime_env():
    return {
        **os.environ,
        "PYTHONPATH": str(ROOT),
        "DATAOPS_MCP_SUBJECT": "agent-context-test",
        "DATAOPS_MCP_ROLES": "viewer",
        "DATAOPS_MCP_BUSINESS_DOMAINS": "sales",
        "DATAOPS_MCP_ENVIRONMENTS": "test",
        "DATAOPS_MCP_CORRELATION_ID": ("01900000-0000-7000-8000-000000000099"),
    }


def test_identity_from_env_is_explicit_bounded_and_fail_closed():
    identity = identity_from_env(runtime_env())

    assert identity.subject == "agent-context-test"
    assert identity.roles == frozenset({"viewer"})
    assert identity.business_domains == frozenset({"sales"})
    assert identity.environments == frozenset({"test"})

    incomplete = runtime_env()
    incomplete.pop("DATAOPS_MCP_BUSINESS_DOMAINS")
    with pytest.raises(ValueError, match="BUSINESS_DOMAINS"):
        identity_from_env(incomplete)

    unsafe = runtime_env()
    unsafe["DATAOPS_MCP_ROLES"] = "viewer,root"
    with pytest.raises(ValueError, match="role"):
        identity_from_env(unsafe)


def test_context_stdio_initializes_and_exposes_only_context_tools():
    messages = [
        {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-06-18",
                "capabilities": {},
                "clientInfo": {"name": "dataops-v53-test", "version": "1.0"},
            },
        },
        {
            "jsonrpc": "2.0",
            "method": "notifications/initialized",
            "params": {},
        },
        {"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}},
    ]
    result = subprocess.run(
        [
            str(ROOT / ".venv/bin/python"),
            str(ROOT / "mcp-servers/dataops_mcp_stdio.py"),
            "--kind",
            "context",
            "--factory",
            "tests.mcp.stdio_fixture:build_context_service",
            "--allow-test-factory",
        ],
        cwd=ROOT,
        env=runtime_env(),
        input="\n".join(json.dumps(item) for item in messages) + "\n",
        capture_output=True,
        text=True,
        timeout=30,
        check=True,
    )
    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"] == ("DataOps Context MCP")
    assert {
        tool["name"] for tool in responses[2]["result"]["tools"]
    } == CONTEXT_TOOL_NAMES


def test_scheduling_stdio_initializes_and_exposes_only_compound_tools():
    messages = [
        {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-06-18",
                "capabilities": {},
                "clientInfo": {"name": "dataops-v53-test", "version": "1.0"},
            },
        },
        {
            "jsonrpc": "2.0",
            "method": "notifications/initialized",
            "params": {},
        },
        {"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}},
    ]
    result = subprocess.run(
        [
            str(ROOT / ".venv/bin/python"),
            str(ROOT / "mcp-servers/dataops_mcp_stdio.py"),
            "--kind",
            "scheduling",
            "--factory",
            "tests.mcp.stdio_fixture:build_scheduling_gateway",
            "--allow-test-factory",
        ],
        cwd=ROOT,
        env=runtime_env(),
        input="\n".join(json.dumps(item) for item in messages) + "\n",
        capture_output=True,
        text=True,
        timeout=30,
    )

    assert result.returncode == 0, result.stderr
    responses = {
        item["id"]: item
        for line in result.stdout.splitlines()
        if line.strip()
        for item in [json.loads(line)]
        if "id" in item
    }
    assert {
        tool["name"] for tool in responses[2]["result"]["tools"]
    } == SCHEDULING_TOOL_NAMES