Parameterized Runs and Backfills
In Parameterized Runs and Backfills you learned how to replay a DAG over historical dates. The question that replay exposes is: will the DAG you wrote actually run? A DAG whose Python fails to import is invisible in the Airflow UI until the scheduler next parses it, which can be minutes after you push. By then the pager is out.
In dbt Tests you wrapped dbt models in not_null, unique, and relationships tests so that broken data failed fast. This chapter does the same for broken DAG code: parse-time tests that catch import errors, syntax mistakes, and common wiring problems before the DAG reaches a scheduler at all; and unit tests that validate the logic inside individual TaskFlow tasks without running Airflow.
<aside> 📘 Week 10 parallel. Week 10's dbt tests answer "is the data correct right now?" Week 12's DAG tests answer "is the orchestration code correct before we run it at all?" Both matter. Both live next to the code they test.
</aside>
By the end of this chapter, you should be able to:
pytest + DagBag that catches import errors and cycles.pytest, no Airflow runtime needed.pytest and in CI with one pytest invocation.The Airflow scheduler imports every .py file in dags/ to build its DAG list. If any of those files raises at import time (a missing dependency, a typo, a wrong argument name), the scheduler silently drops the DAG. You will not see it in the UI, you will not get a retry, you will not get an alert. The first time you find out is when a stakeholder asks why yesterday's dashboard is empty.
Every Airflow project above toy-size has at least one DAG integrity test for exactly this reason. One pytest file, twenty lines, catches the entire class of "DAG did not even load" failures.
When the scheduler can't parse a DAG, the UI looks like this: the DAG list shows 0 Dags with a red "1" badge next to it, and the body of the page reads No Dags found. The file is there, the code is there, but Airflow will not run anything because the import failed.
Airflow DAG list showing 0 Dags, red 1 badge indicating one import error, and "No Dags found" in the body
That is exactly the state your integrity test exists to prevent.
The minimum viable test asserts every DAG in dags/ can be imported without raising. DagBag does the work for you:
# tests/test_dag_integrity.py
from airflow.models import DagBag
def test_no_import_errors():
# Airflow 3's DagBag no longer accepts include_examples=...
dag_bag = DagBag(dag_folder="dags")
assert dag_bag.import_errors == {}, (
f"DAG import errors: {dag_bag.import_errors}"
)
def test_every_dag_has_tags():
dag_bag = DagBag(dag_folder="dags")
for dag_id, dag in dag_bag.dags.items():
assert dag.tags, f"DAG {dag_id} is missing tags"
<aside>
⚠️ If you copy an older snippet that passes include_examples=False into DagBag(...), Astro's current Airflow raises TypeError: got an unexpected keyword argument 'include_examples'. Drop that argument.
</aside>
Run it with pytest inside the Astro container (Astro ships pytest in the runtime image, so no extra pip install is needed):
astro dev pytest tests/test_dag_integrity.py --args "-v"
A green run reports both tests passing:
tests/test_dag_integrity.py::test_no_import_errors PASSED [ 50%]
tests/test_dag_integrity.py::test_every_dag_has_tags PASSED [100%]
===== 2 passed in 2.34s =====
What this catches:
DagBag rejects DAGs with cyclic dependencies at load time.default_args or start_date when required.What this does not catch: logic bugs inside your tasks. That is the next section.
<aside>
💡 Run astro dev pytest locally before every push. A pre-commit hook or CI step that runs this single test saves hours of "my DAG is missing from the UI, why?" debugging.
</aside>
Run the integrity test against your own project, then break it on purpose so you see the test earn its keep.
Step 1: Create the test file. Add tests/test_dag_integrity.py to your Week 12 project with the two tests from above (test_no_import_errors and test_every_dag_has_tags).
Step 2: Run it and confirm green. From the project root:
astro dev pytest tests/test_dag_integrity.py --args "-v"
Both tests should pass, the same 2 passed output shown above.
Step 3: Introduce a deliberate bug. In one DAG file, change an import to a nonexistent module (note the doubled s in postgress):
from airflow.providers.postgres.hooks.postgress import PostgresHook
Step 4: Rerun and read the failure. Run the same command again. The integrity test fails and names the exact broken import, while the tag test still passes for the DAGs that did load:
tests/test_dag_integrity.py::test_no_import_errors FAILED
tests/test_dag_integrity.py::test_every_dag_has_tags PASSED
...
E AssertionError: DAG import errors: {'dags/taxi_pipeline.py':
E "ModuleNotFoundError: No module named
E 'airflow.providers.postgres.hooks.postgress'"}
===== 1 failed, 1 passed in 2.11s =====
The message above is trimmed for readability: the real import_errors value is the full import traceback (Traceback (most recent call last): ...), and the ModuleNotFoundError line is its last line, which is the part that tells you what broke.
Step 5: Revert and confirm idempotency. Fix the typo, then run the same command twice without changing anything. Both runs produce byte-identical output: the tests are idempotent by construction.
One caveat if you jump ahead.
<aside>
💡 If you have already added tests/test_taxi_pipeline.py (next section) when you do the deliberate-bug experiment, the failure mode changes: pytest cannot even collect the test file because its from dags.taxi_pipeline import parquet_url_for line fails on the broken DAG. You get Interrupted: 1 error during collection instead of a clean assertion failure. Both outcomes catch the bug; the collection error is just louder.
</aside>
Task bodies frequently mix pure logic (compute a URL, normalize a row, pick a partition name) with side effects (HTTP, DB writes, file I/O). The pure parts are trivial to unit-test if you pull them out of the @task decorator body into a module-level function that the task then calls.
Your taxi_pipeline.py already has one such helper: parquet_url_for(ds: str) -> str computes the TLC URL for a given logical date. It is a pure function (no HTTP, no DB, no randomness), so you can import it directly from pytest:
# tests/test_taxi_pipeline.py
from dags.taxi_pipeline import parquet_url_for
def test_parquet_url_january():
url = parquet_url_for("2024-01-01")
assert url == (
"<https://d37ci6vzurychx.cloudfront.net/>"
"trip-data/green_tripdata_2024-01.parquet"
)
def test_parquet_url_end_of_month():
# The logical date for a @monthly run is the first of the month,
# but double-check that other days in the month also slice to the
# right year-month prefix.
url = parquet_url_for("2024-01-31")
assert "2024-01.parquet" in url
def test_parquet_url_december_rollover():
url = parquet_url_for("2023-12-01")
assert "2023-12.parquet" in url
Three tests, zero Airflow imports. Run them with astro dev pytest tests/ --args "-v".
This pattern generalizes: if a task body has any non-I/O logic (date math, string normalization, a lookup table, a branching decision), extract it into a module-level function and test that function. The I/O wrapper (ingest_taxi_month) stays untested at the unit-test layer; you verify it via the UI-triggered DAG run.
<aside>
💡 Favor TaskFlow (@task) over classic operators partly because the task bodies are regular Python functions. Every pure helper you can test in isolation is a task body that cannot surprise you at 03:00.
</aside>
If you need to test a task that genuinely has no pure core (it does nothing but I/O), the right test is an integration test with a mocked hook, not a unit test. That is the next-level pattern covered in Going Further; the three-tests-for-parquet_url_for above is all Week 12 needs.
Sometimes the bug is in how tasks are wired together, not in any one task. A structural test asserts the dependency graph looks right:
# tests/test_taxi_pipeline_structure.py
from airflow.models import DagBag
def test_ingest_runs_before_dbt_run():
# Index .dags[...] (the in-memory parse result), not .get_dag(...):
# under Airflow 3, get_dag() reads the serialized DAG from the
# metadata DB, which the ephemeral `astro dev pytest` container has
# not populated, so it raises a SQL error. Indexing .dags stays
# in-process and needs no database.
dag = DagBag(dag_folder="dags").dags["taxi_pipeline"]
ingest = dag.get_task("ingest_taxi_month")
dbt_run = dag.get_task("dbt_run")
assert dbt_run in ingest.downstream_list, (
"dbt_run should run after ingest_taxi_month"
)
Use this sparingly. The grid view in the UI already shows you the wiring visually. Structural tests earn their keep when the DAG is dynamically built (for loops that generate tasks from a config) and a subtle bug could insert the wrong edge.
For a Week 12 project, stop at the three patterns above. Things that are not worth writing tests for now:
<aside> 🤓 Curious Geek: DagBag is what the scheduler uses too
The DagBag class your test imports is the same class the Airflow scheduler instantiates on every parse cycle. When your test says "no import errors," it is literally asserting what the scheduler would conclude two seconds after your next git push. This is why the integrity test catches so many real incidents for so little code.
</aside>
Astro ships pytest inside the runtime image, so astro dev pytest runs your tests in the same Python environment the scheduler uses. No virtualenv, no pip install, no version-skew between your laptop and the container:
# runs all tests in tests/ with verbose output
astro dev pytest tests/ --args "-v"
# runs just the integrity test
astro dev pytest tests/test_dag_integrity.py --args "-v"
<aside>
⚠️ Astro's astro dev init ships a default tests/dags/test_dag_example.py that runs alongside your tests. One of its checks (test_dag_retries) asserts every DAG has retries >= 2 in default_args. The Parameterized Runs and Backfills taxi*pipeline already sets default*args={"retries": 2} to satisfy this (reading ahead to Monitoring and Debugging's retry-configuration discussion); if you wrote your own DAG without retries andastro dev pytest tests/fails on test*dag*retries, that default test is what flagged you.
</aside>
In CI (GitHub Actions, for example) there is no Docker or Astro stack, so you install Airflow and pytest directly and run plain pytest: the integrity test only needs Airflow to be importable, not a running scheduler. This is what the class repo's PR check does:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Install Airflow + project deps
run: |
pip install "apache-airflow==3.0.*"
pip install apache-airflow-providers-standard # bundled in Astro, explicit here
pip install -r requirements.txt pytest
- name: Run DAG integrity tests
env:
AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: "True"
AIRFLOW__CORE__LOAD_EXAMPLES: "False"
run: pytest tests/test_dag_integrity.py -v
The AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION=True environment variable is defensive: it prevents a misconfigured test environment from accidentally scheduling real DAG runs. astro dev pytest (above) is the local convenience; CI uses plain pytest because it is lighter and needs no Docker.
If your test file does not collect or the asserts fire for reasons you cannot explain, fall back to the reference.
<aside>
📦 Stuck? The ch6-tests branch of the reference repo holds this chapter's full tests/ suite plus the matching taxi_pipeline.py, verified with astro dev pytest tests/ --args "-v" on Airflow 3. Diff your own copies against it if a test is not catching what you expect, or if pytest collects nothing. The single-file end-state also lives at assets/dag_snapshots/test_dag_integrity.py.
</aside>
Before pushing your assignment branch, make one of those deliberately-broken DAG files fail the test, then fix it. That round-trip is what you want the test doing for you in CI.
<aside>
💡 Using AI to help: Paste a TaskFlow task function and ask an LLM to draft a pytest test with three input cases: happy path, edge case, and error case (⚠️ Ensure no PII or sensitive company data is included!). Review the generated tests carefully: LLMs often invent helper imports or assert on fields the function does not return.
</aside>
The chapter walked you through writing all three test types against taxi_pipeline. One drill cements them: try to break your own function.
<aside>
⌨️ Hands on: After adding the three unit tests for parquet_url_for, write a fourth test for an input you predict would break the current implementation: a two-digit day like 24-01-01, a leap-year date like 2024-02-29, or a string shorter than 7 characters. Run astro dev pytest tests/test_taxi_pipeline.py --args "-v". If it reveals a bug, decide: fix parquet_url_for, or document the assumption the function is allowed to make.
</aside>
The same integrity-test habit shows up again before you deploy to shared Airflow.
<aside>
📝 Practice: In the week's Practice chapter, Exercise 7 includes a pre-push astro dev pytest tests/test_dag_integrity.py --args "-v" step before the shared deploy. It reuses this chapter's parse-test pattern in the place where it matters operationally.
</aside>
parquet_url_for but not ingest_taxi_month. Why is the helper easier to unit-test, and what would be needed to test the task too?downstream_list) worth writing, and when is it over-engineering?<aside> 🚀 Try it in the widget: Interactive Quiz: Testing DAGs (covering the integrity test, parse-time vs unit tests, and what not to test yet).
</aside>
https://lasse.be/simple-hyf-teach-widget/mcq.html?bank=week_12_ch6_testing_dags_quiz&embed=1
If pytest and the DagBag integrity pattern still feel unfamiliar, this walkthrough sets up DAG testing from scratch.
<aside> 🎬 Struggling with this concept? Watch this beginner-friendly video:
</aside>
https://www.youtube.com/watch?v=Onm2vWy7SUM
The next chapter debugs failed runs from their logs, so it assumes you can run your DAG's tests and trust a green parse before you trigger.
astro dev pytest tests/test_dag_integrity.py --args "-v" passes against your project.parquet_url_for and can explain why it is easier to test than ingest_taxi_month.DagBag and the test-a-dag workflow.astro dev pytest conventions.