16766254352

Committed 05 Aug 2025 04:40PM UTC coverage: 86.892% (-0.02%) from 86.91%

Build # 16766254352

Build Type

push

github

Committed by

web-flow

Commit Message

CFNV2: defer deletions for correcting deploy order (#12936)

Coverage Stats

92 of 99 new or added lines in 6 files covered. (92.93%)

185 existing lines in 21 files now uncovered.

66597 of 76643 relevant lines covered (86.89%)

0.87 hits per line

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

92.37

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

import json
from collections import defaultdict
from typing import Callable, Generator, 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", str())
    )


@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
    ) -> 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)

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

localstack / localstack / 16766254352

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