Unit testing DAGs in Apache Airflow - Practice Problems & Coding Challenges

Challenge - 5 Problems

💻 Command Output

intermediate

Output of a simple DAG test run

You have a DAG with one task that prints 'Hello Airflow'. You run a unit test that triggers this task. What will be the output of the test command?

Apache Airflow

def test_hello_task():
    ti = TaskInstance(task=hello_task, execution_date=datetime.now())
    ti.run()
    print(ti.xcom_pull())

A None

B TaskInstanceError

C SyntaxError

D Hello Airflow

💡 Hint

Think about what xcom_pull returns if no value is pushed.

The task prints 'Hello Airflow' but does not push any XCom value, so ti.xcom_pull() returns None.

🧠 Conceptual

intermediate

Purpose of mocking in DAG unit tests

Why do we use mocking when unit testing Airflow DAGs?

A To deploy DAGs automatically

B To simulate external systems and isolate the DAG logic

C To speed up the Airflow scheduler

D To generate DAG documentation

💡 Hint

Think about how to test code without relying on real external services.

Mocking replaces external dependencies with fake ones so tests focus only on DAG logic.

❓ Troubleshoot

advanced

Diagnosing a failing DAG unit test

A unit test for a DAG task fails with the error: 'TaskInstanceStateError: Task instance is in state None'. What is the most likely cause?

A The task was not properly initialized with an execution date

B The DAG file has syntax errors

C The Airflow webserver is down

D The task has no downstream dependencies

💡 Hint

Check how TaskInstance is created in tests.

TaskInstance requires a valid execution_date to track state; missing it causes this error.

✅ Best Practice

advanced

Best practice for testing task dependencies in DAGs

Which approach best verifies task dependencies in an Airflow DAG unit test?

A Manually inspect the DAG Python file

B Run all tasks and check their logs

C Use Airflow CLI to list all DAGs

D Check the DAG's task dictionary for upstream and downstream task IDs

💡 Hint

Think about programmatic ways to verify dependencies.

Accessing task attributes programmatically allows automated verification of dependencies.

🔀 Workflow

expert

Correct order of steps in unit testing an Airflow DAG

Arrange the following steps in the correct order for unit testing an Airflow DAG:

A 3,1,2,4

B 1,3,2,4

C 1,2,3,4

D 2,1,3,4

💡 Hint

Think about preparing the environment before running tests.

First import DAG/tasks, then mock dependencies, create TaskInstances, then run and assert.

Practice

1. What is the main purpose of unit testing DAGs in Airflow?

easy

A. To catch errors early before running workflows

B. To deploy DAGs to production automatically

C. To monitor task execution times

D. To schedule DAG runs at specific times

Solution

1. Step 1: Understand unit testing purpose

   Unit testing is done to find bugs or errors early in code before actual execution.

2. Step 2: Apply this to Airflow DAGs

   Unit testing DAGs helps catch errors in DAG definitions before running workflows in Airflow.

3. Final Answer:

   To catch errors early before running workflows -> Option A

4. Quick Check:

   Unit testing = catch errors early [OK]

Hint: Unit tests find bugs before running DAGs [OK]

Common Mistakes:

• Confusing unit testing with deployment
• Thinking unit tests schedule DAGs
• Mixing monitoring with testing

2. Which Python code snippet correctly loads DAGs for unit testing using Airflow's DagBag?

easy

A. dagbag = DagBag.load('/path/to/dags')

B. dagbag = DagBag(dag_folder='/path/to/dags')

C. dagbag = DagBag(folder='/path/to/dags')

D. dagbag = DagBag.get('/path/to/dags')

Solution

1. Step 1: Recall DagBag constructor usage

   DagBag takes a keyword argument dag_folder to specify DAGs folder path.

2. Step 2: Check each option

   Only dagbag = DagBag(dag_folder='/path/to/dags') uses correct syntax. Others use invalid methods (A,D) or incorrect keyword argument like folder (B).

3. Final Answer:

   dagbag = DagBag(dag_folder='/path/to/dags') -> Option B

4. Quick Check:

   DagBag(dag_folder=...) correct syntax [OK]

Hint: Use dag_folder keyword to load DAGs in DagBag [OK]

Common Mistakes:

• Using incorrect keyword argument (e.g., folder)
• Using non-existent methods like load() or get()
• Omitting dag_folder keyword

3. Given this test code snippet, what will be the output if the DAG named 'example_dag' has 3 tasks?

dagbag = DagBag(dag_folder='dags')
dag = dagbag.get_dag('example_dag')
print(len(dag.tasks))

medium

A. None

B. 0

C. 3

D. Error: DAG not found

Solution

1. Step 1: Load DAG using DagBag

   DagBag loads DAGs from 'dags' folder; get_dag('example_dag') returns the DAG object if found.

2. Step 2: Count tasks in the DAG

   len(dag.tasks) returns the number of tasks in the DAG, which is 3 as given.

3. Final Answer:

   3 -> Option C

4. Quick Check:

   len(dag.tasks) = 3 [OK]

Hint: len(dag.tasks) gives task count [OK]

Common Mistakes:

• Assuming get_dag returns None if DAG exists
• Confusing tasks with task instances
• Expecting error if DAG is present

4. You wrote this test code but it raises an AttributeError: 'NoneType' object has no attribute 'tasks'. What is the likely cause?

dagbag = DagBag(dag_folder='dags')
dag = dagbag.get_dag('missing_dag')
print(len(dag.tasks))

medium

A. The DAG 'missing_dag' does not exist in the folder

B. DagBag cannot load DAGs from 'dags' folder

C. len() cannot be used on dag.tasks

D. get_dag() returns a list, not a DAG object

Solution

1. Step 1: Analyze get_dag return value

   get_dag returns None if the DAG name is not found in the loaded DAGs.

2. Step 2: Understand AttributeError cause

   Trying to access dag.tasks when dag is None causes AttributeError.

3. Final Answer:

   The DAG 'missing_dag' does not exist in the folder -> Option A

4. Quick Check:

   get_dag returns None if DAG missing [OK]

Hint: Check if get_dag returns None before accessing tasks [OK]

Common Mistakes:

• Assuming get_dag always returns a DAG
• Thinking len() is invalid on tasks list
• Believing DagBag can't load DAGs from folder

5. You want to write a unit test to check that your DAG named 'data_pipeline' has exactly 5 tasks and no import errors. Which code snippet correctly performs this test?

hard

A. dagbag = DagBag(dag_folder='dags')
assert dagbag.import_errors == {}
dag = dagbag.get_dag('data_pipeline')
assert len(dag.tasks) == 0

B. dagbag = DagBag('dags')
assert dagbag.import_errors == None
dag = dagbag.get_dag('data_pipeline')
assert len(dag.tasks) == 5

C. dagbag = DagBag(dag_folder='dags')
assert dagbag.import_errors != {}
dag = dagbag.get_dag('data_pipeline')
assert len(dag.tasks) == 5

D. dagbag = DagBag(dag_folder='dags')
assert dagbag.import_errors == {}
dag = dagbag.get_dag('data_pipeline')
assert len(dag.tasks) == 5

Solution

1. Step 1: Check import errors correctly

   dagbag.import_errors is a dict; empty dict {} means no import errors.

2. Step 2: Verify task count assertion

   len(dag.tasks) == 5 checks that the DAG has exactly 5 tasks.

3. Final Answer:

   dagbag = DagBag(dag_folder='dags') assert dagbag.import_errors == {} dag = dagbag.get_dag('data_pipeline') assert len(dag.tasks) == 5 -> Option D

4. Quick Check:

   import_errors == {} and task count 5 [OK]

Hint: Assert import_errors == {} and task count matches [OK]

Common Mistakes:

• Checking import_errors against None instead of {}
• Asserting wrong task count
• Confusing import_errors presence logic