Apache Airflow / Intermediate: runtime work, control flow and resource budgets / Lesson 12

Testing DAG contracts and safe local logic

Choose tests that establish specific business and graph properties without overstating what an offline check proves.

About 4 min readContent reviewed 2026-10-28

On this page

Lesson goals

Separate pure-function tests, definition checks and integration tests.
Write deterministic fixtures for intervals, duplicates and invalid inputs.

Name the claim before choosing the test

A pure-function test can establish how supplied rows are validated or aggregated. A DAG-definition check can establish task identities, dependency edges or supported imports in a compatible environment. An integration test can exercise task execution, communication and external boundaries. These are different claims and should not be collapsed into a statement that the pipeline was tested. Begin with a concrete failure the test should catch: a boundary row counted twice, a duplicate append or a missing prerequisite. Avoid tests that simply repeat the implementation without checking an independently stated invariant. Small deterministic fixtures often reveal more than a large unexplained sample.

Test an interval filter with edge fixtures

The local helper selects records in a half-open integer interval. Fixtures include one value before the start, one at the start, one just before the end and one exactly at the end. The expected result is stated independently as the two included identifiers. Integer timestamps are a declared simplification, not a substitute for timezone-aware calendar testing. The empty-input case has a defined empty result. If malformed input should fail, test that separately rather than silently dropping it. This example executes as ordinary isolated Python and makes no Airflow imports, database calls or remote artifact reads.

python Offline edge-focused interval test

def select_interval(rows, start, end):
    return [row['id'] for row in rows if start <= row['timestamp'] < end]

rows = [{'id': 'before', 'timestamp': 9}, {'id': 'start', 'timestamp': 10}, {'id': 'inside', 'timestamp': 19}, {'id': 'end', 'timestamp': 20}]
assert select_interval(rows, 10, 20) == ['start', 'inside']
assert select_interval([], 10, 20) == []
print(select_interval(rows, 10, 20))

Expected output

['start', 'inside']

Only start and inside are included under the declared half-open rule.

Review graph edges without pretending to parse Airflow

The JSON fixture declares required task names and edges for a tiny report. A local check can compare that specification with an independently supplied graph representation, but cannot prove an actual DAG file loads unless the supported framework loader is exercised. Airflow 3 marks internal models such as DagBag as internal interfaces for authors, so do not build a teaching claim on direct metadata-model access. A real definition test should follow the selected version's supported tooling in an authorized environment. Keep task IDs and intended trigger rules in the contract because a renamed node or permissive leaf can change operational meaning.

json Synthetic graph acceptance specification

{"tasks":["validate","summarize","publish"],"required_edges":[["validate","summarize"],["summarize","publish"]],"publication_requires":"validated-complete-artifact"}

This states desired graph behavior and is not evidence an Airflow parser accepted a DAG.

Include failures that occur between systems

A meaningful pipeline test matrix includes invalid schema, unavailable input, partial artifact publication, lost destination acknowledgment and retry after a completed effect. Pure fixtures can test transformation decisions, but crash safety at a remote boundary needs the destination's actual supported protocol. Use synthetic identities and disposable authorized resources when such integration work is permitted. Never aim a DAG test command at an existing configured environment merely because its name contains test; some helpers can execute task logic and contact real systems. Review what will run and which connections it resolves before choosing the test surface.

Report exactly what ran and what remains untested

Record the interpreter or framework baseline, fixture identity, expected result and observed outcome. Distinguish compile-only or static review from executed behavior. A passing local assertion does not certify scheduler timing, executor dispatch, secret resolution or provider compatibility. When the implementation changes, rerun checks tied to the changed claim rather than repeatedly exercising unrelated services. This course executes the self-contained Python examples in isolated temporary storage and parses JSON fixtures. Every Airflow DAG or operator fragment remains runnable:false, so learners can inspect real syntax without being told an unavailable scheduler or database was tested.

Common errors and fixes

A pure-function test is reported as a full pipeline integration test.

Why it happens: The test boundary was not stated.
How to fix it: Report the exact inputs, runtime and property verified.
Prevent it next time: Keep local and framework evidence separately labeled.

A test helper unexpectedly contacts a configured service.

Why it happens: Its task execution and connection lookup behavior were not reviewed.
How to fix it: Stop dependent execution and use isolated pure fixtures until an authorized environment exists.
Prevent it next time: Inspect test command side effects before running it.

The interval test omits exact boundaries.

Why it happens: Only an ordinary interior record was covered.
How to fix it: Add start, end, before-start and empty-input fixtures.
Prevent it next time: Derive fixtures from the business invariant.

Check your understanding

1. What does the local select_interval test establish about Airflow scheduling?

Show worked solution

Nothing directly. It establishes selection behavior for supplied Python records and declared integer boundaries. Scheduling, context availability and task execution remain separate claims. Its value is isolating a business invariant so a later integration failure can be diagnosed without confusing row-selection logic with orchestration behavior.

2. Why should a test using an external connection be reviewed even if it only reads data?