• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

localstack / localstack / 16766254352

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

push

github

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

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

84.34
/localstack-core/localstack/services/cloudformation/engine/v2/change_set_model_transform.py
1
import copy
1✔
2
import logging
1✔
3
import os
1✔
4
from typing import Any, Final, Optional, TypedDict
1✔
5

6
import boto3
1✔
7
from botocore.exceptions import ClientError
1✔
8
from samtranslator.translator.transform import transform as transform_sam
1✔
9

10
from localstack.aws.connect import connect_to
1✔
11
from localstack.services.cloudformation.engine.policy_loader import create_policy_loader
1✔
12
from localstack.services.cloudformation.engine.template_preparer import parse_template
1✔
13
from localstack.services.cloudformation.engine.transformers import (
1✔
14
    FailedTransformationException,
15
    execute_macro,
16
)
17
from localstack.services.cloudformation.engine.v2.change_set_model import (
1✔
18
    ChangeType,
19
    Maybe,
20
    NodeGlobalTransform,
21
    NodeParameter,
22
    NodeTransform,
23
    Nothing,
24
    Scope,
25
    is_nothing,
26
)
27
from localstack.services.cloudformation.engine.v2.change_set_model_preproc import (
1✔
28
    ChangeSetModelPreproc,
29
    PreprocEntityDelta,
30
)
31
from localstack.services.cloudformation.stores import get_cloudformation_store
1✔
32
from localstack.services.cloudformation.v2.entities import ChangeSet
1✔
33
from localstack.utils import testutil
1✔
34

35
LOG = logging.getLogger(__name__)
1✔
36

37
SERVERLESS_TRANSFORM = "AWS::Serverless-2016-10-31"
1✔
38
EXTENSIONS_TRANSFORM = "AWS::LanguageExtensions"
1✔
39
SECRETSMANAGER_TRANSFORM = "AWS::SecretsManager-2020-07-23"
1✔
40
INCLUDE_TRANSFORM = "AWS::Include"
1✔
41

42
_SCOPE_TRANSFORM_TEMPLATE_OUTCOME: Final[Scope] = Scope("TRANSFORM_TEMPLATE_OUTCOME")
1✔
43

44

45
# TODO: evaluate the use of subtypes to represent and validate types of transforms
46
class GlobalTransform:
1✔
47
    name: str
1✔
48
    parameters: Maybe[dict]
1✔
49

50
    def __init__(self, name: str, parameters: Maybe[dict]):
1✔
51
        self.name = name
1✔
52
        self.parameters = parameters
1✔
53

54

55
class TransformPreprocParameter(TypedDict):
1✔
56
    # TODO: expand
57
    ParameterKey: str
1✔
58
    ParameterValue: Any
1✔
59
    ParameterType: Optional[str]
1✔
60

61

62
class ChangeSetModelTransform(ChangeSetModelPreproc):
1✔
63
    _before_parameters: Final[dict]
1✔
64
    _after_parameters: Final[dict]
1✔
65
    _before_template: Final[Maybe[dict]]
1✔
66
    _after_template: Final[Maybe[dict]]
1✔
67

68
    def __init__(
1✔
69
        self,
70
        change_set: ChangeSet,
71
        before_parameters: dict,
72
        after_parameters: dict,
73
        before_template: Optional[dict],
74
        after_template: Optional[dict],
75
    ):
76
        super().__init__(change_set=change_set)
1✔
77
        self._before_parameters = before_parameters
1✔
78
        self._after_parameters = after_parameters
1✔
79
        self._before_template = before_template or Nothing
1✔
80
        self._after_template = after_template or Nothing
1✔
81

82
    def visit_node_parameter(
1✔
83
        self, node_parameter: NodeParameter
84
    ) -> PreprocEntityDelta[
85
        dict[str, TransformPreprocParameter], dict[str, TransformPreprocParameter]
86
    ]:
87
        # Enable compatability with v1 util.
88
        # TODO: port v1's SSM parameter resolution
89

90
        parameter_value_delta = super().visit_node_parameter(node_parameter=node_parameter)
1✔
91
        parameter_value_before = parameter_value_delta.before
1✔
92
        parameter_value_after = parameter_value_delta.after
1✔
93

94
        parameter_type_delta = self.visit(node_parameter.type_)
1✔
95
        parameter_type_before = parameter_type_delta.before
1✔
96
        parameter_type_after = parameter_type_delta.after
1✔
97

98
        parameter_key = node_parameter.name
1✔
99

100
        before = Nothing
1✔
101
        if not is_nothing(parameter_value_before):
1✔
102
            before = TransformPreprocParameter(
1✔
103
                ParameterKey=parameter_key,
104
                ParameterValue=parameter_value_before,
105
                ParameterType=parameter_type_before
106
                if not is_nothing(parameter_type_before)
107
                else None,
108
            )
109
        after = Nothing
1✔
110
        if not is_nothing(parameter_value_after):
1✔
111
            after = TransformPreprocParameter(
1✔
112
                ParameterKey=parameter_key,
113
                ParameterValue=parameter_value_after,
114
                ParameterType=parameter_type_after
115
                if not is_nothing(parameter_type_after)
116
                else None,
117
            )
118

119
        return PreprocEntityDelta(before=before, after=after)
1✔
120

121
    # Ported from v1:
122
    @staticmethod
1✔
123
    def _apply_global_serverless_transformation(
1✔
124
        region_name: str, template: dict, parameters: dict
125
    ) -> dict:
126
        """only returns string when parsing SAM template, otherwise None"""
127
        # TODO: we might also want to override the access key ID to account ID
128
        region_before = os.environ.get("AWS_DEFAULT_REGION")
1✔
129
        if boto3.session.Session().region_name is None:
1✔
130
            os.environ["AWS_DEFAULT_REGION"] = region_name
1✔
131
        loader = create_policy_loader()
1✔
132
        # The following transformation function can carry out in-place changes ensure this cannot occur.
133
        template = copy.deepcopy(template)
1✔
134
        parameters = copy.deepcopy(parameters)
1✔
135
        try:
1✔
136
            transformed = transform_sam(template, parameters, loader)
1✔
137
            return transformed
1✔
UNCOV
138
        except Exception as e:
×
139
            raise FailedTransformationException(transformation=SERVERLESS_TRANSFORM, message=str(e))
×
140
        finally:
141
            # Note: we need to fix boto3 region, otherwise AWS SAM transformer fails
142
            os.environ.pop("AWS_DEFAULT_REGION", None)
1✔
143
            if region_before is not None:
1✔
UNCOV
144
                os.environ["AWS_DEFAULT_REGION"] = region_before
×
145

146
    @staticmethod
1✔
147
    def _apply_global_include(
1✔
148
        global_transform: GlobalTransform, template: dict, parameters: dict, account_id, region_name
149
    ) -> dict:
UNCOV
150
        location = global_transform.parameters.get("Location")
×
UNCOV
151
        if not location or not location.startswith("s3://"):
×
UNCOV
152
            raise FailedTransformationException(
×
153
                transformation=INCLUDE_TRANSFORM,
154
                message="Unexpected Location parameter for AWS::Include transformer: %s" % location,
155
            )
156

UNCOV
157
        s3_client = connect_to(aws_access_key_id=account_id, region_name=region_name).s3
×
UNCOV
158
        bucket, _, path = location.removeprefix("s3://").partition("/")
×
UNCOV
159
        try:
×
UNCOV
160
            content = testutil.download_s3_object(s3_client, bucket, path)
×
UNCOV
161
        except ClientError:
×
UNCOV
162
            raise FailedTransformationException(
×
163
                transformation=INCLUDE_TRANSFORM,
164
                message="Error downloading S3 object '%s/%s'" % (bucket, path),
165
            )
UNCOV
166
        try:
×
UNCOV
167
            template_to_include = parse_template(content)
×
UNCOV
168
        except Exception as e:
×
UNCOV
169
            raise FailedTransformationException(transformation=INCLUDE_TRANSFORM, message=str(e))
×
UNCOV
170
        return {**template, **template_to_include}
×
171

172
    @staticmethod
1✔
173
    def _apply_global_macro_transformation(
1✔
174
        account_id: str,
175
        region_name,
176
        global_transform: GlobalTransform,
177
        template: dict,
178
        parameters: dict,
179
    ) -> Optional[dict]:
180
        macro_name = global_transform.name
1✔
181
        macros_store = get_cloudformation_store(
1✔
182
            account_id=account_id, region_name=region_name
183
        ).macros
184
        macro = macros_store.get(macro_name)
1✔
185
        if macro is None:
1✔
UNCOV
186
            raise RuntimeError(f"No definitions for global transform '{macro_name}'")
×
187
        transformation_parameters = global_transform.parameters or dict()
1✔
188
        transformed_template = execute_macro(
1✔
189
            account_id,
190
            region_name,
191
            parsed_template=template,
192
            macro=macro,
193
            stack_parameters=parameters,
194
            transformation_parameters=transformation_parameters,
195
        )
196
        # The type annotation on the v1 util appears to be incorrect.
197
        return transformed_template  # noqa
1✔
198

199
    def _apply_global_transform(
1✔
200
        self, global_transform: GlobalTransform, template: dict, parameters: dict
201
    ) -> dict:
202
        transform_name = global_transform.name
1✔
203
        if transform_name == EXTENSIONS_TRANSFORM:
1✔
204
            # Applied lazily in downstream tasks (see ChangeSetModelPreproc).
205
            transformed_template = template
1✔
206
        elif transform_name == SERVERLESS_TRANSFORM:
1✔
207
            transformed_template = self._apply_global_serverless_transformation(
1✔
208
                region_name=self._change_set.region_name,
209
                template=template,
210
                parameters=parameters,
211
            )
212
        elif transform_name == SECRETSMANAGER_TRANSFORM:
1✔
213
            # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/transform-aws-secretsmanager.html
214
            LOG.warning("%s is not yet supported. Ignoring.", SECRETSMANAGER_TRANSFORM)
×
215
            transformed_template = template
×
216
        elif transform_name == INCLUDE_TRANSFORM:
1✔
UNCOV
217
            transformed_template = self._apply_global_include(
×
218
                global_transform=global_transform,
219
                region_name=self._change_set.region_name,
220
                account_id=self._change_set.account_id,
221
                template=template,
222
                parameters=parameters,
223
            )
224
        else:
225
            transformed_template = self._apply_global_macro_transformation(
1✔
226
                account_id=self._change_set.account_id,
227
                region_name=self._change_set.region_name,
228
                global_transform=global_transform,
229
                template=template,
230
                parameters=parameters,
231
            )
232
        return transformed_template
1✔
233

234
    def transform(self) -> tuple[dict, dict]:
1✔
235
        self._setup_runtime_cache()
1✔
236

237
        node_template = self._change_set.update_model.node_template
1✔
238

239
        parameters_delta = self.visit_node_parameters(node_template.parameters)
1✔
240
        parameters_before = parameters_delta.before
1✔
241
        parameters_after = parameters_delta.after
1✔
242

243
        transform_delta: PreprocEntityDelta[list[GlobalTransform], list[GlobalTransform]] = (
1✔
244
            self.visit_node_transform(node_template.transform)
245
        )
246
        transform_before: Maybe[list[GlobalTransform]] = transform_delta.before
1✔
247
        transform_after: Maybe[list[GlobalTransform]] = transform_delta.after
1✔
248

249
        transformed_before_template = self._before_template
1✔
250
        if transform_before and not is_nothing(self._before_template):
1✔
251
            transformed_before_template = self._before_cache.get(_SCOPE_TRANSFORM_TEMPLATE_OUTCOME)
1✔
252
            if not transformed_before_template:
1✔
UNCOV
253
                transformed_before_template = self._before_template
×
UNCOV
254
                for before_global_transform in transform_before:
×
UNCOV
255
                    if not is_nothing(before_global_transform.name):
×
UNCOV
256
                        transformed_before_template = self._apply_global_transform(
×
257
                            global_transform=before_global_transform,
258
                            parameters=parameters_before,
259
                            template=transformed_before_template,
260
                        )
UNCOV
261
                self._before_cache[_SCOPE_TRANSFORM_TEMPLATE_OUTCOME] = transformed_before_template
×
262

263
        transformed_after_template = self._after_template
1✔
264
        if transform_after and not is_nothing(self._after_template):
1✔
265
            transformed_after_template = self._after_cache.get(_SCOPE_TRANSFORM_TEMPLATE_OUTCOME)
1✔
266
            if not transformed_after_template:
1✔
267
                transformed_after_template = self._after_template
1✔
268
                for after_global_transform in transform_after:
1✔
269
                    if not is_nothing(after_global_transform.name):
1✔
270
                        transformed_after_template = self._apply_global_transform(
1✔
271
                            global_transform=after_global_transform,
272
                            parameters=parameters_after,
273
                            template=transformed_after_template,
274
                        )
275
                self._after_cache[_SCOPE_TRANSFORM_TEMPLATE_OUTCOME] = transformed_after_template
1✔
276

277
        self._save_runtime_cache()
1✔
278

279
        return transformed_before_template, transformed_after_template
1✔
280

281
    def visit_node_global_transform(
1✔
282
        self, node_global_transform: NodeGlobalTransform
283
    ) -> PreprocEntityDelta[GlobalTransform, GlobalTransform]:
284
        change_type = node_global_transform.change_type
1✔
285

286
        name_delta = self.visit(node_global_transform.name)
1✔
287
        parameters_delta = self.visit(node_global_transform.parameters)
1✔
288

289
        before = Nothing
1✔
290
        if change_type != ChangeType.CREATED:
1✔
291
            before = GlobalTransform(name=name_delta.before, parameters=parameters_delta.before)
1✔
292
        after = Nothing
1✔
293
        if change_type != ChangeType.REMOVED:
1✔
294
            after = GlobalTransform(name=name_delta.after, parameters=parameters_delta.after)
1✔
295
        return PreprocEntityDelta(before=before, after=after)
1✔
296

297
    def visit_node_transform(
1✔
298
        self, node_transform: NodeTransform
299
    ) -> PreprocEntityDelta[list[GlobalTransform], list[GlobalTransform]]:
300
        change_type = node_transform.change_type
1✔
301
        before = list() if change_type != ChangeType.CREATED else Nothing
1✔
302
        after = list() if change_type != ChangeType.REMOVED else Nothing
1✔
303
        for change_set_entity in node_transform.global_transforms:
1✔
304
            delta: PreprocEntityDelta[GlobalTransform, GlobalTransform] = self.visit(
1✔
305
                change_set_entity=change_set_entity
306
            )
307
            delta_before = delta.before
1✔
308
            delta_after = delta.after
1✔
309
            if not is_nothing(before) and not is_nothing(delta_before):
1✔
310
                before.append(delta_before)
1✔
311
            if not is_nothing(after) and not is_nothing(delta_after):
1✔
312
                after.append(delta_after)
1✔
313
        return PreprocEntityDelta(before=before, after=after)
1✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc