Unit Testing DAGs with pytest

Integrity tests load every DAG file the way the DAG processor does and assert on the result; unit tests exercise the plain code the tasks call, which is why BookNest keeps its logic in importable modules:

tests/test_dags.py (excerpt) and tests/test_pipeline.py (excerpt)Python

@pytest.fixture(scope="session")
def dagbag():
    return DagBag(dag_folder="/opt/airflow/dags")     # parses files like the DAG processor
def test_no_import_errors(dagbag):
    assert dagbag.import_errors == {}
def test_daily_pipeline_shape(dagbag):
    dag = dagbag.dags["booknest_daily"]
    assert dag.max_active_runs == 1 and dag.catchup is False
    assert dag.get_task("publish").upstream_task_ids == {"model.check"}
def test_trigger_fires_once_the_day_is_complete(tmp_path):
    write_orders(tmp_path / "export.jsonl", "2026-06-30T23:55:02Z", "2026-07-01T00:03:12Z")
    trigger = ExportWatermarkTrigger(str(tmp_path / "export.jsonl"), "2026-06-30", 0.1)
    event = asyncio.run(anext(trigger.run()))
    assert event.payload == {"ds": "2026-06-30", "newest": "2026-07-01T00:03:12Z"}

run7/w6.sh runs them with docker run --rm --network none --entrypoint pytest, mounting dags, include and tests read-only into l2/airflow-booknest-test:3.3.2 (the BookNest image plus pytest 9.1.1): no network, no metadata database, no running stack. The last lines of its output:

Output of 27

tests/test_pipeline.py::test_trigger_fires_once_the_day_is_complete PASSED [100%]
============================== 8 passed in 10.72s ==============================

Two Airflow 3.3 129 details broke the first draft: DagBag no longer takes include_examples (examples follow [core] load_examples), and dagbag.get_dag() queries the metadata database, so the tests read dagbag.dags. Run the suite in CI on every change (Package Managers and DevOps).

MENU