22709357475

Committed 05 Mar 2026 08:35AM UTC coverage: 59.732% (-27.2%) from 86.974%

Build # 22709357475

Build Type

Pull #13880

github

Committed by

web-flow

Commit Message

Merge 28fcab93c into 710618057

Pull Request Pull Request #13880: Firehose: Replace TaggingService

Coverage Stats

12 of 12 new or added lines in 2 files covered. (100.0%)

20464 existing lines in 510 files now uncovered.

45290 of 75822 relevant lines covered (59.73%)

0.6 hits per line

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

18.66

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

import json
from collections import defaultdict
from collections.abc import Callable, Generator
from typing import 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: str | None
    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", []).sort(
        key=lambda change: change.get("ResourceChange", {}).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={"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={"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 Callable, Generator	1✔
4	from typing import 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: str \| None	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✔
UNCOV 27	return NormalizedEvent(	×
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✔
UNCOV 38	def capture(stack_name: str) -> Generator[StackEvent, None, None]:	×
UNCOV 39	resource_states: dict[str, str] = {}	×
UNCOV 40	events = aws_client.cloudformation.describe_stack_events(StackName=stack_name)[	×
41	"StackEvents"
42	]
UNCOV 43	for event in events:	×
44	# TODO: not supported events
UNCOV 45	if event.get("ResourceStatus") in {	×
46	"UPDATE_COMPLETE_CLEANUP_IN_PROGRESS",
47	}:
UNCOV 48	continue	×
49
UNCOV 50	resource = event["LogicalResourceId"]	×
UNCOV 51	status = event["ResourceStatus"]	×
UNCOV 52	if resource not in resource_states:	×
UNCOV 53	yield event	×
UNCOV 54	resource_states[resource] = status	×
UNCOV 55	continue	×
56
UNCOV 57	if status != resource_states[resource]:	×
UNCOV 58	yield event	×
UNCOV 59	resource_states[resource] = status	×
60
UNCOV 61	return capture	×
62
63
64	@pytest.fixture	1✔
65	def capture_per_resource_events(	1✔
66	capture_resource_state_changes,
67	) -> Callable[[str], PerResourceStackEvents]:
UNCOV 68	def capture(stack_name: str) -> dict:	×
UNCOV 69	per_resource_events = defaultdict(list)	×
UNCOV 70	events = capture_resource_state_changes(stack_name)	×
UNCOV 71	for event in events:	×
UNCOV 72	if logical_resource_id := event.get("LogicalResourceId"):	×
UNCOV 73	resource_name = (	×
74	logical_resource_id
75	if logical_resource_id != event.get("StackName")
76	else "Stack"
77	)
UNCOV 78	normalized_event = normalize_event(event)	×
UNCOV 79	per_resource_events[resource_name].append(normalized_event)	×
80
UNCOV 81	for resource_id in per_resource_events:	×
UNCOV 82	per_resource_events[resource_id].sort(key=lambda event: event["Timestamp"])	×
83
UNCOV 84	filtered_per_resource_events = {}	×
UNCOV 85	for resource_id in per_resource_events:	×
UNCOV 86	events = []	×
UNCOV 87	last: tuple[str, str, str] \| None = None	×
88
UNCOV 89	for event in per_resource_events[resource_id]:	×
UNCOV 90	unique_key = (	×
91	event["LogicalResourceId"],
92	event["ResourceStatus"],
93	event["ResourceType"],
94	)
UNCOV 95	if last is None:	×
UNCOV 96	events.append(event)	×
UNCOV 97	last = unique_key	×
UNCOV 98	continue	×
99
UNCOV 100	if unique_key == last:	×
101	continue	×
102
UNCOV 103	events.append(event)	×
UNCOV 104	last = unique_key	×
105
UNCOV 106	filtered_per_resource_events[resource_id] = events	×
107
UNCOV 108	return filtered_per_resource_events	×
109
UNCOV 110	return capture	×
111
112
113	def _normalise_describe_change_set_output(value: DescribeChangeSetOutput) -> None:	1✔
UNCOV 114	value.get("Changes", []).sort(	×
115	key=lambda change: change.get("ResourceChange", {}).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
UNCOV 126	stack_name = f"stack-{short_uid()}"	×
UNCOV 127	change_set_name = f"cs-{short_uid()}"	×
128
UNCOV 129	def inner(	×
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	"""
UNCOV 140	snapshot.add_transformer(snapshot.transform.cloudformation_api())	×
141
UNCOV 142	if isinstance(t1, dict):	×
UNCOV 143	t1 = json.dumps(t1)	×
144	elif isinstance(t1, str):	×
145	with open(t1) as infile:	×
146	t1 = infile.read()	×
UNCOV 147	if isinstance(t2, dict):	×
UNCOV 148	t2 = json.dumps(t2)	×
149	elif isinstance(t2, str):	×
150	with open(t2) as infile:	×
151	t2 = infile.read()	×
152
UNCOV 153	p1 = p1 or {}	×
UNCOV 154	p2 = p2 or {}	×
155
156	# deploy original stack
UNCOV 157	change_set_details = aws_client_no_retry.cloudformation.create_change_set(	×
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	)
UNCOV 169	snapshot.match("create-change-set-1", change_set_details)	×
UNCOV 170	stack_id = change_set_details["StackId"]	×
UNCOV 171	change_set_id = change_set_details["Id"]	×
UNCOV 172	aws_client_no_retry.cloudformation.get_waiter("change_set_create_complete").wait(	×
173	ChangeSetName=change_set_id
174	)
UNCOV 175	cleanups.append(	×
176	lambda: call_safe(
177	aws_client_no_retry.cloudformation.delete_change_set,
178	kwargs={"ChangeSetName": change_set_id},
179	)
180	)
181
UNCOV 182	describe_change_set_with_prop_values = (	×
183	aws_client_no_retry.cloudformation.describe_change_set(
184	ChangeSetName=change_set_id, IncludePropertyValues=True
185	)
186	)
UNCOV 187	_normalise_describe_change_set_output(describe_change_set_with_prop_values)	×
UNCOV 188	snapshot.match("describe-change-set-1-prop-values", describe_change_set_with_prop_values)	×
189
UNCOV 190	describe_change_set_without_prop_values = (	×
191	aws_client_no_retry.cloudformation.describe_change_set(
192	ChangeSetName=change_set_id, IncludePropertyValues=False
193	)
194	)
UNCOV 195	_normalise_describe_change_set_output(describe_change_set_without_prop_values)	×
UNCOV 196	snapshot.match("describe-change-set-1", describe_change_set_without_prop_values)	×
197
UNCOV 198	execute_results = aws_client_no_retry.cloudformation.execute_change_set(	×
199	ChangeSetName=change_set_id
200	)
UNCOV 201	snapshot.match("execute-change-set-1", execute_results)	×
UNCOV 202	aws_client_no_retry.cloudformation.get_waiter("stack_create_complete").wait(	×
203	StackName=stack_id
204	)
205
206	# ensure stack deletion
UNCOV 207	cleanups.append(	×
208	lambda: call_safe(
209	aws_client_no_retry.cloudformation.delete_stack, kwargs={"StackName": stack_id}
210	)
211	)
212
UNCOV 213	describe = aws_client_no_retry.cloudformation.describe_stacks(StackName=stack_id)["Stacks"][	×
214	0
215	]
UNCOV 216	snapshot.match("post-create-1-describe", describe)	×
217
218	# run any custom steps if present
UNCOV 219	if custom_update_step:	×
UNCOV 220	custom_update_step()	×
221
222	# update stack
UNCOV 223	change_set_details = aws_client_no_retry.cloudformation.create_change_set(	×
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	)
UNCOV 235	snapshot.match("create-change-set-2", change_set_details)	×
UNCOV 236	stack_id = change_set_details["StackId"]	×
UNCOV 237	change_set_id = change_set_details["Id"]	×
UNCOV 238	try:	×
UNCOV 239	aws_client_no_retry.cloudformation.get_waiter("change_set_create_complete").wait(	×
240	ChangeSetName=change_set_id
241	)
242	except WaiterError as e:	×
243	desc = aws_client_no_retry.cloudformation.describe_change_set(	×
244	ChangeSetName=change_set_id
245	)
246	raise RuntimeError(f"Change set deployment failed: {desc}") from e	×
247
UNCOV 248	describe_change_set_with_prop_values = (	×
249	aws_client_no_retry.cloudformation.describe_change_set(
250	ChangeSetName=change_set_id, IncludePropertyValues=True
251	)
252	)
UNCOV 253	_normalise_describe_change_set_output(describe_change_set_with_prop_values)	×
UNCOV 254	snapshot.match("describe-change-set-2-prop-values", describe_change_set_with_prop_values)	×
255
UNCOV 256	describe_change_set_without_prop_values = (	×
257	aws_client_no_retry.cloudformation.describe_change_set(
258	ChangeSetName=change_set_id, IncludePropertyValues=False
259	)
260	)
UNCOV 261	_normalise_describe_change_set_output(describe_change_set_without_prop_values)	×
UNCOV 262	snapshot.match("describe-change-set-2", describe_change_set_without_prop_values)	×
263
UNCOV 264	execute_results = aws_client_no_retry.cloudformation.execute_change_set(	×
265	ChangeSetName=change_set_id
266	)
UNCOV 267	snapshot.match("execute-change-set-2", execute_results)	×
UNCOV 268	aws_client_no_retry.cloudformation.get_waiter("stack_update_complete").wait(	×
269	StackName=stack_id
270	)
271
UNCOV 272	describe = aws_client_no_retry.cloudformation.describe_stacks(StackName=stack_id)["Stacks"][	×
273	0
274	]
UNCOV 275	snapshot.match("post-create-2-describe", describe)	×
276
277	# delete stack
UNCOV 278	aws_client_no_retry.cloudformation.delete_stack(StackName=stack_id)	×
UNCOV 279	aws_client_no_retry.cloudformation.get_waiter("stack_delete_complete").wait(	×
280	StackName=stack_id
281	)
UNCOV 282	describe = aws_client_no_retry.cloudformation.describe_stacks(StackName=stack_id)["Stacks"][	×
283	0
284	]
UNCOV 285	snapshot.match("delete-describe", describe)	×
286
UNCOV 287	events = capture_per_resource_events(stack_id)	×
UNCOV 288	snapshot.match("per-resource-events", events)	×
289
UNCOV 290	return stack_id	×
291
UNCOV 292	yield inner	×

localstack / localstack / 22709357475

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