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_reconciliation_cli():
    class Resolver:
        def __init__(self):
            self.limit = None
            self.grace_seconds = None

        def reconcile(self, *, limit, grace_seconds):
            self.limit = limit
            self.grace_seconds = grace_seconds
            return {
                "pending_finalized": 1,
                "pending_deleted": 1,
                "ready_failed": 1,
                "orphans_deleted": 1,
            }

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

    result = app.test_cli_runner().invoke(
        args=[
            "reconcile-rule-artifacts",
            "--limit",
            "25",
            "--grace-seconds",
            "600",
        ]
    )

    assert result.exit_code == 0
    assert "pending_finalized=1" in result.output
    assert "orphans_deleted=1" in result.output
    assert resolver.limit == 25
    assert resolver.grace_seconds == 600