Partially enable PT012 rule (apache#38219)

* Partially enable PT012 rule

* Fixup 'tests/www/test_app.py' tests

* Fixup 'tests/decorators/test_bash.py::TestBashDecorator::test_cwd_is_file' tests

pyproject.toml

@@ -1360,7 +1360,6 @@ ignore = [
     "PT007", # Wrong type of values in @pytest.mark.parametrize
     "PT008", # Use return_value= instead of patching with lambda
     "PT011", # pytest.raises() is too broad, set the match parameter
-    "PT012", # [controversial rule] pytest.raises() block should contain a single simple statement.
     "PT018", # assertion should be broken down into multiple parts
     "PT019", # fixture without value is injected as parameter, use @pytest.mark.usefixtures instead
 ]
@@ -1405,7 +1404,7 @@ combine-as-imports = true
 "dev/breeze/tests/*" = ["TID253", "S101"]
 "tests/*" = ["D", "TID253", "S101"]
 "docker_tests/*" = ["D", "TID253", "S101"]
-"kubernetes_tests/*" = ["D", "TID253", "S101"]
+"kubernetes_tests/*" = ["D", "TID253", "S101", "PT012"]
 "helm_tests/*" = ["D", "TID253", "S101"]
 
 # All of the modules which have an extra license header (i.e. that we copy from another project) need to
@@ -1417,7 +1416,7 @@ combine-as-imports = true
 "tests/providers/elasticsearch/log/elasticmock/__init__.py" = ["E402"]
 "tests/providers/elasticsearch/log/elasticmock/utilities/__init__.py" = ["E402"]
 "tests/providers/openai/hooks/test_openai.py" = ["E402"]
-"tests/providers/openai/operators/test_openai.py" = ["E402"]
+"tests/providers/openai/operators/test_openai.py" = ["E402", "PT012"]
 "tests/providers/qdrant/hooks/test_qdrant.py" = ["E402"]
 "tests/providers/qdrant/operators/test_qdrant.py" = ["E402"]
 "tests/providers/snowflake/operators/test_snowflake_sql.py" = ["E402"]
@@ -1488,6 +1487,78 @@ combine-as-imports = true
 "airflow/providers/smtp/hooks/smtp.py" = ["D105"]
 "airflow/providers/tableau/hooks/tableau.py" = ["D105"]
 
+# All the test modules which do not follow PT012 yet
+"tests/providers/amazon/aws/hooks/test_base_aws.py" = ["PT012"]
+"tests/providers/amazon/aws/hooks/test_datasync.py" = ["PT012"]
+"tests/providers/amazon/aws/hooks/test_eks.py" = ["PT012"]
+"tests/providers/amazon/aws/hooks/test_redshift_data.py" = ["PT012"]
+"tests/providers/amazon/aws/hooks/test_s3.py" = ["PT012"]
+"tests/providers/amazon/aws/operators/test_emr_serverless.py" = ["PT012"]
+"tests/providers/amazon/aws/operators/test_redshift_data.py" = ["PT012"]
+"tests/providers/amazon/aws/sensors/test_glacier.py" = ["PT012"]
+"tests/providers/amazon/aws/sensors/test_glue.py" = ["PT012"]
+"tests/providers/amazon/aws/sensors/test_lambda_function.py" = ["PT012"]
+"tests/providers/amazon/aws/system/utils/test_helpers.py" = ["PT012"]
+"tests/providers/amazon/aws/transfers/test_redshift_to_s3.py" = ["PT012"]
+"tests/providers/amazon/aws/triggers/test_ecs.py" = ["PT012"]
+"tests/providers/amazon/aws/waiters/test_neptune.py" = ["PT012"]
+"tests/providers/apache/beam/hooks/test_beam.py" = ["PT012"]
+"tests/providers/apache/hive/hooks/test_hive.py" = ["PT012"]
+"tests/providers/apache/hive/sensors/test_named_hive_partition.py" = ["PT012"]
+"tests/providers/apache/spark/hooks/test_spark_sql.py" = ["PT012"]
+"tests/providers/celery/sensors/test_celery_queue.py" = ["PT012"]
+"tests/providers/cncf/kubernetes/executors/test_kubernetes_executor.py" = ["PT012"]
+"tests/providers/cncf/kubernetes/hooks/test_kubernetes.py" = ["PT012"]
+"tests/providers/cncf/kubernetes/operators/test_pod.py" = ["PT012"]
+"tests/providers/cncf/kubernetes/utils/test_k8s_resource_iterator.py" = ["PT012"]
+"tests/providers/cncf/kubernetes/utils/test_pod_manager.py" = ["PT012"]
+"tests/providers/common/sql/operators/test_sql.py" = ["PT012"]
+"tests/providers/databricks/hooks/test_databricks.py" = ["PT012"]
+"tests/providers/databricks/operators/test_databricks.py" = ["PT012"]
+"tests/providers/databricks/operators/test_databricks_repos.py" = ["PT012"]
+"tests/providers/databricks/sensors/test_databricks_partition.py" = ["PT012"]
+"tests/providers/datadog/sensors/test_datadog.py" = ["PT012"]
+"tests/providers/dbt/cloud/operators/test_dbt.py" = ["PT012"]
+"tests/providers/fab/auth_manager/cli_commands/test_user_command.py" = ["PT012"]
+"tests/providers/ftp/operators/test_ftp.py" = ["PT012"]
+"tests/providers/google/cloud/hooks/test_bigquery.py" = ["PT012"]
+"tests/providers/google/cloud/hooks/test_cloud_storage_transfer_service.py" = ["PT012"]
+"tests/providers/google/cloud/hooks/test_dataflow.py" = ["PT012"]
+"tests/providers/google/cloud/hooks/test_dataprep.py" = ["PT012"]
+"tests/providers/google/cloud/hooks/test_kubernetes_engine.py" = ["PT012"]
+"tests/providers/google/cloud/hooks/test_pubsub.py" = ["PT012"]
+"tests/providers/google/cloud/operators/test_bigtable.py" = ["PT012"]
+"tests/providers/google/cloud/operators/test_cloud_sql.py" = ["PT012"]
+"tests/providers/google/cloud/operators/test_cloud_storage_transfer_service.py" = ["PT012"]
+"tests/providers/google/cloud/operators/test_compute.py" = ["PT012"]
+"tests/providers/google/cloud/operators/test_datafusion.py" = ["PT012"]
+"tests/providers/google/cloud/operators/test_dataproc.py" = ["PT012"]
+"tests/providers/google/cloud/operators/test_functions.py" = ["PT012"]
+"tests/providers/google/cloud/operators/test_kubernetes_engine.py" = ["PT012"]
+"tests/providers/google/cloud/operators/test_mlengine.py" = ["PT012"]
+"tests/providers/google/cloud/operators/test_spanner.py" = ["PT012"]
+"tests/providers/google/cloud/sensors/test_datafusion.py" = ["PT012"]
+"tests/providers/google/cloud/sensors/test_dataproc.py" = ["PT012"]
+"tests/providers/google/cloud/sensors/test_gcs.py" = ["PT012"]
+"tests/providers/google/cloud/sensors/test_pubsub.py" = ["PT012"]
+"tests/providers/google/cloud/transfers/test_gcs_to_bigquery.py" = ["PT012"]
+"tests/providers/google/cloud/utils/test_credentials_provider.py" = ["PT012"]
+"tests/providers/google/common/hooks/test_base_google.py" = ["PT012"]
+"tests/providers/jenkins/sensors/test_jenkins.py" = ["PT012"]
+"tests/providers/microsoft/azure/hooks/test_data_factory.py" = ["PT012"]
+"tests/providers/microsoft/azure/hooks/test_wasb.py" = ["PT012"]
+"tests/providers/microsoft/psrp/hooks/test_psrp.py" = ["PT012"]
+"tests/providers/oracle/hooks/test_oracle.py" = ["PT012"]
+"tests/providers/papermill/operators/test_papermill.py" = ["PT012"]
+"tests/providers/sftp/hooks/test_sftp.py" = ["PT012"]
+"tests/providers/sftp/operators/test_sftp.py" = ["PT012"]
+"tests/providers/sftp/sensors/test_sftp.py" = ["PT012"]
+"tests/providers/sftp/triggers/test_sftp.py" = ["PT012"]
+"tests/providers/ssh/hooks/test_ssh.py" = ["PT012"]
+"tests/providers/ssh/operators/test_ssh.py" = ["PT012"]
+"tests/providers/telegram/hooks/test_telegram.py" = ["PT012"]
+"tests/providers/telegram/operators/test_telegram.py" = ["PT012"]
+
 [tool.ruff.lint.flake8-tidy-imports]
 # Ban certain modules from being imported at module level, instead requiring
 # that they're imported lazily (e.g., within a function definition).

tests/cli/commands/test_task_command.py

@@ -619,14 +619,13 @@ def test_task_states_for_dag_run_when_dag_run_not_exists(self):
         task_states_for_dag_run should return an AirflowException when invalid dag id is passed
         """
         with pytest.raises(DagRunNotFound):
-            default_date2 = timezone.datetime(2016, 1, 9)
             task_command.task_states_for_dag_run(
                 self.parser.parse_args(
                     [
                         "tasks",
                         "states-for-dag-run",
                         "not_exists_dag",
-                        default_date2.isoformat(),
+                        timezone.datetime(2016, 1, 9).isoformat(),
                         "--output",
                         "json",
                     ]

tests/cli/test_cli_parser.py

@@ -155,9 +155,9 @@ def test_dynamic_conflict_detection(self, cli_commands_mock: MagicMock):
                 args=[],
             )
         ]
+        reload(executor_loader)
         with pytest.raises(CliConflictError, match="test_command"):
             # force re-evaluation of cli commands (done in top level code)
-            reload(executor_loader)
             reload(cli_parser)
 
     def test_falsy_default_value(self):
@@ -229,6 +229,7 @@ def test_positive_int(self):
 
         with pytest.raises(argparse.ArgumentTypeError):
             cli_config.positive_int(allow_zero=False)("0")
+        with pytest.raises(argparse.ArgumentTypeError):
             cli_config.positive_int(allow_zero=True)("-1")
 
     @pytest.mark.parametrize(
@@ -285,8 +286,8 @@ def test_cli_parser_executors(self, executor, expected_args):
     def test_non_existing_directory_raises_when_metavar_is_dir_for_db_export_cleaned(self):
         """Test that the error message is correct when the directory does not exist."""
         with contextlib.redirect_stderr(StringIO()) as stderr:
+            parser = cli_parser.get_parser()
             with pytest.raises(SystemExit):
-                parser = cli_parser.get_parser()
                 parser.parse_args(["db", "export-archived", "--output-path", "/non/existing/directory"])
             error_msg = stderr.getvalue()
 
@@ -302,8 +303,8 @@ def test_invalid_choice_raises_for_export_format_in_db_export_archived_command(
     ):
         """Test that invalid choice raises for export-format in db export-cleaned command."""
         with contextlib.redirect_stderr(StringIO()) as stderr:
+            parser = cli_parser.get_parser()
             with pytest.raises(SystemExit):
-                parser = cli_parser.get_parser()
                 parser.parse_args(
                     ["db", "export-archived", "--export-format", export_format, "--output-path", "mydir"]
                 )

tests/core/test_stats.py

@@ -124,15 +124,14 @@ def test_load_invalid_custom_stats_client(self):
                 ("metrics", "statsd_on"): "True",
                 ("metrics", "statsd_custom_client_path"): f"{__name__}.InvalidCustomStatsd",
             }
-        ), pytest.raises(
-            AirflowConfigException,
-            match=re.escape(
-                "Your custom StatsD client must extend the statsd."
-                "StatsClient in order to ensure backwards compatibility."
-            ),
         ):
             importlib.reload(airflow.stats)
-            airflow.stats.Stats.incr("empty_key")
+            error_message = re.escape(
+                "Your custom StatsD client must extend the statsd."
+                "StatsClient in order to ensure backwards compatibility."
+            )
+            with pytest.raises(AirflowConfigException, match=error_message):
+                airflow.stats.Stats.incr("empty_key")
         importlib.reload(airflow.stats)
 
     def test_load_allow_list_validator(self):

tests/dag_processing/test_job_runner.py

@@ -1461,17 +1461,17 @@ def test_launch_process(self):
         assert os.path.isfile(log_file_loc)
 
     def test_single_parsing_loop_no_parent_signal_conn(self):
+        processor_agent = DagFileProcessorAgent("", 1, timedelta(days=365), [], False, False)
+        processor_agent._process = Mock()
+        processor_agent._parent_signal_conn = None
         with pytest.raises(ValueError, match="Process not started"):
-            processor_agent = DagFileProcessorAgent("", 1, timedelta(days=365), [], False, False)
-            processor_agent._process = Mock()
-            processor_agent._parent_signal_conn = None
             processor_agent.run_single_parsing_loop()
 
     def test_single_parsing_loop_no_process(self):
+        processor_agent = DagFileProcessorAgent("", 1, timedelta(days=365), [], False, False)
+        processor_agent._parent_signal_conn = Mock()
+        processor_agent._process = None
         with pytest.raises(ValueError, match="Process not started"):
-            processor_agent = DagFileProcessorAgent("", 1, timedelta(days=365), [], False, False)
-            processor_agent._parent_signal_conn = Mock()
-            processor_agent._process = None
             processor_agent.run_single_parsing_loop()
 
     def test_single_parsing_loop_process_isnt_alive(self):
@@ -1498,15 +1498,15 @@ def test_get_callbacks_pipe(self):
         assert retval == processor_agent._parent_signal_conn
 
     def test_get_callbacks_pipe_no_parent_signal_conn(self):
+        processor_agent = DagFileProcessorAgent("", 1, timedelta(days=365), [], False, False)
+        processor_agent._parent_signal_conn = None
         with pytest.raises(ValueError, match="Process not started"):
-            processor_agent = DagFileProcessorAgent("", 1, timedelta(days=365), [], False, False)
-            processor_agent._parent_signal_conn = None
             processor_agent.get_callbacks_pipe()
 
     def test_wait_until_finished_no_parent_signal_conn(self):
+        processor_agent = DagFileProcessorAgent("", 1, timedelta(days=365), [], False, False)
+        processor_agent._parent_signal_conn = None
         with pytest.raises(ValueError, match="Process not started"):
-            processor_agent = DagFileProcessorAgent("", 1, timedelta(days=365), [], False, False)
-            processor_agent._parent_signal_conn = None
             processor_agent.wait_until_finished()
 
     def test_wait_until_finished_poll_eof_error(self):
@@ -1519,9 +1519,9 @@ def test_wait_until_finished_poll_eof_error(self):
         assert ret_val is None
 
     def test_heartbeat_no_parent_signal_conn(self):
+        processor_agent = DagFileProcessorAgent("", 1, timedelta(days=365), [], False, False)
+        processor_agent._parent_signal_conn = None
         with pytest.raises(ValueError, match="Process not started"):
-            processor_agent = DagFileProcessorAgent("", 1, timedelta(days=365), [], False, False)
-            processor_agent._parent_signal_conn = None
             processor_agent.heartbeat()
 
     def test_heartbeat_poll_eof_error(self):
@@ -1553,15 +1553,15 @@ def test_heartbeat_poll_process_message(self):
             processor_agent._process_message.assert_called_with("testelem")
 
     def test_process_message_invalid_type(self):
+        message = "xyz"
+        processor_agent = DagFileProcessorAgent("", 1, timedelta(days=365), [], False, False)
         with pytest.raises(RuntimeError, match="Unexpected message received of type str"):
-            message = "xyz"
-            processor_agent = DagFileProcessorAgent("", 1, timedelta(days=365), [], False, False)
             processor_agent._process_message(message)
 
     def test_heartbeat_manager(self):
+        processor_agent = DagFileProcessorAgent("", 1, timedelta(days=365), [], False, False)
+        processor_agent._parent_signal_conn = None
         with pytest.raises(ValueError, match="Process not started"):
-            processor_agent = DagFileProcessorAgent("", 1, timedelta(days=365), [], False, False)
-            processor_agent._parent_signal_conn = None
             processor_agent._heartbeat_manager()
 
     @mock.patch("airflow.utils.process_utils.reap_process_group")

tests/decorators/test_bash.py

@@ -339,14 +339,15 @@ def bash():
 
         assert bash_task.operator.bash_command == NOTSET
 
+        dr = self.dag_maker.create_dagrun()
+        ti = dr.task_instances[0]
         with pytest.raises(AirflowException, match=f"Can not find the cwd: {cwd_path}"):
-            ti, _ = self.execute_task(bash_task)
-
-            self.validate_bash_command_rtif(ti, "echo")
+            ti.run()
+        assert ti.task.bash_command == "echo"
 
     def test_cwd_is_file(self, tmp_path):
         """Verify task failure for user-defined working directory that is actually a file."""
-        cwd_file = tmp_path / "test_file.sh"
+        cwd_file = tmp_path / "testfile.var.env"
         cwd_file.touch()
 
         with self.dag:
@@ -359,28 +360,32 @@ def bash():
 
         assert bash_task.operator.bash_command == NOTSET
 
+        dr = self.dag_maker.create_dagrun()
+        ti = dr.task_instances[0]
         with pytest.raises(AirflowException, match=f"The cwd {cwd_file} must be a directory"):
-            ti, _ = self.execute_task(bash_task)
-
-            self.validate_bash_command_rtif(ti, "echo")
+            ti.run()
+        assert ti.task.bash_command == "echo"
 
     def test_command_not_found(self):
         """Fail task if executed command is not found on path."""
-        with pytest.raises(
-            AirflowException, match="Bash command failed\\. The command returned a non-zero exit code 127\\."
-        ):
-            with self.dag:
 
-                @task.bash
-                def bash():
-                    return "set -e; something-that-isnt-on-path"
+        with self.dag:
 
-                bash_task = bash()
+            @task.bash
+            def bash():
+                return "set -e; something-that-isnt-on-path"
 
-            assert bash_task.operator.bash_command == NOTSET
+            bash_task = bash()
 
-            ti, _ = self.execute_task(bash_task)
-            self.validate_bash_command_rtif(ti, "set -e; something-that-isnt-on-path")
+        assert bash_task.operator.bash_command == NOTSET
+
+        dr = self.dag_maker.create_dagrun()
+        ti = dr.task_instances[0]
+        with pytest.raises(
+            AirflowException, match="Bash command failed\\. The command returned a non-zero exit code 127\\."
+        ):
+            ti.run()
+        assert ti.task.bash_command == "set -e; something-that-isnt-on-path"
 
     def test_multiple_outputs_true(self):
         """Verify setting `multiple_outputs` for a @task.bash-decorated function is ignored."""
@@ -474,7 +479,8 @@ def bash():
 
         assert bash_task.operator.bash_command == NOTSET
 
+        dr = self.dag_maker.create_dagrun()
+        ti = dr.task_instances[0]
         with pytest.raises(AirflowException):
-            ti, _ = self.execute_task(bash_task)
-
-            self.validate_bash_command_rtif(ti, f"{DEFAULT_DATE.date()}; exit 1;")
+            ti.run()
+        assert ti.task.bash_command == f"{DEFAULT_DATE.date()}; exit 1;"

tests/decorators/test_setup_teardown.py

@@ -126,7 +126,7 @@ def test_marking_operator_as_teardown_task(self, dag_maker):
 
     def test_setup_taskgroup_decorator(self, dag_maker):
         with dag_maker():
-            with pytest.raises(
+            with pytest.raises(  # noqa: PT012, check decorators required more than one line
                 expected_exception=AirflowException,
                 match="Task groups cannot be marked as setup or teardown.",
             ):
@@ -144,7 +144,7 @@ def mytask():
 
     def test_teardown_taskgroup_decorator(self, dag_maker):
         with dag_maker():
-            with pytest.raises(
+            with pytest.raises(  # noqa: PT012, check decorators required more than one line
                 expected_exception=AirflowException,
                 match="Task groups cannot be marked as setup or teardown.",
             ):
@@ -989,7 +989,9 @@ def mytask():
         def teardowntask():
             print("teardown")
 
-        with pytest.raises(ValueError, match="All tasks in the list must be either setup or teardown tasks"):
+        with pytest.raises(  # noqa: PT012, check decorators required more than one line
+            ValueError, match="All tasks in the list must be either setup or teardown tasks"
+        ):
             with dag_maker():
                 with setuptask() << context_wrapper([teardowntask(), setuptask2()]):
                     mytask()

tests/jobs/test_backfill_job.py

@@ -1046,10 +1046,10 @@ def test_backfill_depends_on_past_backwards(self):
 
         # raises backwards
         expected_msg = "You cannot backfill backwards because one or more tasks depend_on_past: test_dop_task"
+        executor = MockExecutor()
+        job = Job(executor=executor)
+        job_runner = BackfillJobRunner(job=job, dag=dag, run_backwards=True, **kwargs)
         with pytest.raises(AirflowException, match=expected_msg):
-            executor = MockExecutor()
-            job = Job(executor=executor)
-            job_runner = BackfillJobRunner(job=job, dag=dag, run_backwards=True, **kwargs)
             run_job(job=job, execute_callable=job_runner._execute)
 
     def test_cli_receives_delay_arg(self):

tests/jobs/test_triggerer_job.py

@@ -213,8 +213,8 @@ def test_capacity_decode():
         4 / 2,  # Resolves to a float, in addition to being just plain weird
     ]
     for input_str in variants:
+        job = Job()
         with pytest.raises(ValueError):
-            job = Job()
             TriggererJobRunner(job=job, capacity=input_str)