[AIRFLOW-5560] Allow no confirmation on reset dags (#6197)

- update backfill command to take --yes
- update clear command to use --yes rather than no_confirm

airflow/bin/cli.py

@@ -188,7 +188,7 @@ def backfill(args, dag=None):
                 [dag],
                 start_date=args.start_date,
                 end_date=args.end_date,
-                confirm_prompt=True,
+                confirm_prompt=not args.yes,
                 include_subdags=True,
             )
 
@@ -773,7 +773,7 @@ def clear(args):
         end_date=args.end_date,
         only_failed=args.only_failed,
         only_running=args.only_running,
-        confirm_prompt=not args.no_confirm,
+        confirm_prompt=not args.yes,
         include_subdags=not args.exclude_subdags,
         include_parentdag=not args.exclude_parentdag,
     )
@@ -1799,9 +1799,6 @@ class CLIFactory:
             ("-r", "--only_running"), "Only running jobs", "store_true"),
         'downstream': Arg(
             ("-d", "--downstream"), "Include downstream tasks", "store_true"),
-        'no_confirm': Arg(
-            ("-c", "--no_confirm"),
-            "Do not request confirmation", "store_true"),
         'exclude_subdags': Arg(
             ("-x", "--exclude_subdags"),
             "Exclude subdags", "store_true"),
@@ -2244,7 +2241,7 @@ class CLIFactory:
                             " within the backfill date range.",
                     'args': (
                         'dag_id', 'task_regex', 'start_date', 'end_date',
-                        'mark_success', 'local', 'donot_pickle',
+                        'mark_success', 'local', 'donot_pickle', 'yes',
                         'bf_ignore_dependencies', 'bf_ignore_first_depends_on_past',
                         'subdir', 'pool', 'delay_on_limit', 'dry_run', 'verbose', 'conf',
                         'reset_dag_run', 'rerun_failed_tasks', 'run_backwards'
@@ -2267,7 +2264,7 @@ class CLIFactory:
                     'help': "Clear a set of task instance, as if they never ran",
                     'args': (
                         'dag_id', 'task_regex', 'start_date', 'end_date', 'subdir',
-                        'upstream', 'downstream', 'no_confirm', 'only_failed',
+                        'upstream', 'downstream', 'yes', 'only_failed',
                         'only_running', 'exclude_subdags', 'exclude_parentdag', 'dag_regex'),
                 },
                 {

tests/core.py

@@ -1623,31 +1623,31 @@ class TestCli(unittest.TestCase):
 
     def test_subdag_clear(self):
         args = self.parser.parse_args([
-            'tasks', 'clear', 'example_subdag_operator', '--no_confirm'])
+            'tasks', 'clear', 'example_subdag_operator', '--yes'])
         cli.clear(args)
         args = self.parser.parse_args([
-            'tasks', 'clear', 'example_subdag_operator', '--no_confirm', '--exclude_subdags'])
+            'tasks', 'clear', 'example_subdag_operator', '--yes', '--exclude_subdags'])
         cli.clear(args)
 
     def test_parentdag_downstream_clear(self):
         args = self.parser.parse_args([
-            'tasks', 'clear', 'example_subdag_operator.section-1', '--no_confirm'])
+            'tasks', 'clear', 'example_subdag_operator.section-1', '--yes'])
         cli.clear(args)
         args = self.parser.parse_args([
-            'tasks', 'clear', 'example_subdag_operator.section-1', '--no_confirm',
+            'tasks', 'clear', 'example_subdag_operator.section-1', '--yes',
             '--exclude_parentdag'])
         cli.clear(args)
 
     def test_get_dags(self):
         dags = cli.get_dags(self.parser.parse_args(['tasks', 'clear', 'example_subdag_operator',
-                                                    '-c']))
+                                                    '--yes']))
         self.assertEqual(len(dags), 1)
 
-        dags = cli.get_dags(self.parser.parse_args(['tasks', 'clear', 'subdag', '-dx', '-c']))
+        dags = cli.get_dags(self.parser.parse_args(['tasks', 'clear', 'subdag', '-dx', '--yes']))
         self.assertGreater(len(dags), 1)
 
         with self.assertRaises(AirflowException):
-            cli.get_dags(self.parser.parse_args(['tasks', 'clear', 'foobar', '-dx', '-c']))
+            cli.get_dags(self.parser.parse_args(['tasks', 'clear', 'foobar', '-dx', '--yes']))
 
     def test_process_subdir_path_with_placeholder(self):
         self.assertEqual(os.path.join(settings.DAGS_FOLDER, 'abc'), cli.process_subdir('DAGS_FOLDER/abc'))