This is an automated email from the ASF dual-hosted git repository.
kaxilnaik 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 3fb3fb36ad5 Make ``dag.test`` consistent with ``airflow dags test``
CLI command (#51476)
3fb3fb36ad5 is described below
commit 3fb3fb36ad52df7b33e738175806758bc0d6059f
Author: Kaxil Naik <[email protected]>
AuthorDate: Fri Jun 6 18:54:13 2025 +0530
Make ``dag.test`` consistent with ``airflow dags test`` CLI command (#51476)
---
task-sdk/src/airflow/sdk/definitions/dag.py | 8 ++++++--
1 file changed, 6 insertions(+), 2 deletions(-)
diff --git a/task-sdk/src/airflow/sdk/definitions/dag.py
b/task-sdk/src/airflow/sdk/definitions/dag.py
index 449e101f7f5..87215f5531a 100644
--- a/task-sdk/src/airflow/sdk/definitions/dag.py
+++ b/task-sdk/src/airflow/sdk/definitions/dag.py
@@ -54,7 +54,7 @@ from airflow.models.deadline import DeadlineAlert
from airflow.sdk.bases.operator import BaseOperator
from airflow.sdk.definitions._internal.abstractoperator import AbstractOperator
from airflow.sdk.definitions._internal.node import validate_key
-from airflow.sdk.definitions._internal.types import NOTSET
+from airflow.sdk.definitions._internal.types import NOTSET, ArgNotSet
from airflow.sdk.definitions.asset import AssetAll, BaseAsset
from airflow.sdk.definitions.context import Context
from airflow.sdk.definitions.param import DagParam, ParamsDict
@@ -1014,7 +1014,7 @@ class DAG:
def test(
self,
run_after: datetime | None = None,
- logical_date: datetime | None = None,
+ logical_date: datetime | None | ArgNotSet = NOTSET,
run_conf: dict[str, Any] | None = None,
conn_file_path: str | None = None,
variable_file_path: str | None = None,
@@ -1082,6 +1082,10 @@ class DAG:
with exit_stack:
self.validate()
+
+ # Allow users to explicitly pass None. If it isn't set, we default
to current time.
+ logical_date = logical_date if not isinstance(logical_date,
ArgNotSet) else timezone.utcnow()
+
log.debug("Clearing existing task instances for logical date %s",
logical_date)
# TODO: Replace with calling client.dag_run.clear in Execution API
at some point
SchedulerDAG.clear_dags(
