Add resource provider for AWS::SQS::QueuePolicy

localstack/services/cloudformation/resource_provider.py

@@ -43,7 +43,8 @@
 # by default we use the GenericBaseModel (the legacy model), unless the resource is listed below
 # add your new provider here when you want it to be the default
 PROVIDER_DEFAULTS = {
-    "AWS::SQS::Queue": "ResourceProvider"
+    "AWS::SQS::Queue": "ResourceProvider",
+    "AWS::SQS::QueuePolicy": "ResourceProvider",
     # "AWS::IAM::User": "ResourceProvider",
     # "AWS::SSM::Parameter": "GenericBaseModel",
     # "AWS::OpenSearchService::Domain": "GenericBaseModel",

localstack/services/sqs/resource_providers/aws_sqs_queuepolicy.py

@@ -0,0 +1,114 @@
+# LocalStack Resource Provider Scaffolding v2
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Optional, Type, TypedDict
+
+import localstack.services.cloudformation.provider_utils as util
+from localstack.services.cloudformation.resource_provider import (
+    CloudFormationResourceProviderPlugin,
+    OperationStatus,
+    ProgressEvent,
+    ResourceProvider,
+    ResourceRequest,
+)
+
+
+class SQSQueuePolicyProperties(TypedDict):
+    PolicyDocument: Optional[dict]
+    Queues: Optional[list[str]]
+    Id: Optional[str]
+
+
+REPEATED_INVOCATION = "repeated_invocation"
+
+
+class SQSQueuePolicyProvider(ResourceProvider[SQSQueuePolicyProperties]):
+
+    TYPE = "AWS::SQS::QueuePolicy"  # Autogenerated. Don't change
+    SCHEMA = util.get_schema_path(Path(__file__))  # Autogenerated. Don't change
+
+    def create(
+        self,
+        request: ResourceRequest[SQSQueuePolicyProperties],
+    ) -> ProgressEvent[SQSQueuePolicyProperties]:
+        """
+        Create a new resource.
+
+        Primary identifier fields:
+          - /properties/Id
+
+        Required properties:
+          - PolicyDocument
+          - Queues
+
+        Read-only properties:
+          - /properties/Id
+
+        """
+        model = request.desired_state
+        sqs = request.aws_client_factory.sqs
+        for queue in model.get("Queues", []):
+            policy = json.dumps(model["PolicyDocument"])
+            sqs.set_queue_attributes(QueueUrl=queue, Attributes={"Policy": policy})
+
+        physical_resource_id = util.generate_default_name(
+            stack_name=request.stack_name, logical_resource_id=request.logical_resource_id
+        )
+        model["Id"] = physical_resource_id
+
+        return ProgressEvent(
+            status=OperationStatus.SUCCESS,
+            resource_model=model,
+            custom_context=request.custom_context,
+        )
+
+    def read(
+        self,
+        request: ResourceRequest[SQSQueuePolicyProperties],
+    ) -> ProgressEvent[SQSQueuePolicyProperties]:
+        """
+        Fetch resource information
+        """
+        raise NotImplementedError
+
+    def delete(
+        self,
+        request: ResourceRequest[SQSQueuePolicyProperties],
+    ) -> ProgressEvent[SQSQueuePolicyProperties]:
+        """
+        Delete a resource
+        """
+        sqs = request.aws_client_factory.sqs
+        for queue in request.previous_state["Queues"]:
+            try:
+                sqs.set_queue_attributes(QueueUrl=queue, Attributes={"Policy": ""})
+
+            except sqs.exceptions.ClientError as err:
+                if "AWS.SimpleQueueService.NonExistentQueue" != err.response["Error"]["Code"]:
+                    return ProgressEvent(status=OperationStatus.FAILED, resource_model={})
+
+        return ProgressEvent(
+            status=OperationStatus.SUCCESS,
+            resource_model={},
+        )
+
+    def update(
+        self,
+        request: ResourceRequest[SQSQueuePolicyProperties],
+    ) -> ProgressEvent[SQSQueuePolicyProperties]:
+        """
+        Update a resource
+        """
+        raise NotImplementedError
+
+
+class SQSQueuePolicyProviderPlugin(CloudFormationResourceProviderPlugin):
+    name = "AWS::SQS::QueuePolicy"
+
+    def __init__(self):
+        self.factory: Optional[Type[ResourceProvider]] = None
+
+    def load(self):
+        self.factory = SQSQueuePolicyProvider

localstack/services/sqs/resource_providers/aws_sqs_queuepolicy.schema.json

@@ -0,0 +1,30 @@
+{
+  "typeName": "AWS::SQS::QueuePolicy",
+  "description": "Resource Type definition for AWS::SQS::QueuePolicy",
+  "additionalProperties": false,
+  "properties": {
+    "Id": {
+      "type": "string"
+    },
+    "PolicyDocument": {
+      "type": "object"
+    },
+    "Queues": {
+      "type": "array",
+      "uniqueItems": false,
+      "items": {
+        "type": "string"
+      }
+    }
+  },
+  "required": [
+    "PolicyDocument",
+    "Queues"
+  ],
+  "primaryIdentifier": [
+    "/properties/Id"
+  ],
+  "readOnlyProperties": [
+    "/properties/Id"
+  ]
+}