Back to Repositories

Validating Pod Launcher Role Configurations in Apache Airflow

This test suite validates the pod launcher role configuration in Apache Airflow’s Helm charts. It ensures proper RBAC settings and role bindings across different executor types and namespace configurations.

Test Coverage Overview

The test suite provides comprehensive coverage of pod launcher role configurations in Airflow’s Helm deployment.

Key areas tested include:

Role binding validation for different executor types (Celery, Kubernetes, Local)
RBAC permission verification for scheduler and worker accounts
Multi-namespace mode configurations
Role and RoleBinding naming conventions

Implementation Analysis

The testing approach utilizes pytest’s parametrize feature for systematic validation of multiple configuration scenarios.

Implementation highlights:

Uses jmespath for JSON/YAML parsing and assertion validation
Leverages helm_template_generator for chart rendering
Implements parametrized test cases for comprehensive coverage

Technical Details

Testing infrastructure includes:

pytest framework for test organization
helm_template_generator for Helm chart rendering
jmespath for structured data querying
Custom test fixtures for chart configuration
RBAC-specific template validation

Best Practices Demonstrated

The test suite exemplifies several testing best practices in Helm chart validation.

Notable practices include:

Systematic parameter variation for comprehensive coverage
Clear separation of test cases by functionality
Explicit assertion messages for debugging
Efficient test organization using pytest classes
Thorough validation of security-critical RBAC configurations

apache/airflow

helm_tests/airflow_aux/test_pod_launcher_role.py

            
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
from __future__ import annotations

import jmespath
import pytest

from tests.charts.helm_template_generator import render_chart


class TestPodLauncher:
    """Tests pod launcher."""

    @pytest.mark.parametrize(
        "executor, rbac, allow, expected_accounts",
        [
            ("CeleryKubernetesExecutor", True, True, ["scheduler", "worker"]),
            ("KubernetesExecutor", True, True, ["scheduler", "worker"]),
            ("CeleryExecutor", True, True, ["worker"]),
            ("LocalExecutor", True, True, ["scheduler"]),
            ("LocalExecutor", False, False, []),
        ],
    )
    def test_pod_launcher_role(self, executor, rbac, allow, expected_accounts):
        docs = render_chart(
            values={
                "rbac": {"create": rbac},
                "allowPodLaunching": allow,
                "executor": executor,
            },
            show_only=["templates/rbac/pod-launcher-rolebinding.yaml"],
        )
        if expected_accounts:
            for idx, suffix in enumerate(expected_accounts):
                assert f"release-name-airflow-{suffix}" == jmespath.search(f"subjects[{idx}].name", docs[0])
        else:
            assert docs == []

    @pytest.mark.parametrize(
        "multiNamespaceMode, namespace, expectedRole, expectedRoleBinding",
        [
            (
                True,
                "namespace",
                "namespace-release-name-pod-launcher-role",
                "namespace-release-name-pod-launcher-rolebinding",
            ),
            (
                True,
                "other-ns",
                "other-ns-release-name-pod-launcher-role",
                "other-ns-release-name-pod-launcher-rolebinding",
            ),
            (False, "namespace", "release-name-pod-launcher-role", "release-name-pod-launcher-rolebinding"),
        ],
    )
    def test_pod_launcher_rolebinding_multi_namespace(
        self, multiNamespaceMode, namespace, expectedRole, expectedRoleBinding
    ):
        docs = render_chart(
            namespace=namespace,
            values={"webserver": {"allowPodLogReading": True}, "multiNamespaceMode": multiNamespaceMode},
            show_only=["templates/rbac/pod-launcher-rolebinding.yaml"],
        )

        actualRoleBinding = jmespath.search("metadata.name", docs[0])
        assert actualRoleBinding == expectedRoleBinding

        actualRoleRef = jmespath.search("roleRef.name", docs[0])
        assert actualRoleRef == expectedRole

        actualKind = jmespath.search("kind", docs[0])
        actualRoleRefKind = jmespath.search("roleRef.kind", docs[0])
        if multiNamespaceMode:
            assert actualKind == "ClusterRoleBinding"
            assert actualRoleRefKind == "ClusterRole"
        else:
            assert actualKind == "RoleBinding"
            assert actualRoleRefKind == "Role"

    @pytest.mark.parametrize(
        "multiNamespaceMode, namespace, expectedRole",
        [
            (True, "namespace", "namespace-release-name-pod-launcher-role"),
            (True, "other-ns", "other-ns-release-name-pod-launcher-role"),
            (False, "namespace", "release-name-pod-launcher-role"),
        ],
    )
    def test_pod_launcher_role_multi_namespace(self, multiNamespaceMode, namespace, expectedRole):
        docs = render_chart(
            namespace=namespace,
            values={"webserver": {"allowPodLogReading": True}, "multiNamespaceMode": multiNamespaceMode},
            show_only=["templates/rbac/pod-launcher-role.yaml"],
        )

        actualRole = jmespath.search("metadata.name", docs[0])
        assert actualRole == expectedRole

        actualKind = jmespath.search("kind", docs[0])
        if multiNamespaceMode:
            assert actualKind == "ClusterRole"
        else:
            assert actualKind == "Role"