533d4262-4f08-49b7-b7ba-18c46e51ac1a

Committed 02 Jun 2025 06:43PM UTC coverage: 86.752% (+0.1%) from 86.654%

Build # 533d4262-4f08-49b7-b7ba-18c46e51ac1a

Build Type

push

circleci

Committed by

web-flow

Commit Message

APIGW: implement Canary Deployments CRUD logic (#12694)

Coverage Stats

142 of 147 new or added lines in 3 files covered. (96.6%)

187 existing lines in 16 files now uncovered.

64937 of 74854 relevant lines covered (86.75%)

0.87 hits per line

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

90.0

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

import json
from collections import defaultdict
from typing import Callable

import pytest

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

PerResourceStackEvents = dict[str, list[StackEvent]]


@pytest.fixture
def capture_per_resource_events(
    aws_client: ServiceLevelClientFactory,
) -> Callable[[str], PerResourceStackEvents]:
    def capture(stack_name: str) -> PerResourceStackEvents:
        events = aws_client.cloudformation.describe_stack_events(StackName=stack_name)[
            "StackEvents"
        ]
        per_resource_events = defaultdict(list)
        for event in events:
            if logical_resource_id := event.get("LogicalResourceId"):
                per_resource_events[logical_resource_id].append(event)
        return 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
    ):
        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",
            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()],
        )
        snapshot.match("create-change-set-2", 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
        )

        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)

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

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

    yield inner

1	import json	1✔
2	from collections import defaultdict	1✔
3	from typing import Callable	1✔
4
5	import pytest	1✔
6
7	from localstack.aws.api.cloudformation import DescribeChangeSetOutput, StackEvent	1✔
8	from localstack.aws.connect import ServiceLevelClientFactory	1✔
9	from localstack.utils.functions import call_safe	1✔
10	from localstack.utils.strings import short_uid	1✔
11
12	PerResourceStackEvents = dict[str, list[StackEvent]]	1✔
13
14
15	@pytest.fixture	1✔
16	def capture_per_resource_events(	1✔
17	aws_client: ServiceLevelClientFactory,
18	) -> Callable[[str], PerResourceStackEvents]:
19	def capture(stack_name: str) -> PerResourceStackEvents:	1✔
20	events = aws_client.cloudformation.describe_stack_events(StackName=stack_name)[	1✔
21	"StackEvents"
22	]
23	per_resource_events = defaultdict(list)	1✔
24	for event in events:	1✔
25	if logical_resource_id := event.get("LogicalResourceId"):	×
26	per_resource_events[logical_resource_id].append(event)	×
27	return per_resource_events	1✔
28
29	return capture	1✔
30
31
32	def _normalise_describe_change_set_output(value: DescribeChangeSetOutput) -> None:	1✔
33	value.get("Changes", list()).sort(	1✔
34	key=lambda change: change.get("ResourceChange", dict()).get("LogicalResourceId", str())
35	)
36
37
38	@pytest.fixture	1✔
39	def capture_update_process(aws_client_no_retry, cleanups, capture_per_resource_events):	1✔
40	"""
41	Fixture to deploy a new stack (via creating and executing a change set), then updating the
42	stack with a second template (via creating and executing a change set).
43	"""
44
45	stack_name = f"stack-{short_uid()}"	1✔
46	change_set_name = f"cs-{short_uid()}"	1✔
47
48	def inner(	1✔
49	snapshot, t1: dict \| str, t2: dict \| str, p1: dict \| None = None, p2: dict \| None = None
50	):
51	snapshot.add_transformer(snapshot.transform.cloudformation_api())	1✔
52
53	if isinstance(t1, dict):	1✔
54	t1 = json.dumps(t1)	1✔
55	elif isinstance(t1, str):	×
56	with open(t1) as infile:	×
UNCOV 57	t1 = infile.read()	×
58	if isinstance(t2, dict):	1✔
59	t2 = json.dumps(t2)	1✔
UNCOV 60	elif isinstance(t2, str):	×
UNCOV 61	with open(t2) as infile:	×
UNCOV 62	t2 = infile.read()	×
63
64	p1 = p1 or {}	1✔
65	p2 = p2 or {}	1✔
66
67	# deploy original stack
68	change_set_details = aws_client_no_retry.cloudformation.create_change_set(	1✔
69	StackName=stack_name,
70	ChangeSetName=change_set_name,
71	TemplateBody=t1,
72	ChangeSetType="CREATE",
73	Parameters=[{"ParameterKey": k, "ParameterValue": v} for (k, v) in p1.items()],
74	)
75	snapshot.match("create-change-set-1", change_set_details)	1✔
76	stack_id = change_set_details["StackId"]	1✔
77	change_set_id = change_set_details["Id"]	1✔
78	aws_client_no_retry.cloudformation.get_waiter("change_set_create_complete").wait(	1✔
79	ChangeSetName=change_set_id
80	)
81	cleanups.append(	1✔
82	lambda: call_safe(
83	aws_client_no_retry.cloudformation.delete_change_set,
84	kwargs=dict(ChangeSetName=change_set_id),
85	)
86	)
87
88	describe_change_set_with_prop_values = (	1✔
89	aws_client_no_retry.cloudformation.describe_change_set(
90	ChangeSetName=change_set_id, IncludePropertyValues=True
91	)
92	)
93	_normalise_describe_change_set_output(describe_change_set_with_prop_values)	1✔
94	snapshot.match("describe-change-set-1-prop-values", describe_change_set_with_prop_values)	1✔
95
96	describe_change_set_without_prop_values = (	1✔
97	aws_client_no_retry.cloudformation.describe_change_set(
98	ChangeSetName=change_set_id, IncludePropertyValues=False
99	)
100	)
101	_normalise_describe_change_set_output(describe_change_set_without_prop_values)	1✔
102	snapshot.match("describe-change-set-1", describe_change_set_without_prop_values)	1✔
103
104	execute_results = aws_client_no_retry.cloudformation.execute_change_set(	1✔
105	ChangeSetName=change_set_id
106	)
107	snapshot.match("execute-change-set-1", execute_results)	1✔
108	aws_client_no_retry.cloudformation.get_waiter("stack_create_complete").wait(	1✔
109	StackName=stack_id
110	)
111
112	# ensure stack deletion
113	cleanups.append(	1✔
114	lambda: call_safe(
115	aws_client_no_retry.cloudformation.delete_stack, kwargs=dict(StackName=stack_id)
116	)
117	)
118
119	describe = aws_client_no_retry.cloudformation.describe_stacks(StackName=stack_id)["Stacks"][	1✔
120	0
121	]
122	snapshot.match("post-create-1-describe", describe)	1✔
123
124	# update stack
125	change_set_details = aws_client_no_retry.cloudformation.create_change_set(	1✔
126	StackName=stack_name,
127	ChangeSetName=change_set_name,
128	TemplateBody=t2,
129	ChangeSetType="UPDATE",
130	Parameters=[{"ParameterKey": k, "ParameterValue": v} for (k, v) in p2.items()],
131	)
132	snapshot.match("create-change-set-2", change_set_details)	1✔
133	stack_id = change_set_details["StackId"]	1✔
134	change_set_id = change_set_details["Id"]	1✔
135	aws_client_no_retry.cloudformation.get_waiter("change_set_create_complete").wait(	1✔
136	ChangeSetName=change_set_id
137	)
138
139	describe_change_set_with_prop_values = (	1✔
140	aws_client_no_retry.cloudformation.describe_change_set(
141	ChangeSetName=change_set_id, IncludePropertyValues=True
142	)
143	)
144	_normalise_describe_change_set_output(describe_change_set_with_prop_values)	1✔
145	snapshot.match("describe-change-set-2-prop-values", describe_change_set_with_prop_values)	1✔
146
147	describe_change_set_without_prop_values = (	1✔
148	aws_client_no_retry.cloudformation.describe_change_set(
149	ChangeSetName=change_set_id, IncludePropertyValues=False
150	)
151	)
152	_normalise_describe_change_set_output(describe_change_set_without_prop_values)	1✔
153	snapshot.match("describe-change-set-2", describe_change_set_without_prop_values)	1✔
154
155	execute_results = aws_client_no_retry.cloudformation.execute_change_set(	1✔
156	ChangeSetName=change_set_id
157	)
158	snapshot.match("execute-change-set-2", execute_results)	1✔
159	aws_client_no_retry.cloudformation.get_waiter("stack_update_complete").wait(	1✔
160	StackName=stack_id
161	)
162
163	describe = aws_client_no_retry.cloudformation.describe_stacks(StackName=stack_id)["Stacks"][	1✔
164	0
165	]
166	snapshot.match("post-create-2-describe", describe)	1✔
167
168	events = capture_per_resource_events(stack_name)	1✔
169	snapshot.match("per-resource-events", events)	1✔
170
171	# delete stack
172	aws_client_no_retry.cloudformation.delete_stack(StackName=stack_id)	1✔
173	aws_client_no_retry.cloudformation.get_waiter("stack_delete_complete").wait(	1✔
174	StackName=stack_id
175	)
176	describe = aws_client_no_retry.cloudformation.describe_stacks(StackName=stack_id)["Stacks"][	1✔
177	0
178	]
179	snapshot.match("delete-describe", describe)	1✔
180
181	yield inner	1✔

localstack / localstack / 533d4262-4f08-49b7-b7ba-18c46e51ac1a

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