A data pipeline can fail before a task ever runs: an Airflow DAG might contain a cycle or an operator might be missing a required argument. pytest runs Python tests that catch such problems before deployment. It supports both unit tests for individual functions or operators and integration tests for components working together. section 10.1 section 10.1.1
A small pytest workflow:
- Install it with `pip install pytest`. Put tests in a `tests/` directory; name test files with a `test_` prefix so pytest discovers them. section 10.1.1
- Run a directory with `pytest tests`, a file with `pytest tests/dags/test_example.py`, or one function with `pytest tests/dags/test_example.py::test_example`. section 10.1.1 section 10.1.4
- For DAG integrity, parameterize a test over DAG files, load each with Airflow’s `DagBag`, and fail on logged errors or files containing no DAGs. section 10.1.1
Test an operator without running Airflow
A test can instantiate `BashOperator(task_id="test", bash_command="echo 'hello!'")`, call `execute(context={})`, and assert that the result is `"hello!"`. This exercises the operator method directly, not Airflow’s live task lifecycle. If an operator needs a connection that should not be shared across tests, the `pytest-mock` fixture can replace the connection lookup and check how it was called. section 10.1.4 section 10.1.4
Listing 10.3 DAG integrity test
import glob import os import pytest from airflow.models import DagBag DAG_PATH = os.path.join( os.path.dirname(__file__), "..", "..", "dags/**/*.py" ) DAG_FILES = glob.glob(DAG_PATH, recursive=True) @pytest.mark.parametrize("dag_file", DAG_FILES) def test_dag_integrity(dag_file, caplog): """Test integrity of DAGs.""" DagBag(dag_folder=dag_file, include_examples=False) # The DagBag's function catches and logs all exceptions, so we ➥ parse the logs to raise them here for record in caplog.records: if record.levelname == "ERROR": raise record.exc_info[1] elif "assumed to contain no DAGs" in record.message: assert False, f"No DAGs found in {dag_file}"
The DAG integrity test shows how pytest parameterization and captured logs turn Airflow parse errors into test failures. section 10.1.1
