DataOps-platform/tests/mcp/test_deployment_contract.py

import json
import subprocess
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
COMPOSE = ROOT / "deploy/docker/docker-compose.yml"
DOCKERFILE = ROOT / "deploy/docker/mcp.Dockerfile"
ENV_EXAMPLE = ROOT / "deploy/docker/.env.example"


def test_dataops_mcp_image_is_python_311_and_runs_without_root():
    source = DOCKERFILE.read_text(encoding="utf-8")

    assert source.startswith("FROM python:3.11-slim")
    assert "mcp-servers/requirements.txt" in source
    assert "mcp-servers/dataops_mcp_stdio.py" in source
    assert "USER dataops" in source


def test_compose_exposes_only_profiled_stdio_dataops_mcp_processes():
    compose = json.loads(
        subprocess.run(
            [
                "docker",
                "compose",
                "-f",
                str(COMPOSE),
                "--profile",
                "dataops-mcp",
                "config",
                "--format",
                "json",
            ],
            cwd=ROOT,
            capture_output=True,
            text=True,
            check=True,
        ).stdout
    )
    services = compose["services"]
    scheduling = services["dataops-scheduling-mcp"]
    context = services["dataops-context-mcp"]

    for service in (scheduling, context):
        assert service["profiles"] == ["dataops-mcp"]
        assert service["stdin_open"] is True
        assert "ports" not in service
        assert service["read_only"] is True
        assert service["environment"]["DATAOPS_MCP_DATABASE_URL"].startswith(
            "postgresql://"
        )
    assert (
        scheduling["environment"]["RUNNER_TASK_TOKEN_SECRET"]
        == (services["runner"]["environment"]["RUNNER_TASK_TOKEN_SECRET"])
    )
    assert scheduling["environment"]["KESTRA_BASE_URL"] == ("http://kestra:8080/api/v1")
    assert "app.core.mcp.bootstrap:build_scheduling_gateway" in scheduling["command"]
    assert "app.core.mcp.bootstrap:build_context_service" in context["command"]


def test_mcp_identity_scope_is_explicit_in_environment_template():
    source = ENV_EXAMPLE.read_text(encoding="utf-8")

    for name in (
        "DATAOPS_MCP_SUBJECT",
        "DATAOPS_MCP_ROLES",
        "DATAOPS_MCP_BUSINESS_DOMAINS",
        "DATAOPS_MCP_ENVIRONMENTS",
    ):
        assert f"{name}=" in source