Testing custom operators in Apache Airflow - Interactive Code Practice

Practice - 5 Tasks

1fill in blank

easy

Complete the code to import the base class for custom Airflow operators.

Apache Airflow

from airflow.operators.[1] import BaseOperator

Drag options to blanks, or click blank then click option'

Abash

Bbase

Cpython

Ddummy

Attempts:

3 left

💡 Hint

Common Mistakes

Importing BaseOperator from bash or python operators instead of base.

Trying to import BaseOperator directly from airflow.operators.

The BaseOperator class is imported from airflow.operators.base. This is the base class to create custom operators.

2fill in blank

medium

Complete the code to define the execute method in a custom operator.

Apache Airflow

def execute(self, context):
    self.[1]('Running custom operator logic')

Drag options to blanks, or click blank then click option'

Alog.info

Breturn

Cprint

Draise

Attempts:

3 left

💡 Hint

Common Mistakes

Using print which outputs to stdout instead of structured task logs.

Using log.info without self..

Using return instead of logging.

The execute method runs the operator's logic. Using self.log.info properly logs a message to Airflow's task logger during execution.

3fill in blank

hard

Fix the error in the custom operator's init method by completing the super call.

Apache Airflow

def __init__(self, param1, **kwargs):
    super().[1](param1=param1, **kwargs)

Drag options to blanks, or click blank then click option'

Arun

Bexecute

Cstart

D__init__

Attempts:

3 left

💡 Hint

Common Mistakes

Calling super().execute instead of __init__.

Forgetting to call the parent class's init method.

The __init__ method of the parent class must be called to properly initialize the operator.

4fill in blank

hard

Complete the code to create a test case that checks if the custom operator runs without errors.

Apache Airflow

def test_custom_operator():
    operator = CustomOperator(task_id='test_task')
    result = operator.[1]({})
    assert result is None

Drag options to blanks, or click blank then click option'

Astart

Brun

Cexecute

Dinit

Attempts:

3 left

💡 Hint

Common Mistakes

Calling __init__ or run instead of execute.

Not passing the context dictionary to the execute method.

To test a custom operator, instantiate it and call its execute method with an empty context dictionary. For operators that don't return a value, the result is None.

5fill in blank

hard

Fill all three blanks to complete a test that mocks the logging inside the custom operator.

Apache Airflow

import unittest.mock as mock

def test_logging_in_operator():
    operator = CustomOperator(task_id='log_test')
    with mock.patch.object(operator, 'log') as mock_log:
        operator.[1]({})
        mock_log.[2].assert_called_with('[3]')

Drag options to blanks, or click blank then click option'

Aexecute

Binfo

CRunning custom operator logic

Ddebug

Attempts:

3 left

💡 Hint

Common Mistakes

Using debug instead of info for logging.

Not mocking the correct log method or message.

Wrong patch target for the operator's log.

The test mocks the operator's log instance to check if self.log.info is called with the expected message during execution.

Practice

1. What is the main reason to write tests for custom Airflow operators?

easy

A. To ensure the operator works as expected before running in production

B. To make the DAG run faster

C. To reduce the number of tasks in the DAG

D. To automatically generate documentation

Solution

1. Step 1: Understand the purpose of testing

   Testing is done to verify that code behaves correctly and avoids errors in production.

2. Step 2: Apply this to custom operators

   Custom operators are pieces of code that need validation to ensure they perform their tasks as intended.

3. Final Answer:

   To ensure the operator works as expected before running in production -> Option A

4. Quick Check:

   Testing = Verify correctness [OK]

Hint: Testing checks correctness before production use [OK]

Common Mistakes:

• Thinking tests speed up DAG execution
• Confusing testing with documentation generation
• Believing tests reduce task count

2. Which Python module is commonly used to write tests for Airflow custom operators?

easy

A. requests

B. flask

C. unittest

D. pandas

Solution

1. Step 1: Identify common testing modules in Python

   Python's built-in module for testing is unittest, widely used for writing test cases.

2. Step 2: Match with Airflow testing practices

   Airflow tests often use unittest to create test classes and methods for operators.

3. Final Answer:

   unittest -> Option C

4. Quick Check:

   Python testing = unittest [OK]

Hint: Use unittest for Python testing by default [OK]

Common Mistakes:

• Choosing unrelated modules like flask or pandas
• Confusing HTTP libraries with testing tools

3. Given this test snippet for a custom operator:

from airflow.models import DagBag

def test_operator_load():
    dagbag = DagBag()
    assert 'my_custom_dag' in dagbag.dags

What does this test check?

medium

A. If the DAG named 'my_custom_dag' is loaded successfully

B. If the operator runs without errors

C. If the DAG tasks execute in order

D. If the DAG has no syntax errors

Solution

1. Step 1: Understand DagBag usage

   DagBag loads all DAGs from the Airflow folder and stores them in dagbag.dags dictionary.

2. Step 2: Analyze the assertion

   The test asserts that 'my_custom_dag' is a key in dagbag.dags, meaning it was loaded.

3. Final Answer:

   If the DAG named 'my_custom_dag' is loaded successfully -> Option A

4. Quick Check:

   DagBag loads DAGs = check DAG presence [OK]

Hint: DagBag.dags keys are loaded DAG ids [OK]

Common Mistakes:

• Thinking it tests operator execution
• Confusing DAG loading with task order
• Assuming it checks syntax errors directly

4. You wrote a test for a custom operator but it fails with AttributeError: 'NoneType' object has no attribute 'execute'. What is the likely cause?

medium

A. The DAG file has syntax errors

B. The operator instance was not created properly in the test

C. The Airflow scheduler is not running

D. The test file is missing import statements

Solution

1. Step 1: Understand the error meaning

   AttributeError on 'NoneType' means the variable expected to be an object is actually None.

2. Step 2: Relate to operator instance creation

   If the operator instance was not created or assigned, calling execute() on None causes this error.

3. Final Answer:

   The operator instance was not created properly in the test -> Option B

4. Quick Check:

   None.execute error = missing instance [OK]

Hint: Check if operator object is None before calling methods [OK]

Common Mistakes:

• Blaming DAG syntax errors for this runtime error
• Thinking scheduler status affects unit tests
• Ignoring missing or wrong object creation

5. You want to test a custom operator that makes an HTTP request. Which approach is best to avoid real network calls during tests?

hard

A. Write the test to call the real HTTP endpoint every time

B. Run the test only when connected to the internet

C. Skip the test if the HTTP server is down

D. Use mocking to replace the HTTP call with a fake response

Solution

1. Step 1: Understand testing external calls

   Real network calls can be slow, flaky, or cause side effects, so they should be avoided in unit tests.

2. Step 2: Use mocking for external dependencies

   Mocking replaces the real HTTP call with a controlled fake response, making tests fast and reliable.

3. Final Answer:

   Use mocking to replace the HTTP call with a fake response -> Option D

4. Quick Check:

   Mock external calls = reliable tests [OK]

Hint: Mock external calls to avoid real network in tests [OK]

Common Mistakes:

• Running tests dependent on internet availability
• Skipping tests instead of fixing design
• Calling real endpoints causing flaky tests