DataOps-platform/tests/runner/test_bootstrap.py

import pytest

from app.runner.bootstrap import runner_settings_from_env


def test_runner_settings_require_signing_secret_and_explicit_dependencies(
    monkeypatch,
):
    required = {
        "DATABASE_URL": "postgresql://platform",
        "NEO4J_URI": "bolt://neo4j:7687",
        "NEO4J_USER": "neo4j",
        "NEO4J_PASSWORD": "test",
        "DATASOURCE_CREDENTIAL_MASTER_KEY": (
            "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY="
        ),
        "DATASOURCE_CREDENTIAL_KEY_VERSION": "v1",
        "RUNNER_TASK_TOKEN_SECRET": "x" * 32,
        "RUNNER_HTTP_ALLOWED_HOSTS": "events.internal, audit.internal",
        "RUNNER_MINIO_HOST": "minio:9000",
        "RUNNER_MINIO_USER": "runner-test",
        "RUNNER_MINIO_PASSWORD": "runner-test-password",
        "RUNNER_MINIO_BUCKET": "dataops-rules",
    }
    for key, value in required.items():
        monkeypatch.setenv(key, value)

    settings = runner_settings_from_env()

    assert settings.runtime.pool_size == 1
    assert settings.runtime.max_overflow == 1
    assert settings.runtime.max_idle_pools == 4
    assert settings.allowed_http_hosts == {
        "events.internal",
        "audit.internal",
    }
    assert settings.artifact_bucket == "dataops-rules"
    assert "postgresql://platform" not in repr(settings)
    assert required["RUNNER_TASK_TOKEN_SECRET"] not in repr(settings)
    assert required["RUNNER_MINIO_PASSWORD"] not in repr(settings)


def test_runner_refuses_to_start_without_a_task_token_secret(monkeypatch):
    monkeypatch.delenv("RUNNER_TASK_TOKEN_SECRET", raising=False)

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