This is an automated email from the ASF dual-hosted git repository.
henry3260 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 63fc32fd14b Fix DAG.cli() crashing on dags test --show-dagrun (#72810)
63fc32fd14b is described below
commit 63fc32fd14bbbb20abfa60301abcd2f33d49726f
Author: Y-C <[email protected]>
AuthorDate: Wed Sep 16 02:30:44 2026 +0800
Fix DAG.cli() crashing on dags test --show-dagrun (#72810)
* Fix DAG.cli() crashing on dags test --show-dagrun
A Dag file run as a script goes through DAG.cli(), whose parser drops the
dag_id positional and passes the Dag object to the handler instead. dag_test
never needed dag_id from the parsed arguments until it rendered the run: the
task instance query read args.dag_id, so --show-dagrun, --save-dagrun and
--imgcat-dagrun raised AttributeError after the Dag had already run, while
the plain command and the regular airflow dags test path worked.
Filtering on the resolved Dag's id is the same value on the regular path and
the only one available from DAG.cli().
* Update airflow-core/tests/unit/cli/commands/test_dag_command.py
Co-authored-by: rjgoyln <[email protected]>
* Update airflow-core/tests/unit/cli/commands/test_dag_command.py
---------
Co-authored-by: Eason09053360
<[email protected]>
Co-authored-by: rjgoyln <[email protected]>
Co-authored-by: Henry Chen <[email protected]>
---
.../src/airflow/cli/commands/dag_command.py | 2 +-
.../tests/unit/cli/commands/test_dag_command.py | 21 ++++++++++++++++++++-
2 files changed, 21 insertions(+), 2 deletions(-)
diff --git a/airflow-core/src/airflow/cli/commands/dag_command.py
b/airflow-core/src/airflow/cli/commands/dag_command.py
index fe83a42fd3e..7ff7e1e5e15 100644
--- a/airflow-core/src/airflow/cli/commands/dag_command.py
+++ b/airflow-core/src/airflow/cli/commands/dag_command.py
@@ -851,7 +851,7 @@ def dag_test(args, dag: DAG | None = None, *, session:
Session = NEW_SESSION) ->
if show_dagrun or imgcat or filename:
tis = session.scalars(
select(TaskInstance).where(
- TaskInstance.dag_id == args.dag_id,
+ TaskInstance.dag_id == dag.dag_id,
TaskInstance.run_id == dr.run_id,
)
).all()
diff --git a/airflow-core/tests/unit/cli/commands/test_dag_command.py
b/airflow-core/tests/unit/cli/commands/test_dag_command.py
index 6223cf43c53..592ee1a50cc 100644
--- a/airflow-core/tests/unit/cli/commands/test_dag_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_dag_command.py
@@ -961,9 +961,10 @@ class TestCliDags:
)
@mock.patch("airflow.cli.commands.dag_command.render_dag",
return_value=MagicMock(source="SOURCE"))
- @mock.patch("airflow.cli.commands.dag_command.get_bagged_dag")
+ @mock.patch("airflow.cli.commands.dag_command.get_bagged_dag",
autospec=True)
def test_dag_test_show_dag(self, mock_get_dag, mock_render_dag,
stdout_capture):
mock_get_dag.return_value.test.return_value.run_id =
"__test_dag_test_show_dag_fake_dag_run_run_id__"
+ mock_get_dag.return_value.dag_id = "example_bash_operator"
cli_args = self.parser.parse_args(
["dags", "test", "example_bash_operator",
DEFAULT_DATE.isoformat(), "--show-dagrun"]
@@ -988,6 +989,24 @@ class TestCliDags:
mock_render_dag.assert_has_calls([mock.call(mock_get_dag.return_value,
tis=[])])
assert "SOURCE" in output
+ @mock.patch("airflow.cli.commands.dag_command.render_dag", autospec=True)
+ @mock.patch.object(DAG, "test", autospec=True)
+ def test_dag_test_show_dag_from_dag_cli(self, mock_test, mock_render_dag,
dag_maker, stdout_capture):
+ """``DAG.cli()`` passes the Dag positionally and its parser drops
``dag_id``."""
+ with dag_maker("dag_cli_show_dagrun", schedule=None) as dag:
+ EmptyOperator(task_id="only_task")
+ mock_test.return_value = dag_maker.create_dagrun(run_id="dag_cli_run")
+
+ parser = cli_parser.get_parser(dag_parser=True)
+ with stdout_capture:
+ dag_command.dag_test(parser.parse_args(["dags", "test",
"--show-dagrun"]), dag)
+
+ mock_render_dag.assert_called_once()
+ assert mock_render_dag.call_args.args[0] is dag
+ assert [(ti.dag_id, ti.task_id, ti.run_id) for ti in
mock_render_dag.call_args.kwargs["tis"]] == [
+ ("dag_cli_show_dagrun", "only_task", "dag_cli_run")
+ ]
+
@mock.patch("airflow.dag_processing.dagbag.BundleDagBag")
def test_dag_test_with_bundle_name(self, mock_dagbag,
configure_dag_bundles):
"""Test that DAG can be tested using bundle name."""
