diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt index 2b7c13ee66036..ae56f4109449f 100644 --- a/docs/spelling_wordlist.txt +++ b/docs/spelling_wordlist.txt @@ -449,6 +449,7 @@ deferrable deidentify DeidentifyTemplate del +deletable delim deliverability deltalake diff --git a/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py b/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py index ea0d1272db8c6..d1a1b9e54e571 100644 --- a/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py +++ b/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py @@ -27,6 +27,7 @@ from airflow.providers.amazon.aws.hooks.redshift_cluster import RedshiftHook from airflow.providers.amazon.aws.operators.base_aws import AwsBaseOperator from airflow.providers.amazon.aws.triggers.redshift_cluster import ( + RedshiftClusterSettledTrigger, RedshiftCreateClusterSnapshotTrigger, RedshiftCreateClusterTrigger, RedshiftDeleteClusterTrigger, @@ -836,7 +837,9 @@ class RedshiftDeleteClusterOperator(AwsBaseOperator[RedshiftHook]): https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html :param poll_interval: Time (in seconds) to wait between two consecutive calls to check cluster state :param deferrable: Run operator in the deferrable mode. - :param max_attempts: (Deferrable mode only) The maximum number of attempts to be made + :param max_attempts: The maximum number of attempts to be made. In deferrable mode this bounds the + async wait for a busy cluster to settle before the delete is re-issued; combined with + ``poll_interval`` the default gives a ~15 minute window, long enough to outlast a pause/resize. """ template_fields: Sequence[str] = aws_template_fields( @@ -864,15 +867,18 @@ def __init__( self.final_cluster_snapshot_identifier = final_cluster_snapshot_identifier self.wait_for_completion = wait_for_completion self.poll_interval = poll_interval - # These parameters are added to keep trying if there is a running operation in the cluster - # If there is a running operation in the cluster while trying to delete it, a InvalidClusterStateFault - # is thrown. In such case, retrying - self._attempts = 10 + # Retry the delete while the cluster is mid-transition (InvalidClusterStateFault) until it + # settles into a deletable state. + self._attempts = 60 self._attempt_interval = 15 self.deferrable = deferrable self.max_attempts = max_attempts def execute(self, context: Context): + if self.deferrable: + self._delete_or_defer_until_settled() + return + while self._attempts: try: self.hook.delete_cluster( @@ -897,36 +903,82 @@ def execute(self, context: Context): else: raise - if self.deferrable: - cluster_state = self.hook.cluster_status(cluster_identifier=self.cluster_identifier) - if cluster_state == "cluster_not_found": - self.log.info("Cluster deleted successfully") - elif cluster_state in ("creating", "modifying"): - raise AirflowException( - f"Unable to delete cluster since cluster is currently in status: {cluster_state}" - ) - else: - self.defer( - timeout=timedelta(seconds=self.max_attempts * self.poll_interval + 60), - trigger=RedshiftDeleteClusterTrigger( - cluster_identifier=self.cluster_identifier, - waiter_delay=self.poll_interval, - waiter_max_attempts=self.max_attempts, - aws_conn_id=self.aws_conn_id, - region_name=self.region_name, - verify=self.verify, - botocore_config=self.botocore_config, - ), - method_name="execute_complete", - ) - - elif self.wait_for_completion: + if self.wait_for_completion: waiter = self.hook.conn.get_waiter("cluster_deleted") waiter.wait( ClusterIdentifier=self.cluster_identifier, WaiterConfig={"Delay": self.poll_interval, "MaxAttempts": self.max_attempts}, ) + def _delete_or_defer_until_settled(self) -> None: + """ + Issue the delete once (deferrable mode), then defer. + + If accepted, defer to :class:`RedshiftDeleteClusterTrigger` to await deletion. If the cluster is + busy (``InvalidClusterStateFault``), defer to :class:`RedshiftClusterSettledTrigger`; the + ``_retry_delete_when_settled`` callback re-issues the delete once it settles. + """ + try: + self.hook.delete_cluster( + cluster_identifier=self.cluster_identifier, + skip_final_cluster_snapshot=self.skip_final_cluster_snapshot, + final_cluster_snapshot_identifier=self.final_cluster_snapshot_identifier, + ) + except self.hook.conn.exceptions.InvalidClusterStateFault: + self.log.info( + "Cluster %s is busy; deferring until it settles into a deletable state.", + self.cluster_identifier, + ) + self.defer( + timeout=timedelta(seconds=self.max_attempts * self.poll_interval + 60), + trigger=RedshiftClusterSettledTrigger( + cluster_identifier=self.cluster_identifier, + waiter_delay=self.poll_interval, + waiter_max_attempts=self.max_attempts, + aws_conn_id=self.aws_conn_id, + region_name=self.region_name, + verify=self.verify, + botocore_config=self.botocore_config, + ), + method_name="_retry_delete_when_settled", + ) + return + + self._defer_until_deleted() + + def _defer_until_deleted(self) -> None: + """Defer to the delete-completion waiter, short-circuiting if the cluster is already gone.""" + cluster_state = self.hook.cluster_status(cluster_identifier=self.cluster_identifier) + if cluster_state == "cluster_not_found": + self.log.info("Cluster deleted successfully") + return + self.defer( + timeout=timedelta(seconds=self.max_attempts * self.poll_interval + 60), + trigger=RedshiftDeleteClusterTrigger( + cluster_identifier=self.cluster_identifier, + waiter_delay=self.poll_interval, + waiter_max_attempts=self.max_attempts, + aws_conn_id=self.aws_conn_id, + region_name=self.region_name, + verify=self.verify, + botocore_config=self.botocore_config, + ), + method_name="execute_complete", + ) + + def _retry_delete_when_settled(self, context: Context, event: dict[str, Any] | None = None) -> None: + """ + Re-issue the delete once the cluster has settled, then defer until deletion completes. + + Callback for :class:`RedshiftClusterSettledTrigger`. If the delete is still rejected because of a + race (the cluster re-entered a transitional state), defer to the settle-wait trigger again. + """ + validated_event = validate_execute_complete_event(event) + if validated_event["status"] != "success": + raise AirflowException(f"Error waiting for cluster to become deletable: {validated_event}") + + self._delete_or_defer_until_settled() + def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> None: validated_event = validate_execute_complete_event(event) diff --git a/providers/amazon/src/airflow/providers/amazon/aws/triggers/redshift_cluster.py b/providers/amazon/src/airflow/providers/amazon/aws/triggers/redshift_cluster.py index ebd32a2b42388..ce6d29154fb4d 100644 --- a/providers/amazon/src/airflow/providers/amazon/aws/triggers/redshift_cluster.py +++ b/providers/amazon/src/airflow/providers/amazon/aws/triggers/redshift_cluster.py @@ -343,3 +343,56 @@ async def run(self) -> AsyncIterator[TriggerEvent]: await asyncio.sleep(self.poke_interval) except Exception as e: yield TriggerEvent({"status": "error", "message": str(e)}) + + +class RedshiftClusterSettledTrigger(AwsBaseWaiterTrigger): + """ + Wait until a Redshift cluster settles into a non-transitional (deletable) lifecycle. + + A ``delete_cluster`` call is rejected with ``InvalidClusterStateFault`` while an operation is in + flight (e.g. a pause or resize). Because a busy cluster can settle into *different* terminal states + depending on the in-flight operation (a ``pausing`` cluster becomes ``paused``; a ``resizing`` cluster + becomes ``available``), this trigger uses the custom ``cluster_deletable`` waiter, which has one + ``success`` acceptor per deletable ``ClusterStatus`` value rather than a single target state. + + :param cluster_identifier: unique identifier of a cluster + :param waiter_delay: The amount of time in seconds to wait between attempts. + :param waiter_max_attempts: The maximum number of attempts to be made. + :param aws_conn_id: The Airflow connection used for AWS credentials. + :param region_name: The AWS region where the cluster is. Used to build the hook. + :param verify: Whether or not to verify SSL certificates. Used to build the hook. + :param botocore_config: Configuration dictionary for the botocore client. Used to build the hook. + """ + + def __init__( + self, + *, + cluster_identifier: str, + aws_conn_id: str | None = "aws_default", + region_name: str | None = None, + waiter_delay: int = 30, + waiter_max_attempts: int = 30, + **kwargs, + ): + super().__init__( + serialized_fields={"cluster_identifier": cluster_identifier}, + waiter_name="cluster_deletable", + waiter_args={"ClusterIdentifier": cluster_identifier}, + failure_message="Error while waiting for the redshift cluster to become deletable", + status_message="Waiting for redshift cluster to settle into a deletable state", + status_queries=["Clusters[].ClusterStatus"], + return_value=None, + waiter_delay=waiter_delay, + waiter_max_attempts=waiter_max_attempts, + aws_conn_id=aws_conn_id, + region_name=region_name, + **kwargs, + ) + + def hook(self) -> AwsGenericHook: + return RedshiftHook( + aws_conn_id=self.aws_conn_id, + region_name=self.region_name, + verify=self.verify, + config=self.botocore_config, + ) diff --git a/providers/amazon/src/airflow/providers/amazon/aws/waiters/redshift.json b/providers/amazon/src/airflow/providers/amazon/aws/waiters/redshift.json index 8165eb3fc439a..389da67068e90 100644 --- a/providers/amazon/src/airflow/providers/amazon/aws/waiters/redshift.json +++ b/providers/amazon/src/airflow/providers/amazon/aws/waiters/redshift.json @@ -50,6 +50,61 @@ "state": "failure" } ] + }, + "cluster_deletable": { + "operation": "DescribeClusters", + "delay": 30, + "maxAttempts": 60, + "acceptors": [ + { + "matcher": "pathAll", + "argument": "Clusters[].ClusterStatus", + "expected": "available", + "state": "success" + }, + { + "matcher": "pathAll", + "argument": "Clusters[].ClusterStatus", + "expected": "paused", + "state": "success" + }, + { + "matcher": "pathAll", + "argument": "Clusters[].ClusterStatus", + "expected": "incompatible-hsm", + "state": "success" + }, + { + "matcher": "pathAll", + "argument": "Clusters[].ClusterStatus", + "expected": "incompatible-restore", + "state": "success" + }, + { + "matcher": "pathAll", + "argument": "Clusters[].ClusterStatus", + "expected": "incompatible-network", + "state": "success" + }, + { + "matcher": "pathAll", + "argument": "Clusters[].ClusterStatus", + "expected": "incompatible-parameters", + "state": "success" + }, + { + "matcher": "pathAll", + "argument": "Clusters[].ClusterStatus", + "expected": "hardware-failure", + "state": "success" + }, + { + "matcher": "error", + "argument": "Clusters[].ClusterStatus", + "expected": "ClusterNotFound", + "state": "success" + } + ] } } } diff --git a/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_cluster.py b/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_cluster.py index db6466125a035..c329b9431dd97 100644 --- a/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_cluster.py +++ b/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_cluster.py @@ -34,6 +34,7 @@ RedshiftResumeClusterOperator, ) from airflow.providers.amazon.aws.triggers.redshift_cluster import ( + RedshiftClusterSettledTrigger, RedshiftCreateClusterSnapshotTrigger, RedshiftDeleteClusterTrigger, RedshiftPauseClusterTrigger, @@ -785,14 +786,65 @@ def test_delete_cluster_multiple_attempts_fail(self, _, mock_conn, mock_delete_c with pytest.raises(returned_exception): redshift_operator.execute(None) - assert mock_delete_cluster.call_count == 10 + assert mock_delete_cluster.call_count == 60 + + def test_busy_retry_defaults(self): + redshift_operator = RedshiftDeleteClusterOperator( + task_id="task_test", + cluster_identifier="test_cluster", + aws_conn_id="aws_conn_test", + ) + # 60 * 15s = ~15 min window, long enough to outlast a cluster pause/resize. + assert redshift_operator._attempts == 60 + assert redshift_operator._attempt_interval == 15 + + @mock.patch.object(RedshiftHook, "delete_cluster") + @mock.patch.object(RedshiftHook, "conn") + @mock.patch("time.sleep", return_value=None) + def test_delete_cluster_exhausts_busy_retries_then_raises(self, _, mock_conn, mock_delete_cluster): + exception = boto3.client("redshift").exceptions.InvalidClusterStateFault({}, "test") + returned_exception = type(exception) + mock_conn.exceptions.InvalidClusterStateFault = returned_exception + mock_delete_cluster.side_effect = exception + + redshift_operator = RedshiftDeleteClusterOperator( + task_id="task_test", + cluster_identifier="test_cluster", + aws_conn_id="aws_conn_test", + wait_for_completion=False, + ) + redshift_operator._attempts = 3 + with pytest.raises(returned_exception): + redshift_operator.execute(None) + + assert mock_delete_cluster.call_count == 3 + + @mock.patch.object(RedshiftHook, "delete_cluster") + @mock.patch.object(RedshiftHook, "conn") + @mock.patch("time.sleep", return_value=None) + def test_delete_cluster_succeeds_on_second_attempt(self, _, mock_conn, mock_delete_cluster): + exception = boto3.client("redshift").exceptions.InvalidClusterStateFault({}, "test") + returned_exception = type(exception) + mock_conn.exceptions.InvalidClusterStateFault = returned_exception + mock_delete_cluster.side_effect = [exception, True] + + redshift_operator = RedshiftDeleteClusterOperator( + task_id="task_test", + cluster_identifier="test_cluster", + aws_conn_id="aws_conn_test", + wait_for_completion=False, + ) + redshift_operator._attempt_interval = 0 + redshift_operator.execute(None) + + assert mock_delete_cluster.call_count == 2 @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.cluster_status") @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.delete_cluster") def test_delete_cluster_deferrable_mode(self, mock_delete_cluster, mock_cluster_status): - """Test delete cluster operator with defer when deferrable param is true""" + """When the delete is accepted, deferrable mode waits for deletion to complete.""" mock_delete_cluster.return_value = True - mock_cluster_status.return_value = "available" + mock_cluster_status.return_value = "deleting" delete_cluster = RedshiftDeleteClusterOperator( task_id="task_test", cluster_identifier="test_cluster", @@ -806,16 +858,82 @@ def test_delete_cluster_deferrable_mode(self, mock_delete_cluster, mock_cluster_ assert isinstance(exc.value.trigger, RedshiftDeleteClusterTrigger), ( "Trigger is not a RedshiftDeleteClusterTrigger" ) + # Delete is attempted exactly once (no synchronous busy-retry loop in deferrable mode). + mock_delete_cluster.assert_called_once() - @mock.patch("airflow.providers.amazon.aws.operators.redshift_cluster.RedshiftDeleteClusterOperator.defer") @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.cluster_status") @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.delete_cluster") - def test_delete_cluster_deferrable_mode_in_paused_state( - self, mock_delete_cluster, mock_cluster_status, mock_defer + def test_delete_cluster_deferrable_mode_already_gone(self, mock_delete_cluster, mock_cluster_status): + """When the cluster is already gone after the delete, deferrable mode completes without deferring.""" + mock_delete_cluster.return_value = True + mock_cluster_status.return_value = "cluster_not_found" + delete_cluster = RedshiftDeleteClusterOperator( + task_id="task_test", + cluster_identifier="test_cluster", + deferrable=True, + wait_for_completion=False, + ) + + # No TaskDeferred is raised; the operator returns normally. + delete_cluster.execute(context=None) + mock_delete_cluster.assert_called_once() + + @mock.patch.object(RedshiftHook, "conn") + @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.delete_cluster") + def test_delete_cluster_deferrable_mode_busy_defers_to_settle_trigger( + self, mock_delete_cluster, mock_conn ): - """Test delete cluster operator with defer when deferrable param is true""" + """A busy cluster (InvalidClusterStateFault) defers to the settle-wait trigger, not a sync loop.""" + exception = boto3.client("redshift").exceptions.InvalidClusterStateFault({}, "test") + mock_conn.exceptions.InvalidClusterStateFault = type(exception) + mock_delete_cluster.side_effect = exception + + delete_cluster = RedshiftDeleteClusterOperator( + task_id="task_test", + cluster_identifier="test_cluster", + deferrable=True, + wait_for_completion=False, + ) + + with pytest.raises(TaskDeferred) as exc: + delete_cluster.execute(context=None) + + assert isinstance(exc.value.trigger, RedshiftClusterSettledTrigger), ( + "Trigger is not a RedshiftClusterSettledTrigger" + ) + assert exc.value.method_name == "_retry_delete_when_settled" + # Delete attempted once; the synchronous busy-retry loop never runs in deferrable mode. + mock_delete_cluster.assert_called_once() + + @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.cluster_status") + @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.delete_cluster") + def test_retry_delete_when_settled_reissues_delete(self, mock_delete_cluster, mock_cluster_status): + """The settle-wait callback re-issues the delete and defers to the delete-complete trigger.""" mock_delete_cluster.return_value = True - mock_cluster_status.return_value = "creating" + mock_cluster_status.return_value = "deleting" + delete_cluster = RedshiftDeleteClusterOperator( + task_id="task_test", + cluster_identifier="test_cluster", + deferrable=True, + wait_for_completion=False, + ) + + with pytest.raises(TaskDeferred) as exc: + delete_cluster._retry_delete_when_settled( + context=None, event={"status": "success", "message": "Cluster settled"} + ) + + assert isinstance(exc.value.trigger, RedshiftDeleteClusterTrigger) + mock_delete_cluster.assert_called_once() + + @mock.patch.object(RedshiftHook, "conn") + @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.delete_cluster") + def test_retry_delete_when_settled_redefers_on_race(self, mock_delete_cluster, mock_conn): + """If the cluster re-enters a transitional state (race), the callback re-defers to settle-wait.""" + exception = boto3.client("redshift").exceptions.InvalidClusterStateFault({}, "test") + mock_conn.exceptions.InvalidClusterStateFault = type(exception) + mock_delete_cluster.side_effect = exception + delete_cluster = RedshiftDeleteClusterOperator( task_id="task_test", cluster_identifier="test_cluster", @@ -823,10 +941,50 @@ def test_delete_cluster_deferrable_mode_in_paused_state( wait_for_completion=False, ) + with pytest.raises(TaskDeferred) as exc: + delete_cluster._retry_delete_when_settled( + context=None, event={"status": "success", "message": "Cluster settled"} + ) + + assert isinstance(exc.value.trigger, RedshiftClusterSettledTrigger) + assert exc.value.method_name == "_retry_delete_when_settled" + + def test_retry_delete_when_settled_error_event_raises(self): + """A non-success event from the settle-wait trigger raises AirflowException.""" + delete_cluster = RedshiftDeleteClusterOperator( + task_id="task_test", + cluster_identifier="test_cluster", + deferrable=True, + wait_for_completion=False, + ) with pytest.raises(AirflowException): - delete_cluster.execute(context=None) + delete_cluster._retry_delete_when_settled( + context=None, event={"status": "error", "message": "timed out"} + ) - assert not mock_defer.called + @mock.patch.object(RedshiftHook, "delete_cluster") + @mock.patch.object(RedshiftHook, "conn") + @mock.patch("time.sleep", return_value=None) + def test_delete_cluster_sync_mode_still_uses_busy_retry_loop( + self, mock_sleep, mock_conn, mock_delete_cluster + ): + """Sync mode (deferrable=False) must keep the synchronous busy-retry loop unchanged.""" + exception = boto3.client("redshift").exceptions.InvalidClusterStateFault({}, "test") + mock_conn.exceptions.InvalidClusterStateFault = type(exception) + mock_delete_cluster.side_effect = [exception, exception, True] + + redshift_operator = RedshiftDeleteClusterOperator( + task_id="task_test", + cluster_identifier="test_cluster", + aws_conn_id="aws_conn_test", + deferrable=False, + wait_for_completion=False, + ) + redshift_operator.execute(None) + + # Three synchronous attempts, and time.sleep was used between the retries. + assert mock_delete_cluster.call_count == 3 + assert mock_sleep.called def test_delete_cluster_execute_complete_success(self): """Asserts that logging occurs as expected""" diff --git a/providers/amazon/tests/unit/amazon/aws/triggers/test_redshift_cluster.py b/providers/amazon/tests/unit/amazon/aws/triggers/test_redshift_cluster.py index a55494ce8ab7c..42f0b5f1f805d 100644 --- a/providers/amazon/tests/unit/amazon/aws/triggers/test_redshift_cluster.py +++ b/providers/amazon/tests/unit/amazon/aws/triggers/test_redshift_cluster.py @@ -23,6 +23,7 @@ import pytest from airflow.providers.amazon.aws.triggers.redshift_cluster import ( + RedshiftClusterSettledTrigger, RedshiftClusterTrigger, RedshiftCreateClusterSnapshotTrigger, RedshiftCreateClusterTrigger, @@ -152,43 +153,64 @@ async def test_redshift_cluster_sensor_trigger_exception(self, mock_cluster_stat RedshiftCreateClusterTrigger, 15, 999999, + "cluster_available", id="RedshiftCreateClusterTrigger", ), pytest.param( RedshiftPauseClusterTrigger, 15, 999999, + "cluster_paused", id="RedshiftPauseClusterTrigger", ), pytest.param( RedshiftCreateClusterSnapshotTrigger, 15, 999999, + "snapshot_available", id="RedshiftCreateClusterSnapshotTrigger", ), pytest.param( RedshiftResumeClusterTrigger, 15, 999999, + "cluster_resumed", id="RedshiftResumeClusterTrigger", ), pytest.param( RedshiftDeleteClusterTrigger, 30, 30, + "cluster_deleted", id="RedshiftDeleteClusterTrigger", ), + pytest.param( + RedshiftClusterSettledTrigger, + 30, + 30, + "cluster_deletable", + id="RedshiftClusterSettledTrigger", + ), ] class TestRedshiftWaiterTriggers: - """Tests for the five Redshift triggers that inherit from ``AwsBaseWaiterTrigger``.""" + """Tests for the Redshift triggers that inherit from ``AwsBaseWaiterTrigger``.""" + + @pytest.mark.parametrize( + ("trigger_cls", "default_delay", "default_max_attempts", "waiter_name"), + WAITER_TRIGGER_PARAMS, + ) + def test_waiter_name(self, trigger_cls, default_delay, default_max_attempts, waiter_name): + trigger = trigger_cls(cluster_identifier="test_cluster") + assert trigger.waiter_name == waiter_name + assert trigger.waiter_args == {"ClusterIdentifier": "test_cluster"} @pytest.mark.parametrize( - ("trigger_cls", "default_delay", "default_max_attempts"), + ("trigger_cls", "default_delay", "default_max_attempts", "waiter_name"), WAITER_TRIGGER_PARAMS, ) - def test_serialization(self, trigger_cls, default_delay, default_max_attempts): + def test_serialization(self, trigger_cls, default_delay, default_max_attempts, waiter_name): trigger = trigger_cls( cluster_identifier="test_cluster", aws_conn_id="aws_default", @@ -206,11 +228,11 @@ def test_serialization(self, trigger_cls, default_delay, default_max_attempts): } @pytest.mark.parametrize( - ("trigger_cls", "default_delay", "default_max_attempts"), + ("trigger_cls", "default_delay", "default_max_attempts", "waiter_name"), WAITER_TRIGGER_PARAMS, ) def test_serialization_with_verify_and_botocore_config( - self, trigger_cls, default_delay, default_max_attempts + self, trigger_cls, default_delay, default_max_attempts, waiter_name ): trigger = trigger_cls( cluster_identifier="test_cluster", @@ -225,12 +247,12 @@ def test_serialization_with_verify_and_botocore_config( assert "region_name" not in kwargs @pytest.mark.parametrize( - ("trigger_cls", "default_delay", "default_max_attempts"), + ("trigger_cls", "default_delay", "default_max_attempts", "waiter_name"), WAITER_TRIGGER_PARAMS, ) @mock.patch("airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftHook") def test_hook_propagates_verify_and_botocore_config( - self, mock_hook_cls, trigger_cls, default_delay, default_max_attempts + self, mock_hook_cls, trigger_cls, default_delay, default_max_attempts, waiter_name ): trigger = trigger_cls( cluster_identifier="test_cluster",