Implementation:Dagster io Dagster Helm Test Fixtures

Knowledge Sources	Dagster_io_Dagster Dagster K8s Deployment
Domains	Kubernetes, Testing, Helm, Integration_Testing
Last Updated	2026-02-10 12:30 GMT

Overview

Concrete tool for provisioning Kubernetes namespaces, ConfigMaps, Secrets, and Helm chart installations as pytest fixtures for Dagster integration tests.

Description

The helm.py module (1,079 lines) provides the core pytest fixture infrastructure for Dagster's Kubernetes integration test suites. It contains session-scoped fixtures that create isolated Kubernetes test namespaces, populate them with ConfigMaps and Secrets, install the Dagster Helm chart with various configurations (CeleryK8sRunLauncher, K8sRunLauncher, user deployments subchart), and set up port forwarding to the webserver. Key context managers include helm_chart (Celery-based), helm_chart_for_k8s_run_launcher (K8s native), and helm_chart_for_user_deployments_subchart. The module handles both RabbitMQ and Redis as Celery backends and supports run monitoring, subchart deployments, and LocalStack S3 emulation.

Usage

Import these fixtures in Celery-K8s or K8s integration test conftest.py files. The fixtures are session-scoped and designed to be composed: namespace creates isolation, configmaps and secrets populate test data, and helm_namespace or helm_namespaces_for_k8s_run_launcher install the full Helm chart. The webserver_url fixture provides a forwarded URL for API calls.

Code Reference

Source Location

• Repository: Dagster_io_Dagster
• File: integration_tests/python_modules/dagster-k8s-test-infra/dagster_k8s_test_infra/helm.py
• Lines: 1-1079

Signature

# Key fixtures and context managers:

@pytest.fixture(scope="session")
def should_cleanup(pytestconfig) -> bool: ...

@pytest.fixture(scope="session")
def namespace(cluster_provider, pytestconfig, should_cleanup) -> str: ...

@pytest.fixture(scope="session")
def configmaps(namespace, should_cleanup) -> None: ...

@pytest.fixture(scope="session")
def secrets(namespace, should_cleanup) -> None: ...

@pytest.fixture(scope="session")
def celery_backend(request) -> str: ...

@pytest.fixture(scope="session")
def helm_namespace(
    dagster_docker_image, cluster_provider, namespace,
    should_cleanup, configmaps, aws_configmap, secrets, celery_backend,
) -> str: ...

@pytest.fixture(scope="session")
def helm_namespaces_for_k8s_run_launcher(
    dagster_docker_image, cluster_provider, namespace,
    should_cleanup, configmaps, aws_configmap, secrets, request,
) -> Tuple[str, str]: ...

@contextmanager
def helm_chart(namespace, docker_image, celery_backend, should_cleanup=True): ...

@contextmanager
def helm_chart_for_k8s_run_launcher(
    system_namespace, user_code_namespace, docker_image,
    enable_subchart, should_cleanup=True, run_monitoring=False,
): ...

@contextmanager
def helm_chart_for_user_deployments_subchart(namespace, docker_image, should_cleanup=True): ...

@pytest.fixture(scope="session")
def webserver_url(helm_namespace) -> str: ...

Import

from dagster_k8s_test_infra.helm import (
    helm_chart,
    helm_chart_for_k8s_run_launcher,
    helm_chart_for_user_deployments_subchart,
)

I/O Contract

Inputs

Name	Type	Required	Description
dagster_docker_image	str	Yes	Docker image tag (e.g., repo:tag) for Dagster components
cluster_provider	fixture	Yes	Kubernetes cluster provider (Kind or existing)
namespace	str	Yes	Kubernetes namespace for test isolation
should_cleanup	bool	Yes	Whether to tear down resources after tests
celery_backend	str	No	Either "rabbitmq" or "redis" for Celery broker
docker_image	str	Yes	Docker image in repository:tag format
enable_subchart	bool	No	Whether to enable the dagster-user-deployments subchart
run_monitoring	bool	No	Whether to enable run monitoring daemon configuration

Outputs

Name	Type	Description
namespace	str	Kubernetes namespace name (e.g., dagster-test-3fcd70)
helm_namespace	str	Namespace with Helm chart installed and all pods ready
helm_namespaces_for_k8s_run_launcher	Tuple[str, str]	Tuple of (user_code_namespace, system_namespace)
webserver_url	str	Port-forwarded URL to the dagster-webserver (e.g., http://localhost:12345)

Usage Examples

Using Helm Fixtures in a K8s Integration Test

import pytest

# The fixtures are session-scoped and auto-compose via pytest dependency injection

def test_k8s_run_launcher(
    user_code_namespace_for_k8s_run_launcher,
    webserver_url_for_k8s_run_launcher,
):
    """Test that runs can be launched via the K8sRunLauncher."""
    import requests

    # The webserver is already running and port-forwarded
    response = requests.get(f"{webserver_url_for_k8s_run_launcher}/server_info")
    assert response.status_code == 200
    assert "dagster" in response.text

Using the Celery Helm Chart Fixture

def test_celery_k8s_integration(helm_namespace, webserver_url):
    """Test with CeleryK8sRunLauncher Helm deployment.

    The helm_namespace fixture installs the Dagster Helm chart with
    Celery workers (parametrized for both RabbitMQ and Redis backends).
    """
    import requests

    response = requests.get(f"{webserver_url}/server_info")
    assert "dagster" in response.text

Related Pages

• Principle:Dagster_io_Dagster_Kubernetes_Helm_Test_Infrastructure