How do you test Airflow DAGs before they reach production?

Detailed explanation

Layer	What it proves	Speed	Tooling
Integrity	Every file imports, no cycles, no duplicate dag_id, team rules hold, structure is as intended	Seconds	DagBag + pytest
Unit	Task logic, custom operators, hooks, timetables and callbacks behave correctly	Seconds	.function, plain functions, operator.execute(context) with mocked hooks
Integration	The DAG runs end to end, tasks hand over the right values, reruns are idempotent	Minutes	dag.test(), airflow dags test <dag_id> <date>, containers or a staging account

Why the DagBag matters: it imports each file the way the DAG processor does, records import errors instead of raising, runs the cycle check, and exposes each DAG’s attributes for rule checks. A plain import would miss cycles and would stop at the first broken file.

Rules worth asserting: an owner, tags, retries >= 1, an explicit catchup, a fixed start_date, no schedule_interval (gone in Airflow 3), and a short parse time per file. Pair them with the ruff AIR rules, which flag Airflow-specific problems such as imports removed in Airflow 3.

Example

A small DAG folder with one good DAG and two that a review should catch:

import os, tempfile, textwrap
from airflow.dag_processing.dagbag import DagBag

dags_dir = tempfile.mkdtemp(prefix="dags_")
files = {
    "orders_daily.py": '''
        from datetime import datetime, timedelta
        from airflow.sdk import dag, task

        @dag(schedule="@daily", start_date=datetime(2026, 1, 1), catchup=False, tags=["sales"],
             default_args={"owner": "sales-data", "retries": 2, "retry_delay": timedelta(minutes=5)})
        def orders_daily():
            @task
            def extract(ds=None) -> list[dict]:
                return [{"id": 1, "amount": 40.0}, {"id": 1, "amount": 40.0}, {"id": 2, "amount": -3.0}]

            @task
            def clean(rows: list[dict]) -> list[dict]:
                seen, out = set(), []
                for r in rows:
                    if r["id"] not in seen and r["amount"] >= 0:
                        seen.add(r["id"])
                        out.append(r)
                return out

            @task
            def count(rows: list[dict]) -> int:
                return len(rows)

            count(clean(extract()))

        orders_daily()
    ''',
    "broken_dag.py": '''
        from airflow.sdk import DAG
        import package_nobody_installed
    ''',
    "no_owner.py": '''
        from datetime import datetime
        from airflow.sdk import DAG
        from airflow.providers.standard.operators.empty import EmptyOperator

        with DAG("no_owner", schedule=None, start_date=datetime(2026, 1, 1)):
            EmptyOperator(task_id="noop")
    ''',
}
for name, body in files.items():
    with open(os.path.join(dags_dir, name), "w") as fh:
        fh.write(textwrap.dedent(body))

bag = DagBag(dag_folder=dags_dir)            # in a real repo: DagBag(dag_folder="dags/")

def test_no_import_errors():
    errors = {os.path.basename(k): v.strip().splitlines()[-1] for k, v in bag.import_errors.items()}
    assert not errors, errors

def test_dag_rules():
    problems = []
    for dag_id, d in sorted(bag.dags.items()):
        if not d.tags:
            problems.append(f"{dag_id}: no tags")
        if any(t.owner in (None, "", "airflow") for t in d.tasks):
            problems.append(f"{dag_id}: no owner")
        if any(t.retries < 1 for t in d.tasks):
            problems.append(f"{dag_id}: no retries")
    assert not problems, problems

def test_orders_structure():
    d = bag.dags["orders_daily"]
    assert d.task_ids == ["extract", "clean", "count"]
    assert d.get_task("count").upstream_task_ids == {"clean"}

for test in (test_no_import_errors, test_dag_rules, test_orders_structure):
    try:
        test()
        print(f"PASS {test.__name__}")
    except AssertionError as exc:
        print(f"FAIL {test.__name__}: {exc}")

FAIL test_no_import_errors: {'broken_dag.py': "ModuleNotFoundError: No module named 'package_nobody_installed'"}
FAIL test_dag_rules: ['no_owner: no tags', 'no_owner: no owner', 'no_owner: no retries']
PASS test_orders_structure

Airflow’s own log lines (including the traceback for broken_dag.py) are left out of the output above. The missing package would have been an import error on the DAG processor, and the DAG would have vanished from the UI; the rule test catches a DAG that passes import but breaks team standards.

Unit test the logic without a scheduler. A TaskFlow task keeps the original function as .python_callable on the task (and .function on the decorated object):

clean = bag.dags["orders_daily"].get_task("clean").python_callable

def test_clean_dedupes_and_drops_negatives():
    rows = [{"id": 1, "amount": 40.0}, {"id": 1, "amount": 40.0}, {"id": 2, "amount": -3.0}]
    assert clean(rows) == [{"id": 1, "amount": 40.0}]

test_clean_dedupes_and_drops_negatives()
print("PASS test_clean_dedupes_and_drops_negatives")

PASS test_clean_dedupes_and_drops_negatives

Integration test with dag.test(), which runs every task in one process with no scheduler. In Airflow 3 it needs the DAG to be registered in the metadata database, which happens when you run python dags/orders_daily.py (with if __name__ == "__main__": dag.test() at the bottom) or airflow dags test orders_daily 2026-03-01. Here a helper does the registration the DAG processor would normally do:

import contextlib, io
import pendulum
from airflow.dag_processing.bundles.manager import DagBundlesManager
from airflow.dag_processing.dagbag import sync_bag_to_db
from airflow.models.xcom import XComModel
from airflow.utils.session import create_session

def run_dag(dag, **kwargs):
    with contextlib.redirect_stdout(io.StringIO()):
        DagBundlesManager().sync_bundles_to_db()
        sync_bag_to_db(bag, "dags-folder", None)
        dr = dag.test(**kwargs)
    with create_session() as session:
        states = {ti.task_id: str(ti.state) for ti in dr.get_task_instances(session=session)}
        result = session.query(XComModel).filter_by(dag_id=dag.dag_id, run_id=dr.run_id,
                                                    task_id="count", key="return_value").one()
        return str(dr.state), dict(sorted(states.items())), XComModel.deserialize_value(result)

print(run_dag(bag.dags["orders_daily"], logical_date=pendulum.datetime(2026, 3, 1, tz="UTC")))

('success', {'clean': 'success', 'count': 'success', 'extract': 'success'}, 1)

Trade-offs and pitfalls

• Only testing imports. Rule and structure assertions catch the mistakes that import fine.
• CI on different versions from production: tests pass and parsing fails after deploy. Install with the Airflow constraints file or build the production image.
• Tests that need real cloud credentials get skipped or flake. Mock hooks, use local stand-ins (SQLite, MinIO, LocalStack-style emulators) or a dedicated test account.
• Wall-clock assertions. Always pass a fixed logical_date.
• Integration tests are slower; run them for critical DAGs or on merge, not on every commit.

Follow-up answers

• DagBag vs import: it runs the same parsing as production, collects every file’s errors, checks for cycles and duplicate IDs, and gives you DAG objects to assert on.
• Custom operator: instantiate it, patch its hook (for example with unittest.mock.patch), call execute(context={...}) with the fields it reads, and assert on the hook calls and return value.
• DebugExecutor replacement: dag.test() and airflow dags test, which run everything in one process so breakpoints work.
• Idempotency test: run the DAG (or task) twice for the same logical date against a test database and assert the target is identical after both runs.