Resolve PT017 in tests (#37837)

pyproject.toml

@@ -1350,7 +1350,6 @@ ignore = [
     "PT011", # pytest.raises() is too broad, set the match parameter
     "PT012", # [controversial rule] pytest.raises() block should contain a single simple statement.
     "PT015", # assertion always fails, replace with pytest.fail()
-    "PT017", # raise exception in except block, use pytest.raises() instead
     "PT018", # assertion should be broken down into multiple parts
     "PT019", # fixture without value is injected as parameter, use @pytest.mark.usefixtures instead
 ]

tests/models/test_baseoperator.py

@@ -192,25 +192,11 @@ def test_trigger_rule_validation(self):
         )
 
         # An operator with default trigger rule and a fail-stop dag should be allowed
-        try:
-            BaseOperator(
-                task_id="test_valid_trigger_rule", dag=fail_stop_dag, trigger_rule=DEFAULT_TRIGGER_RULE
-            )
-        except FailStopDagInvalidTriggerRule as exception:
-            assert (
-                False
-            ), f"BaseOperator raises exception with fail-stop dag & default trigger rule: {exception}"
-
+        BaseOperator(task_id="test_valid_trigger_rule", dag=fail_stop_dag, trigger_rule=DEFAULT_TRIGGER_RULE)
         # An operator with non default trigger rule and a non fail-stop dag should be allowed
-        try:
-            BaseOperator(
-                task_id="test_valid_trigger_rule", dag=non_fail_stop_dag, trigger_rule=TriggerRule.DUMMY
-            )
-        except FailStopDagInvalidTriggerRule as exception:
-            assert (
-                False
-            ), f"BaseOperator raises exception with non fail-stop dag & non-default trigger rule: {exception}"
-
+        BaseOperator(
+            task_id="test_valid_trigger_rule", dag=non_fail_stop_dag, trigger_rule=TriggerRule.ALWAYS
+        )
         # An operator with non default trigger rule and a fail stop dag should not be allowed
         with pytest.raises(FailStopDagInvalidTriggerRule):
             BaseOperator(

tests/models/test_dag.py

@@ -1842,17 +1842,12 @@ def test_dag_add_task_checks_trigger_rule(self):
         from airflow.utils.trigger_rule import TriggerRule
 
         task_with_non_default_trigger_rule = EmptyOperator(
-            task_id="task_with_non_default_trigger_rule", trigger_rule=TriggerRule.DUMMY
+            task_id="task_with_non_default_trigger_rule", trigger_rule=TriggerRule.ALWAYS
         )
         non_fail_stop_dag = DAG(
             dag_id="test_dag_add_task_checks_trigger_rule", start_date=DEFAULT_DATE, fail_stop=False
         )
-        try:
-            non_fail_stop_dag.add_task(task_with_non_default_trigger_rule)
-        except FailStopDagInvalidTriggerRule as exception:
-            assert (
-                False
-            ), f"dag add_task() raises FailStopDagInvalidTriggerRule for non fail stop dag: {exception}"
+        non_fail_stop_dag.add_task(task_with_non_default_trigger_rule)
 
         # a fail stop dag should allow default trigger rule
         from airflow.models.abstractoperator import DEFAULT_TRIGGER_RULE
@@ -1863,12 +1858,7 @@ def test_dag_add_task_checks_trigger_rule(self):
         task_with_default_trigger_rule = EmptyOperator(
             task_id="task_with_default_trigger_rule", trigger_rule=DEFAULT_TRIGGER_RULE
         )
-        try:
-            fail_stop_dag.add_task(task_with_default_trigger_rule)
-        except FailStopDagInvalidTriggerRule as exception:
-            assert (
-                False
-            ), f"dag.add_task() raises exception for fail-stop dag & default trigger rule: {exception}"
+        fail_stop_dag.add_task(task_with_default_trigger_rule)
 
         # a fail stop dag should not allow a non-default trigger rule
         with pytest.raises(FailStopDagInvalidTriggerRule):

tests/providers/cncf/kubernetes/executors/test_kubernetes_executor.py

@@ -1837,7 +1837,7 @@ def test_recover_from_resource_too_old(self):
         def effect():
             yield "500"
             while True:
-                yield Exception("sentinel")
+                yield SystemError("sentinel")
 
         mock_underscore_run.side_effect = effect()
 
@@ -1846,22 +1846,17 @@ def effect():
         with mock.patch(
             "airflow.providers.cncf.kubernetes.executors.kubernetes_executor_utils.get_kube_client"
         ):
-            try:
+            with pytest.raises(SystemError, match="sentinel"):
                 # self.watcher._run() is mocked and return "500" as last resource_version
                 self.watcher.run()
-                assert False, "Should have raised Exception"
-            except Exception as e:
-                assert e.args == ("sentinel",)
 
             # both resource_version should be 0 after _run raises an exception
             assert self.watcher.resource_version == "0"
             assert ResourceVersion().resource_version[self.test_namespace] == "0"
 
             # check that in the next run, _run is invoked with resource_version = 0
             mock_underscore_run.reset_mock()
-            try:
+            with pytest.raises(SystemError, match="sentinel"):
                 self.watcher.run()
-            except Exception as e:
-                assert e.args == ("sentinel",)
 
             mock_underscore_run.assert_called_once_with(mock.ANY, "0", mock.ANY, mock.ANY)

tests/providers/microsoft/psrp/hooks/test_psrp.py

@@ -146,18 +146,14 @@ def test_invoke(self, runspace_pool, powershell, ws_man, logging_level):
             on_output_callback=on_output_callback,
             **options,
         ) as hook, patch.object(type(hook), "log") as logger:
-            try:
+            with pytest.raises(AirflowException, match="Process had one or more errors"):
                 with hook.invoke() as ps:
                     assert ps.state == PSInvocationState.NOT_STARTED
 
                     # We're simulating an error in order to test error
                     # handling as well as the logging of error exception
                     # details.
                     ps.had_errors = True
-            except AirflowException as exc:
-                assert str(exc) == "Process had one or more errors"
-            else:
-                self.fail("Expected an error")
             assert ps.state == PSInvocationState.COMPLETED
 
         assert on_output_callback.mock_calls == [call("output")]

tests/providers/mysql/operators/test_mysql.py

@@ -94,10 +94,8 @@ def test_overwrite_schema(self, client):
 
             from MySQLdb import OperationalError
 
-            try:
+            with pytest.raises(OperationalError, match="Unknown database 'foobar'"):
                 op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE, ignore_ti_state=True)
-            except OperationalError as e:
-                assert "Unknown database 'foobar'" in str(e)
 
     def test_mysql_operator_resolve_parameters_template_json_file(self, tmp_path):
         path = tmp_path / "testfile.json"

tests/providers/postgres/operators/test_postgres.py

@@ -93,10 +93,8 @@ def test_overwrite_database(self):
 
         from psycopg2 import OperationalError
 
-        try:
+        with pytest.raises(OperationalError, match='database "foobar" does not exist'):
             op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE, ignore_ti_state=True)
-        except OperationalError as e:
-            assert 'database "foobar" does not exist' in str(e)
 
     def test_runtime_parameter_setting(self):
         """

tests/providers/sqlite/operators/test_sqlite.py

@@ -70,9 +70,6 @@ def test_sqlite_operator_with_invalid_sql(self):
 
         from sqlite3 import OperationalError
 
-        try:
-            op = SqliteOperator(task_id="sqlite_operator_with_multiple_statements", sql=sql, dag=self.dag)
+        op = SqliteOperator(task_id="sqlite_operator_with_multiple_statements", sql=sql, dag=self.dag)
+        with pytest.raises(OperationalError, match="no such table: test_airflow2"):
             op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE, ignore_ti_state=True)
-            pytest.fail("An exception should have been thrown")
-        except OperationalError as e:
-            assert "no such table: test_airflow2" in str(e)

tests/sensors/test_external_task_sensor.py

@@ -579,10 +579,9 @@ def test_external_task_sensor_fn_multiple_execution_dates(self):
         # We need to test for an AirflowException explicitly since
         # AirflowSensorTimeout is a subclass that will be raised if this does
         # not execute properly.
-        try:
+        with pytest.raises(AirflowException) as ex_ctx:
             task_chain_with_failure.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE, ignore_ti_state=True)
-        except AirflowException as ex:
-            assert type(ex) == AirflowException
+        assert type(ex_ctx.value) is AirflowException
 
     def test_external_task_sensor_delta(self):
         self.add_time_sensor()

tests/utils/test_file.py

@@ -161,14 +161,12 @@ def test_find_path_from_directory_fails_on_recursive_link(self, test_dir):
 
         ignore_list_file = ".airflowignore"
 
-        try:
+        error_message = (
+            f"Detected recursive loop when walking DAG directory {test_dir}: "
+            f"{Path(recursing_tgt).resolve()} has appeared more than once."
+        )
+        with pytest.raises(RuntimeError, match=error_message):
             list(find_path_from_directory(test_dir, ignore_list_file, ignore_file_syntax="glob"))
-            assert False, "Walking a self-recursive tree should fail"
-        except RuntimeError as err:
-            assert str(err) == (
-                f"Detected recursive loop when walking DAG directory {test_dir}: "
-                f"{Path(recursing_tgt).resolve()} has appeared more than once."
-            )
 
     def test_might_contain_dag_with_default_callable(self):
         file_path_with_dag = os.path.join(TEST_DAGS_FOLDER, "test_scheduler_dags.py")