What is the recommended approach for testing custom operators outside an Airflow code base?

[tuzonghua]

If I'm building and distributing an internal Airflow provider, for example, a repository that only contains custom operators, hooks, etc., but no DAGs, what is the best way to write integration tests for that code? Official Airflow providers often use the dag_maker fixture as a convenient way to write unit tests (example) that create an in-memory DAG. However, the official documentation suggests dag.test(), which requires a separate DAG file to exist in a configured DAG bundle. If the latter is the answer, will official Airflow providers be migrating to that approach as well?

[Labeeb2339]

For an internal provider repository with no production DAGs, I would keep two layers separate:

• Use Airflow's dag_maker pytest fixture for operator/hook integration tests. It creates an in-memory DAG and gives the test the Airflow context needed to create a DagRun/TaskInstance and exercise the operator, so a throwaway DAG file or configured DAG bundle is not required.
• Use dag.test() when the thing being tested is an actual DAG file as a whole: task dependencies, templating, and the end-to-end execution flow. It is not a replacement for unit-testing a reusable operator class.

A provider test can therefore follow the same shape as the official provider tests: construct the DAG inside the test with dag_maker, instantiate the custom operator under that DAG, create/run the task instance using the test fixture, and mock only the external service boundary. Put a small real DAG under a test/example-dags area only when you specifically want to validate DAG parsing or dag.test() behavior.

For tests that must exercise a real executor, scheduler, or external service, use the provider's system/integration-test setup separately; dag.test() defaults to a local simulated run unless use_executor=True is supplied. The current docs describe that distinction here: https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/debug.html#testing-dags-with-dag-test

[christianalberto]

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)

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:

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.