Airflow Testing: DAG Passed Review, Failed Prod Fix

The DAG passed review with two approvals and a green thread. It failed in prod within ninety seconds.

The connection it needed didn't exist outside the author's laptop. Parsing doesn't check connections, so nothing caught it.

DAGs are code and deserve tests like code. You'll build that suite here.

Why DAGs Need Tests

DAGs are code with production blast radius. A broken import halts scheduling for every DAG, not just the broken file. That shared fate makes testing a fleet concern, not a style preference.

Tests also run faster than incidents. A 45-second DagBag suite beats a 2-hour scheduler outage followed by a rollback and a postmortem.

Start with the cheapest gate that catches the most: parse tests on every PR. Layer slower tests as files change.

DAG-Load Tests With DagBag

DagBag loads every file in dags/ and reports import errors, cycles, and duplicate ids. Assert zero import errors and fleet conventions (tags, retries, catchup=False) in one fast test. Astronomer's parametrized pattern turns each file into its own test case: get_import_errors() yields (path, error) tuples and a parametrized test_file_imports raises per file, so CI names the exact broken file instead of one red blob.

Run it per PR over the whole folder; 150 files still finish under a minute. Failures point at exact tracebacks, not mystery scheduler silence.

Extend with custom rules: APPROVED_TAGS membership, at least one task per DAG, trigger_rule all_success where your standard demands it. Conventions enforced in tests survive team growth.

import pytest
from airflow.models import DagBag

def test_no_import_errors():
    bag = DagBag(dag_folder="dags/", include_examples=False)
    assert not bag.import_errors, f"import errors: {bag.import_errors}"

def test_dag_has_tags_and_retries():
    bag = DagBag(dag_folder="dags/", include_examples=False)
    assert bag.dags, "no DAGs loaded"
    for dag_id, dag in bag.dags.items():
        assert dag.tags, f"{dag_id} missing tags"
        assert dag.default_args.get("retries", 0) >= 1, f"{dag_id} needs retries"
        assert dag.catchup is False, f"{dag_id} must keep catchup=False"

Task Unit Tests That Run in Milliseconds

Unit tests target task callables as plain functions with mocked hooks. No scheduler, no broker, no database. Milliseconds per case means hundreds of cases without pain.

Structure tasks for testability: pure functions taking dates and hooks as args, side effects isolated at the edges. Functions that build SQL or transform frames test trivially; monoliths that import hooks at top level don't.

Mock at the hook boundary. Assert SQL contains the partition key, empty inputs produce zero-row results, and errors raise instead of silently returning None.

from unittest.mock import MagicMock
from dags.sales_tasks import build_partition_sql

def test_build_partition_sql_uses_logical_date():
    sql = build_partition_sql("2026-08-01")
    assert "ds='2026-08-01'" in sql
    assert "DELETE FROM sales" in sql  # idempotent reruns

def test_load_handles_empty_with_mock_hook():
    mock_hook = MagicMock()
    mock_hook.get_records.return_value = []
    # task callable takes hook as arg; no real DB touched
    from dags.sales_tasks import load_day
    result = load_day("2026-08-01", hook=mock_hook)
    assert result["rows"] == 0
    mock_hook.run.assert_called()

Airflow Dags Test for Single-Run Validation

airflow dags test runs one DAG for one logical date end to end without the scheduler. Templates render, tasks order, XComs flow. It's the dress rehearsal before staging. Since 2.10 it skips sensors and friends with --mark-success-pattern 'sensor.*' and can run tasks under their real executors with --use-executor. Its sibling airflow tasks test runs a single task instance with no dependency checks and no DB writes, which isolates one operator fast.

Run it on changed DAGs pre-merge with a fixed date: airflow dags test sales_daily 2026-08-01. Read task logs for template and key errors that unit tests miss.

Keep it scoped: one date, changed DAGs only. Full-history runs belong in staging, not in PR CI.

Testing With Mocked Connections

Connections are the top prod-only failure. Mock hooks in unit tests, then assert connection ids exist per environment in a dedicated CI test. Parse-green plus connection-missing is still red. Data quality deserves the same treatment inside the DAG: SQL check operators for relational asserts, Great Expectations suites for JSON-defined expectations across DBs and frames, Soda Core YAML checks for warehouse tables. Wire them with dependencies and branching so bad data halts or notifies instead of flowing downstream.

Provide test doubles via env: staging connections in CI secrets, prod ids asserted without values. Never commit real credentials to make tests pass.

Stage with real credentials before prod. Mocked tests prove logic; staging proves parity. Incremental loads keep each quality gate cheap: small partitions fail small.

The CI Gate Before Merge

The CI gate is lint plus DagBag parse plus unit tests on every PR, dag test on changed DAGs, staging run before prod. Merges and deploys block on red; no exceptions for urgent DAGs. Matrix the suites (unit, integration, dag-integrity) across parallel runners with pip caching so the gate clears in minutes, and split requirements so CI doesn't reinstall the world per run.

Debug interactively with dag.test(): add if __name__ == '__main__': dag.test() (or dag_object.test() for @dag style) and run the file under VSCode, PyCharm, or pdb with breakpoints. Pass execution_date, conn_file_path, variable_file_path, and run_conf dicts to rehearse dates and configs, plus mark_success_pattern to skip sensors and use_executor=True to honor real executors.

Urgent DAGs need gates most. Incident pressure produces the sloppiest imports, and sloppy imports halt the scheduler mid-incident. Measure the gate: PRs should clear in under 10 minutes. Slower gates get skipped; skipped gates cause outages.