Add remaining resource provider migrations (localstack#10368)

localstack/services/cloudformation/resource_providers/aws_cloudformation_macro.py

@@ -0,0 +1,102 @@
+# LocalStack Resource Provider Scaffolding v2
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Optional, TypedDict
+
+import localstack.services.cloudformation.provider_utils as util
+from localstack.services.cloudformation.resource_provider import (
+    OperationStatus,
+    ProgressEvent,
+    ResourceProvider,
+    ResourceRequest,
+)
+from localstack.services.cloudformation.stores import get_cloudformation_store
+
+
+class CloudFormationMacroProperties(TypedDict):
+    FunctionName: Optional[str]
+    Name: Optional[str]
+    Description: Optional[str]
+    Id: Optional[str]
+    LogGroupName: Optional[str]
+    LogRoleARN: Optional[str]
+
+
+REPEATED_INVOCATION = "repeated_invocation"
+
+
+class CloudFormationMacroProvider(ResourceProvider[CloudFormationMacroProperties]):
+    TYPE = "AWS::CloudFormation::Macro"  # Autogenerated. Don't change
+    SCHEMA = util.get_schema_path(Path(__file__))  # Autogenerated. Don't change
+
+    def create(
+        self,
+        request: ResourceRequest[CloudFormationMacroProperties],
+    ) -> ProgressEvent[CloudFormationMacroProperties]:
+        """
+        Create a new resource.
+
+        Primary identifier fields:
+          - /properties/Id
+
+        Required properties:
+          - FunctionName
+          - Name
+
+        Create-only properties:
+          - /properties/Name
+
+        Read-only properties:
+          - /properties/Id
+
+
+
+        """
+        model = request.desired_state
+
+        # TODO: fix or validate that we want to keep this here.
+        #  AWS::CloudFormation:: resources need special handling since they seem to require access to internal APIs
+        store = get_cloudformation_store(request.account_id, request.region_name)
+        store.macros[model["Name"]] = model
+        model["Id"] = model["Name"]
+
+        return ProgressEvent(status=OperationStatus.SUCCESS, resource_model=model)
+
+    def read(
+        self,
+        request: ResourceRequest[CloudFormationMacroProperties],
+    ) -> ProgressEvent[CloudFormationMacroProperties]:
+        """
+        Fetch resource information
+
+
+        """
+        raise NotImplementedError
+
+    def delete(
+        self,
+        request: ResourceRequest[CloudFormationMacroProperties],
+    ) -> ProgressEvent[CloudFormationMacroProperties]:
+        """
+        Delete a resource
+
+
+        """
+        model = request.desired_state
+
+        store = get_cloudformation_store(request.account_id, request.region_name)
+        store.macros.pop(model["Name"], None)
+
+        return ProgressEvent(status=OperationStatus.SUCCESS, resource_model=model)
+
+    def update(
+        self,
+        request: ResourceRequest[CloudFormationMacroProperties],
+    ) -> ProgressEvent[CloudFormationMacroProperties]:
+        """
+        Update a resource
+
+
+        """
+        raise NotImplementedError

localstack/services/cloudformation/resource_providers/aws_cloudformation_macro.schema.json

@@ -0,0 +1,38 @@
+{
+  "typeName": "AWS::CloudFormation::Macro",
+  "description": "Resource Type definition for AWS::CloudFormation::Macro",
+  "additionalProperties": false,
+  "properties": {
+    "Id": {
+      "type": "string"
+    },
+    "Description": {
+      "type": "string"
+    },
+    "FunctionName": {
+      "type": "string"
+    },
+    "LogGroupName": {
+      "type": "string"
+    },
+    "LogRoleARN": {
+      "type": "string"
+    },
+    "Name": {
+      "type": "string"
+    }
+  },
+  "required": [
+    "FunctionName",
+    "Name"
+  ],
+  "createOnlyProperties": [
+    "/properties/Name"
+  ],
+  "primaryIdentifier": [
+    "/properties/Id"
+  ],
+  "readOnlyProperties": [
+    "/properties/Id"
+  ]
+}

localstack/services/cloudformation/resource_providers/aws_cloudformation_macro_plugin.py

@@ -0,0 +1,20 @@
+from typing import Optional, Type
+
+from localstack.services.cloudformation.resource_provider import (
+    CloudFormationResourceProviderPlugin,
+    ResourceProvider,
+)
+
+
+class CloudFormationMacroProviderPlugin(CloudFormationResourceProviderPlugin):
+    name = "AWS::CloudFormation::Macro"
+
+    def __init__(self):
+        self.factory: Optional[Type[ResourceProvider]] = None
+
+    def load(self):
+        from localstack.services.cloudformation.resource_providers.aws_cloudformation_macro import (
+            CloudFormationMacroProvider,
+        )
+
+        self.factory = CloudFormationMacroProvider

localstack/services/cloudformation/resource_providers/aws_cloudformation_waitcondition.py

@@ -0,0 +1,83 @@
+# LocalStack Resource Provider Scaffolding v2
+from __future__ import annotations
+
+import uuid
+from pathlib import Path
+from typing import Optional, TypedDict
+
+import localstack.services.cloudformation.provider_utils as util
+from localstack.services.cloudformation.resource_provider import (
+    OperationStatus,
+    ProgressEvent,
+    ResourceProvider,
+    ResourceRequest,
+)
+
+
+class CloudFormationWaitConditionProperties(TypedDict):
+    Count: Optional[int]
+    Data: Optional[dict]
+    Handle: Optional[str]
+    Id: Optional[str]
+    Timeout: Optional[str]
+
+
+REPEATED_INVOCATION = "repeated_invocation"
+
+
+class CloudFormationWaitConditionProvider(ResourceProvider[CloudFormationWaitConditionProperties]):
+    TYPE = "AWS::CloudFormation::WaitCondition"  # Autogenerated. Don't change
+    SCHEMA = util.get_schema_path(Path(__file__))  # Autogenerated. Don't change
+
+    def create(
+        self,
+        request: ResourceRequest[CloudFormationWaitConditionProperties],
+    ) -> ProgressEvent[CloudFormationWaitConditionProperties]:
+        """
+        Create a new resource.
+
+        Primary identifier fields:
+          - /properties/Id
+
+        Read-only properties:
+          - /properties/Data
+          - /properties/Id
+
+        """
+        model = request.desired_state
+        model["Data"] = {}  # TODO
+        model["Id"] = f"{request.stack_id}/{uuid.uuid4()}/{request.logical_resource_id}"
+        return ProgressEvent(status=OperationStatus.SUCCESS, resource_model=model)
+
+    def read(
+        self,
+        request: ResourceRequest[CloudFormationWaitConditionProperties],
+    ) -> ProgressEvent[CloudFormationWaitConditionProperties]:
+        """
+        Fetch resource information
+
+
+        """
+        raise NotImplementedError
+
+    def delete(
+        self,
+        request: ResourceRequest[CloudFormationWaitConditionProperties],
+    ) -> ProgressEvent[CloudFormationWaitConditionProperties]:
+        """
+        Delete a resource
+
+
+        """
+        return ProgressEvent(status=OperationStatus.SUCCESS, resource_model={})  # NO-OP
+
+    def update(
+        self,
+        request: ResourceRequest[CloudFormationWaitConditionProperties],
+    ) -> ProgressEvent[CloudFormationWaitConditionProperties]:
+        """
+        Update a resource
+
+
+        """
+        raise NotImplementedError

localstack/services/cloudformation/resource_providers/aws_cloudformation_waitcondition.schema.json

@@ -0,0 +1,29 @@
+{
+  "typeName": "AWS::CloudFormation::WaitCondition",
+  "description": "Resource Type definition for AWS::CloudFormation::WaitCondition",
+  "additionalProperties": false,
+  "properties": {
+    "Id": {
+      "type": "string"
+    },
+    "Data": {
+      "type": "object"
+    },
+    "Count": {
+      "type": "integer"
+    },
+    "Handle": {
+      "type": "string"
+    },
+    "Timeout": {
+      "type": "string"
+    }
+  },
+  "readOnlyProperties": [
+    "/properties/Data",
+    "/properties/Id"
+  ],
+  "primaryIdentifier": [
+    "/properties/Id"
+  ]
+}

localstack/services/cloudformation/resource_providers/aws_cloudformation_waitcondition_plugin.py

@@ -0,0 +1,20 @@
+from typing import Optional, Type
+
+from localstack.services.cloudformation.resource_provider import (
+    CloudFormationResourceProviderPlugin,
+    ResourceProvider,
+)
+
+
+class CloudFormationWaitConditionProviderPlugin(CloudFormationResourceProviderPlugin):
+    name = "AWS::CloudFormation::WaitCondition"
+
+    def __init__(self):
+        self.factory: Optional[Type[ResourceProvider]] = None
+
+    def load(self):
+        from localstack.services.cloudformation.resource_providers.aws_cloudformation_waitcondition import (
+            CloudFormationWaitConditionProvider,
+        )
+
+        self.factory = CloudFormationWaitConditionProvider

localstack/services/cloudformation/resource_providers/aws_cloudformation_waitconditionhandle.py

@@ -0,0 +1,94 @@
+# LocalStack Resource Provider Scaffolding v2
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Optional, TypedDict
+
+import localstack.services.cloudformation.provider_utils as util
+from localstack.services.cloudformation.resource_provider import (
+    OperationStatus,
+    ProgressEvent,
+    ResourceProvider,
+    ResourceRequest,
+)
+
+
+class CloudFormationWaitConditionHandleProperties(TypedDict):
+    Id: Optional[str]
+
+
+REPEATED_INVOCATION = "repeated_invocation"
+
+
+class CloudFormationWaitConditionHandleProvider(
+    ResourceProvider[CloudFormationWaitConditionHandleProperties]
+):
+    TYPE = "AWS::CloudFormation::WaitConditionHandle"  # Autogenerated. Don't change
+    SCHEMA = util.get_schema_path(Path(__file__))  # Autogenerated. Don't change
+
+    def create(
+        self,
+        request: ResourceRequest[CloudFormationWaitConditionHandleProperties],
+    ) -> ProgressEvent[CloudFormationWaitConditionHandleProperties]:
+        """
+        Create a new resource.
+
+        Primary identifier fields:
+          - /properties/Id
+
+
+
+
+
+        Read-only properties:
+          - /properties/Id
+
+
+
+        """
+        # TODO: properly test this and fix s3 bucket usage
+        model = request.desired_state
+
+        s3 = request.aws_client_factory.s3
+        region = s3.meta.region_name
+
+        bucket = f"cloudformation-waitcondition-{region}"
+        waitcondition_url = s3.generate_presigned_url(
+            "put_object", Params={"Bucket": bucket, "Key": request.stack_id}
+        )
+        model["Id"] = waitcondition_url
+
+        return ProgressEvent(status=OperationStatus.SUCCESS, resource_model=model)
+
+    def read(
+        self,
+        request: ResourceRequest[CloudFormationWaitConditionHandleProperties],
+    ) -> ProgressEvent[CloudFormationWaitConditionHandleProperties]:
+        """
+        Fetch resource information
+
+
+        """
+        raise NotImplementedError
+
+    def delete(
+        self,
+        request: ResourceRequest[CloudFormationWaitConditionHandleProperties],
+    ) -> ProgressEvent[CloudFormationWaitConditionHandleProperties]:
+        """
+        Delete a resource
+
+
+        """
+        return ProgressEvent(status=OperationStatus.SUCCESS, resource_model={})
+
+    def update(
+        self,
+        request: ResourceRequest[CloudFormationWaitConditionHandleProperties],
+    ) -> ProgressEvent[CloudFormationWaitConditionHandleProperties]:
+        """
+        Update a resource
+
+
+        """
+        raise NotImplementedError