10 Testing

This chapter explains why testing is essential for Airflow projects and how Airflow’s orchestration role and reliance on external systems make testing nuanced. It frames testing around unit and integration levels, and standardizes on pytest for its fixtures and simplicity. A practical baseline is establishing DAG integrity tests to catch early mistakes such as cycles, duplicate task IDs, or missing operator arguments, and running them automatically in a CI/CD pipeline alongside static checks (e.g., Flake8, Pylint, Black). The chapter also outlines how to structure test directories so discovery is effortless and consistent across projects.

It then shows how to write unit tests for operators by calling execute with a minimal or crafted context, and how to isolate behavior by mocking external dependencies with pytest-mock, including validating call counts and arguments and patching at the point of use. For file-based operators, pytest’s tmp_path fixture enables clean, temporary I/O. When templating or task context is required, tests assign tasks to a DAG and invoke run to mimic real execution, which in turn requires an initialized metastore; reusable pytest fixtures and unique DAG IDs keep tests independent and reproducible. Together, these patterns let you verify input validation, error handling, and operator logic without a running Airflow deployment.

For integration scenarios, the chapter demonstrates testing against real services in containers using pytest-docker-tools: fetch an image, start a container with randomized host ports, provide environment variables and initialization scripts (e.g., to set up a Postgres schema), run the operator end to end, and assert persisted outcomes. It extends this to full-DAG testing with dag.test from within tests, and discusses trade-offs between mocking external systems and emulating them with containers. Beyond local emulation, it introduces environment strategies such as Whirl for orchestrating dependencies and DTAP-style setups with branch-based promotion. Finally, it shows how tests accelerate development via IDE breakpoints or pdb and how to embed them in CI for fast, reliable feedback.

DAG cycle error displayed by Airflow

Example Python package structure

Test directory structure following the structure in figure 10.2

DAG_PATH points to the directory holding all DAG files.

A parameterized test runs a test for every dag_file.

mock_get contains several properties that can be used to validate the behavior. (Screenshot was taken using the Python debugger in PyCharm.)

Converting JSON to CSV format

Running an operator involves several steps. In section 9.1, we test only step 5 and manually provide runtime task context to operator.execute() if needed.

Calling task.run() results in task run details stored in the database.

Breakdown of converting JSON data to Postgres queries

Process of running a test with pytest-docker-tools. Running Docker containers during tests enables testing against real systems. The life cycle of the Docker container is managed by pytest-docker-tools, and the user must implement the test.

Setting a breakpoint in an IDE. This screenshot was taken in PyCharm, but any IDE allows you to set breakpoints and debug.

Starting a test in debug mode so that it stops at breakpoints

Debugging allows us to inspect the state of the program at the set breakpoint. Here we inspect the values of ratings.

Debugging on the command line with PDB

• A DAG integrity test filters basic errors in your DAGs.
• Unit testing verifies the correctness of individual operators.
• Pytest and plug-ins provide several useful constructs for testing, such as temporary directories and plug-ins for managing Docker containers during tests.
• Operators that don’t use task instance context can simply run with execute().
• Operators that do use task instance context must run together with a DAG.
• You can simulate an entire DAG using dag.test()
• For integration testing, you must simulate your production environment as closely as possible.
• Different tests catch different errors. Before implementing any test, consider what exactly you are aiming to test. Answering that question will guide you towards the test type.
• Tests should be run in your CI/CD pipeline if possible to ensure issues are caught early.

FAQ