[v3-3-test] Fix DAG.cli() crashing on dags pause and unpause (#72109)

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)}"

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"])