Enable PT006 rule to airflow-core tests (security, lineage, jobs, executors, datasets) (#57913)

Co-authored-by: Kalyan R <kalyan.ben10@live.com>

Author: xchwan

airflow-core/tests/unit/datasets/test_dataset.py

@@ -22,7 +22,7 @@
 
 
 @pytest.mark.parametrize(
-    "module_path, attr_name, expected_value, warning_message",
+    ("module_path", "attr_name", "expected_value", "warning_message"),
     (
         pytest.param(
             "airflow",

airflow-core/tests/unit/executors/test_base_executor.py

@@ -125,7 +125,7 @@ def test_gauge_executor_metrics_single_executor(mock_stats_gauge, mock_trigger_t
 
 
 @pytest.mark.parametrize(
-    "executor_class, executor_name",
+    ("executor_class", "executor_name"),
     [(LocalExecutor, "LocalExecutor")],
 )
 @mock.patch("airflow.executors.local_executor.LocalExecutor.sync")
@@ -285,7 +285,7 @@ def test_parser_add_command(mock_add_command, mock_get_cli_command):
     mock_add_command.assert_called_once()
 
 
-@pytest.mark.parametrize("loop_duration, total_tries", [(0.5, 12), (1.0, 7), (1.7, 4), (10, 2)])
+@pytest.mark.parametrize(("loop_duration", "total_tries"), [(0.5, 12), (1.0, 7), (1.7, 4), (10, 2)])
 def test_running_retry_attempt_type(loop_duration, total_tries):
     """
     Verify can_try_again returns True until at least 5 seconds have passed.

airflow-core/tests/unit/executors/test_local_executor.py

@@ -169,7 +169,7 @@ def test_clean_stop_on_signal(self):
             executor.end()
 
     @pytest.mark.parametrize(
-        ["conf_values", "expected_server"],
+        ("conf_values", "expected_server"),
         [
             (
                 {

airflow-core/tests/unit/jobs/test_base_job.py

@@ -94,7 +94,7 @@ def abort():
         assert job.end_date is not None
 
     @pytest.mark.parametrize(
-        "job_runner, job_type,job_heartbeat_sec",
+        ("job_runner", "job_type", "job_heartbeat_sec"),
         [(SchedulerJobRunner, "scheduler", "11"), (TriggererJobRunner, "triggerer", "9")],
     )
     def test_heart_rate_after_fetched_from_db(self, job_runner, job_type, job_heartbeat_sec):
@@ -116,7 +116,7 @@ def test_heart_rate_after_fetched_from_db(self, job_runner, job_type, job_heartb
             session.rollback()
 
     @pytest.mark.parametrize(
-        "job_runner, job_type,job_heartbeat_sec",
+        ("job_runner", "job_type", "job_heartbeat_sec"),
         [(SchedulerJobRunner, "scheduler", "11"), (TriggererJobRunner, "triggerer", "9")],
     )
     def test_heart_rate_via_constructor_persists(self, job_runner, job_type, job_heartbeat_sec):

airflow-core/tests/unit/jobs/test_scheduler_job.py

@@ -950,7 +950,7 @@ def test_find_executable_task_instances_pool(self, dag_maker):
         session.rollback()
 
     @pytest.mark.parametrize(
-        "state, total_executed_ti",
+        ("state", "total_executed_ti"),
         [
             (DagRunState.SUCCESS, 0),
             (DagRunState.FAILED, 0),
@@ -1824,7 +1824,7 @@ def test_enqueue_task_instances_sets_ti_state_to_None_if_dagrun_in_finish_state(
         mock_queue_workload.assert_not_called()
 
     @pytest.mark.parametrize(
-        "task1_exec, task2_exec",
+        ("task1_exec", "task2_exec"),
         [
             ("default_exec", "default_exec"),
             ("default_exec", "secondary_exec"),
@@ -1967,7 +1967,7 @@ def _create_dagruns():
             assert ti.state == State.QUEUED
 
     @pytest.mark.parametrize(
-        "task1_exec, task2_exec",
+        ("task1_exec", "task2_exec"),
         [
             ("default_exec", "default_exec"),
             ("default_exec", "secondary_exec"),
@@ -2106,7 +2106,7 @@ def _create_dagruns():
         session.rollback()
 
     @pytest.mark.parametrize(
-        "task1_exec, task2_exec",
+        ("task1_exec", "task2_exec"),
         [
             ("default_exec", "default_exec"),
             ("default_exec", "secondary_exec"),
@@ -2641,7 +2641,7 @@ def test_runs_are_created_after_max_active_runs_was_reached(self, dag_maker, ses
         assert len(dag_runs) == 2
 
     @pytest.mark.parametrize(
-        "ti_state, final_ti_span_status",
+        ("ti_state", "final_ti_span_status"),
         [
             pytest.param(State.SUCCESS, SpanStatus.ENDED, id="dr_ended_successfully"),
             pytest.param(State.RUNNING, SpanStatus.ACTIVE, id="dr_still_running"),
@@ -2766,7 +2766,7 @@ def test_end_spans_of_externally_ended_ops(self, dag_maker):
         assert self.job_runner.active_spans.get("ti:" + ti.id) is None
 
     @pytest.mark.parametrize(
-        "state, final_span_status",
+        ("state", "final_span_status"),
         [
             pytest.param(State.SUCCESS, SpanStatus.ENDED, id="dr_ended_successfully"),
             pytest.param(State.RUNNING, SpanStatus.NEEDS_CONTINUANCE, id="dr_still_running"),
@@ -2956,7 +2956,7 @@ def test_dagrun_timeout_fails_run_and_update_next_dagrun(self, dag_maker):
         )
 
     @pytest.mark.parametrize(
-        "state, expected_callback_msg", [(State.SUCCESS, "success"), (State.FAILED, "task_failure")]
+        ("state", "expected_callback_msg"), [(State.SUCCESS, "success"), (State.FAILED, "task_failure")]
     )
     def test_dagrun_callbacks_are_called(self, state, expected_callback_msg, dag_maker, session):
         """
@@ -3001,7 +3001,7 @@ def test_dagrun_callbacks_are_called(self, state, expected_callback_msg, dag_mak
         session.close()
 
     @pytest.mark.parametrize(
-        "state, expected_callback_msg", [(State.SUCCESS, "success"), (State.FAILED, "task_failure")]
+        ("state", "expected_callback_msg"), [(State.SUCCESS, "success"), (State.FAILED, "task_failure")]
     )
     def test_dagrun_plugins_are_notified(self, state, expected_callback_msg, dag_maker, session):
         """
@@ -3156,7 +3156,7 @@ def test_dagrun_callbacks_are_not_added_when_callbacks_are_not_defined(self, sta
 
         session.rollback()
 
-    @pytest.mark.parametrize("state, msg", [[State.SUCCESS, "success"], [State.FAILED, "task_failure"]])
+    @pytest.mark.parametrize(("state", "msg"), [[State.SUCCESS, "success"], [State.FAILED, "task_failure"]])
     def test_dagrun_callbacks_are_added_when_callbacks_are_defined(self, state, msg, dag_maker):
         """
         Test if on_*_callback are defined on DAG, Callbacks ARE registered and sent to DAG Processor
@@ -3278,7 +3278,7 @@ def test_do_not_schedule_removed_task(self, dag_maker, session):
         assert len(new_tis) == 0
 
     @pytest.mark.parametrize(
-        "ti_states, run_state",
+        ("ti_states", "run_state"),
         [
             (["failed", "success"], "failed"),
             (["success", "success"], "success"),
@@ -4257,7 +4257,7 @@ def test_should_update_dag_next_dagruns(self, provide_run_count: bool, kwargs: d
         assert actual == should_update
 
     @pytest.mark.parametrize(
-        "run_type, expected",
+        ("run_type", "expected"),
         [
             (DagRunType.MANUAL, False),
             (DagRunType.SCHEDULED, True),
@@ -4498,7 +4498,7 @@ def dict_from_obj(obj):
 
     @pytest.mark.need_serialized_dag
     @pytest.mark.parametrize(
-        "disable, enable",
+        ("disable", "enable"),
         [
             pytest.param({"is_stale": True}, {"is_stale": False}, id="active"),
             pytest.param({"is_paused": True}, {"is_paused": False}, id="paused"),
@@ -5613,7 +5613,7 @@ def _running_counts():
         assert session.scalar(select(func.count()).where(DagRun.dag_id == dag1_dag_id)) == 36
 
     @pytest.mark.parametrize(
-        "pause_it, expected_running",
+        ("pause_it", "expected_running"),
         [
             (True, 0),
             (False, 3),
@@ -5820,7 +5820,7 @@ def test_no_dagruns_would_stuck_in_running(self, dag_maker):
         assert DagRun.find(run_id="dr1_run_2")[0].state == State.RUNNING
 
     @pytest.mark.parametrize(
-        "state, start_date, end_date",
+        ("state", "start_date", "end_date"),
         [
             [State.NONE, None, None],
             [
@@ -5862,7 +5862,7 @@ def test_dag_file_processor_process_task_instances(self, state, start_date, end_
             assert ti.state == State.SCHEDULED
 
     @pytest.mark.parametrize(
-        "state,start_date,end_date",
+        ("state", "start_date", "end_date"),
         [
             [State.NONE, None, None],
             [
@@ -5908,7 +5908,7 @@ def test_dag_file_processor_process_task_instances_with_max_active_tis_per_dag(
             assert ti.state == State.SCHEDULED
 
     @pytest.mark.parametrize(
-        "state,start_date,end_date",
+        ("state", "start_date", "end_date"),
         [
             [State.NONE, None, None],
             [
@@ -5954,7 +5954,7 @@ def test_dag_file_processor_process_task_instances_with_max_active_tis_per_dagru
             assert ti.state == State.SCHEDULED
 
     @pytest.mark.parametrize(
-        "state, start_date, end_date",
+        ("state", "start_date", "end_date"),
         [
             [State.NONE, None, None],
             [
@@ -6712,7 +6712,7 @@ def test_asset_orphaning_ignore_orphaned_assets(self, dag_maker, session):
         assert [asset.updated_at for asset in orphaned] == updated_at_timestamps
 
     @pytest.mark.parametrize(
-        "paused, stale, expected_classpath",
+        ("paused", "stale", "expected_classpath"),
         [
             pytest.param(
                 False,
@@ -6931,7 +6931,7 @@ def test_scheduler_passes_context_from_server_on_heartbeat_timeout(self, dag_mak
         assert callback_request.context_from_server.max_tries == ti.max_tries
 
     @pytest.mark.parametrize(
-        "retries,callback_kind,expected",
+        ("retries", "callback_kind", "expected"),
         [
             (1, "retry", TaskInstanceState.UP_FOR_RETRY),
             (0, "failure", TaskInstanceState.FAILED),
@@ -7178,7 +7178,7 @@ def per_test(self) -> Generator:
         self.clean_db()
 
     @pytest.mark.parametrize(
-        "expected_query_count, dag_count, task_count",
+        ("expected_query_count", "dag_count", "task_count"),
         [
             (21, 1, 1),  # One DAG with one task per DAG file.
             (21, 1, 5),  # One DAG with five tasks per DAG file.
@@ -7244,7 +7244,7 @@ def test_execute_queries_count_with_harvested_dags(
 
     @pytest.mark.flaky(reruns=3, reruns_delay=5)
     @pytest.mark.parametrize(
-        "expected_query_counts, dag_count, task_count, start_ago, schedule, shape",
+        ("expected_query_counts", "dag_count", "task_count", "start_ago", "schedule", "shape"),
         [
             # One DAG with one task per DAG file.
             ([10, 10, 10, 10], 1, 1, "1d", "None", "no_structure"),

airflow-core/tests/unit/jobs/test_triggerer_job.py

@@ -279,7 +279,7 @@ def send_msg_spy(self, msg, *args, **kwargs):
 
 
 @pytest.mark.parametrize(
-    "trigger, watcher_count, trigger_count",
+    ("trigger", "watcher_count", "trigger_count"),
     [
         (TimeDeltaTrigger(datetime.timedelta(days=7)), 0, 1),
         (FileDeleteTrigger("/tmp/foo.txt", poke_interval=1), 1, 0),

airflow-core/tests/unit/lineage/test_hook.py

@@ -206,7 +206,7 @@ class FakePlugin(plugins_manager.AirflowPlugin):
 
 
 @pytest.mark.parametrize(
-    "has_readers, expected_class",
+    ("has_readers", "expected_class"),
     [
         (True, HookLineageCollector),
         (False, NoOpCollector),

airflow-core/tests/unit/security/test_kerberos.py

@@ -33,7 +33,7 @@
 
 class TestKerberos:
     @pytest.mark.parametrize(
-        "kerberos_config, expected_cmd",
+        ("kerberos_config", "expected_cmd"),
         [
             (
                 {("kerberos", "reinit_frequency"): "42"},