Testing DAGs

Analogy categoryAnalogies

Why Test DAGs at All?

A DAG file is Python that executes on every scheduler parse, so a syntax error, a bad import, or a misconfigured operator argument can break DAG loading entirely — and that failure often isn't caught until the file is deployed to a live scheduler. Testing DAGs means validating three separate layers: that the file parses without exceptions (integrity), that the DAG's structure matches intent (dependencies, schedule, task count), and that individual task logic produces correct results.

🏏

Cricket analogy: It's like a team not just checking that all eleven players show up (the file parses) but also verifying the batting order matches the game plan (structure) and that each player's technique is sound in the nets (task logic) before the match starts.

DAG Integrity Tests

The most common and highest-leverage test suite loops over every .py file in the DAGs folder, imports it with a DagBag, and asserts dagbag.import_errors is empty. This single test catches the majority of real-world production incidents: a bad import, a missing Variable, a typo in an operator kwarg. It should run in CI on every pull request, before any DAG reaches the scheduler, since import errors otherwise surface silently as a red banner in the Airflow UI.

🏏

Cricket analogy: It's like a ground inspector checking every pitch on the circuit is fit for play before the season starts, catching cracked surfaces before a single ball is bowled rather than during a live match.

python

import glob
import pytest
from airflow.models import DagBag

@pytest.fixture(scope="session")
def dagbag():
    return DagBag(dag_folder="dags/", include_examples=False)

def test_no_import_errors(dagbag):
    assert len(dagbag.import_errors) == 0, (
        f"DAG import failures found:\n{dagbag.import_errors}"
    )

def test_all_dags_have_owner_and_retries(dagbag):
    for dag_id, dag in dagbag.dags.items():
        assert dag.default_args.get("owner"), f"{dag_id} missing owner"
        assert dag.default_args.get("retries", 0) >= 1, f"{dag_id} has no retries"

def test_expected_dag_count(dagbag):
    expected = len(glob.glob("dags/ingest_*.py"))
    assert len(dagbag.dags) >= expected

Testing Task Logic in Isolation

Beyond structural checks, the callable behind a PythonOperator or a TaskFlow @task function is ordinary Python and should be unit-testable independent of Airflow entirely — extract the transformation logic into a plain function, pass in fixture data, and assert on the output, mocking any hooks or connections. For tasks that must exercise real Airflow context (templated fields, XCom push/pull, Jinja rendering), use dag.test() (Airflow 2.5+) or construct a TaskInstance manually and call .run() against a local SQLite metadata database.

🏏

Cricket analogy: It's like a batting coach testing a player's cover-drive technique against a bowling machine in isolation before ever facing a live bowler in a match situation, isolating the skill from match pressure.

Use dag.test() (available from Airflow 2.5) to run an entire DAG locally against a lightweight SQLite backend without a live scheduler — useful for fast integration-style tests in CI that exercise task dependencies and XCom passing end to end.

Common Pitfall: Testing Against a Shared Environment

A frequent mistake is running DAG tests against a shared staging Airflow instance with real connections, which makes tests slow, flaky, and capable of triggering real side effects (sending emails, writing to production tables) if a test DAG is accidentally unpaused. Tests should run against an isolated SQLite or ephemeral Postgres metadata database with mocked hooks and connections defined via AIRFLOW_CONN_* environment variables scoped only to the test process.

🏏

Cricket analogy: It's like running batting practice on the actual match pitch the day before a final instead of a separate net — you risk damaging the surface everyone depends on for the real game.

Never point DAG unit or integration tests at a shared staging or production Airflow metadata database. Use an isolated SQLite database (AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=sqlite:///:memory: for simple cases, or a fresh ephemeral Postgres container in CI) and mock every external connection so tests can't send real emails, hit real APIs, or write to real tables.

DAG testing has three layers: import/integrity, structural correctness, and individual task logic.
A DagBag-based test asserting import_errors is empty catches the majority of real-world DAG deployment failures and should gate every pull request.
Extract transformation logic into plain, Airflow-independent functions so it can be unit-tested with ordinary pytest fixtures and mocks.
dag.test() (Airflow 2.5+) runs a full DAG locally against SQLite without a live scheduler, useful for fast CI integration tests.
Never run DAG tests against a shared staging/production metadata database or real connections — use isolated SQLite/ephemeral Postgres and mocked hooks.
Structural tests (owner set, retries configured, expected task count) catch configuration drift that import tests alone would miss.
CI should run DAG tests on every pull request before merge, not after deployment to a live scheduler.