DataOps-platform/tests/runner/test_bootstrap.py

import pytest
from flask import Flask

from app.runner.bootstrap import (
    register_artifact_cleanup_cli,
    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()


def test_runner_registers_bounded_artifact_cleanup_cli():
    class Resolver:
        def __init__(self):
            self.limit = None

        def cleanup_expired(self, *, limit):
            self.limit = limit
            return 3

    resolver = Resolver()
    app = Flask("runner-cleanup-test")
    register_artifact_cleanup_cli(app, resolver)

    result = app.test_cli_runner().invoke(
        args=["cleanup-rule-artifacts", "--limit", "25"]
    )

    assert result.exit_code == 0
    assert result.output.strip() == "removed 3 expired rule artifacts"
    assert resolver.limit == 25