DataOps-platform/tests/test_kestra_local_contract.py

import json
import subprocess
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
COMPOSE = ROOT / "deploy/docker/docker-compose.yml"
KESTRA_CONFIG = ROOT / "deploy/docker/kestra/application.yml"
POSTGRES_BOOTSTRAP = ROOT / "deploy/docker/postgres/init/000-init.sql"
KESTRA_BOOTSTRAP = (
    ROOT / "deploy/docker/postgres/init/010-kestra-runtime.sql"
)
KESTRA_SOAK = ROOT / "scripts/validation/kestra_queue_soak.sh"


def _compose_config():
    result = subprocess.run(
        [
            "docker",
            "compose",
            "-f",
            str(COMPOSE),
            "--profile",
            "kestra-mcp",
            "config",
            "--format",
            "json",
        ],
        cwd=ROOT,
        check=True,
        capture_output=True,
        text=True,
    )
    return json.loads(result.stdout)


def test_kestra_and_mcp_images_are_immutable_and_network_is_isolated():
    config = _compose_config()
    kestra = config["services"]["kestra"]
    mcp = config["services"]["kestra-mcp"]

    assert kestra["image"] == "kestra/kestra:v1.3.20"
    assert "--worker-thread=16" in kestra["command"]
    assert mcp["image"] == (
        "ghcr.io/kestra-io/mcp-server-python@"
        "sha256:1fd62028f60914af31e244ddb0086ed5017420313e2c8547ea9c01b8d24dfc1d"
    )
    assert kestra["ports"] == [
        {
            "mode": "ingress",
            "target": 8080,
            "published": "18080",
            "host_ip": "127.0.0.1",
            "protocol": "tcp",
        }
    ]
    assert "ports" not in mcp
    assert set(kestra["networks"]) == {"dataops-test-net"}
    assert set(mcp["networks"]) == {"dataops-test-net"}
    assert mcp["profiles"] == ["kestra-mcp"]
    assert mcp["stdin_open"] is True


def test_kestra_uses_its_own_database_and_mcp_has_a_restricted_tool_surface():
    config = _compose_config()
    kestra_environment = config["services"]["kestra"]["environment"]
    mcp_environment = config["services"]["kestra-mcp"]["environment"]

    assert kestra_environment["KESTRA_DB_NAME"] == "kestra"
    assert kestra_environment["KESTRA_DB_HOST"] == "postgres"
    assert kestra_environment["KESTRA_DB_USERNAME"] == "kestra_runtime"
    assert config["services"]["kestra"]["restart"] == "unless-stopped"
    assert (
        config["services"]["kestra"]["depends_on"]["kestra-db-init"][
            "condition"
        ]
        == "service_completed_successfully"
    )
    assert mcp_environment["KESTRA_BASE_URL"] == "http://kestra:8080/api/v1"
    assert mcp_environment["KESTRA_TENANT_ID"] == "main"
    assert mcp_environment["KESTRA_MCP_LOG_LEVEL"] == "ERROR"
    assert set(mcp_environment["KESTRA_MCP_DISABLED_TOOLS"].split(",")) == {
        "ee",
        "files",
        "kv",
        "namespace",
    }

    combined = json.dumps(
        {
            "kestra": config["services"]["kestra"],
            "kestra-mcp": config["services"]["kestra-mcp"],
        },
        sort_keys=True,
    )
    for forbidden in (
        "source-postgres",
        "source-mysql",
        "source-test-password",
        "source-root-test-password",
        "DATASOURCE_CREDENTIAL_MASTER_KEY",
    ):
        assert forbidden not in combined


def test_kestra_configuration_and_fresh_database_bootstrap_are_present():
    assert KESTRA_CONFIG.is_file()
    source = KESTRA_CONFIG.read_text(encoding="utf-8")
    assert "datasources:" in source
    assert "postgres:" in source
    assert (
        "url: jdbc:postgresql://${KESTRA_DB_HOST}:5432/"
        "${KESTRA_DB_NAME}?tcpKeepAlive=true"
    ) in source
    assert "maximum-pool-size: 20" in source
    assert "minimum-idle: 4" in source
    assert "connection-test-query: SELECT 1" in source
    assert "basic-auth:" in source
    assert "${KESTRA_USERNAME}" in source
    assert "${KESTRA_PASSWORD}" in source
    assert "type: local" in source
    assert "tutorial-flows:" in source
    assert "enabled: false" in source

    bootstrap = POSTGRES_BOOTSTRAP.read_text(encoding="utf-8")
    assert "CREATE DATABASE kestra" in bootstrap
    dedicated = KESTRA_BOOTSTRAP.read_text(encoding="utf-8")
    assert "CREATE ROLE kestra_runtime LOGIN" in dedicated
    assert "ALTER DATABASE kestra OWNER TO kestra_runtime" in dedicated
    assert "pg_catalog" in dedicated

    soak = KESTRA_SOAK.read_text(encoding="utf-8")
    assert "KESTRA_SOAK_SECONDS:-1800" in soak
    assert "executions/${namespace}/${flow_id}" in soak
    assert "dataops-kestra-soak-" in soak
    assert 'restart postgres' in soak
    assert 'restart kestra' in soak
    assert "controlled_restart_completed" in soak
    assert 'kind:"soak_start"' in soak
    assert 'kind:"soak_complete"' in soak
    assert 'logs --since "$soak_started_at"' in soak
    assert "postgres_restart_count_before" in soak
    assert "kestra_restart_count_before" in soak
    assert "Unexpected automatic container restart" in soak
    assert "explicit `docker restart` does not consume RestartCount" in soak
    assert "broken pipe" in soak
    assert "terminating connection due to administrator command" in soak
    assert "every other FATAL line" in soak


def test_kestra_mcp_is_documented_as_stdio_not_an_http_endpoint():
    readme = (ROOT / "deploy/docker/README.md").read_text(encoding="utf-8")
    assert "http://127.0.0.1:18080" in readme
    assert "Kestra MCP" in readme
    assert "stdio" in readme
    assert "docker compose --profile kestra-mcp run --rm -T kestra-mcp" in readme