GitHub user christianalberto added a comment to the discussion: What is the
recommended approach for testing custom operators outside an Airflow code base?
For unit and integration testing in a standalone package or custom internal
provider (where you maintain custom operators, hooks, and sensors without
production DAGs), **the recommended approach remains using the `dag_maker`
fixture (or constructing an in-memory DAG) and calling
`operator.execute(context)` directly.**
You do **not** need to migrate to `dag.test()` or maintain actual DAG files
inside configured DAG bundles to test your custom operator logic.
---
### Why `dag_maker` over `dag.test()`?
* **`dag_maker` (Pytest Fixture):** Designed specifically to isolate
task/operator logic. It constructs a `DAG` and `TaskInstance` in memory within
the test execution context without requiring file system parsing or DAG bundle
discovery.
* **`dag.test()`:** Aimed primarily at end-to-end integration testing of **full
DAG workflows**, validating DAG parsing, DAG Bundle resolution, and task
dependency execution flow. For testing standalone library operators, it
introduces unnecessary overhead.
* **Airflow Core Practice:** Official Airflow providers continue to use
`dag_maker` extensively in their test suites to test hooks and operators in
isolation.
---
### Implementation Examples (`pytest`)
In your custom provider repository, structure your unit and integration tests
by executing `.execute()` directly:
#### 1. Unit Test (Mocked Connections / External Services)
```python
import pytest
from airflow.utils.state import TaskInstanceState
from my_custom_provider.operators.my_operator import MyCustomOperator
def test_my_custom_operator_unit(dag_maker):
# 1. Create in-memory DAG & Task
with dag_maker(dag_id="test_dag"):
task = MyCustomOperator(
task_id="test_task",
my_param="hello"
)
# 2. Instantiate TaskInstance & Context
ti = dag_maker.create_dagrun().get_task_instance(task.task_id)
context = ti.get_template_context()
# 3. Execute operator logic directly
result = task.execute(context)
# 4. Assertions
assert result == "EXPECTED_OUTPUT"
```
2. Integration Test (Real Connections / Localstack / DB)
When testing real interactions against external databases or local service
containers:
```python
import pytest
from airflow.models import Connection
from airflow.utils.session import create_session
from my_custom_provider.operators.my_operator import MyCustomOperator
@pytest.fixture
def setup_airflow_connection():
"""Seeds a temporary test connection into Airflow's metadata DB."""
conn = Connection(
conn_id="my_default_conn",
conn_type="http",
host="localhost",
port=8080,
)
with create_session() as session:
session.add(conn)
session.commit()
yield
# Cleanup connection after test
with create_session() as session:
session.query(Connection).filter(Connection.conn_id ==
"my_default_conn").delete()
def test_my_custom_operator_integration(dag_maker, setup_airflow_connection):
with dag_maker(dag_id="test_integration_dag"):
task = MyCustomOperator(
task_id="test_integration_task",
connection_id="my_default_conn"
)
ti = dag_maker.create_dagrun().get_task_instance(task.task_id)
context = ti.get_template_context()
# Execute directly against target service
result = task.execute(context)
assert result["status"] == 200
```
**Summary Recommendation**
1. Stick with dag_maker to create the execution context and task instances in
memory.
2. Invoke operator.execute(context) directly instead of triggering a full
Airflow DAG runner.
3. Keep dag.test() for end-to-end DAG execution tests in production pipelines,
rather than unit testing provider libraries.
GitHub link:
https://github.com/apache/airflow/discussions/72120#discussioncomment-18177955
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
