From e9d3da87512abae3c032eee7f45c7a9e63e6d5b9 Mon Sep 17 00:00:00 2001 From: Y-C Date: Sun, 6 Sep 2026 00:18:47 +0800 Subject: [PATCH] [v3-3-test] Fix DAG.cli() crashing on dags pause and unpause (#72109) * Fix DAG.cli() crashing on dags pause and unpause A Dag file run as a script goes through DAG.cli(), whose parser drops --dag-id and instead dispatches every subcommand with the Dag object as a second positional argument. The pause and unpause handlers never accepted it, so Dag authors got a TypeError instead of the command. The handlers are only reachable this way from a Dag file, which is why the regular `airflow dags pause` path has always worked and no test covered the difference. * Ignore --treat-dag-id-as-regex when DAG.cli() supplies the Dag The Dag-scoped parser drops --dag-id but keeps --treat-dag-id-as-regex, so the flag survives into a context where the user has no pattern left to supply. The Dag's own id was then read back as a pattern, and dag_ids may contain dots, so an unanchored match could pause unrelated Dags. Normalising the flag next to the dag_id it guards covers both of its readers, including the confirmation prompt, rather than guarding each reader in turn and leaving the next one to be found later. * Update airflow-core/tests/unit/cli/commands/test_dag_command.py --------- (cherry picked from commit bcc430f58359dd68f7dfa2e4fa1d547ea6a2e04e) Co-authored-by: Y-C Co-authored-by: Eason09053360 <185830721+Eason09053360@users.noreply.github.com> Co-authored-by: Henry Chen --- .../src/airflow/cli/commands/dag_command.py | 16 +++++++---- .../unit/cli/commands/test_dag_command.py | 28 +++++++++++++++++++ 2 files changed, 38 insertions(+), 6 deletions(-) diff --git a/airflow-core/src/airflow/cli/commands/dag_command.py b/airflow-core/src/airflow/cli/commands/dag_command.py index 7cfbe294b1971..c31b989d1e627 100644 --- a/airflow-core/src/airflow/cli/commands/dag_command.py +++ b/airflow-core/src/airflow/cli/commands/dag_command.py @@ -243,23 +243,27 @@ def _bulk_clear_runs( @cli_utils.action_cli @deprecated_for_airflowctl("airflowctl dags pause") @providers_configuration_loaded -def dag_pause(args) -> None: +def dag_pause(args, dag: DAG | None = None) -> None: """Pauses a DAG.""" - set_is_paused(True, args) + set_is_paused(True, args, dag) @cli_utils.action_cli @deprecated_for_airflowctl("airflowctl dags unpause") @providers_configuration_loaded -def dag_unpause(args) -> None: +def dag_unpause(args, dag: DAG | None = None) -> None: """Unpauses a DAG.""" - set_is_paused(False, args) + set_is_paused(False, args, dag) @providers_configuration_loaded @provide_session -def set_is_paused(is_paused: bool, args, *, session: Session = NEW_SESSION) -> None: +def set_is_paused(is_paused: bool, args, dag: DAG | None = None, *, session: Session = NEW_SESSION) -> None: """Set is_paused for DAG by a given dag_id.""" + if dag: + # A Dag object fully determines the target, so pattern matching has nothing left to match on. + args.dag_id = dag.dag_id + args.treat_dag_id_as_regex = False query = select(DagModel) if args.treat_dag_id_as_regex: query = query.where(DagModel.dag_id.regexp_match(args.dag_id)) @@ -274,7 +278,7 @@ def set_is_paused(is_paused: bool, args, *, session: Session = NEW_SESSION) -> N return if not args.yes and args.treat_dag_id_as_regex: - dags_ids = [dag.dag_id for dag in matched_dags] + dags_ids = [dag_model.dag_id for dag_model in matched_dags] question = ( f"You are about to {'un' if not is_paused else ''}pause {len(dags_ids)} DAGs:\n" f"{','.join(dags_ids)}" 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 e4d2886faf0eb..db7bf2d14adf7 100644 --- a/airflow-core/tests/unit/cli/commands/test_dag_command.py +++ b/airflow-core/tests/unit/cli/commands/test_dag_command.py @@ -564,6 +564,34 @@ def test_pause(self): dag_command.dag_unpause(args) assert not DagModel.get_dagmodel("example_bash_operator").is_paused + def test_pause_unpause_from_dag_cli(self): + """``DAG.cli()`` passes the Dag positionally and its parser drops ``--dag-id``.""" + parser = cli_parser.get_parser(dag_parser=True) + dag = DAG("example_bash_operator") + + dag_command.dag_pause(parser.parse_args(["dags", "pause"]), dag) + assert DagModel.get_dagmodel("example_bash_operator").is_paused + + dag_command.dag_unpause(parser.parse_args(["dags", "unpause"]), dag) + assert not DagModel.get_dagmodel("example_bash_operator").is_paused + + @mock.patch("airflow.cli.commands.dag_command.ask_yesno") + def test_pause_from_dag_cli_ignores_treat_dag_id_as_regex(self, mock_yesno): + """The Dag fixes the target, so its dag_id must not be read back as a pattern.""" + target = DAG("dag.cli_regex_target") + sync_dag_to_db(target) + sync_dag_to_db(DAG("dagXcli_regex_target")) + parser = cli_parser.get_parser(dag_parser=True) + + dag_command.dag_pause(parser.parse_args(["dags", "pause", "--treat-dag-id-as-regex"]), target) + + mock_yesno.assert_not_called() + assert DagModel.get_dagmodel("dag.cli_regex_target").is_paused + assert not DagModel.get_dagmodel("dagXcli_regex_target").is_paused + + clear_db_dags() + self.setup_class() + @mock.patch("airflow.cli.commands.dag_command.ask_yesno") def test_pause_regex(self, mock_yesno): args = self.parser.parse_args(["dags", "pause", "^example_.*$", "--treat-dag-id-as-regex"])