Functional Tests

Functional Test Writing Guide

Overview

Functional tests run against a live Kubernetes cluster (kind) with the Helm chart installed. Unlike chart tests, they verify real runtime behavior: running processes, user identity, network reachability, and configuration values.

Critical Rules

1. Always run tests with uv run — never python3 -m pytest or python -m pytest
2. Pass --topology to bin/reset-local-dev (unified/control/data) before running — pytest itself needs no env var, it infers topology from the test's own location on disk
3. Always use uppercase kubeconfig constants from tests.utils.k8s — KUBECONFIG_UNIFIED, KUBECONFIG_CONTROL, KUBECONFIG_DATA (not lowercase variants)
4. Run bin/reset-local-dev before the first test run to set up the cluster

Installation Scenarios

Three scenarios exist, each with its own test directory:

Cross-scenario tests (applicable to all planes) belong in tests/functional/shared/. This directory does not yet exist — create it (with an __init__.py) when adding the first shared test, then add an entry point to each scenario's conftest if needed.

Local Setup Workflow

# 1. Set up the cluster for your chosen topology (downloads tools, generates certs, launches kind, installs chart)
bin/reset-local-dev --topology=unified   # or: control, data

# 2. Run the tests (does NOT tear down the cluster — re-run freely while iterating)
uv run pytest tests/functional/unified

# 3. See helper file paths (kubeconfig, etc.)
make show-test-helper-files

Makefile shortcuts run setup + tests in one step:

make test-functional-unified
make test-functional-control
make test-functional-data

Enable verbose debug output (helm install --debug, kubectl -v=9):

export DEBUG=1

Stored artifacts (outside the repo, consistent across runs):

• Tools: ~/.local/share/astronomer-software/bin
• Kubeconfigs: ~/.local/share/astronomer-software/kubeconfig/{unified,control,data}
• Certs: ~/.local/share/astronomer-software/certs (auto-renewed if expiring within 4 weeks)

Test Organization

tests/functional/
├── conftest.py                        # Shared fixtures (k8s clients, named pod hosts)
├── unified/
│   ├── conftest.py                    # unified-specific fixtures (if any)
│   ├── test_config.py                 # Configuration and behavior assertions
│   ├── test_container_user_is_not_root.py
│   ├── test_network_security.py       # Port-scan test (complex one-off, do not replicate pattern)
│   └── test_container_read_only_root.py
├── control/
│   ├── conftest.py
│   ├── test_control.py
│   ├── test_pod_configs.py
│   └── test_container_user_is_not_root.py
├── data/
│   ├── test_data.py
│   └── test_container_user_is_not_root.py
└── shared/                            # Create when adding first cross-scenario test
    ├── __init__.py
    └── test_<name>.py

Kubeconfig Helpers

Always import from tests.utils.k8s:

from tests.utils.k8s import KUBECONFIG_UNIFIED, KUBECONFIG_CONTROL, KUBECONFIG_DATA

These resolve to ~/.local/share/astronomer-software/kubeconfig/<scenario>.

Known bug: tests/functional/control/test_container_user_is_not_root.py imports kubeconfig_control (lowercase), which does not exist in tests.utils.k8s. Fix this to KUBECONFIG_CONTROL whenever you touch that file.

Shared Fixtures

tests/functional/conftest.py provides these fixtures (all scope="function"):

Writing Tests

Assert command output in a container

def test_prometheus_user(prometheus):
    user = prometheus.check_output("whoami")
    assert user == "nobody", f"Expected 'nobody', got '{user}'"

Assert a file exists and has expected content

def test_dashboard_config_mounted(grafana):
    f = grafana.file("/etc/grafana/provisioning/dashboards/dashboard.yaml")
    assert f.exists
    assert f.is_file
    content = grafana.check_output("cat /etc/grafana/provisioning/dashboards/dashboard.yaml")
    assert "apiVersion: 1" in content
    assert "providers:" in content

Assert containers do not run as root

import pytest
import testinfra
from tests.utils.k8s import KUBECONFIG_UNIFIED, get_pod_running_containers

container_ignore_list = ["kube-state", "houston", "astro-ui"]


def test_container_user_is_not_root():
    containers = get_pod_running_containers(kubeconfig=KUBECONFIG_UNIFIED, namespace="astronomer")
    for container in containers.values():
        if container["_name"] in container_ignore_list:
            pytest.skip(f"Unsupported container: {container['_name']}")
        host = testinfra.get_host(
            f"kubectl://{container['pod_name']}?container={container['_name']}&namespace={container['namespace']}",
            kubeconfig=KUBECONFIG_UNIFIED,
        )
        user = host.user()
        assert user.name != "root"
        assert user.uid != 0
        assert user.gid != 0

Use the Kubernetes API directly

def test_ensure_feature_disabled(k8s_core_v1_client):
    pods = k8s_core_v1_client.list_namespaced_pod("astronomer")
    should_not_run = ["prometheus-postgres-exporter"]
    for pod in pods.items:
        for feature in should_not_run:
            if feature in pod.metadata.name:
                raise ValueError(f"Expected '{feature}' to be disabled")

Parse JSON config from a container process

import json


def test_houston_config(houston_api):
    data = houston_api.check_output("echo \"config = require('config'); console.log(JSON.stringify(config))\" | node -")
    config = json.loads(data)
    assert "url" not in config["nats"]
    assert len(config["nats"]["servers"]) > 0

Flaky Tests

Use @pytest.mark.flaky for tests that depend on eventually-consistent cluster state (e.g. network reachability, pod readiness):

@pytest.mark.flaky(reruns=20, reruns_delay=10)
def test_houston_can_reach_prometheus(houston_api):
    assert houston_api.check_output("wget --timeout=5 -qO- http://astronomer-prometheus.astronomer.svc.cluster.local:9090/targets")

• reruns: max retry attempts on failure
• reruns_delay: seconds between retries
• Use sparingly — only when the cluster genuinely needs time to converge

Utility Functions

From tests.utils.k8s:

get_pod_running_containers(namespace, kubeconfig=None) -> dict Returns {pod_name_container_name: container_info} for all ready containers. Each value includes pod_name, namespace, and _name (container name).

get_pod_by_label_selector(namespace, label_selector, kubeconfig) -> str Returns the name of the first pod matching the given label selector. Asserts at least one pod is found.

What NOT to Do

• Do not hardcode kubeconfig paths — always use the constants from tests.utils.k8s
• Do not run with python -m pytest — always use uv run pytest
• Do not replicate the class-based structure of test_network_security.py for ordinary tests — that file is a one-off for a specialized port-scan workflow
• Do not add tests directly to tests/functional/ root — tests belong in a scenario subdirectory or shared/