16820655284

Committed 07 Aug 2025 05:03PM UTC coverage: 86.841% (-0.05%) from 86.892%

Build # 16820655284

Build Type

push

github

Committed by

web-flow

Commit Message

CFNV2: support CDK bootstrap and deployment (#12967)

Coverage Stats

32 of 38 new or added lines in 5 files covered. (84.21%)

2013 existing lines in 125 files now uncovered.

66606 of 76699 relevant lines covered (86.84%)

0.87 hits per line

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

92.54

/localstack-core/localstack/testing/pytest/cloudformation/fixtures.py

import json
from collections import defaultdict
from collections.abc import Generator
from typing import Callable, Optional, TypedDict

import pytest
from botocore.exceptions import WaiterError

from localstack.aws.api.cloudformation import DescribeChangeSetOutput, StackEvent
from localstack.aws.connect import ServiceLevelClientFactory
from localstack.utils.functions import call_safe
from localstack.utils.strings import short_uid


class NormalizedEvent(TypedDict):
    PhysicalResourceId: Optional[str]
    LogicalResourceId: str
    ResourceType: str
    ResourceStatus: str
    Timestamp: str


PerResourceStackEvents = dict[str, list[NormalizedEvent]]


def normalize_event(event: StackEvent) -> NormalizedEvent:
    return NormalizedEvent(
        PhysicalResourceId=event.get("PhysicalResourceId"),
        LogicalResourceId=event.get("LogicalResourceId"),
        ResourceType=event.get("ResourceType"),
        ResourceStatus=event.get("ResourceStatus"),
        Timestamp=event.get("Timestamp"),
    )


@pytest.fixture
def capture_resource_state_changes(aws_client: ServiceLevelClientFactory):
    def capture(stack_name: str) -> Generator[StackEvent, None, None]:
        resource_states: dict[str, str] = {}
        events = aws_client.cloudformation.describe_stack_events(StackName=stack_name)[
            "StackEvents"
        ]
        for event in events:
            # TODO: not supported events
            if event.get("ResourceStatus") in {
                "UPDATE_COMPLETE_CLEANUP_IN_PROGRESS",
            }:
                continue

            resource = event["LogicalResourceId"]
            status = event["ResourceStatus"]
            if resource not in resource_states:
                yield event
                resource_states[resource] = status
                continue

            if status != resource_states[resource]:
                yield event
                resource_states[resource] = status

    return capture


@pytest.fixture
def capture_per_resource_events(
    capture_resource_state_changes,
) -> Callable[[str], PerResourceStackEvents]:
    def capture(stack_name: str) -> dict:
        per_resource_events = defaultdict(list)
        events = capture_resource_state_changes(stack_name)
        for event in events:
            if logical_resource_id := event.get("LogicalResourceId"):
                resource_name = (
                    logical_resource_id
                    if logical_resource_id != event.get("StackName")
                    else "Stack"
                )
                normalized_event = normalize_event(event)
                per_resource_events[resource_name].append(normalized_event)

        for resource_id in per_resource_events:
            per_resource_events[resource_id].sort(key=lambda event: event["Timestamp"])

        filtered_per_resource_events = {}
        for resource_id in per_resource_events:
            events = []
            last: tuple[str, str, str] | None = None

            for event in per_resource_events[resource_id]:
                unique_key = (
                    event["LogicalResourceId"],
                    event["ResourceStatus"],
                    event["ResourceType"],
                )
                if last is None:
                    events.append(event)
                    last = unique_key
                    continue

                if unique_key == last:
                    continue

                events.append(event)
                last = unique_key

            filtered_per_resource_events[resource_id] = events

        return filtered_per_resource_events

    return capture


def _normalise_describe_change_set_output(value: DescribeChangeSetOutput) -> None:
    value.get("Changes", list()).sort(
        key=lambda change: change.get("ResourceChange", dict()).get("LogicalResourceId", "")
    )


@pytest.fixture
def capture_update_process(aws_client_no_retry, cleanups, capture_per_resource_events):
    """
    Fixture to deploy a new stack (via creating and executing a change set), then updating the
    stack with a second template (via creating and executing a change set).
    """

    stack_name = f"stack-{short_uid()}"
    change_set_name = f"cs-{short_uid()}"

    def inner(
        snapshot,
        t1: dict | str,
        t2: dict | str,
        p1: dict | None = None,
        p2: dict | None = None,
        custom_update_step: Callable[[], None] | None = None,
    ) -> str:
        """
        :return: stack id
        """
        snapshot.add_transformer(snapshot.transform.cloudformation_api())

        if isinstance(t1, dict):
            t1 = json.dumps(t1)
        elif isinstance(t1, str):
            with open(t1) as infile:
                t1 = infile.read()
        if isinstance(t2, dict):
            t2 = json.dumps(t2)
        elif isinstance(t2, str):
            with open(t2) as infile:
                t2 = infile.read()

        p1 = p1 or {}
        p2 = p2 or {}

        # deploy original stack
        change_set_details = aws_client_no_retry.cloudformation.create_change_set(
            StackName=stack_name,
            ChangeSetName=change_set_name,
            TemplateBody=t1,
            ChangeSetType="CREATE",
            Capabilities=[
                "CAPABILITY_IAM",
                "CAPABILITY_NAMED_IAM",
                "CAPABILITY_AUTO_EXPAND",
            ],
            Parameters=[{"ParameterKey": k, "ParameterValue": v} for (k, v) in p1.items()],
        )
        snapshot.match("create-change-set-1", change_set_details)
        stack_id = change_set_details["StackId"]
        change_set_id = change_set_details["Id"]
        aws_client_no_retry.cloudformation.get_waiter("change_set_create_complete").wait(
            ChangeSetName=change_set_id
        )
        cleanups.append(
            lambda: call_safe(
                aws_client_no_retry.cloudformation.delete_change_set,
                kwargs=dict(ChangeSetName=change_set_id),
            )
        )

        describe_change_set_with_prop_values = (
            aws_client_no_retry.cloudformation.describe_change_set(
                ChangeSetName=change_set_id, IncludePropertyValues=True
            )
        )
        _normalise_describe_change_set_output(describe_change_set_with_prop_values)
        snapshot.match("describe-change-set-1-prop-values", describe_change_set_with_prop_values)

        describe_change_set_without_prop_values = (
            aws_client_no_retry.cloudformation.describe_change_set(
                ChangeSetName=change_set_id, IncludePropertyValues=False
            )
        )
        _normalise_describe_change_set_output(describe_change_set_without_prop_values)
        snapshot.match("describe-change-set-1", describe_change_set_without_prop_values)

        execute_results = aws_client_no_retry.cloudformation.execute_change_set(
            ChangeSetName=change_set_id
        )
        snapshot.match("execute-change-set-1", execute_results)
        aws_client_no_retry.cloudformation.get_waiter("stack_create_complete").wait(
            StackName=stack_id
        )

        # ensure stack deletion
        cleanups.append(
            lambda: call_safe(
                aws_client_no_retry.cloudformation.delete_stack, kwargs=dict(StackName=stack_id)
            )
        )

        describe = aws_client_no_retry.cloudformation.describe_stacks(StackName=stack_id)["Stacks"][
            0
        ]
        snapshot.match("post-create-1-describe", describe)

        # run any custom steps if present
        if custom_update_step:
            custom_update_step()

        # update stack
        change_set_details = aws_client_no_retry.cloudformation.create_change_set(
            StackName=stack_name,
            ChangeSetName=change_set_name,
            TemplateBody=t2,
            ChangeSetType="UPDATE",
            Parameters=[{"ParameterKey": k, "ParameterValue": v} for (k, v) in p2.items()],
            Capabilities=[
                "CAPABILITY_IAM",
                "CAPABILITY_NAMED_IAM",
                "CAPABILITY_AUTO_EXPAND",
            ],
        )
        snapshot.match("create-change-set-2", change_set_details)
        stack_id = change_set_details["StackId"]
        change_set_id = change_set_details["Id"]
        try:
            aws_client_no_retry.cloudformation.get_waiter("change_set_create_complete").wait(
                ChangeSetName=change_set_id
            )
        except WaiterError as e:
            desc = aws_client_no_retry.cloudformation.describe_change_set(
                ChangeSetName=change_set_id
            )
            raise RuntimeError(f"Change set deployment failed: {desc}") from e

        describe_change_set_with_prop_values = (
            aws_client_no_retry.cloudformation.describe_change_set(
                ChangeSetName=change_set_id, IncludePropertyValues=True
            )
        )
        _normalise_describe_change_set_output(describe_change_set_with_prop_values)
        snapshot.match("describe-change-set-2-prop-values", describe_change_set_with_prop_values)

        describe_change_set_without_prop_values = (
            aws_client_no_retry.cloudformation.describe_change_set(
                ChangeSetName=change_set_id, IncludePropertyValues=False
            )
        )
        _normalise_describe_change_set_output(describe_change_set_without_prop_values)
        snapshot.match("describe-change-set-2", describe_change_set_without_prop_values)

        execute_results = aws_client_no_retry.cloudformation.execute_change_set(
            ChangeSetName=change_set_id
        )
        snapshot.match("execute-change-set-2", execute_results)
        aws_client_no_retry.cloudformation.get_waiter("stack_update_complete").wait(
            StackName=stack_id
        )

        describe = aws_client_no_retry.cloudformation.describe_stacks(StackName=stack_id)["Stacks"][
            0
        ]
        snapshot.match("post-create-2-describe", describe)

        # delete stack
        aws_client_no_retry.cloudformation.delete_stack(StackName=stack_id)
        aws_client_no_retry.cloudformation.get_waiter("stack_delete_complete").wait(
            StackName=stack_id
        )
        describe = aws_client_no_retry.cloudformation.describe_stacks(StackName=stack_id)["Stacks"][
            0
        ]
        snapshot.match("delete-describe", describe)

        events = capture_per_resource_events(stack_id)
        snapshot.match("per-resource-events", events)

        return stack_id

    yield inner

1	import json	1✔
2	from collections import defaultdict	1✔
3	from collections.abc import Generator	1✔
4	from typing import Callable, Optional, TypedDict	1✔
5
6	import pytest	1✔
7	from botocore.exceptions import WaiterError	1✔
8
9	from localstack.aws.api.cloudformation import DescribeChangeSetOutput, StackEvent	1✔
10	from localstack.aws.connect import ServiceLevelClientFactory	1✔
11	from localstack.utils.functions import call_safe	1✔
12	from localstack.utils.strings import short_uid	1✔
13
14
15	class NormalizedEvent(TypedDict):	1✔
16	PhysicalResourceId: Optional[str]	1✔
17	LogicalResourceId: str	1✔
18	ResourceType: str	1✔
19	ResourceStatus: str	1✔
20	Timestamp: str	1✔
21
22
23	PerResourceStackEvents = dict[str, list[NormalizedEvent]]	1✔
24
25
26	def normalize_event(event: StackEvent) -> NormalizedEvent:	1✔
27	return NormalizedEvent(	1✔
28	PhysicalResourceId=event.get("PhysicalResourceId"),
29	LogicalResourceId=event.get("LogicalResourceId"),
30	ResourceType=event.get("ResourceType"),
31	ResourceStatus=event.get("ResourceStatus"),
32	Timestamp=event.get("Timestamp"),
33	)
34
35
36	@pytest.fixture	1✔
37	def capture_resource_state_changes(aws_client: ServiceLevelClientFactory):	1✔
38	def capture(stack_name: str) -> Generator[StackEvent, None, None]:	1✔
39	resource_states: dict[str, str] = {}	1✔
40	events = aws_client.cloudformation.describe_stack_events(StackName=stack_name)[	1✔
41	"StackEvents"
42	]
43	for event in events:	1✔
44	# TODO: not supported events
45	if event.get("ResourceStatus") in {	1✔
46	"UPDATE_COMPLETE_CLEANUP_IN_PROGRESS",
47	}:
48	continue	1✔
49
50	resource = event["LogicalResourceId"]	1✔
51	status = event["ResourceStatus"]	1✔
52	if resource not in resource_states:	1✔
53	yield event	1✔
54	resource_states[resource] = status	1✔
55	continue	1✔
56
57	if status != resource_states[resource]:	1✔
58	yield event	1✔
59	resource_states[resource] = status	1✔
60
61	return capture	1✔
62
63
64	@pytest.fixture	1✔
65	def capture_per_resource_events(	1✔
66	capture_resource_state_changes,
67	) -> Callable[[str], PerResourceStackEvents]:
68	def capture(stack_name: str) -> dict:	1✔
69	per_resource_events = defaultdict(list)	1✔
70	events = capture_resource_state_changes(stack_name)	1✔
71	for event in events:	1✔
72	if logical_resource_id := event.get("LogicalResourceId"):	1✔
73	resource_name = (	1✔
74	logical_resource_id
75	if logical_resource_id != event.get("StackName")
76	else "Stack"
77	)
78	normalized_event = normalize_event(event)	1✔
79	per_resource_events[resource_name].append(normalized_event)	1✔
80
81	for resource_id in per_resource_events:	1✔
82	per_resource_events[resource_id].sort(key=lambda event: event["Timestamp"])	1✔
83
84	filtered_per_resource_events = {}	1✔
85	for resource_id in per_resource_events:	1✔
86	events = []	1✔
87	last: tuple[str, str, str] \| None = None	1✔
88
89	for event in per_resource_events[resource_id]:	1✔
90	unique_key = (	1✔
91	event["LogicalResourceId"],
92	event["ResourceStatus"],
93	event["ResourceType"],
94	)
95	if last is None:	1✔
96	events.append(event)	1✔
97	last = unique_key	1✔
98	continue	1✔
99
100	if unique_key == last:	1✔
UNCOV 101	continue	×
102
103	events.append(event)	1✔
104	last = unique_key	1✔
105
106	filtered_per_resource_events[resource_id] = events	1✔
107
108	return filtered_per_resource_events	1✔
109
110	return capture	1✔
111
112
113	def _normalise_describe_change_set_output(value: DescribeChangeSetOutput) -> None:	1✔
114	value.get("Changes", list()).sort(	1✔
115	key=lambda change: change.get("ResourceChange", dict()).get("LogicalResourceId", "")
116	)
117
118
119	@pytest.fixture	1✔
120	def capture_update_process(aws_client_no_retry, cleanups, capture_per_resource_events):	1✔
121	"""
122	Fixture to deploy a new stack (via creating and executing a change set), then updating the
123	stack with a second template (via creating and executing a change set).
124	"""
125
126	stack_name = f"stack-{short_uid()}"	1✔
127	change_set_name = f"cs-{short_uid()}"	1✔
128
129	def inner(	1✔
130	snapshot,
131	t1: dict \| str,
132	t2: dict \| str,
133	p1: dict \| None = None,
134	p2: dict \| None = None,
135	custom_update_step: Callable[[], None] \| None = None,
136	) -> str:
137	"""
138	:return: stack id
139	"""
140	snapshot.add_transformer(snapshot.transform.cloudformation_api())	1✔
141
142	if isinstance(t1, dict):	1✔
143	t1 = json.dumps(t1)	1✔
144	elif isinstance(t1, str):	×
145	with open(t1) as infile:	×
UNCOV 146	t1 = infile.read()	×
147	if isinstance(t2, dict):	1✔
148	t2 = json.dumps(t2)	1✔
UNCOV 149	elif isinstance(t2, str):	×
UNCOV 150	with open(t2) as infile:	×
UNCOV 151	t2 = infile.read()	×
152
153	p1 = p1 or {}	1✔
154	p2 = p2 or {}	1✔
155
156	# deploy original stack
157	change_set_details = aws_client_no_retry.cloudformation.create_change_set(	1✔
158	StackName=stack_name,
159	ChangeSetName=change_set_name,
160	TemplateBody=t1,
161	ChangeSetType="CREATE",
162	Capabilities=[
163	"CAPABILITY_IAM",
164	"CAPABILITY_NAMED_IAM",
165	"CAPABILITY_AUTO_EXPAND",
166	],
167	Parameters=[{"ParameterKey": k, "ParameterValue": v} for (k, v) in p1.items()],
168	)
169	snapshot.match("create-change-set-1", change_set_details)	1✔
170	stack_id = change_set_details["StackId"]	1✔
171	change_set_id = change_set_details["Id"]	1✔
172	aws_client_no_retry.cloudformation.get_waiter("change_set_create_complete").wait(	1✔
173	ChangeSetName=change_set_id
174	)
175	cleanups.append(	1✔
176	lambda: call_safe(
177	aws_client_no_retry.cloudformation.delete_change_set,
178	kwargs=dict(ChangeSetName=change_set_id),
179	)
180	)
181
182	describe_change_set_with_prop_values = (	1✔
183	aws_client_no_retry.cloudformation.describe_change_set(
184	ChangeSetName=change_set_id, IncludePropertyValues=True
185	)
186	)
187	_normalise_describe_change_set_output(describe_change_set_with_prop_values)	1✔
188	snapshot.match("describe-change-set-1-prop-values", describe_change_set_with_prop_values)	1✔
189
190	describe_change_set_without_prop_values = (	1✔
191	aws_client_no_retry.cloudformation.describe_change_set(
192	ChangeSetName=change_set_id, IncludePropertyValues=False
193	)
194	)
195	_normalise_describe_change_set_output(describe_change_set_without_prop_values)	1✔
196	snapshot.match("describe-change-set-1", describe_change_set_without_prop_values)	1✔
197
198	execute_results = aws_client_no_retry.cloudformation.execute_change_set(	1✔
199	ChangeSetName=change_set_id
200	)
201	snapshot.match("execute-change-set-1", execute_results)	1✔
202	aws_client_no_retry.cloudformation.get_waiter("stack_create_complete").wait(	1✔
203	StackName=stack_id
204	)
205
206	# ensure stack deletion
207	cleanups.append(	1✔
208	lambda: call_safe(
209	aws_client_no_retry.cloudformation.delete_stack, kwargs=dict(StackName=stack_id)
210	)
211	)
212
213	describe = aws_client_no_retry.cloudformation.describe_stacks(StackName=stack_id)["Stacks"][	1✔
214	0
215	]
216	snapshot.match("post-create-1-describe", describe)	1✔
217
218	# run any custom steps if present
219	if custom_update_step:	1✔
220	custom_update_step()	1✔
221
222	# update stack
223	change_set_details = aws_client_no_retry.cloudformation.create_change_set(	1✔
224	StackName=stack_name,
225	ChangeSetName=change_set_name,
226	TemplateBody=t2,
227	ChangeSetType="UPDATE",
228	Parameters=[{"ParameterKey": k, "ParameterValue": v} for (k, v) in p2.items()],
229	Capabilities=[
230	"CAPABILITY_IAM",
231	"CAPABILITY_NAMED_IAM",
232	"CAPABILITY_AUTO_EXPAND",
233	],
234	)
235	snapshot.match("create-change-set-2", change_set_details)	1✔
236	stack_id = change_set_details["StackId"]	1✔
237	change_set_id = change_set_details["Id"]	1✔
238	try:	1✔
239	aws_client_no_retry.cloudformation.get_waiter("change_set_create_complete").wait(	1✔
240	ChangeSetName=change_set_id
241	)
UNCOV 242	except WaiterError as e:	×
UNCOV 243	desc = aws_client_no_retry.cloudformation.describe_change_set(	×
244	ChangeSetName=change_set_id
245	)
UNCOV 246	raise RuntimeError(f"Change set deployment failed: {desc}") from e	×
247
248	describe_change_set_with_prop_values = (	1✔
249	aws_client_no_retry.cloudformation.describe_change_set(
250	ChangeSetName=change_set_id, IncludePropertyValues=True
251	)
252	)
253	_normalise_describe_change_set_output(describe_change_set_with_prop_values)	1✔
254	snapshot.match("describe-change-set-2-prop-values", describe_change_set_with_prop_values)	1✔
255
256	describe_change_set_without_prop_values = (	1✔
257	aws_client_no_retry.cloudformation.describe_change_set(
258	ChangeSetName=change_set_id, IncludePropertyValues=False
259	)
260	)
261	_normalise_describe_change_set_output(describe_change_set_without_prop_values)	1✔
262	snapshot.match("describe-change-set-2", describe_change_set_without_prop_values)	1✔
263
264	execute_results = aws_client_no_retry.cloudformation.execute_change_set(	1✔
265	ChangeSetName=change_set_id
266	)
267	snapshot.match("execute-change-set-2", execute_results)	1✔
268	aws_client_no_retry.cloudformation.get_waiter("stack_update_complete").wait(	1✔
269	StackName=stack_id
270	)
271
272	describe = aws_client_no_retry.cloudformation.describe_stacks(StackName=stack_id)["Stacks"][	1✔
273	0
274	]
275	snapshot.match("post-create-2-describe", describe)	1✔
276
277	# delete stack
278	aws_client_no_retry.cloudformation.delete_stack(StackName=stack_id)	1✔
279	aws_client_no_retry.cloudformation.get_waiter("stack_delete_complete").wait(	1✔
280	StackName=stack_id
281	)
282	describe = aws_client_no_retry.cloudformation.describe_stacks(StackName=stack_id)["Stacks"][	1✔
283	0
284	]
285	snapshot.match("delete-describe", describe)	1✔
286
287	events = capture_per_resource_events(stack_id)	1✔
288	snapshot.match("per-resource-events", events)	1✔
289
290	return stack_id	1✔
291
292	yield inner	1✔

localstack / localstack / 16820655284

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