What should a DAG test actually assert?

Three layers. That every file imports without error and has no cycles, that each task's business logic is correct in isolation, and that the graph wires tasks together as intended. The first two catch nearly everything and are cheap to run in CI.

How do I test a task without Airflow running?

Keep the logic in a plain function and have the task call it. Then the function is testable with ordinary pytest and no Airflow context at all. A task whose logic is inline in the operator is much harder to test.

What does dag.test() do?

It runs the whole DAG in your current process against the configured database, executing tasks in order with real XComs. It is the closest thing to an integration test without a scheduler, and it works well in CI against a throwaway SQLite database.

How do I stop a broken DAG reaching production?

Run the DAG validation test in CI on every pull request. An import error only surfaces in the UI as a missing DAG, so without a CI check a syntax error can silently remove a pipeline from your schedule.

Testing Airflow DAGs | SkillByExample

An Airflow DAG is Python, so it is testable — but the failure that bites hardest is not a logic bug. It is a DAG that fails to import, disappears from the scheduler, and takes a pipeline down without any alert.

Layer 1: does every DAG import

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


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


def test_no_import_errors(dagbag):
    """A DAG that fails to import vanishes from the scheduler with no alert."""
    assert not dagbag.import_errors, (
        "DAG import failures:\n"
        + "\n".join(f"  {path}: {err}" for path, err in dagbag.import_errors.items())
    )


def test_dags_were_found(dagbag):
    assert len(dagbag.dags) > 0, "no DAGs found — check dag_folder"


def test_every_dag_has_tags_and_owner(dagbag):
    for dag_id, dag in dagbag.dags.items():
        assert dag.tags, f"{dag_id} has no tags"
        assert dag.default_args.get("owner") not in (None, "airflow"), \
            f"{dag_id} has no real owner"


def test_no_cycles(dagbag):
    from airflow.utils.dag_cycle_tester import check_cycle
    for dag_id, dag in dagbag.dags.items():
        check_cycle(dag)          # raises AirflowDagCycleException on failure

pytest tests/test_dag_validation.py -v

tests/test_dag_validation.py::test_no_import_errors PASSED               [ 25%]
tests/test_dag_validation.py::test_dags_were_found PASSED               [ 50%]
tests/test_dag_validation.py::test_every_dag_has_tags_and_owner PASSED  [ 75%]
tests/test_dag_validation.py::test_no_cycles PASSED                     [100%]

============================== 4 passed in 3.42s ===============================

Break something and see what it catches:

# dags/broken.py
from airflow import DAG
import nonexistent_module

tests/test_dag_validation.py::test_no_import_errors FAILED

E   AssertionError: DAG import failures:
E     /repo/dags/broken.py: Traceback (most recent call last):
E       File "/repo/dags/broken.py", line 2, in <module>
E         import nonexistent_module
E     ModuleNotFoundError: No module named 'nonexistent_module'

This single test is the highest-value one you will write. Without it, that file simply stops producing a DAG and nothing tells you.

Cycles too:

    a >> b >> c >> a

E   airflow.exceptions.AirflowDagCycleException:
E   Cycle detected in DAG: task a is part of a cycle

Layer 2: unit-test the logic, not the operator

The key move is keeping business logic out of the DAG file:

# dags/lib/transforms.py — plain Python, no Airflow imports
def bucket_revenue(amount: float) -> str:
    if amount < 0:
        raise ValueError(f"negative amount: {amount}")
    if amount >= 1000:
        return "enterprise"
    if amount >= 100:
        return "standard"
    return "small"


def summarise(rows: list[dict]) -> dict:
    if not rows:
        return {"count": 0, "total": 0.0, "buckets": {}}
    buckets: dict[str, int] = {}
    for r in rows:
        b = bucket_revenue(r["amount"])
        buckets[b] = buckets.get(b, 0) + 1
    return {
        "count": len(rows),
        "total": round(sum(r["amount"] for r in rows), 2),
        "buckets": buckets,
    }

# dags/daily_summary.py
from datetime import datetime
from airflow.decorators import dag, task
from lib.transforms import summarise


@dag(dag_id="daily_summary", start_date=datetime(2026, 2, 1),
     schedule="@daily", catchup=False, tags=["reporting"],
     default_args={"owner": "data-team"})
def pipeline():

    @task
    def extract() -> list[dict]:
        return [{"amount": 1200.0}, {"amount": 250.0}, {"amount": 45.0}]

    @task
    def transform(rows: list[dict]) -> dict:
        return summarise(rows)          # one line — the logic lives elsewhere

    @task
    def load(summary: dict) -> None:
        print(f"loading {summary}")

    load(transform(extract()))


pipeline()

Now the logic tests need no Airflow at all:

# tests/test_transforms.py
import pytest
from dags.lib.transforms import bucket_revenue, summarise


@pytest.mark.parametrize("amount,expected", [
    (0.0,     "small"),
    (99.99,   "small"),
    (100.0,   "standard"),
    (999.99,  "standard"),
    (1000.0,  "enterprise"),
])
def test_bucket_boundaries(amount, expected):
    assert bucket_revenue(amount) == expected


def test_negative_amount_rejected():
    with pytest.raises(ValueError, match="negative amount"):
        bucket_revenue(-1.0)


def test_summarise_empty():
    assert summarise([]) == {"count": 0, "total": 0.0, "buckets": {}}


def test_summarise_counts_buckets():
    rows = [{"amount": 1200.0}, {"amount": 250.0}, {"amount": 45.0}, {"amount": 2000.0}]
    assert summarise(rows) == {
        "count": 4,
        "total": 3495.0,
        "buckets": {"enterprise": 2, "standard": 1, "small": 1},
    }

pytest tests/test_transforms.py -v

tests/test_transforms.py::test_bucket_boundaries[0.0-small] PASSED       [ 11%]
tests/test_transforms.py::test_bucket_boundaries[99.99-small] PASSED     [ 22%]
tests/test_transforms.py::test_bucket_boundaries[100.0-standard] PASSED  [ 33%]
tests/test_transforms.py::test_bucket_boundaries[999.99-standard] PASSED [ 44%]
tests/test_transforms.py::test_bucket_boundaries[1000.0-enterprise] PASSED [ 55%]
tests/test_transforms.py::test_negative_amount_rejected PASSED           [ 66%]
tests/test_transforms.py::test_summarise_empty PASSED                    [ 77%]
tests/test_transforms.py::test_summarise_counts_buckets PASSED           [ 88%]

============================== 8 passed in 0.21s ===============================

Two-tenths of a second, no database, no scheduler. That speed is what makes the tests get run.

Layer 3: does the graph wire up correctly

# tests/test_dag_structure.py
def test_task_ids(dagbag):
    dag = dagbag.get_dag("daily_summary")
    assert set(dag.task_ids) == {"extract", "transform", "load"}


def test_dependencies(dagbag):
    dag = dagbag.get_dag("daily_summary")
    assert dag.get_task("extract").downstream_task_ids == {"transform"}
    assert dag.get_task("transform").downstream_task_ids == {"load"}
    assert dag.get_task("load").downstream_task_ids == set()


def test_schedule_and_catchup(dagbag):
    dag = dagbag.get_dag("daily_summary")
    assert dag.schedule_interval == "@daily"
    assert dag.catchup is False, "catchup=True can queue hundreds of runs on unpause"

tests/test_dag_structure.py::test_task_ids PASSED         [ 33%]
tests/test_dag_structure.py::test_dependencies PASSED     [ 66%]
tests/test_dag_structure.py::test_schedule_and_catchup PASSED [100%]

The catchup assertion has saved real clusters. It is a one-line test against a mistake that queues a year of runs.

Running the whole DAG in-process

# tests/test_dag_run.py
from airflow.models import DagBag


def test_dag_runs_end_to_end():
    dag = DagBag(dag_folder="dags/", include_examples=False).get_dag("daily_summary")
    dag.test()          # executes every task in order, in this process

pytest tests/test_dag_run.py -v -s

[2026-02-15 14:22:01] {dagrun.py:698} INFO - DagRun Finished: dag_id=daily_summary,
  state=success, run_id=manual__2026-02-15T14:22:00+00:00
[2026-02-15 14:22:01] {taskinstance.py:1400} INFO - Marking task as SUCCESS. task_id=extract
[2026-02-15 14:22:01] {taskinstance.py:1400} INFO - Marking task as SUCCESS. task_id=transform
[2026-02-15 14:22:01] {logging_mixin.py:188} INFO - loading {'count': 3, 'total': 1495.0,
  'buckets': {'enterprise': 1, 'standard': 1, 'small': 1}}
[2026-02-15 14:22:01] {taskinstance.py:1400} INFO - Marking task as SUCCESS. task_id=load

============================== 1 passed in 4.83s ===============================

Real XComs, real ordering, no scheduler. dag.test() is the closest thing to an integration test that fits in CI.

Mocking external systems

# tests/test_with_mocks.py
from unittest.mock import patch, MagicMock
from dags.lib.loaders import load_to_warehouse


@patch("dags.lib.loaders.PostgresHook")
def test_load_uses_upsert(mock_hook_class):
    mock_hook = MagicMock()
    mock_hook_class.return_value = mock_hook

    load_to_warehouse([{"day": "2026-02-14", "count": 1284}])

    mock_hook.run.assert_called_once()
    sql = mock_hook.run.call_args[0][0]
    assert "ON CONFLICT" in sql, "writes must be idempotent — Airflow retries tasks"


@patch("dags.lib.loaders.PostgresHook")
def test_load_skips_empty(mock_hook_class):
    load_to_warehouse([])
    mock_hook_class.return_value.run.assert_not_called()

tests/test_with_mocks.py::test_load_uses_upsert PASSED   [ 50%]
tests/test_with_mocks.py::test_load_skips_empty PASSED   [100%]

Asserting on ON CONFLICT encodes a real operational requirement: Airflow retries, so a non-idempotent insert will duplicate rows on the second attempt.

In CI

# .github/workflows/dags.yml
name: DAG checks
on: [pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    env:
      AIRFLOW_HOME: ${{ github.workspace }}/.airflow
      AIRFLOW__CORE__LOAD_EXAMPLES: "False"
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install
        run: |
          pip install "apache-airflow==2.10.4" \
            --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.10.4/constraints-3.11.txt"
          pip install pytest

      - name: Initialise metadata DB
        run: airflow db migrate

      - name: Run tests
        run: pytest tests/ -v

Run pytest tests/ -v
============================= test session starts ==============================
collected 16 items

tests/test_dag_validation.py ....                                        [ 25%]
tests/test_dag_structure.py ...                                          [ 43%]
tests/test_transforms.py ........                                        [ 93%]
tests/test_dag_run.py .                                                  [100%]

============================== 16 passed in 12.71s =============================

A test-friendly DAG

Do	Instead of
Logic in `lib/`, tasks call it	logic inline in the operator
`{{ ds }}` for dates	`datetime.now()`
Read Variables inside tasks	reading at module level
Parameters via `op_kwargs` or arguments	globals set at import
Idempotent writes (`ON CONFLICT`)	plain `INSERT`

Each of these makes a DAG both more testable and more correct at runtime — the same properties that let a task be unit-tested let it be retried and backfilled safely.

Practice

1. Add a DAG file with a syntax error and run the validation test.

E   AssertionError: DAG import failures:
E     /repo/dags/broken.py: invalid syntax (broken.py, line 12)

Without this test the DAG just disappears from the UI. Make it the first test in every Airflow repo, and run it on every pull request.

2. Create a cycle and see what check_cycle reports.

E   airflow.exceptions.AirflowDagCycleException: Cycle detected in DAG: task a is part of a cycle

Airflow also refuses to load such a DAG at runtime, but catching it in CI means the branch never merges.

3. Test bucket_revenue at exactly 100 and 1000.

test_bucket_boundaries[100.0-standard] PASSED
test_bucket_boundaries[1000.0-enterprise] PASSED

Boundaries are where the bugs are. Parametrised tests make the boundary cases explicit in the test names, so a failure reads as [1000.0-enterprise] FAILED rather than a line number.

4. Run dag.test() on a DAG whose task raises.

[2026-02-15 14:31:02] {taskinstance.py:1938} ERROR - Task failed with exception
ValueError: negative amount: -5.0
[2026-02-15 14:31:02] {dagrun.py:698} INFO - DagRun Finished: state=failed

The exception propagates and the run is marked failed, so the test fails — which is what you want. Note dag.test() honours retries, so a task with retries=3 will take three attempts before the test reports failure.

Next: the patterns that keep a DAG healthy once it is running every day.