Fix RedshiftResumeClusterOperator deferrable implementation

airflow/providers/amazon/aws/operators/redshift_cluster.py

@@ -440,7 +440,7 @@ def execute(self, context: Context):
                     aws_conn_id=self.aws_conn_id,
                     cluster_identifier=self.cluster_identifier,
                     attempts=self._attempts,
-                    operation_type="pause_cluster",
+                    operation_type="resume_cluster",
                 ),
                 method_name="execute_complete",
             )
@@ -470,7 +470,7 @@ def execute_complete(self, context: Context, event: Any = None) -> None:
                 raise AirflowException(msg)
             elif "status" in event and event["status"] == "success":
                 self.log.info("%s completed successfully.", self.task_id)
-                self.log.info("Paused cluster successfully")
+                self.log.info("Resumed cluster successfully")
         else:
             raise AirflowException("No event received from trigger")
 

airflow/providers/amazon/aws/triggers/redshift_cluster.py

@@ -70,6 +70,16 @@ async def run(self) -> AsyncIterator["TriggerEvent"]:
                     else:
                         if self.attempts < 1:
                             yield TriggerEvent({"status": "error", "message": f"{self.task_id} failed"})
+                elif self.operation_type == "resume_cluster":
+                    response = await hook.resume_cluster(
+                        cluster_identifier=self.cluster_identifier,
+                        polling_period_seconds=self.poll_interval,
+                    )
+                    if response:
+                        yield TriggerEvent(response)
+                    else:
+                        error_message = f"{self.task_id} failed"
+                        yield TriggerEvent({"status": "error", "message": error_message})
                 else:
                     yield TriggerEvent(f"{self.operation_type} is not supported")
             except Exception as e:

tests/providers/amazon/aws/deferrable/hooks/test_redshift_cluster.py

@@ -82,3 +82,44 @@ async def test_get_cluster_status_exception(self, cluster_status):
             "message": "An error occurred (SomeServiceException) when calling the "
             "redshift operation: Details/context around the exception or error",
         }
+
+    @pytest.mark.asyncio
+    @async_mock.patch("aiobotocore.client.AioBaseClient._make_api_call")
+    async def test_resume_cluster(self, mock_make_api_call):
+        """Test Resume cluster async hook function by mocking return value of resume_cluster"""
+
+        hook = RedshiftAsyncHook(aws_conn_id="test_aws_connection_id")
+        await hook.resume_cluster(cluster_identifier="redshift_cluster_1")
+        mock_make_api_call.assert_called_once_with(
+            "ResumeCluster", {"ClusterIdentifier": "redshift_cluster_1"}
+        )
+
+    @pytest.mark.asyncio
+    @async_mock.patch(
+        "airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftAsyncHook.get_client_async"
+    )
+    async def test_resume_cluster_exception(self, mock_client):
+        """Test Resume cluster async hook function with exception by mocking return value of resume_cluster"""
+        mock_client.return_value.__aenter__.return_value.resume_cluster.side_effect = ClientError(
+            {
+                "Error": {
+                    "Code": "SomeServiceException",
+                    "Message": "Details/context around the exception or error",
+                },
+                "ResponseMetadata": {
+                    "RequestId": "1234567890ABCDEF",
+                    "HostId": "host ID data will appear here as a hash",
+                    "HTTPStatusCode": 500,
+                    "HTTPHeaders": {"header metadata key/values will appear here"},
+                    "RetryAttempts": 0,
+                },
+            },
+            operation_name="redshift",
+        )
+        hook = RedshiftAsyncHook(aws_conn_id="test_aws_connection_id")
+        result = await hook.resume_cluster(cluster_identifier="test")
+        assert result == {
+            "status": "error",
+            "message": "An error occurred (SomeServiceException) when calling the "
+            "redshift operation: Details/context around the exception or error",
+        }

tests/providers/amazon/aws/deferrable/triggers/test_redshift_cluster.py

@@ -90,3 +90,78 @@ async def test_pause_trigger_failure(self, mock_pause_cluster):
         )
         task = [i async for i in trigger.run()]
         assert TriggerEvent({"status": "error", "message": "Test exception"}) in task
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "operation_type,return_value,response",
+        [
+            (
+                "resume_cluster",
+                {"status": "error", "message": "test error"},
+                TriggerEvent({"status": "error", "message": "test error"}),
+            ),
+        ],
+    )
+    @async_mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftAsyncHook.resume_cluster")
+    async def test_resume_trigger_run_error(
+        self, mock_resume_cluster, operation_type, return_value, response
+    ):
+        """Test RedshiftClusterTrigger resume cluster with success"""
+        mock_resume_cluster.return_value = return_value
+        trigger = RedshiftClusterTrigger(
+            task_id=TASK_ID,
+            poll_interval=POLLING_PERIOD_SECONDS,
+            aws_conn_id="test_redshift_conn_id",
+            cluster_identifier="mock_cluster_identifier",
+            operation_type=operation_type,
+            attempts=1,
+        )
+        generator = trigger.run()
+        actual = await generator.asend(None)
+        assert response == actual
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "operation_type,return_value,response",
+        [
+            (
+                "resume_cluster",
+                {"status": "success", "cluster_state": "available"},
+                TriggerEvent({"status": "success", "cluster_state": "available"}),
+            ),
+        ],
+    )
+    @async_mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftAsyncHook.resume_cluster")
+    async def test_resume_trigger_run_success(
+        self, mock_resume_cluster, operation_type, return_value, response
+    ):
+        """Test RedshiftClusterTrigger resume cluster with success"""
+        mock_resume_cluster.return_value = return_value
+        trigger = RedshiftClusterTrigger(
+            task_id=TASK_ID,
+            poll_interval=POLLING_PERIOD_SECONDS,
+            aws_conn_id="test_redshift_conn_id",
+            cluster_identifier="mock_cluster_identifier",
+            operation_type=operation_type,
+            attempts=1,
+        )
+        generator = trigger.run()
+        actual = await generator.asend(None)
+        assert response == actual
+
+    @pytest.mark.asyncio
+    @async_mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftAsyncHook.resume_cluster")
+    async def test_resume_trigger_failure(self, mock_resume_cluster):
+        """Test RedshiftClusterTrigger resume cluster with failure status"""
+        mock_resume_cluster.side_effect = Exception("Test exception")
+        trigger = RedshiftClusterTrigger(
+            task_id=TASK_ID,
+            poll_interval=POLLING_PERIOD_SECONDS,
+            aws_conn_id="test_redshift_conn_id",
+            cluster_identifier="mock_cluster_identifier",
+            operation_type="resume_cluster",
+            attempts=1,
+        )
+        task = [i async for i in trigger.run()]
+        assert len(task) == 1
+        assert TriggerEvent({"status": "error", "message": "Test exception"}) in task