18505123992

Committed 14 Oct 2025 05:30PM UTC coverage: 86.888% (-0.01%) from 86.899%

Build # 18505123992

Build Type

push

github

Committed by

web-flow

Commit Message

S3: fix `aws-global` validation in CreateBucket (#13250)

Coverage Stats

10 of 10 new or added lines in 4 files covered. (100.0%)

831 existing lines in 40 files now uncovered.

68028 of 78294 relevant lines covered (86.89%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.35

/localstack-core/localstack/services/cloudformation/resource_providers/aws_cloudformation_stack.py

# LocalStack Resource Provider Scaffolding v2
from __future__ import annotations

from pathlib import Path
from typing import TypedDict

import localstack.services.cloudformation.provider_utils as util
from localstack.services.cloudformation.resource_provider import (
    OperationStatus,
    ProgressEvent,
    ResourceProvider,
    ResourceRequest,
)


class CloudFormationStackProperties(TypedDict):
    TemplateURL: str | None
    Id: str | None
    NotificationARNs: list[str] | None
    Parameters: dict | None
    Tags: list[Tag] | None
    TimeoutInMinutes: int | None


class Tag(TypedDict):
    Key: str | None
    Value: str | None


REPEATED_INVOCATION = "repeated_invocation"


class CloudFormationStackProvider(ResourceProvider[CloudFormationStackProperties]):
    TYPE = "AWS::CloudFormation::Stack"  # Autogenerated. Don't change
    SCHEMA = util.get_schema_path(Path(__file__))  # Autogenerated. Don't change

    def create(
        self,
        request: ResourceRequest[CloudFormationStackProperties],
    ) -> ProgressEvent[CloudFormationStackProperties]:
        """
        Create a new resource.

        Primary identifier fields:
          - /properties/Id

        Required properties:
          - TemplateURL



        Read-only properties:
          - /properties/Id



        """
        model = request.desired_state

        # TODO: validations

        if not request.custom_context.get(REPEATED_INVOCATION):
            if not model.get("StackName"):
                model["StackName"] = util.generate_default_name(
                    request.stack_name, request.logical_resource_id
                )

            create_params = util.select_attributes(
                model,
                [
                    "StackName",
                    "Parameters",
                    "NotificationARNs",
                    "TemplateURL",
                    "TimeoutInMinutes",
                    "Tags",
                ],
            )

            create_params["Capabilities"] = [
                "CAPABILITY_IAM",
                "CAPABILITY_NAMED_IAM",
                "CAPABILITY_AUTO_EXPAND",
            ]

            create_params["Parameters"] = [
                {
                    "ParameterKey": k,
                    "ParameterValue": str(v).lower() if isinstance(v, bool) else str(v),
                }
                for k, v in create_params.get("Parameters", {}).items()
            ]

            result = request.aws_client_factory.cloudformation.create_stack(**create_params)
            model["Id"] = result["StackId"]

            request.custom_context[REPEATED_INVOCATION] = True
            return ProgressEvent(
                status=OperationStatus.IN_PROGRESS,
                resource_model=model,
                custom_context=request.custom_context,
            )

        stack = request.aws_client_factory.cloudformation.describe_stacks(StackName=model["Id"])[
            "Stacks"
        ][0]
        match stack["StackStatus"]:
            case "CREATE_COMPLETE":
                # only store nested stack outputs when we know the deploy has completed
                model["Outputs"] = {
                    o["OutputKey"]: o["OutputValue"] for o in stack.get("Outputs", [])
                }
                return ProgressEvent(
                    status=OperationStatus.SUCCESS,
                    resource_model=model,
                    custom_context=request.custom_context,
                )
            case "CREATE_IN_PROGRESS":
                return ProgressEvent(
                    status=OperationStatus.IN_PROGRESS,
                    resource_model=model,
                    custom_context=request.custom_context,
                )
            case "CREATE_FAILED":
                return ProgressEvent(
                    status=OperationStatus.FAILED,
                    resource_model=model,
                    custom_context=request.custom_context,
                )
            case _:
                raise Exception(f"Unexpected status: {stack['StackStatus']}")

    def read(
        self,
        request: ResourceRequest[CloudFormationStackProperties],
    ) -> ProgressEvent[CloudFormationStackProperties]:
        """
        Fetch resource information


        """
        raise NotImplementedError

    def delete(
        self,
        request: ResourceRequest[CloudFormationStackProperties],
    ) -> ProgressEvent[CloudFormationStackProperties]:
        """
        Delete a resource
        """

        model = request.desired_state
        if not request.custom_context.get(REPEATED_INVOCATION):
            request.aws_client_factory.cloudformation.delete_stack(StackName=model["Id"])

            request.custom_context[REPEATED_INVOCATION] = True
            return ProgressEvent(
                status=OperationStatus.IN_PROGRESS,
                resource_model=model,
                custom_context=request.custom_context,
            )

        try:
            stack = request.aws_client_factory.cloudformation.describe_stacks(
                StackName=model["Id"]
            )["Stacks"][0]
        except Exception as e:
            if "does not exist" in str(e):
                return ProgressEvent(
                    status=OperationStatus.SUCCESS,
                    resource_model=model,
                    custom_context=request.custom_context,
                )
            raise e

        match stack["StackStatus"]:
            case "DELETE_COMPLETE":
                return ProgressEvent(
                    status=OperationStatus.SUCCESS,
                    resource_model=model,
                    custom_context=request.custom_context,
                )
            case "DELETE_IN_PROGRESS":
                return ProgressEvent(
                    status=OperationStatus.IN_PROGRESS,
                    resource_model=model,
                    custom_context=request.custom_context,
                )
            case "DELETE_FAILED":
                return ProgressEvent(
                    status=OperationStatus.FAILED,
                    resource_model=model,
                    custom_context=request.custom_context,
                )
            case _:
                raise Exception(f"Unexpected status: {stack['StackStatus']}")

    def update(
        self,
        request: ResourceRequest[CloudFormationStackProperties],
    ) -> ProgressEvent[CloudFormationStackProperties]:
        """
        Update a resource


        """
        raise NotImplementedError

    def list(
        self,
        request: ResourceRequest[CloudFormationStackProperties],
    ) -> ProgressEvent[CloudFormationStackProperties]:
        resources = request.aws_client_factory.cloudformation.describe_stacks()
        return ProgressEvent(
            status=OperationStatus.SUCCESS,
            resource_models=[
                CloudFormationStackProperties(Id=resource["StackId"])
                for resource in resources["Stacks"]
            ],
        )

1	# LocalStack Resource Provider Scaffolding v2
2	from __future__ import annotations	1✔
3
4	from pathlib import Path	1✔
5	from typing import TypedDict	1✔
6
7	import localstack.services.cloudformation.provider_utils as util	1✔
8	from localstack.services.cloudformation.resource_provider import (	1✔
9	OperationStatus,
10	ProgressEvent,
11	ResourceProvider,
12	ResourceRequest,
13	)
14
15
16	class CloudFormationStackProperties(TypedDict):	1✔
17	TemplateURL: str \| None	1✔
18	Id: str \| None	1✔
19	NotificationARNs: list[str] \| None	1✔
20	Parameters: dict \| None	1✔
21	Tags: list[Tag] \| None	1✔
22	TimeoutInMinutes: int \| None	1✔
23
24
25	class Tag(TypedDict):	1✔
26	Key: str \| None	1✔
27	Value: str \| None	1✔
28
29
30	REPEATED_INVOCATION = "repeated_invocation"	1✔
31
32
33	class CloudFormationStackProvider(ResourceProvider[CloudFormationStackProperties]):	1✔
34	TYPE = "AWS::CloudFormation::Stack" # Autogenerated. Don't change	1✔
35	SCHEMA = util.get_schema_path(Path(__file__)) # Autogenerated. Don't change	1✔
36
37	def create(	1✔
38	self,
39	request: ResourceRequest[CloudFormationStackProperties],
40	) -> ProgressEvent[CloudFormationStackProperties]:
41	"""
42	Create a new resource.
43
44	Primary identifier fields:
45	- /properties/Id
46
47	Required properties:
48	- TemplateURL
49
50
51
52	Read-only properties:
53	- /properties/Id
54
55
56
57	"""
58	model = request.desired_state	1✔
59
60	# TODO: validations
61
62	if not request.custom_context.get(REPEATED_INVOCATION):	1✔
63	if not model.get("StackName"):	1✔
64	model["StackName"] = util.generate_default_name(	1✔
65	request.stack_name, request.logical_resource_id
66	)
67
68	create_params = util.select_attributes(	1✔
69	model,
70	[
71	"StackName",
72	"Parameters",
73	"NotificationARNs",
74	"TemplateURL",
75	"TimeoutInMinutes",
76	"Tags",
77	],
78	)
79
80	create_params["Capabilities"] = [	1✔
81	"CAPABILITY_IAM",
82	"CAPABILITY_NAMED_IAM",
83	"CAPABILITY_AUTO_EXPAND",
84	]
85
86	create_params["Parameters"] = [	1✔
87	{
88	"ParameterKey": k,
89	"ParameterValue": str(v).lower() if isinstance(v, bool) else str(v),
90	}
91	for k, v in create_params.get("Parameters", {}).items()
92	]
93
94	result = request.aws_client_factory.cloudformation.create_stack(**create_params)	1✔
95	model["Id"] = result["StackId"]	1✔
96
97	request.custom_context[REPEATED_INVOCATION] = True	1✔
98	return ProgressEvent(	1✔
99	status=OperationStatus.IN_PROGRESS,
100	resource_model=model,
101	custom_context=request.custom_context,
102	)
103
104	stack = request.aws_client_factory.cloudformation.describe_stacks(StackName=model["Id"])[	1✔
105	"Stacks"
106	][0]
107	match stack["StackStatus"]:	1✔
108	case "CREATE_COMPLETE":	1✔
109	# only store nested stack outputs when we know the deploy has completed
110	model["Outputs"] = {	1✔
111	o["OutputKey"]: o["OutputValue"] for o in stack.get("Outputs", [])
112	}
113	return ProgressEvent(	1✔
114	status=OperationStatus.SUCCESS,
115	resource_model=model,
116	custom_context=request.custom_context,
117	)
118	case "CREATE_IN_PROGRESS":	1✔
119	return ProgressEvent(	1✔
120	status=OperationStatus.IN_PROGRESS,
121	resource_model=model,
122	custom_context=request.custom_context,
123	)
124	case "CREATE_FAILED":	1✔
125	return ProgressEvent(	1✔
126	status=OperationStatus.FAILED,
127	resource_model=model,
128	custom_context=request.custom_context,
129	)
130	case _:	×
131	raise Exception(f"Unexpected status: {stack['StackStatus']}")	×
132
133	def read(	1✔
134	self,
135	request: ResourceRequest[CloudFormationStackProperties],
136	) -> ProgressEvent[CloudFormationStackProperties]:
137	"""
138	Fetch resource information
139
140
141	"""
142	raise NotImplementedError
143
144	def delete(	1✔
145	self,
146	request: ResourceRequest[CloudFormationStackProperties],
147	) -> ProgressEvent[CloudFormationStackProperties]:
148	"""
149	Delete a resource
150	"""
151
152	model = request.desired_state	1✔
153	if not request.custom_context.get(REPEATED_INVOCATION):	1✔
154	request.aws_client_factory.cloudformation.delete_stack(StackName=model["Id"])	1✔
155
156	request.custom_context[REPEATED_INVOCATION] = True	1✔
157	return ProgressEvent(	1✔
158	status=OperationStatus.IN_PROGRESS,
159	resource_model=model,
160	custom_context=request.custom_context,
161	)
162
163	try:	1✔
164	stack = request.aws_client_factory.cloudformation.describe_stacks(	1✔
165	StackName=model["Id"]
166	)["Stacks"][0]
167	except Exception as e:	×
168	if "does not exist" in str(e):	×
169	return ProgressEvent(	×
170	status=OperationStatus.SUCCESS,
171	resource_model=model,
172	custom_context=request.custom_context,
173	)
174	raise e	×
175
176	match stack["StackStatus"]:	1✔
177	case "DELETE_COMPLETE":	1✔
178	return ProgressEvent(	1✔
179	status=OperationStatus.SUCCESS,
180	resource_model=model,
181	custom_context=request.custom_context,
182	)
183	case "DELETE_IN_PROGRESS":	1✔
184	return ProgressEvent(	1✔
185	status=OperationStatus.IN_PROGRESS,
186	resource_model=model,
187	custom_context=request.custom_context,
188	)
UNCOV 189	case "DELETE_FAILED":	×
190	return ProgressEvent(	×
191	status=OperationStatus.FAILED,
192	resource_model=model,
193	custom_context=request.custom_context,
194	)
UNCOV 195	case _:	×
UNCOV 196	raise Exception(f"Unexpected status: {stack['StackStatus']}")	×
197
198	def update(	1✔
199	self,
200	request: ResourceRequest[CloudFormationStackProperties],
201	) -> ProgressEvent[CloudFormationStackProperties]:
202	"""
203	Update a resource
204
205
206	"""
207	raise NotImplementedError
208
209	def list(	1✔
210	self,
211	request: ResourceRequest[CloudFormationStackProperties],
212	) -> ProgressEvent[CloudFormationStackProperties]:
213	resources = request.aws_client_factory.cloudformation.describe_stacks()	×
214	return ProgressEvent(	×
215	status=OperationStatus.SUCCESS,
216	resource_models=[
217	CloudFormationStackProperties(Id=resource["StackId"])
218	for resource in resources["Stacks"]
219	],
220	)

localstack / localstack / 18505123992

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous