chart-tests

Use ao escrever, editar, revisar ou executar testes de chart Helm para o repositório Astronomer APC. Abrange padrões pytest, uso de render_chart(), sub-chart…

npx skills add https://github.com/astronomer/astronomer --skill chart-tests

Chart Test Writing Guide

Critical Rules

  1. Always run tests with uv run — never python3 -m pytest or python bin/...
  2. Sub-chart values MUST be nested under the sub-chart name (see Values Nesting)
  3. No helm unittest plugin — all tests are pytest-based using render_chart()
  4. One render_chart() call per test function — never call it multiple times in the same function to cover different value combinations (see One Condition Per Test)

Test Organization

tests/
├── chart_tests/              # Helm template rendering tests (main focus)
│   ├── test_<component>.py   # One file per component
│   ├── conftest.py           # Shared fixtures
│   └── test_data/            # Feature configs, expected outputs
├── functional/               # End-to-end cluster tests
├── k8s_schema/               # Cached Kubernetes API schemas
└── utils/
    ├── chart.py              # render_chart() and helpers
    ├── fixtures.py           # Common fixtures
    └── __init__.py           # get_containers_by_name(), get_all_features(), etc.

Writing Tests

Basic Pattern

import pytest
from tests import supported_k8s_versions
from tests.utils.chart import render_chart

DEPLOYMENT_FILE = "charts/grafana/templates/grafana-deployment.yaml"


@pytest.mark.parametrize("kube_version", supported_k8s_versions)
def test_some_feature(kube_version):
    """Brief description of what is being tested."""
    docs = render_chart(
        kube_version=kube_version,
        show_only=[DEPLOYMENT_FILE],
        values={"grafana": {"enabled": True}},
    )
    assert len(docs) == 1
    assert docs[0]["kind"] == "Deployment"

Sub-Chart Values Nesting — CRITICAL

Templates in charts/<subchart>/templates/ belong to a sub-chart. Values for those templates must be nested under the sub-chart's top-level key:

# ❌ WRONG — will not override sub-chart values
values = {"houston": {"replicas": 3}}

# ✅ CORRECT — nest under the sub-chart name
values = {"astronomer": {"houston": {"replicas": 3}}}

# ✅ EXAMPLE — disable a feature in the astronomer sub-chart
docs = render_chart(
    values={"astronomer": {"dpLink": {"enabled": False}}},
    show_only=["charts/astronomer/templates/dp-link/dp-link-deployment.yaml"],
)
assert len(docs) == 0

Top-level charts (e.g. nginx, grafana, prometheus) use their chart name directly:

values = {"nginx": {"serviceType": "LoadBalancer"}}
values = {"grafana": {"extraEnvVars": [...]}}

Using show_only

Always use show_only to target the specific template being tested:

docs = render_chart(
    show_only=[
        "charts/nginx/templates/controlplane/nginx-cp-service.yaml",
        "charts/nginx/templates/dataplane/nginx-dp-service.yaml",
    ]
)

Parametrized Tests

Always parametrize over supported_k8s_versions and over relevant values axes:

@pytest.mark.parametrize("kube_version", supported_k8s_versions)
@pytest.mark.parametrize("plane_mode,docs_count", [("control", 1), ("unified", 1), ("data", 0)])
def test_deployment_should_render(kube_version, plane_mode, docs_count):
    docs = render_chart(
        kube_version=kube_version,
        show_only=[DEPLOYMENT_FILE],
        values={"global": {"plane": {"mode": plane_mode}}},
    )
    assert len(docs) == docs_count

One Condition Per Test

Never call render_chart() more than once inside the same test function to check several value combinations (e.g. default, feature-enabled, feature-disabled). Each render_chart() call is its own test condition — hiding several behind one function name makes failures ambiguous (which call failed?) and hides the test matrix from pytest --collect-only / -k filtering. Use parametrization when the calls only differ by a value/expectation, or separate test functions when they differ conceptually:

# ❌ WRONG — three renders buried in one function
def test_secretstore_rule(kube_version):
    docs = render_chart(kube_version=kube_version, show_only=[ROLE_FILE])
    assert absent(docs[0])

    docs = render_chart(kube_version=kube_version, values={"global": {"dataPlaneFailover": {"enabled": True}}}, show_only=[ROLE_FILE])
    assert present(docs[0])

    docs = render_chart(kube_version=kube_version, values={"global": {"dataPlaneFailover": {"enabled": False}}}, show_only=[ROLE_FILE])
    assert absent(docs[0])
# ✅ CORRECT — parametrize when only the value/expectation changes
@pytest.mark.parametrize("kube_version", supported_k8s_versions)
@pytest.mark.parametrize(
    "dataplane_failover_enabled,rule_expected",
    [(None, False), (True, True), (False, False)],
    ids=["default", "enabled", "disabled"],
)
def test_secretstore_rule(kube_version, dataplane_failover_enabled, rule_expected):
    values = {}
    if dataplane_failover_enabled is not None:
        values = {"global": {"dataPlaneFailover": {"enabled": dataplane_failover_enabled}}}
    docs = render_chart(kube_version=kube_version, values=values, show_only=[ROLE_FILE])
    assert (expected_rule in docs[0]["rules"]) == rule_expected

Or, if an existing test function already renders with the exact values you need (e.g. a test for dataPlaneFailover.enabled: True that targets a related template), prefer adding your assertion to that template's doc in the existing show_only list over writing a new render_chart() call.

Testing with All Features Enabled

get_all_features() enables as many compatible features as possible. Not all features can be enabled simultaneously due to incompatibilities.

from tests.utils import get_all_features

def test_with_all_features():
    docs = render_chart(values=get_all_features())
    kinds = [doc["kind"] for doc in docs]
    assert "Deployment" in kinds

Testing Probe Customization

Every container must support customizable livenessProbe and readinessProbe. When adding a new component:

  1. Add its probes to tests/chart_tests/test_data/enable_all_probes.yaml
  2. Run tests with that file to verify probes are rendered correctly

Cross-cutting invariants — ALWAYS guard with a cross-cutting test

Some requirements must hold for every container/pod/object the cluster renders, not just one component — typically because an admission controller (e.g. a Gatekeeper/OPA constraint) rejects the whole install if a single object violates them. Examples: every container must define a startupProbe (allow-with-probes, PINF-691), allowPrivilegeEscalation: false (PINF-585/713), no forbidden Service fields (PINF-692).

For requirements like these, a per-component test is not enough — a new component added later silently reintroduces the violation. Always add a single cross-cutting test that:

  1. Renders with get_all_features() (what a real install presents to the admission controller), and
  2. Iterates every container across every pod-manager kind and asserts the invariant, so any new component that violates it fails the suite automatically.

Use get_chart_containers() or get_containers_by_name() over the filtered doc list. test_probes.py (TestStartupProbes, TestCustomProbes) and test_security_context_override.py are the reference patterns. Write this test first (TDD): it should fail (red) listing every offending container before you implement the fix, then pass (green) once coverage is complete.


ConfigMap Scripts

Scripts embedded in ConfigMaps must follow these conventions:

  1. Static content only — scripts must not use Helm templating to conditionally modify their content based on chart values. The rendered output must be identical regardless of what values are passed.

  2. Environment variable inputs — all runtime configuration must be passed as environment variables defined in the container spec (via env or envFrom), not baked into the script at render time.

  3. Stored as files on disk — scripts must be committed as real files in the repository (e.g. under charts/<subchart>/files/) so they can be linted and reviewed like any other source file.

  4. Included via .Files.Get — scripts must be included in ConfigMap templates using .Files.Get, not inline Helm template blocks:

    # ✅ CORRECT
    apiVersion: v1
    kind: ConfigMap
    metadata:
      name: {{ include "chart.fullname" . }}-scripts
    data:
      my-script.sh: {{ .Files.Get "files/my-script.sh" | quote }}
    
    # ❌ WRONG — inline script with template logic
    data:
      my-script.sh: |
        #!/bin/sh
        {{- if .Values.someFlag }}
        do_something
        {{- end }}
    

Test Utilities

render_chart(values, show_only, kube_version, validate_objects)

Renders the chart via helm template and returns parsed YAML documents.

ParameterTypeDescription
valuesdictValues merged with chart defaults
show_onlylist[str]Templates to render (filters output)
kube_versionstrK8s version for schema validation
validate_objectsboolValidate against K8s schemas (default True)
from tests.utils.chart import render_chart

docs = render_chart(
    values={"nginx": {"enabled": True}},
    show_only=["charts/nginx/templates/controlplane/nginx-cp-service.yaml"],
    kube_version="1.31.0",
)

get_containers_by_name(doc, *, include_init_containers=False)

Returns {name: container_dict} for all containers in a pod manager doc (Deployment, StatefulSet, DaemonSet, Job, CronJob). Pass include_init_containers=True to also include init containers.

from tests.utils import get_containers_by_name

c_by_name = get_containers_by_name(doc, include_init_containers=True)
assert c_by_name["grafana"]["securityContext"] == {"readOnlyRootFilesystem": True}
assert c_by_name["bootstrapper"]["securityContext"] == {"readOnlyRootFilesystem": True}

get_all_features()

Returns a values dict with most components enabled.

from tests.utils import get_all_features

Other utilities in tests/utils/__init__.py

  • get_env_vars_dict(container_env) — converts env list to {name: value} dict
  • get_service_ports_by_name(doc) — returns service ports keyed by name
  • get_pod_template(doc) — extracts pod template from any pod manager
  • get_service_account_name_from_doc(doc) — returns the serviceAccountName
  • dot_notation_to_dict(dotted_string, default_value) — builds nested dict from dot notation

Running Tests

Correct examples

# Full suite in parallel (fastest — use for full runs)
uv run pytest tests/chart_tests/ -n auto --quiet

# Full suite, verbose
uv run pytest tests/chart_tests/ --verbose

# Single file
uv run pytest tests/chart_tests/test_grafana.py --verbose

# Tests matching a pattern
uv run pytest tests/chart_tests/ -k "test_service" --verbose

# Single test
uv run pytest tests/chart_tests/test_grafana.py::test_deployment_should_render --verbose

# Verbose output, stop on first failure
uv run pytest tests/chart_tests/ -vv --capture=no --maxfail=1

# Iterate on failures: re-run only last-failed tests
uv run pytest tests/chart_tests/ --maxfail=1 --lf

Tip: -n auto uses all CPU cores. Omit it when running a single file to avoid subprocess overhead.

Incorrect examples

# ❌ WRONG — we do not need to activate the venv manually, and we do not run pytest without `uv run`
.venv/bin/activate && pytest tests/chart_tests/

# ❌ WRONG — do not create virtual environment with python, do not use pip install. Use `uv run` to run all python files in the repo.
python3 -m venv .venv && .venv/bin/pip install -r requirements-dev.txt && .venv/bin/python3 -m pytest tests/chart_tests/

Kubernetes Schema Validation

Tests validate rendered manifests against cached K8s OpenAPI schemas in tests/k8s_schema/v<version>-standalone/. Validation runs by default; disable with validate_objects=False.

def test_custom_resource():
    docs = render_chart(
        show_only=["charts/airflow-operator/templates/crds/airflow.yaml"],
        validate_objects=True,
    )
    for doc in docs:
        assert doc["kind"] == "CustomResourceDefinition"

Mais skills de astronomer

airflow
astronomer
Consulte, gerencie e solucione problemas de DAGs, execuções, tarefas e configuração de sistema do Apache Airflow. Suporta mais de 30 comandos para inspeção de DAGs, gerenciamento de execuções, registro de tarefas, consultas de configuração e acesso direto à API REST. Gerencie múltiplas instâncias do Airflow com configuração persistente; descubra automaticamente implantações locais e Astro. Dispare execuções de DAG de forma síncrona (aguardando conclusão) ou assíncrona, diagnostique falhas, limpe execuções para repetição e acesse logs de tarefas com filtragem por repetição/índice de mapa. Saída...
official
airflow-hitl
astronomer
Portões de aprovação humana, entradas de formulário e ramificações em DAGs do Airflow usando operadores adiáveis. Quatro tipos de operadores: ApprovalOperator para decisões de aprovar/rejeitar, HITLOperator para seleção de múltiplas opções com formulários, HITLBranchOperator para roteamento de tarefas orientado por humanos e HITLEntryOperator para coleta de dados de formulário. Todos os operadores são adiáveis, liberando slots de worker enquanto aguardam resposta humana via a aba Ações Necessárias da interface do Airflow ou API REST. Suporta recursos opcionais incluindo personalização...
official
airflow-state-store
astronomer
Persists task and asset state across retries and DAG runs using Airflow 3.3's AIP-103 key/value stores (`task_state_store`, `asset_state_store`) and the…
official
analyzing-data
astronomer
Consulte seu data warehouse para responder perguntas de negócios com padrões em cache e mapeamentos de conceitos. Suporta busca de padrões e cache para tipos de perguntas repetidas, com registro de resultados para melhorar consultas futuras. Inclui cache de mapeamento conceito-tabela e descoberta de esquemas de tabela via INFORMATION_SCHEMA ou grep no código-fonte. Fornece funções de kernel run_sql() e run_sql_pandas() que retornam DataFrames Polars ou Pandas para análise. Comandos CLI para gerenciar caches de conceitos, padrões e tabelas, além de...
official
annotating-task-lineage
astronomer
Anotar tarefas do Airflow com linhagem de dados usando inlets e outlets. Suporta objetos OpenLineage Dataset, Assets do Airflow e Datasets do Airflow para definir entradas e saídas em bancos de dados, data warehouses e armazenamento em nuvem. Use como fallback quando operadores não possuem extratores OpenLineage integrados; segue um sistema de precedência de quatro níveis onde extratores personalizados e métodos OpenLineage têm prioridade. Inclui auxiliares de nomenclatura de datasets para Snowflake, BigQuery, S3 e PostgreSQL para garantir consistência...
official
authoring-dags
astronomer
Fluxo de trabalho guiado para criação de DAGs do Apache Airflow com integração de validação e testes. Abordagem estruturada em seis fases: descobrir o ambiente e padrões existentes, planejar a estrutura da DAG, implementar seguindo as melhores práticas, validar com comandos da CLI af, testar com consentimento do usuário e iterar em correções. Comandos da CLI para descoberta (af config connections, af config providers, af dags list) e validação (af dags errors, af dags get, af dags explore) fornecem feedback imediato sobre a DAG...
official
authoring-go-sdk-tasks
astronomer
Writes Airflow task logic in Go using the Airflow Go SDK. Use when the user wants to implement Airflow tasks in Go, asks about `BundleProvider`/`RegisterDags`,…
official
authoring-java-sdk-tasks
astronomer
Escreve a lógica de tarefas do Airflow em Java, Kotlin ou qualquer linguagem JVM usando o Airflow Java SDK. Use quando o usuário quiser implementar tarefas do Airflow em Java/JVM, pedir…
official