DataOps-platform/tests/mcp/test_bootstrap.py

import pytest

from app.core.mcp.bootstrap import (
    build_context_service,
    build_scheduling_gateway,
)
from app.core.mcp.context import ContextService
from app.core.mcp.gateway import SchedulingGateway
from app.core.mcp.persistence import PostgresContextRepository


def scheduling_environment(monkeypatch):
    values = {
        "DATAOPS_MCP_DATABASE_URL": "postgresql://dataops:secret@db/dataops",
        "KESTRA_BASE_URL": "http://kestra:8080/api/v1",
        "KESTRA_USERNAME": "scheduler",
        "KESTRA_PASSWORD": "kestra-secret",
        "KESTRA_TENANT_ID": "main",
        "KESTRA_HTTP_TIMEOUT_SECONDS": "15",
        "RUNNER_TASK_TOKEN_SECRET": "x" * 32,
        "RUNNER_TASK_TOKEN_TTL_SECONDS": "60",
    }
    for key, value in values.items():
        monkeypatch.setenv(key, value)
    return values


def test_scheduling_bootstrap_is_fail_closed_and_keeps_secrets_out_of_repr(
    monkeypatch,
):
    values = scheduling_environment(monkeypatch)

    gateway = build_scheduling_gateway()

    assert isinstance(gateway, SchedulingGateway)
    assert gateway.engine.base_url == values["KESTRA_BASE_URL"]
    assert gateway.engine.tenant_id == "main"
    rendered = repr(gateway)
    assert values["KESTRA_PASSWORD"] not in rendered
    assert values["RUNNER_TASK_TOKEN_SECRET"] not in rendered
    assert values["DATAOPS_MCP_DATABASE_URL"] not in rendered


def test_scheduling_bootstrap_requires_every_security_dependency(monkeypatch):
    scheduling_environment(monkeypatch)
    monkeypatch.delenv("RUNNER_TASK_TOKEN_SECRET")

    with pytest.raises(ValueError, match="RUNNER_TASK_TOKEN_SECRET"):
        build_scheduling_gateway()


def test_context_bootstrap_uses_the_persistent_read_model(monkeypatch):
    monkeypatch.setenv(
        "DATAOPS_MCP_DATABASE_URL",
        "postgresql://dataops:secret@db/dataops",
    )

    service = build_context_service()

    assert isinstance(service, ContextService)
    assert isinstance(service.repository, PostgresContextRepository)
    assert "postgresql://dataops:secret" not in repr(service)


@pytest.mark.parametrize(
    ("name", "value"),
    [
        ("KESTRA_HTTP_TIMEOUT_SECONDS", "0"),
        ("KESTRA_HTTP_TIMEOUT_SECONDS", "301"),
        ("RUNNER_TASK_TOKEN_TTL_SECONDS", "0"),
        ("RUNNER_TASK_TOKEN_TTL_SECONDS", "301"),
    ],
)
def test_scheduling_bootstrap_rejects_unsafe_numeric_limits(monkeypatch, name, value):
    scheduling_environment(monkeypatch)
    monkeypatch.setenv(name, value)

    with pytest.raises(ValueError, match=name):
        build_scheduling_gateway()