Testing Airflow Locally: A Setup That Catches Bugs Before Production Does

The DAG that taught me to test was two hundred lines and imported a helper module. Somebody renamed a function in that module. The DAG file still parsed on their machine because their editor had the old .pyc around. It did not parse on the scheduler.

Airflow’s response to an unparseable DAG is to leave the previous version in place and log the error in the DAG processor. The DAG kept showing green in the UI, running yesterday’s code, for four days. Nobody noticed until a column that the new code was supposed to add never showed up in a report.

A one-line pytest would have caught it. We did not have one, because “testing Airflow is hard” is something everyone says and mostly means “end-to-end testing Airflow is hard,” which is true and is also not where the bugs are.

Here is the four-layer setup I put on every Airflow project. The first two layers take an hour to build and catch most of what actually breaks.

Layer 1: does it parse

This is the highest-value test in the entire pyramid and it is about fifteen lines.

# tests/test_dag_integrity.py
import pytest
from airflow.models import DagBag


@pytest.fixture(scope="session")
def dagbag():
    # include_examples=False or you will test Airflow's tutorials.
    return DagBag(dag_folder="dags/", include_examples=False)


def test_no_import_errors(dagbag):
    assert not dagbag.import_errors, (
        "DAG import failures:\n"
        + "\n".join(f"  {f}: {e}" for f, e in dagbag.import_errors.items())
    )


def test_dags_were_found(dagbag):
    # Guards against a path typo silently testing an empty folder.
    assert len(dagbag.dags) > 0

The second test matters more than it looks. A wrong dag_folder gives you a DagBag with zero DAGs and zero import errors, and the suite passes forever while testing nothing.

While you have a DagBag, enforce your conventions. These are the policy checks that would otherwise live in a code review checklist that gets skipped when the reviewer is busy.

# tests/test_dag_conventions.py
import pytest

REQUIRED_TAGS = {"finance", "marketing", "platform", "product"}


def test_all_dags_have_an_owner(dagbag):
    for dag_id, dag in dagbag.dags.items():
        owner = dag.default_args.get("owner", "airflow")
        assert owner != "airflow", f"{dag_id} has the default owner"


def test_all_dags_have_a_team_tag(dagbag):
    for dag_id, dag in dagbag.dags.items():
        assert REQUIRED_TAGS & set(dag.tags), f"{dag_id} has no team tag"


def test_catchup_is_explicit(dagbag):
    # The default changed in Airflow 3. Never rely on it.
    for dag_id, dag in dagbag.dags.items():
        assert "catchup" in dag.params or dag.catchup is not None, dag_id


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


@pytest.mark.parametrize("forbidden", ["execution_date", "next_ds", "prev_ds"])
def test_no_removed_context_keys(forbidden):
    import pathlib
    hits = [
        str(p) for p in pathlib.Path("dags").rglob("*.py")
        if forbidden in p.read_text(encoding="utf-8")
    ]
    assert not hits, f"{forbidden} used in: {hits}"

That last one saved me real time during the Airflow 3 migration, because removed template keys fail at render time in production rather than at parse time anywhere useful.

Parse tests are fast if you keep top-level DAG code cheap. If they take a minute, that is itself the finding — something in your DAG files is hitting a network or a database at import, and the scheduler is doing it every thirty seconds.

Layer 2: unit tests on logic you extracted from the DAG

You cannot meaningfully unit test a PythonOperator callable that opens a connection, transforms data, and writes a table in one function. So do not write that function.

# include/transforms/orders.py — plain Python, no Airflow imports
from dataclasses import dataclass
from datetime import date, datetime


@dataclass(frozen=True)
class Order:
    order_id: str
    ordered_at: datetime
    amount_cents: int
    status: str


def classify_status(shipped_at, refunded_at) -> str:
    """Business rule: a refund after shipment classifies as returned."""
    if refunded_at is not None:
        return "returned"
    if shipped_at is not None:
        return "shipped"
    return "placed"


def partition_path(bucket: str, logical: date | None, fallback: date) -> str:
    """Airflow 3: logical_date can be None for manual/asset-triggered runs."""
    effective = logical or fallback
    return f"s3://{bucket}/orders/dt={effective:%Y-%m-%d}/"

# tests/test_transforms.py
from datetime import date, datetime
from include.transforms.orders import classify_status, partition_path


def test_refund_after_shipment_is_returned():
    assert classify_status(datetime(2026, 1, 5), datetime(2026, 1, 9)) == "returned"


def test_shipped_without_refund():
    assert classify_status(datetime(2026, 1, 5), None) == "shipped"


def test_partition_path_falls_back_when_logical_date_is_none():
    assert partition_path("lake", None, date(2026, 5, 21)) == \
        "s3://lake/orders/dt=2026-05-21/"

No Airflow, no fixtures, no database. These run in milliseconds and they test the part that has business consequences. The DAG file becomes a thin wiring layer:

# dags/orders.py
import pendulum
from airflow.sdk import dag, task
from include.transforms.orders import partition_path

@dag(schedule="@daily", start_date=pendulum.datetime(2026, 1, 1, tz="UTC"), catchup=False)
def orders():

    @task
    def resolve_path(**context) -> str:
        return partition_path(
            bucket="lake",
            logical=context.get("logical_date"),
            fallback=context["dag_run"].run_after.date(),
        )

    @task
    def load(path: str) -> None:
        ...

    load(resolve_path())

orders()

The rule: anything with a branch belongs outside the DAG file. If the DAG file contains an if about data, move it.

flowchart TD
  l1["Parse tests (seconds)"] --> l2["Unit tests on extracted logic (seconds)"]
  l2 --> l3["Task tests with mocked hooks (tens of seconds)"]
  l3 --> l4["dag.test() against local infra (minutes)"]
  l4 --> stg["Staging environment (hours)"]

Each layer is roughly ten times slower and catches roughly ten times fewer bugs. Build them bottom-up.

Layer 3: task tests with mocked hooks

Some logic genuinely lives in the operator boundary — a query built from parameters, an S3 key pattern, a retry decision. Test those by mocking the hook, not the network.

# tests/test_load_task.py
from unittest.mock import MagicMock, patch

from include.tasks.warehouse import merge_partition


@patch("include.tasks.warehouse.SnowflakeHook")
def test_merge_uses_partition_predicate(mock_hook_cls):
    mock_hook = MagicMock()
    mock_hook_cls.return_value = mock_hook

    merge_partition(table="raw.orders", partition_date="2026-05-21")

    sql = mock_hook.run.call_args[0][0]
    # The predicate is what makes the merge prune. Assert it exists.
    assert "dt = '2026-05-21'" in sql
    assert "merge into raw.orders" in sql.lower()


@patch("include.tasks.warehouse.SnowflakeHook")
def test_merge_refuses_empty_partition_date(mock_hook_cls):
    import pytest
    with pytest.raises(ValueError):
        merge_partition(table="raw.orders", partition_date="")
    mock_hook_cls.return_value.run.assert_not_called()

Mock at the hook class, one level below your own code, so you are testing the SQL you generate rather than a mock of your own function. A test that requires real credentials is a test that gets an @pytest.mark.skipif within a month and then never runs again.

Layer 4: dag.test() against local infrastructure

dag.test() executes an entire DAG in a single process with no scheduler, no workers, and no queue. It is the fastest honest end-to-end run you can get, and it is a debugger-friendly one — you can put a breakpoint inside a task and it will hit.

# tests/test_dag_end_to_end.py
import pendulum
import pytest
from airflow.models import DagBag


@pytest.mark.integration
def test_orders_dag_runs_end_to_end(monkeypatch):
    monkeypatch.setenv(
        "AIRFLOW_CONN_WAREHOUSE",
        "postgres://airflow:airflow@localhost:5433/warehouse",
    )
    monkeypatch.setenv(
        "AIRFLOW_CONN_LAKE",
        "aws://?endpoint_url=http%3A%2F%2Flocalhost%3A9000"
        "&aws_access_key_id=minio&aws_secret_access_key=minio123",
    )

    dag = DagBag(dag_folder="dags/", include_examples=False).get_dag("orders")
    dag.test(logical_date=pendulum.datetime(2026, 5, 21, tz="UTC"))

Point it at real local services rather than mocks. The bugs this layer catches are integration bugs — a SQL dialect difference, a serialized XCom that is not JSON-serializable, a path with a double slash — and mocks hide all of them.

# docker-compose.test.yml — the dependencies, not Airflow itself
services:
  warehouse:
    image: postgres:16
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: warehouse
    ports: ["5433:5432"]
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U airflow"]
      interval: 5s
      retries: 10

  minio:
    image: minio/minio
    command: server /data --console-address ":9001"
    environment:
      MINIO_ROOT_USER: minio
      MINIO_ROOT_PASSWORD: minio123
    ports: ["9000:9000", "9001:9001"]

Two services, no Airflow containers, starts in about eight seconds. For warehouse-shaped assertions without a warehouse, DuckDB is often enough — the pattern is in DuckDB for local data pipelines.

Run these separately from the fast suite:

pytest tests/ -m "not integration"     # every PR, ~10 seconds
pytest tests/ -m integration           # on merge to main

Pinning: test the Airflow you actually run

The most frustrating class of local-testing failure is the one where the test passes and production breaks because the provider version differs. Provider packages change operator arguments between minor versions more often than people expect.

# requirements.txt — pin against the official constraints file
apache-airflow==3.0.4
apache-airflow-providers-amazon==9.5.0
apache-airflow-providers-snowflake==6.1.0
apache-airflow-providers-standard==1.0.0

# Install exactly what the deployment installs.
AIRFLOW_VERSION=3.0.4
PYTHON_VERSION=3.12
pip install "apache-airflow==${AIRFLOW_VERSION}" \
  --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"
pip install -r requirements.txt

The constraints file is not optional. Airflow has a large dependency tree and unconstrained resolution produces a local environment that does not match any deployment anywhere.

# .github/workflows/ci.yml
name: airflow-ci
on: [pull_request]
jobs:
  fast:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: pip
      - run: pip install -r requirements.txt -r requirements-dev.txt
      - run: ruff check dags/ include/ tests/
      - run: pytest tests/ -m "not integration" -q

  integration:
    runs-on: ubuntu-latest
    needs: fast
    steps:
      - uses: actions/checkout@v4
      - run: docker compose -f docker-compose.test.yml up -d --wait
      - run: pip install -r requirements.txt -r requirements-dev.txt
      - run: pytest tests/ -m integration -q

Where teams get this wrong

No parse test. The single cheapest test in the stack, missing from most repos I open. An unparseable DAG runs stale code silently, which is worse than failing.

Expensive code at DAG top level. A database query outside a task function runs on every parse — every thirty seconds, per DAG file, in production. It also makes your parse test slow, which is how you find it.

Mocking your own functions. A test that patches the thing it is testing asserts that Python can call a mock. Mock at the boundary: hooks, clients, the filesystem.

Integration tests in the PR suite. A ninety-second suite gets run locally and then skipped. Split fast from slow and gate them differently.

Testing against different provider versions than production. The operator argument that exists in 9.5.0 and not 8.x will find you. Pin with constraints in both places.

Only testing scheduled runs. Manual and asset-triggered runs can have a null logical_date. If every test passes a date, the None path is never exercised — and it is the most common Airflow 3 production break, as noted in data-aware scheduling.

FAQ

Do I need a full local Airflow to develop DAGs?

For writing and testing them, no — dag.test() plus pytest covers most of it and is much faster. Run a local Airflow when you need to verify scheduler behavior specifically: catchup, concurrency limits, pool contention, or that a triggerer picks up deferred tasks.

How do I test deferrable operators locally?

dag.test() handles the defer-and-resume cycle in-process, so a deferrable sensor works. What it does not test is a real triggerer under load, which is where the operational failures live — see deferrable operators.

How do I test dynamic task mapping?

Assert on the structure of the expanded list rather than running every index: unit test the function that builds the fan-out list, including your upper-bound guard. Then run dag.test() with a small input to verify the wiring. The sizing concerns are in dynamic task mapping.

Should I test that a DAG produces correct data?

Not in Airflow’s test suite. Data correctness belongs in the transform layer’s tests, whether that is dbt tests or Python data quality checks. Airflow tests should verify orchestration: does it parse, does it wire correctly, does each task do the one thing it claims.

What about testing connections and variables?

Set them as environment variables in the test, using the AIRFLOW_CONN_* and AIRFLOW_VAR_* naming. Never read a real connection in a test. If a DAG requires a Variable at parse time, that is a separate problem — parse-time Variable lookups hit the database on every scheduler loop.

How long should the fast suite take?

Under fifteen seconds for a repo with a hundred DAGs. If it is slower, profile the parse test. Slow parsing in the test suite is slow parsing in the scheduler, and the scheduler does it far more often than CI does.

What this means for your pipelines

Airflow testing has a reputation for being impractical, and that reputation comes from teams attempting layer four first. Standing up a full Airflow with a scheduler, a triggerer, workers, and a metadata database in CI is genuinely painful, and it is also the layer with the worst ratio of effort to bugs found.

The bugs are further down. A DAG that does not parse. A function that was renamed. A branch in a transform that nobody exercised. A path built from a date that can be None. Every one of those is caught by a test that runs in under a second and needs no infrastructure, and the only thing standing between most teams and that suite is an afternoon of extracting logic out of DAG files into plain Python modules.

Build it bottom-up. Parse test first, today. Then move your business logic into include/ and test it directly. Add hook-mocked tests for the SQL you generate. Add dag.test() against Postgres and MinIO for the handful of DAGs where integration bugs would be expensive. And pin your providers with the constraints file in both environments, because a green suite testing a different Airflow than the one you deploy is the most demoralizing kind of green there is.