This is an automated email from the ASF dual-hosted git repository.
amoghrajesh pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 297b0c7abc1 Update the docs recommending dag.test() for custom
operator tests (#71542)
297b0c7abc1 is described below
commit 297b0c7abc16a6c11570944559e1b427ffdff987
Author: Amogh Desai <[email protected]>
AuthorDate: Fri Aug 14 11:22:27 2026 +0530
Update the docs recommending dag.test() for custom operator tests (#71542)
---
airflow-core/docs/best-practices.rst | 69 ++++++++++++++++++++++-------
airflow-core/docs/core-concepts/debug.rst | 13 ++++++
airflow-core/docs/howto/custom-operator.rst | 23 ++++++++++
airflow-core/docs/tutorial/fundamentals.rst | 6 ++-
4 files changed, 92 insertions(+), 19 deletions(-)
diff --git a/airflow-core/docs/best-practices.rst
b/airflow-core/docs/best-practices.rst
index 52011031cd9..cc4c31bc2b6 100644
--- a/airflow-core/docs/best-practices.rst
+++ b/airflow-core/docs/best-practices.rst
@@ -733,6 +733,8 @@ the example_python_operator.py above so the actual parsing
time is about ~ 0.62
You can look into :ref:`Testing a Dag <testing>` for details on how to test
individual operators.
+.. _best_practices:unit_tests:
+
Unit tests
-----------
@@ -786,29 +788,62 @@ This is an example test want to verify the structure of a
code-generated Dag aga
**Unit test for custom operator:**
+To unit test an operator or a sensor, call it directly. You do not need a Dag
run, a Dag, or a
+metadata database — instantiate the operator and call ``execute()`` with the
context keys your
+code actually reads:
+
+.. code-block:: python
+
+ def test_my_custom_operator_execute():
+ op = MyCustomOperator(task_id="my_custom_operator_task",
prefix="s3://bucket/some/prefix")
+
+ assert op.execute(context={}) == "expected return value"
+
+For a sensor, call ``poke()`` and assert on the returned boolean:
+
+.. code-block:: python
+
+ def test_my_custom_sensor_poke():
+ sensor = MyCustomSensor(task_id="my_custom_sensor_task",
key="some-key")
+
+ assert sensor.poke(context={}) is True
+
+If your operator renders templated fields, render them before asserting:
+
+.. code-block:: python
+
+ op.render_template_fields(context={"ds": "2021-09-13"})
+ assert op.prefix == "s3://bucket/2021-09-13"
+
+For a deferrable operator, assert that it defers with the trigger you expect,
then drive the
+resume path directly:
+
.. code-block:: python
- import pendulum
+ from airflow.sdk.exceptions import TaskDeferred
+
+
+ def test_my_custom_operator_defers():
+ op = MyCustomOperator(task_id="my_custom_operator_task",
deferrable=True)
- from airflow.sdk import DAG, TaskInstanceState
+ with pytest.raises(TaskDeferred) as exc:
+ op.execute(context={})
+ assert isinstance(exc.value.trigger, MyCustomTrigger)
+ # Drive the method the trigger resumes into.
+ getattr(op, exc.value.method_name)(context={}, event={"status":
"success"})
+
+.. note::
- def test_my_custom_operator_execute_no_trigger(dag):
- TEST_TASK_ID = "my_custom_operator_task"
- with DAG(
- dag_id="my_custom_operator_dag",
- schedule="@daily",
- start_date=pendulum.datetime(2021, 9, 13, tz="UTC"),
- ) as dag:
- MyCustomOperator(
- task_id=TEST_TASK_ID,
- prefix="s3://bucket/some/prefix",
- )
+ ``TaskInstance.run()`` and ``TaskInstance.render_templates()`` were
removed in Airflow 3.2 —
+ ``TaskInstance`` has been an internal class since Airflow 3.0. Replace
``ti.run()`` with
+ ``op.execute(context)`` and ``ti.render_templates(context)`` with
+ ``op.render_template_fields(context)`` as shown above.
- dagrun = dag.test()
- ti = dagrun.get_task_instance(task_id=TEST_TASK_ID)
- assert ti.state == TaskInstanceState.SUCCESS
- # Assert something related to tasks results: ti.xcom_pull()
+To exercise a whole Dag run rather than a single operator, see
+:ref:`Testing Dags with dag.test() <concepts:debugging>`. That is an
integration test: it needs a
+metadata database and a Dag that Airflow can serialize, so it is not a
substitute for the unit
+tests above.
Self-Checks
diff --git a/airflow-core/docs/core-concepts/debug.rst
b/airflow-core/docs/core-concepts/debug.rst
index 0bed4ec3f2e..3ee12bea395 100644
--- a/airflow-core/docs/core-concepts/debug.rst
+++ b/airflow-core/docs/core-concepts/debug.rst
@@ -29,6 +29,19 @@ serialized python process.
This approach can be used with any supported database (including a local
SQLite database) and will
*fail fast* as all tasks run in a single process.
+``dag.test()`` executes a real Dag run, so it has two prerequisites:
+
+* an initialized metadata database (``airflow db migrate``), and
+* a Dag that Airflow can serialize, which means the Dag must be defined in a
file inside a
+ configured :doc:`Dag bundle </administration-and-deployment/dag-bundles>` —
by default, your
+ Dags folder. ``dag.test()`` re-parses the bundle and serializes the Dag for
you.
+
+A Dag built in memory — for instance inside a pytest function, in a file
outside your Dags folder —
+is not in any bundle, so nothing gets serialized and the call fails with
+``Cannot create DagRun for DAG <dag_id> because the dag is not serialized``.
To test operator
+behaviour without a Dag run at all, call the operator directly as described in
+:ref:`Unit tests <best_practices:unit_tests>`.
+
To set up ``dag.test``, add these two lines to the bottom of your Dag file:
.. code-block:: python
diff --git a/airflow-core/docs/howto/custom-operator.rst
b/airflow-core/docs/howto/custom-operator.rst
index b5e14128ca0..f6793d4e8d0 100644
--- a/airflow-core/docs/howto/custom-operator.rst
+++ b/airflow-core/docs/howto/custom-operator.rst
@@ -149,6 +149,8 @@ Override ``custom_operator_name`` to change the displayed
name to something othe
custom_operator_name = "Howdy"
# ...
+.. _custom-operator/template-fields:
+
Templating
----------
You can use :ref:`Jinja templates <concepts:jinja-templating>` to parameterize
your operator.
@@ -449,3 +451,24 @@ An example of a sensor that keeps internal state and
cannot be used with resched
is
:class:`airflow.providers.google.cloud.sensors.gcs.GCSUploadSessionCompleteSensor`.
It polls the number of objects at a prefix (this number is the internal state
of the sensor)
and succeeds when there a certain amount of time has passed without the number
of objects changing.
+
+Testing your operator
+---------------------
+
+Instantiate your operator and call ``execute()`` — or ``poke()`` for a sensor
— with the context
+keys your code reads. This needs no Dag, no Dag run and no metadata database:
+
+.. code-block:: python
+
+ def test_hello_operator():
+ op = HelloOperator(task_id="hello", name="Bob")
+
+ assert op.execute(context={}) == "Hello Bob"
+
+Use ``op.render_template_fields(context)`` if you need to assert on rendered
+:ref:`template fields <custom-operator/template-fields>`.
+
+Reach for :ref:`dag.test() <concepts:debugging>` only when you want to
exercise a whole Dag run,
+which is an integration test: it requires a metadata database and a Dag
defined in a file that
+Airflow can serialize. See :ref:`Unit tests <best_practices:unit_tests>` for
the full set of
+patterns, including deferrable operators.
diff --git a/airflow-core/docs/tutorial/fundamentals.rst
b/airflow-core/docs/tutorial/fundamentals.rst
index 749008e36be..14ecb5bf77c 100644
--- a/airflow-core/docs/tutorial/fundamentals.rst
+++ b/airflow-core/docs/tutorial/fundamentals.rst
@@ -293,8 +293,10 @@ This command will provide detailed logs and execute your
bash command.
Keep in mind that the ``airflow tasks test`` command runs task instances
locally, outputs their logs to stdout, and
doesn't track state in the database. This is a handy way to test individual
task instances.
-Similarly, ``airflow dags test`` runs a single Dag run without registering any
state in the database, which is useful
-for testing your entire Dag locally.
+Similarly, ``airflow dags test`` runs a single Dag run locally, which is
useful for testing your entire Dag. Unlike
+``airflow tasks test``, it creates a real Dag run and records task state in
the metadata database, so it needs an
+initialized database and a Dag that Airflow can serialize from your Dags
folder. See
+:ref:`Testing Dags with dag.test() <concepts:debugging>`.
What's Next?
-------------
