Airflow Testing: The DAG That Passed Review, Failed Prod

DAGs get a free pass that no other code gets. A Python service that failed to import would never make it past the build — but a DAG that references a connection that doesn't exist, or imports a module only dev has, sails through review and dies on its first prod run. Reviewers read diffs; they don't run anything.

The uncomfortable truth: a DAG that parses is not a DAG that works. Parsing proves the file is readable Python with a valid DAG object. It proves nothing about the connection, the SQL, the credentials, or the provider version. Every layer between "imports fine" and "loads data" is untested ground unless you build tests for it.

The fix is boring and familiar: treat the DAG folder like a package with tests. DagBag tests for structure, unit tests for task callables, airflow dags test for real single-run validation, mocked connections for isolation, and a CI gate that runs all of it before merge.

If it can't pass in CI, it doesn't get the chance to fail in prod.

That's the whole discipline. Now the specifics.

1. Why DAGs Need Tests: They're Code

A DAG file is a Python module that gets imported by a scheduler process, repeatedly, forever. It has imports, side effects, dependencies, and a runtime contract. By every definition that matters, it's code — so it deserves the code treatment: tests, CI, and a merge gate.

The difference from normal code is the failure profile. A broken web service fails loudly in front of users. A broken DAG fails at 2 AM, silently, and the damage is measured in missing data or double-loaded tables. The stakes argue for more testing discipline, not less.

There's also the parse trap: the scheduler imports every DAG file constantly, and an import-time crash in one file can degrade or stop scheduling entirely. A DAG that fails to import isn't just a broken DAG — it's a threat to every other DAG on the box.

That's why the test pyramid here starts with the cheapest, highest-value check: does the file even load?

2. DAG-Load Tests With DagBag

DagBag is Airflow's own loader — the same machinery the scheduler uses to parse the DAG folder. Loading your DAGs through a DagBag in a test reproduces exactly what the scheduler does, which is exactly the failure you want to catch first.

The two assertions that matter: no import errors, and expected DAGs present with the right schedule. import_errors is the smoking gun — it captures every file that failed to parse, with the traceback, so the test fails with the actual problem.

Structure assertions earn their keep next: dag_id matches the file, schedule matches the contract, default_args carry the retry policy. These catch the silent regressions — someone flattening a retry, someone renaming a task that a dataset trigger depends on.

Keep dag-load tests fast. They run on every commit; if they take minutes, they'll be skipped, then deleted, then missed.

Put the DagBag in a session-scoped pytest fixture in conftest.py so it is built once and shared across the whole suite. For run-level assertions — actually executing a task — use the documented pattern of dag.create_dagrun(state=DagRunState.RUNNING, run_type=DagRunType.MANUAL) followed by ti.run(ignore_ti_state=True) on the task instance: a real task execution without a scheduler.

# tests/test_market_etl_dag.py — full pytest file: DagBag + task tests
import pytest
from airflow.models import DagBag

DAG_FOLDER = "dags/"


def test_market_etl_loads_without_errors():
    dagbag = DagBag(dag_folder=DAG_FOLDER, include_examples=False)
    assert not dagbag.import_errors


def test_market_etl_structure():
    dagbag = DagBag(dag_folder=DAG_FOLDER)
    dag = dagbag.get_dag("market_etl")
    assert dag is not None
    assert dag.schedule == "@daily"
    assert dag.default_args["retries"] == 3
    assert set(dag.task_ids) == {
        "extract_market_data",
        "transform_daily_metrics",
        "load_orders_to_warehouse",
    }


def test_extract_task_writes_expected_keys():
    from dags.market_etl import extract_market_data

    result = extract_market_data(api_base_url="https://api.test.local", limit=10)
    assert set(result.keys()) == {"records", "fetched_at"}
    assert len(result["records"]) == 10


def test_transform_drops_nulls():
    from dags.market_etl import transform_daily_metrics

    rows = [{"order_id": 1, "amount": 12.5}, {"order_id": 2, "amount": None}]
    out = transform_daily_metrics(rows)
    assert len(out) == 1
    assert out[0]["amount"] == 12.5

3. Task Unit Tests: The Callable Is the Contract

Graph tests tell you the shape; task tests tell you the behavior. The strongest move in Airflow testing is to write tasks as plain callables — TaskFlow style — and unit-test the callable directly. No scheduler, no XCom, no DB: pure function, pure assertions.

Test the edges that production hits: empty inputs, null fields, API failures, retry-worthy exceptions. The transform task that silently accepts None is the task that poisons your warehouse for a month. A one-line assertion catches it.

Keep the callable testable: no globals, no implicit connections inside the function, inputs passed in, outputs returned. If the function needs a hook or a client, inject it or mock it at the boundary — which is exactly what the next section covers.

Task tests are where the incident in this article dies: the callable either resolves its connection or it doesn't, and you'll know before the merge, not after the first run.

4. airflow dags test: Single-Run Validation

Unit tests prove the pieces; airflow dags test proves the assembled DAG runs end to end — once, locally, against your real hooks and a real (or dev) backend. It executes the whole graph in-process for a given logical date, exactly the way a scheduled run would.

This is the command that would have caught our incident: run it with the connection the task actually uses, and the missing-connection error surfaces in five seconds instead of at 2 AM. It's the closest thing to a dress rehearsal that Airflow ships.

Use it in CI on every merge to the staging branch. Slow down the schedule for the test, keep the run scoped to one logical date, and treat any non-success state as a build failure.

It doesn't replace unit tests — it complements them. Unit tests catch logic; dags test catches wiring.

Two upgrades to know. In Airflow 2.10+, airflow dags test accepts --mark-success-pattern to skip specific tasks by id and --use-executor to run tasks with your configured executors instead of in-process. And there's an in-process sibling, dag.test() — call it on the DAG object (or the @dag return value) and every task runs in one serialized Python process with no scheduler at all, ideal for stepping through a DAG in an IDE debugger.

# Single-run validation — the dress rehearsal
airflow dags test market_etl 2026-07-15

# It runs every task in order, in one process:
# [2026-07-15 00:00:00] extract_market_data → success
# [2026-07-15 00:00:05] transform_daily_metrics → success
# [2026-07-15 00:00:12] load_orders_to_warehouse → failed
# !! connection snowflake_prod not found

# Verify the connections the DAG depends on exist:
airflow connections list | grep snowflake

# Re-run only the failed task against the same logical date:
airflow tasks test market_etl load_orders_to_warehouse 2026-07-15

5. Testing With Mocked Connections

Unit tests must never touch real infrastructure — that's what makes them fast, deterministic, and safe to run on every commit. The technique is mocking at the hook boundary: patch the hook or the HTTP client the task uses, and assert on how it was called.

The pattern from our test file is the template: patch SnowflakeHook, assert it was constructed with the right connection id, and assert run() got the SQL you intended. You're testing the task's contract with the outside world without ever opening a socket.

Two disciplines make this stick. First, mock at the boundary — mock the hook, not the internals of the task, or the test starts testing your mocks. Second, assert on the call: the task that quietly loses its connection id passes a happy-path test unless you assert the argument.

This is also where the incident dies permanently: a test that asserts load_orders_to_warehouse uses snowflake_prod would have failed the moment the connection id drifted from the prod name.

# tests/test_connection_contract.py
from unittest.mock import patch

from dags.market_etl import load_orders_to_warehouse


def test_load_binds_the_prod_connection():
    with patch("dags.market_etl.SnowflakeHook") as mock_hook:
        mock_hook.return_value.run.return_value = None
        load_orders_to_warehouse(rows=[{"order_id": 1, "amount": 12.5}])

        # The call is the contract:
        mock_hook.assert_called_once_with(snowflake_conn_id="snowflake_prod")
        mock_hook.return_value.run.assert_called_once()


# The CI gate that would have caught the original incident:
#   airflow connections get snowflake_prod   (must exit 0 in CI)

6. The CI Gate: Lint + Parse + Unit Tests Before Merge

Tests only matter if they run, and they only run if the gate blocks. The merge gate is three layers: lint and import checks, the pytest suite, and an environment validation pass — all before the merge button is worth anything.

Layer one is static: ruff or flake8 for the DAG folder, plus the import check that every file parses in the production image. Layer two is the suite: DagBag tests, task unit tests, mocked-connection tests — the files from this article. Layer three is the environment contract: connections that DAGs reference exist (airflow connections get), provider versions match, and airflow dags test runs green on the staging branch.

Gate placement matters. Lint and unit tests on every commit; dags test and environment validation on every staging merge. The expensive checks go where they protect the most, and the cheap ones block everywhere.

No test suite catches everything. But the suite here catches exactly what bit us — a DAG that parsed, reviewed green, and died on a connection that didn't exist. That's the incident this article is about, and it's the incident the gate is built to kill.

One CI knob keeps the suite honest: set AIRFLOW__CORE__UNIT_TEST_MODE=true in the test environment. Unit-test mode silences scheduler-side effects and keeps Airflow quiet and deterministic, so the suite behaves the same on every runner.

# .github/workflows/dag-tests.yml — the merge gate
set -euo pipefail

# 1. Lint + parse — every commit
ruff check dags/ tests/
airflow dags list | grep market_etl

# 2. Unit + dag-load tests — every commit
pytest tests/ -x -q --tb=short

# 3. Environment contract — staging merge only
airflow connections get snowflake_prod > /dev/null
airflow providers list | grep "apache-airflow-providers-snowflake"

# 4. Dress rehearsal — staging merge only
airflow dags test market_etl 2026-07-15

7. Incident Deep Dive: Failure Flow → Fix

Replay the timeline. The DAG landed as a pull request with a full code review — reviewers checked naming, style, and the shape of the graph. Nobody ran it. The file imported cleanly on every laptop involved, so the merge sailed.

Deploy was a simple copy to prod. The scheduler parsed the file, registered the DAG, and the graph rendered beautifully in the UI. At 01:00 the first scheduled run fired. Task one ran, task two ran, task three called SnowflakeHook with snowflake_prod — and the hook found no such connection. Run failed in seconds. The load that the downstream pipeline depended on never happened, and nothing alerted until the data team noticed missing rows.

Parsing succeeded the entire time. That's the incident: the scheduler's import path was clean, and every human review step verified the file that didn't contain the bug. The bug lived in the environment, invisible to parse and invisible to diff.

The fix was the discipline in this article: DagBag tests prove the file loads, task unit tests prove the callables behave, mocked-connection tests pin the contract (snowflake_prod, exactly), airflow dags test dress-rehearses the run, and the CI gate refuses merges until all of it is green — with connections verified against the prod environment contract.