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

localstack / localstack / 18505123992

14 Oct 2025 05:30PM UTC coverage: 86.888% (-0.01%) from 86.899%
18505123992

push

github

web-flow
S3: fix `aws-global` validation in CreateBucket (#13250)

10 of 10 new or added lines in 4 files covered. (100.0%)

831 existing lines in 40 files now uncovered.

68028 of 78294 relevant lines covered (86.89%)

0.87 hits per line

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

95.11
/localstack-core/localstack/services/cloudformation/engine/v2/change_set_model_preproc.py
1
from __future__ import annotations
1✔
2

3
import base64
1✔
4
import copy
1✔
5
import re
1✔
6
from collections.abc import Callable
1✔
7
from typing import Any, Final, Generic, TypeVar
1✔
8

9
from botocore.exceptions import ClientError
1✔
10

11
from localstack import config
1✔
12
from localstack.aws.api.cloudformation import ResourceStatus
1✔
13
from localstack.aws.api.ec2 import AvailabilityZoneList, DescribeAvailabilityZonesResult
1✔
14
from localstack.aws.connect import connect_to
1✔
15
from localstack.services.cloudformation.engine.v2.change_set_model import (
1✔
16
    ChangeSetEntity,
17
    ChangeType,
18
    Maybe,
19
    NodeArray,
20
    NodeCondition,
21
    NodeDependsOn,
22
    NodeDivergence,
23
    NodeIntrinsicFunction,
24
    NodeMapping,
25
    NodeObject,
26
    NodeOutput,
27
    NodeOutputs,
28
    NodeParameter,
29
    NodeParameters,
30
    NodeProperties,
31
    NodeProperty,
32
    NodeResource,
33
    NodeResources,
34
    NodeTemplate,
35
    Nothing,
36
    NothingType,
37
    Scope,
38
    TerminalValue,
39
    TerminalValueCreated,
40
    TerminalValueModified,
41
    TerminalValueRemoved,
42
    TerminalValueUnchanged,
43
    is_nothing,
44
)
45
from localstack.services.cloudformation.engine.v2.change_set_model_visitor import (
1✔
46
    ChangeSetModelVisitor,
47
)
48
from localstack.services.cloudformation.engine.v2.resolving import (
1✔
49
    REGEX_DYNAMIC_REF,
50
    extract_dynamic_reference,
51
    perform_dynamic_reference_lookup,
52
)
53
from localstack.services.cloudformation.engine.validations import ValidationError
1✔
54
from localstack.services.cloudformation.stores import (
1✔
55
    exports_map,
56
)
57
from localstack.services.cloudformation.v2.entities import ChangeSet
1✔
58
from localstack.services.cloudformation.v2.types import ResolvedResource
1✔
59
from localstack.utils.aws.arns import get_partition
1✔
60
from localstack.utils.numbers import to_number
1✔
61
from localstack.utils.objects import get_value_from_path
1✔
62
from localstack.utils.run import to_str
1✔
63
from localstack.utils.strings import to_bytes
1✔
64
from localstack.utils.urls import localstack_host
1✔
65

66
_AWS_URL_SUFFIX = localstack_host().host  # The value in AWS is "amazonaws.com"
1✔
67

68
_PSEUDO_PARAMETERS: Final[set[str]] = {
1✔
69
    "AWS::Partition",
70
    "AWS::AccountId",
71
    "AWS::Region",
72
    "AWS::StackName",
73
    "AWS::StackId",
74
    "AWS::URLSuffix",
75
    "AWS::NoValue",
76
    "AWS::NotificationARNs",
77
}
78

79
TBefore = TypeVar("TBefore")
1✔
80
TAfter = TypeVar("TAfter")
1✔
81
_T = TypeVar("_T")
1✔
82

83
REGEX_OUTPUT_APIGATEWAY = re.compile(
1✔
84
    rf"^(https?://.+\.execute-api\.)(?:[^-]+-){{2,3}}\d\.(amazonaws\.com|{_AWS_URL_SUFFIX})/?(.*)$"
85
)
86
MOCKED_REFERENCE = "unknown"
1✔
87

88
VALID_LOGICAL_RESOURCE_ID_RE = re.compile(r"^[A-Za-z0-9]+$")
1✔
89

90

91
class PreprocEntityDelta(Generic[TBefore, TAfter]):
1✔
92
    before: Maybe[TBefore]
1✔
93
    after: Maybe[TAfter]
1✔
94

95
    def __init__(self, before: Maybe[TBefore] = Nothing, after: Maybe[TAfter] = Nothing):
1✔
96
        self.before = before
1✔
97
        self.after = after
1✔
98

99
    def __eq__(self, other):
1✔
100
        if not isinstance(other, PreprocEntityDelta):
×
101
            return False
×
102
        return self.before == other.before and self.after == other.after
×
103

104

105
class PreprocProperties:
1✔
106
    properties: dict[str, Any]
1✔
107

108
    def __init__(self, properties: dict[str, Any]):
1✔
109
        self.properties = properties
1✔
110

111
    def __eq__(self, other):
1✔
112
        if not isinstance(other, PreprocProperties):
1✔
113
            return False
×
114
        return self.properties == other.properties
1✔
115

116

117
class PreprocResource:
1✔
118
    logical_id: str
1✔
119
    physical_resource_id: str | None
1✔
120
    condition: bool | None
1✔
121
    resource_type: str
1✔
122
    properties: PreprocProperties
1✔
123
    depends_on: list[str] | None
1✔
124
    requires_replacement: bool
1✔
125
    status: ResourceStatus | None
1✔
126

127
    def __init__(
1✔
128
        self,
129
        logical_id: str,
130
        physical_resource_id: str,
131
        condition: bool | None,
132
        resource_type: str,
133
        properties: PreprocProperties,
134
        depends_on: list[str] | None,
135
        requires_replacement: bool,
136
        status: ResourceStatus | None = None,
137
    ):
138
        self.logical_id = logical_id
1✔
139
        self.physical_resource_id = physical_resource_id
1✔
140
        self.condition = condition
1✔
141
        self.resource_type = resource_type
1✔
142
        self.properties = properties
1✔
143
        self.depends_on = depends_on
1✔
144
        self.requires_replacement = requires_replacement
1✔
145
        self.status = status
1✔
146

147
    @staticmethod
1✔
148
    def _compare_conditions(c1: bool, c2: bool):
1✔
149
        # The lack of condition equates to a true condition.
150
        c1 = c1 if isinstance(c1, bool) else True
1✔
151
        c2 = c2 if isinstance(c2, bool) else True
1✔
152
        return c1 == c2
1✔
153

154
    def __eq__(self, other):
1✔
155
        if not isinstance(other, PreprocResource):
1✔
156
            return False
1✔
157
        return all(
1✔
158
            [
159
                self.logical_id == other.logical_id,
160
                self._compare_conditions(self.condition, other.condition),
161
                self.resource_type == other.resource_type,
162
                self.properties == other.properties,
163
            ]
164
        )
165

166

167
class PreprocOutput:
1✔
168
    name: str
1✔
169
    value: Any
1✔
170
    export: Any | None
1✔
171
    condition: bool | None
1✔
172

173
    def __init__(self, name: str, value: Any, export: Any | None, condition: bool | None):
1✔
174
        self.name = name
1✔
175
        self.value = value
1✔
176
        self.export = export
1✔
177
        self.condition = condition
1✔
178

179
    def __eq__(self, other):
1✔
180
        if not isinstance(other, PreprocOutput):
×
181
            return False
×
182
        return all(
×
183
            [
184
                self.name == other.name,
185
                self.value == other.value,
186
                self.export == other.export,
187
                self.condition == other.condition,
188
            ]
189
        )
190

191

192
class ChangeSetModelPreproc(ChangeSetModelVisitor):
1✔
193
    _change_set: Final[ChangeSet]
1✔
194
    _before_resolved_resources: Final[dict]
1✔
195
    _before_cache: Final[dict[Scope, Any]]
1✔
196
    _after_cache: Final[dict[Scope, Any]]
1✔
197

198
    def __init__(self, change_set: ChangeSet):
1✔
199
        self._change_set = change_set
1✔
200
        self._before_resolved_resources = change_set.stack.resolved_resources
1✔
201
        self._before_cache = {}
1✔
202
        self._after_cache = {}
1✔
203

204
    def _setup_runtime_cache(self) -> None:
1✔
205
        runtime_cache_key = self.__class__.__name__
1✔
206

207
        self._before_cache.clear()
1✔
208
        self._after_cache.clear()
1✔
209

210
        before_runtime_cache = self._change_set.update_model.before_runtime_cache
1✔
211
        if cache := before_runtime_cache.get(runtime_cache_key):
1✔
212
            self._before_cache.update(cache)
1✔
213

214
        after_runtime_cache = self._change_set.update_model.after_runtime_cache
1✔
215
        if cache := after_runtime_cache.get(runtime_cache_key):
1✔
216
            self._after_cache.update(cache)
×
217

218
    def _save_runtime_cache(self) -> None:
1✔
219
        runtime_cache_key = self.__class__.__name__
1✔
220

221
        before_runtime_cache = self._change_set.update_model.before_runtime_cache
1✔
222
        before_runtime_cache[runtime_cache_key] = copy.deepcopy(self._before_cache)
1✔
223

224
        after_runtime_cache = self._change_set.update_model.after_runtime_cache
1✔
225
        after_runtime_cache[runtime_cache_key] = copy.deepcopy(self._after_cache)
1✔
226

227
    def process(self) -> None:
1✔
228
        self._setup_runtime_cache()
1✔
229
        node_template = self._change_set.update_model.node_template
1✔
230
        node_conditions = self._change_set.update_model.node_template.conditions
1✔
231
        self.visit(node_conditions)
1✔
232
        self.visit(node_template)
1✔
233
        self._save_runtime_cache()
1✔
234

235
    def _get_node_resource_for(
1✔
236
        self, resource_name: str, node_template: NodeTemplate
237
    ) -> NodeResource:
238
        # TODO: this could be improved with hashmap lookups if the Node contained bindings and not lists.
239
        for node_resource in node_template.resources.resources:
1✔
240
            if node_resource.name == resource_name:
1✔
241
                self.visit(node_resource)
1✔
242
                return node_resource
1✔
243
        raise ValidationError(
1✔
244
            f"Template format error: Unresolved resource dependencies [{resource_name}] in the Resources block of the template"
245
        )
246

247
    def _get_node_property_for(
1✔
248
        self, property_name: str, node_resource: NodeResource
249
    ) -> NodeProperty | None:
250
        # TODO: this could be improved with hashmap lookups if the Node contained bindings and not lists.
251
        for node_property in node_resource.properties.properties:
1✔
252
            if node_property.name == property_name:
1✔
253
                self.visit(node_property)
1✔
254
                return node_property
1✔
255
        return None
1✔
256

257
    def _deployed_property_value_of(
1✔
258
        self, resource_logical_id: str, property_name: str, resolved_resources: dict
259
    ) -> Any:
260
        # We have to override this function to make sure it does not try to access the
261
        # resolved resource
262

263
        # Before we can obtain deployed value for a resource, we need to first ensure to
264
        # process the resource if this wasn't processed already. Ideally, values should only
265
        # be accessible through delta objects, to ensure computation is always complete at
266
        # every level.
267
        _ = self._get_node_resource_for(
1✔
268
            resource_name=resource_logical_id,
269
            node_template=self._change_set.update_model.node_template,
270
        )
271
        resolved_resource = resolved_resources.get(resource_logical_id)
1✔
272
        if resolved_resource is None:
1✔
273
            raise RuntimeError(
1✔
274
                f"No deployed instances of resource '{resource_logical_id}' were found"
275
            )
276
        properties = resolved_resource.get("Properties", {})
1✔
277
        # TODO support structured properties, e.g. NestedStack.Outputs.OutputName
278
        property_value: Any | None = get_value_from_path(properties, property_name)
1✔
279

280
        if property_value:
1✔
281
            if not isinstance(property_value, (str, list, dict)):
1✔
282
                # Str: Standard expected type. TODO validate bools and numbers
283
                # List: Multiple resource types can return a list of values e.g. AWS::EC2::VPC.
284
                # Dict: Custom resources in CloudFormation can return arbitrary data structures.
285
                raise RuntimeError(
×
286
                    f"Accessing property '{property_name}' from '{resource_logical_id}' resulted in a non-string value nor list"
287
                )
288
            return property_value
1✔
289
        elif config.CFN_IGNORE_UNSUPPORTED_RESOURCE_TYPES:
1✔
290
            return MOCKED_REFERENCE
1✔
291

292
        return property_value
×
293

294
    def _before_deployed_property_value_of(
1✔
295
        self, resource_logical_id: str, property_name: str
296
    ) -> Any:
297
        return self._deployed_property_value_of(
1✔
298
            resource_logical_id=resource_logical_id,
299
            property_name=property_name,
300
            resolved_resources=self._before_resolved_resources,
301
        )
302

303
    def _after_deployed_property_value_of(
1✔
304
        self, resource_logical_id: str, property_name: str
305
    ) -> str | None:
306
        return self._before_deployed_property_value_of(
1✔
307
            resource_logical_id=resource_logical_id, property_name=property_name
308
        )
309

310
    def _get_node_mapping(self, map_name: str) -> NodeMapping:
1✔
311
        mappings: list[NodeMapping] = self._change_set.update_model.node_template.mappings.mappings
1✔
312
        # TODO: another scenarios suggesting property lookups might be preferable.
313
        for mapping in mappings:
1✔
314
            if mapping.name == map_name:
1✔
315
                self.visit(mapping)
1✔
316
                return mapping
1✔
317
        raise RuntimeError(f"Undefined '{map_name}' mapping")
×
318

319
    def _get_node_parameter_if_exists(self, parameter_name: str) -> Maybe[NodeParameter]:
1✔
320
        parameters: list[NodeParameter] = (
1✔
321
            self._change_set.update_model.node_template.parameters.parameters
322
        )
323
        # TODO: another scenarios suggesting property lookups might be preferable.
324
        for parameter in parameters:
1✔
325
            if parameter.name == parameter_name:
1✔
326
                self.visit(parameter)
1✔
327
                return parameter
1✔
328
        return Nothing
1✔
329

330
    def _get_node_condition_if_exists(self, condition_name: str) -> Maybe[NodeCondition]:
1✔
331
        conditions: list[NodeCondition] = (
1✔
332
            self._change_set.update_model.node_template.conditions.conditions
333
        )
334
        # TODO: another scenarios suggesting property lookups might be preferable.
335
        for condition in conditions:
1✔
336
            if condition.name == condition_name:
1✔
337
                self.visit(condition)
1✔
338
                return condition
1✔
339
        return Nothing
×
340

341
    def _resolve_condition(self, logical_id: str) -> PreprocEntityDelta:
1✔
342
        node_condition = self._get_node_condition_if_exists(condition_name=logical_id)
1✔
343
        if isinstance(node_condition, NodeCondition):
1✔
344
            condition_delta = self.visit(node_condition)
1✔
345
            return condition_delta
1✔
346
        raise RuntimeError(f"No condition '{logical_id}' was found.")
×
347

348
    def _resolve_pseudo_parameter(self, pseudo_parameter_name: str) -> Any:
1✔
349
        match pseudo_parameter_name:
1✔
350
            case "AWS::Partition":
1✔
351
                return get_partition(self._change_set.region_name)
1✔
352
            case "AWS::AccountId":
1✔
353
                return self._change_set.stack.account_id
1✔
354
            case "AWS::Region":
1✔
355
                return self._change_set.stack.region_name
1✔
356
            case "AWS::StackName":
1✔
357
                return self._change_set.stack.stack_name
1✔
358
            case "AWS::StackId":
1✔
359
                return self._change_set.stack.stack_id
1✔
360
            case "AWS::URLSuffix":
1✔
361
                return _AWS_URL_SUFFIX
1✔
362
            case "AWS::NoValue":
×
363
                return None
×
364
            case _:
×
365
                raise RuntimeError(f"The use of '{pseudo_parameter_name}' is currently unsupported")
×
366

367
    def _resolve_reference(self, logical_id: str) -> PreprocEntityDelta:
1✔
368
        if logical_id in _PSEUDO_PARAMETERS:
1✔
369
            pseudo_parameter_value = self._resolve_pseudo_parameter(
1✔
370
                pseudo_parameter_name=logical_id
371
            )
372
            # Pseudo parameters are constants within the lifecycle of a template.
373
            return PreprocEntityDelta(before=pseudo_parameter_value, after=pseudo_parameter_value)
1✔
374

375
        node_parameter = self._get_node_parameter_if_exists(parameter_name=logical_id)
1✔
376
        if isinstance(node_parameter, NodeParameter):
1✔
377
            parameter_delta = self.visit(node_parameter)
1✔
378
            return parameter_delta
1✔
379

380
        node_resource = self._get_node_resource_for(
1✔
381
            resource_name=logical_id, node_template=self._change_set.update_model.node_template
382
        )
383
        resource_delta = self.visit(node_resource)
1✔
384
        before = resource_delta.before
1✔
385
        after = resource_delta.after
1✔
386
        return PreprocEntityDelta(before=before, after=after)
1✔
387

388
    def _resolve_mapping(
1✔
389
        self, map_name: str, top_level_key: str, second_level_key
390
    ) -> PreprocEntityDelta:
391
        # TODO: add support for nested intrinsic functions, and KNOWN AFTER APPLY logical ids.
392
        node_mapping: NodeMapping = self._get_node_mapping(map_name=map_name)
1✔
393
        top_level_value = node_mapping.bindings.bindings.get(top_level_key)
1✔
394
        if not isinstance(top_level_value, NodeObject):
1✔
395
            error_key = "::".join([map_name, top_level_key, second_level_key])
1✔
396
            raise ValidationError(f"Template error: Unable to get mapping for {error_key}")
1✔
397
        second_level_value = top_level_value.bindings.get(second_level_key)
1✔
398
        if not isinstance(second_level_value, (TerminalValue, NodeArray, NodeObject)):
1✔
399
            error_key = "::".join([map_name, top_level_key, second_level_key])
1✔
400
            raise ValidationError(f"Template error: Unable to get mapping for {error_key}")
1✔
401
        mapping_value_delta = self.visit(second_level_value)
1✔
402
        return mapping_value_delta
1✔
403

404
    def visit(self, change_set_entity: ChangeSetEntity) -> PreprocEntityDelta:
1✔
405
        entity_scope = change_set_entity.scope
1✔
406
        if entity_scope in self._before_cache and entity_scope in self._after_cache:
1✔
407
            before = self._before_cache[entity_scope]
1✔
408
            after = self._after_cache[entity_scope]
1✔
409
            return PreprocEntityDelta(before=before, after=after)
1✔
410
        delta = super().visit(change_set_entity=change_set_entity)
1✔
411
        if isinstance(delta, PreprocEntityDelta):
1✔
412
            delta = self._maybe_perform_replacements(delta)
1✔
413
            self._before_cache[entity_scope] = delta.before
1✔
414
            self._after_cache[entity_scope] = delta.after
1✔
415
        return delta
1✔
416

417
    def _maybe_perform_replacements(self, delta: PreprocEntityDelta) -> PreprocEntityDelta:
1✔
418
        delta = self._maybe_perform_static_replacements(delta)
1✔
419
        delta = self._maybe_perform_dynamic_replacements(delta)
1✔
420
        return delta
1✔
421

422
    def _maybe_perform_static_replacements(self, delta: PreprocEntityDelta) -> PreprocEntityDelta:
1✔
423
        return self._maybe_perform_on_delta(delta, self._perform_static_replacements)
1✔
424

425
    def _maybe_perform_dynamic_replacements(self, delta: PreprocEntityDelta) -> PreprocEntityDelta:
1✔
426
        return self._maybe_perform_on_delta(delta, self._perform_dynamic_replacements)
1✔
427

428
    def _maybe_perform_on_delta(
1✔
429
        self, delta: PreprocEntityDelta | None, f: Callable[[_T], _T]
430
    ) -> PreprocEntityDelta | None:
431
        if isinstance(delta.before, str):
1✔
432
            delta.before = f(delta.before)
1✔
433
        if isinstance(delta.after, str):
1✔
434
            delta.after = f(delta.after)
1✔
435
        return delta
1✔
436

437
    def _perform_dynamic_replacements(self, value: _T) -> _T:
1✔
438
        if not isinstance(value, str):
1✔
439
            return value
×
440

441
        if dynamic_ref := extract_dynamic_reference(value):
1✔
442
            new_value = perform_dynamic_reference_lookup(
1✔
443
                reference=dynamic_ref,
444
                account_id=self._change_set.account_id,
445
                region_name=self._change_set.region_name,
446
            )
447
            if new_value:
1✔
448
                # We need to use a function here, to avoid backslash processing by regex.
449
                # From the regex sub documentation:
450
                # repl can be a string or a function; if it is a string, any backslash escapes in it are processed.
451
                # Using a function, we can avoid this processing.
452
                return REGEX_DYNAMIC_REF.sub(lambda _: new_value, value)
1✔
453

454
        return value
1✔
455

456
    @staticmethod
1✔
457
    def _perform_static_replacements(value: str) -> str:
1✔
458
        api_match = REGEX_OUTPUT_APIGATEWAY.match(value)
1✔
459
        if api_match and value not in config.CFN_STRING_REPLACEMENT_DENY_LIST:
1✔
460
            prefix = api_match[1]
1✔
461
            host = api_match[2]
1✔
462
            path = api_match[3]
1✔
463
            port = localstack_host().port
1✔
464
            value = f"{prefix}{host}:{port}/{path}"
1✔
465
            return value
1✔
466

467
        return value
1✔
468

469
    def _cached_apply(
1✔
470
        self, scope: Scope, arguments_delta: PreprocEntityDelta, resolver: Callable[[Any], Any]
471
    ) -> PreprocEntityDelta:
472
        """
473
        Applies the resolver function to the given input delta if and only if the required
474
        values are not already present in the runtime caches. This function handles both
475
        the 'before' and 'after' components of the delta independently.
476

477
        The resolver function receives either the 'before' or 'after' value from the input
478
        delta and returns a resolved value. If the result returned by the resolver is
479
        itself a PreprocEntityDelta, the function automatically extracts the appropriate
480
        component from it:  the 'before' value if the input was 'before', and the 'after'
481
        value if the input was 'after'.
482

483
        This function only reads from the cache and does not update it. It is the caller's
484
        responsibility to handle caching, either manually or via the upstream visit method
485
        of this class.
486

487
        Args:
488
            scope (Scope): The current scope used as a key for cache lookup.
489
            arguments_delta (PreprocEntityDelta): The delta containing 'before' and 'after' values to resolve.
490
            resolver (Callable[[Any], Any]): Function to apply on uncached 'before' or 'after' argument values.
491

492
        Returns:
493
            PreprocEntityDelta: A new delta with resolved 'before' and 'after' values.
494
        """
495

496
        # TODO: Update all visit_* methods in this class and its subclasses to use this function.
497
        #       This ensures maximal reuse of precomputed 'before' (and 'after') values from
498
        #       prior runtimes on the change sets template, thus avoiding unnecessary recomputation.
499

500
        arguments_before = arguments_delta.before
1✔
501
        arguments_after = arguments_delta.after
1✔
502

503
        before = self._before_cache.get(scope, Nothing)
1✔
504
        if is_nothing(before) and not is_nothing(arguments_before):
1✔
505
            before = resolver(arguments_before)
1✔
506
            if isinstance(before, PreprocEntityDelta):
1✔
507
                before = before.before
1✔
508

509
        after = self._after_cache.get(scope, Nothing)
1✔
510
        if is_nothing(after) and not is_nothing(arguments_after):
1✔
511
            after = resolver(arguments_after)
1✔
512
            if isinstance(after, PreprocEntityDelta):
1✔
513
                after = after.after
1✔
514

515
        return PreprocEntityDelta(before=before, after=after)
1✔
516

517
    def visit_node_property(self, node_property: NodeProperty) -> PreprocEntityDelta:
1✔
518
        return self.visit(node_property.value)
1✔
519

520
    def visit_terminal_value_modified(
1✔
521
        self, terminal_value_modified: TerminalValueModified
522
    ) -> PreprocEntityDelta:
523
        return PreprocEntityDelta(
1✔
524
            before=terminal_value_modified.value,
525
            after=terminal_value_modified.modified_value,
526
        )
527

528
    def visit_terminal_value_created(
1✔
529
        self, terminal_value_created: TerminalValueCreated
530
    ) -> PreprocEntityDelta:
531
        return PreprocEntityDelta(after=terminal_value_created.value)
1✔
532

533
    def visit_terminal_value_removed(
1✔
534
        self, terminal_value_removed: TerminalValueRemoved
535
    ) -> PreprocEntityDelta:
536
        return PreprocEntityDelta(before=terminal_value_removed.value)
1✔
537

538
    def visit_terminal_value_unchanged(
1✔
539
        self, terminal_value_unchanged: TerminalValueUnchanged
540
    ) -> PreprocEntityDelta:
541
        return PreprocEntityDelta(
1✔
542
            before=terminal_value_unchanged.value,
543
            after=terminal_value_unchanged.value,
544
        )
545

546
    def visit_node_divergence(self, node_divergence: NodeDivergence) -> PreprocEntityDelta:
1✔
547
        before_delta = self.visit(node_divergence.value)
1✔
548
        after_delta = self.visit(node_divergence.divergence)
1✔
549
        return PreprocEntityDelta(before=before_delta.before, after=after_delta.after)
1✔
550

551
    def visit_node_object(self, node_object: NodeObject) -> PreprocEntityDelta:
1✔
552
        node_change_type = node_object.change_type
1✔
553
        before = {} if node_change_type != ChangeType.CREATED else Nothing
1✔
554
        after = {} if node_change_type != ChangeType.REMOVED else Nothing
1✔
555
        for name, change_set_entity in node_object.bindings.items():
1✔
556
            delta: PreprocEntityDelta = self.visit(change_set_entity=change_set_entity)
1✔
557
            delta_before = delta.before
1✔
558
            delta_after = delta.after
1✔
559
            if not is_nothing(before) and not is_nothing(delta_before):
1✔
560
                before[name] = delta_before
1✔
561
            if not is_nothing(after) and not is_nothing(delta_after):
1✔
562
                after[name] = delta_after
1✔
563
        return PreprocEntityDelta(before=before, after=after)
1✔
564

565
    def _resolve_attribute(self, arguments: str | list[str], select_before: bool) -> str:
1✔
566
        # TODO: add arguments validation.
567
        arguments_list: list[str]
568
        if isinstance(arguments, str):
1✔
569
            arguments_list = arguments.split(".")
1✔
570
        else:
571
            arguments_list = arguments
1✔
572

573
        if len(arguments_list) != 2:
1✔
574
            raise ValidationError(
1✔
575
                "Template error: every Fn::GetAtt object requires two non-empty parameters, the resource name and the resource attribute"
576
            )
577

578
        logical_name_of_resource = arguments_list[0]
1✔
579
        attribute_name = arguments_list[1]
1✔
580

581
        node_resource = self._get_node_resource_for(
1✔
582
            resource_name=logical_name_of_resource,
583
            node_template=self._change_set.update_model.node_template,
584
        )
585

586
        if not is_nothing(node_resource.condition_reference):
1✔
587
            condition = self._get_node_condition_if_exists(node_resource.condition_reference.value)
1✔
588
            evaluation_result = self._resolve_condition(condition.name)
1✔
589

590
            if select_before and not evaluation_result.before:
1✔
591
                raise ValidationError(
1✔
592
                    f"Template format error: Unresolved resource dependencies [{logical_name_of_resource}] in the Resources block of the template"
593
                )
594

595
            if not select_before and not evaluation_result.after:
1✔
596
                raise ValidationError(
×
597
                    f"Template format error: Unresolved resource dependencies [{logical_name_of_resource}] in the Resources block of the template"
598
                )
599

600
        node_property: NodeProperty | None = self._get_node_property_for(
1✔
601
            property_name=attribute_name, node_resource=node_resource
602
        )
603
        if node_property is not None:
1✔
604
            # The property is statically defined in the template and its value can be computed.
605
            property_delta = self.visit(node_property)
1✔
606
            value = property_delta.before if select_before else property_delta.after
1✔
607
        else:
608
            # The property is not statically defined and must therefore be available in
609
            # the properties deployed set.
610
            if select_before:
1✔
611
                value = self._before_deployed_property_value_of(
1✔
612
                    resource_logical_id=logical_name_of_resource,
613
                    property_name=attribute_name,
614
                )
615
            else:
616
                value = self._after_deployed_property_value_of(
1✔
617
                    resource_logical_id=logical_name_of_resource,
618
                    property_name=attribute_name,
619
                )
620
        return value
1✔
621

622
    def visit_node_intrinsic_function_fn_get_att(
1✔
623
        self, node_intrinsic_function: NodeIntrinsicFunction
624
    ) -> PreprocEntityDelta:
625
        # TODO: validate the return value according to the spec.
626
        arguments_delta = self.visit(node_intrinsic_function.arguments)
1✔
627
        before_arguments: Maybe[str | list[str]] = arguments_delta.before
1✔
628
        after_arguments: Maybe[str | list[str]] = arguments_delta.after
1✔
629

630
        before = self._before_cache.get(node_intrinsic_function.scope, Nothing)
1✔
631
        if is_nothing(before) and not is_nothing(before_arguments):
1✔
632
            before = self._resolve_attribute(arguments=before_arguments, select_before=True)
1✔
633

634
        after = self._after_cache.get(node_intrinsic_function.scope, Nothing)
1✔
635
        if is_nothing(after) and not is_nothing(after_arguments):
1✔
636
            after = self._resolve_attribute(arguments=after_arguments, select_before=False)
1✔
637

638
        return PreprocEntityDelta(before=before, after=after)
1✔
639

640
    def visit_node_intrinsic_function_fn_equals(
1✔
641
        self, node_intrinsic_function: NodeIntrinsicFunction
642
    ) -> PreprocEntityDelta:
643
        # TODO: add argument shape validation.
644
        def _compute_fn_equals(args: list[Any]) -> bool:
1✔
645
            return args[0] == args[1]
1✔
646

647
        arguments_delta = self.visit(node_intrinsic_function.arguments)
1✔
648

649
        if isinstance(arguments_delta.after, list) and len(arguments_delta.after) != 2:
1✔
650
            raise ValidationError(
×
651
                "Template error: every Fn::Equals object requires a list of 2 string parameters."
652
            )
653

654
        delta = self._cached_apply(
1✔
655
            scope=node_intrinsic_function.scope,
656
            arguments_delta=arguments_delta,
657
            resolver=_compute_fn_equals,
658
        )
659
        return delta
1✔
660

661
    def visit_node_intrinsic_function_fn_if(
1✔
662
        self, node_intrinsic_function: NodeIntrinsicFunction
663
    ) -> PreprocEntityDelta:
664
        # `if` needs to be short-circuiting i.e. if the condition is True we don't evaluate the
665
        # False branch. If the condition is False, we don't evaluate the True branch.
666
        if len(node_intrinsic_function.arguments.array) != 3:
1✔
667
            raise ValueError(
×
668
                f"Incorrectly constructed Fn::If usage, expected 3 arguments, found {len(node_intrinsic_function.arguments.array)}"
669
            )
670

671
        condition_delta = self.visit(node_intrinsic_function.arguments.array[0])
1✔
672
        if_delta = PreprocEntityDelta()
1✔
673
        if not is_nothing(condition_delta.before):
1✔
674
            node_condition = self._get_node_condition_if_exists(
1✔
675
                condition_name=condition_delta.before
676
            )
677
            condition_value = self.visit(node_condition).before
1✔
678
            if condition_value:
1✔
679
                arg_delta = self.visit(node_intrinsic_function.arguments.array[1])
1✔
680
            else:
681
                arg_delta = self.visit(node_intrinsic_function.arguments.array[2])
1✔
682
            if_delta.before = arg_delta.before
1✔
683

684
        if not is_nothing(condition_delta.after):
1✔
685
            node_condition = self._get_node_condition_if_exists(
1✔
686
                condition_name=condition_delta.after
687
            )
688
            condition_value = self.visit(node_condition).after
1✔
689
            if condition_value:
1✔
690
                arg_delta = self.visit(node_intrinsic_function.arguments.array[1])
1✔
691
            else:
692
                arg_delta = self.visit(node_intrinsic_function.arguments.array[2])
1✔
693
            if_delta.after = arg_delta.after
1✔
694

695
        return if_delta
1✔
696

697
    def visit_node_intrinsic_function_fn_and(
1✔
698
        self, node_intrinsic_function: NodeIntrinsicFunction
699
    ) -> PreprocEntityDelta:
700
        def _compute_fn_and(args: list[bool]) -> bool:
1✔
701
            result = all(args)
1✔
702
            return result
1✔
703

704
        arguments_delta = self.visit(node_intrinsic_function.arguments)
1✔
705
        delta = self._cached_apply(
1✔
706
            scope=node_intrinsic_function.scope,
707
            arguments_delta=arguments_delta,
708
            resolver=_compute_fn_and,
709
        )
710
        return delta
1✔
711

712
    def visit_node_intrinsic_function_fn_or(
1✔
713
        self, node_intrinsic_function: NodeIntrinsicFunction
714
    ) -> PreprocEntityDelta:
715
        def _compute_fn_or(args: list[bool]):
1✔
716
            result = any(args)
1✔
717
            return result
1✔
718

719
        arguments_delta = self.visit(node_intrinsic_function.arguments)
1✔
720
        delta = self._cached_apply(
1✔
721
            scope=node_intrinsic_function.scope,
722
            arguments_delta=arguments_delta,
723
            resolver=_compute_fn_or,
724
        )
725
        return delta
1✔
726

727
    def visit_node_intrinsic_function_fn_not(
1✔
728
        self, node_intrinsic_function: NodeIntrinsicFunction
729
    ) -> PreprocEntityDelta:
730
        def _compute_fn_not(arg: list[bool] | bool) -> bool:
1✔
731
            # Is the argument ever a lone boolean?
732
            if isinstance(arg, list):
1✔
733
                return not arg[0]
1✔
734
            else:
735
                return not arg
×
736

737
        arguments_delta = self.visit(node_intrinsic_function.arguments)
1✔
738
        delta = self._cached_apply(
1✔
739
            scope=node_intrinsic_function.scope,
740
            arguments_delta=arguments_delta,
741
            resolver=_compute_fn_not,
742
        )
743
        return delta
1✔
744

745
    def visit_node_intrinsic_function_fn_sub(
1✔
746
        self, node_intrinsic_function: NodeIntrinsicFunction
747
    ) -> PreprocEntityDelta:
748
        def _compute_sub(args: str | list[Any], select_before: bool) -> str:
1✔
749
            # TODO: add further schema validation.
750
            string_template: str
751
            sub_parameters: dict
752
            if isinstance(args, str):
1✔
753
                string_template = args
1✔
754
                sub_parameters = {}
1✔
755
            elif (
1✔
756
                isinstance(args, list)
757
                and len(args) == 2
758
                and isinstance(args[0], str)
759
                and isinstance(args[1], dict)
760
            ):
761
                string_template = args[0]
1✔
762
                sub_parameters = args[1]
1✔
763
            else:
764
                raise RuntimeError(
×
765
                    "Invalid arguments shape for Fn::Sub, expected a String "
766
                    f"or a Tuple of String and Map but got '{args}'"
767
                )
768
            sub_string = string_template
1✔
769
            template_variable_names = re.findall("\\${([^}]+)}", string_template)
1✔
770
            for template_variable_name in template_variable_names:
1✔
771
                template_variable_value = Nothing
1✔
772

773
                # Try to resolve the variable name as pseudo parameter.
774
                if template_variable_name in _PSEUDO_PARAMETERS:
1✔
775
                    template_variable_value = self._resolve_pseudo_parameter(
1✔
776
                        pseudo_parameter_name=template_variable_name
777
                    )
778

779
                # Try to resolve the variable name as an entry to the defined parameters.
780
                elif template_variable_name in sub_parameters:
1✔
781
                    template_variable_value = sub_parameters[template_variable_name]
1✔
782

783
                # Try to resolve the variable name as GetAtt.
784
                elif "." in template_variable_name:
1✔
785
                    try:
1✔
786
                        template_variable_value = self._resolve_attribute(
1✔
787
                            arguments=template_variable_name, select_before=select_before
788
                        )
789
                    except RuntimeError:
1✔
790
                        pass
1✔
791

792
                # Try to resolve the variable name as Ref.
793
                else:
794
                    try:
1✔
795
                        resource_delta = self._resolve_reference(logical_id=template_variable_name)
1✔
796
                        template_variable_value = (
1✔
797
                            resource_delta.before if select_before else resource_delta.after
798
                        )
799
                        if isinstance(template_variable_value, PreprocResource):
1✔
800
                            template_variable_value = template_variable_value.physical_resource_id
1✔
801
                    except RuntimeError:
×
802
                        pass
×
803

804
                if is_nothing(template_variable_value):
1✔
805
                    raise RuntimeError(
1✔
806
                        f"Undefined variable name in Fn::Sub string template '{template_variable_name}'"
807
                    )
808

809
                if not isinstance(template_variable_value, str):
1✔
810
                    template_variable_value = str(template_variable_value)
1✔
811

812
                sub_string = sub_string.replace(
1✔
813
                    f"${{{template_variable_name}}}", template_variable_value
814
                )
815

816
            # FIXME: the following type reduction is ported from v1; however it appears as though such
817
            #        reduction is not performed by the engine, and certainly not at this depth given the
818
            #        lack of context. This section should be removed with Fn::Sub always retuning a string
819
            #        and the resource providers reviewed.
820
            account_id = self._change_set.account_id
1✔
821
            is_another_account_id = sub_string.isdigit() and len(sub_string) == len(account_id)
1✔
822
            if sub_string == account_id or is_another_account_id:
1✔
823
                result = sub_string
1✔
824
            elif sub_string.isdigit():
1✔
825
                result = int(sub_string)
1✔
826
            else:
827
                try:
1✔
828
                    result = float(sub_string)
1✔
829
                except ValueError:
1✔
830
                    result = sub_string
1✔
831
            return result
1✔
832

833
        arguments_delta = self.visit(node_intrinsic_function.arguments)
1✔
834
        arguments_before = arguments_delta.before
1✔
835
        arguments_after = arguments_delta.after
1✔
836
        before = self._before_cache.get(node_intrinsic_function.scope, Nothing)
1✔
837
        if is_nothing(before) and not is_nothing(arguments_before):
1✔
838
            before = _compute_sub(args=arguments_before, select_before=True)
1✔
839
        after = self._after_cache.get(node_intrinsic_function.scope, Nothing)
1✔
840
        if is_nothing(after) and not is_nothing(arguments_after):
1✔
841
            after = _compute_sub(args=arguments_after, select_before=False)
1✔
842
        return PreprocEntityDelta(before=before, after=after)
1✔
843

844
    def visit_node_intrinsic_function_fn_join(
1✔
845
        self, node_intrinsic_function: NodeIntrinsicFunction
846
    ) -> PreprocEntityDelta:
847
        # TODO: add support for schema validation.
848
        # TODO: add tests for joining non string values.
849
        def _compute_fn_join(args: list[Any]) -> str | NothingType:
1✔
850
            if not (isinstance(args, list) and len(args) == 2):
1✔
851
                return Nothing
1✔
852
            delimiter: str = str(args[0])
1✔
853
            values: list[Any] = args[1]
1✔
854
            if not isinstance(values, list):
1✔
855
                # shortcut if values is the empty string, for example:
856
                # {"Fn::Join": ["", {"Ref": <parameter>}]}
857
                # CDK bootstrap does this
858
                if values == "":
1✔
859
                    return ""
1✔
860
                raise RuntimeError(f"Invalid arguments list definition for Fn::Join: '{args}'")
1✔
861
            str_values: list[str] = []
1✔
862
            for value in values:
1✔
863
                if value is None:
1✔
864
                    continue
1✔
865
                str_value = str(value)
1✔
866
                str_values.append(str_value)
1✔
867
            join_result = delimiter.join(str_values)
1✔
868
            return join_result
1✔
869

870
        arguments_delta = self.visit(node_intrinsic_function.arguments)
1✔
871
        delta = self._cached_apply(
1✔
872
            scope=node_intrinsic_function.scope,
873
            arguments_delta=arguments_delta,
874
            resolver=_compute_fn_join,
875
        )
876
        return delta
1✔
877

878
    def visit_node_intrinsic_function_fn_select(
1✔
879
        self, node_intrinsic_function: NodeIntrinsicFunction
880
    ):
881
        # TODO: add further support for schema validation
882
        def _compute_fn_select(args: list[Any]) -> Any:
1✔
883
            values = args[1]
1✔
884
            # defer evaluation if the selection list contains unresolved elements (e.g., unresolved intrinsics)
885
            if isinstance(values, list) and not all(isinstance(value, str) for value in values):
1✔
886
                raise RuntimeError("Fn::Select list contains unresolved elements")
1✔
887

888
            if not isinstance(values, list) or not values:
1✔
889
                raise ValidationError(
1✔
890
                    "Template error: Fn::Select requires a list argument with two elements: an integer index and a list"
891
                )
892
            try:
1✔
893
                index: int = int(args[0])
1✔
894
            except ValueError as e:
1✔
895
                raise ValidationError(
1✔
896
                    "Template error: Fn::Select requires a list argument with two elements: an integer index and a list"
897
                ) from e
898

899
            values_len = len(values)
1✔
900
            if index < 0 or index >= values_len:
1✔
901
                raise ValidationError(
1✔
902
                    "Template error: Fn::Select requires a list argument with two elements: an integer index and a list"
903
                )
904
            selection = values[index]
1✔
905
            return selection
1✔
906

907
        arguments_delta = self.visit(node_intrinsic_function.arguments)
1✔
908
        delta = self._cached_apply(
1✔
909
            scope=node_intrinsic_function.scope,
910
            arguments_delta=arguments_delta,
911
            resolver=_compute_fn_select,
912
        )
913
        return delta
1✔
914

915
    def visit_node_intrinsic_function_fn_split(
1✔
916
        self, node_intrinsic_function: NodeIntrinsicFunction
917
    ):
918
        # TODO: add further support for schema validation
919
        def _compute_fn_split(args: list[Any]) -> Any:
1✔
920
            delimiter = args[0]
1✔
921
            if not isinstance(delimiter, str) or not delimiter:
1✔
922
                raise RuntimeError(f"Invalid delimiter value for Fn::Split: '{delimiter}'")
×
923
            source_string = args[1]
1✔
924
            if not isinstance(source_string, str):
1✔
925
                raise RuntimeError(f"Invalid source string value for Fn::Split: '{source_string}'")
1✔
926
            split_string = source_string.split(delimiter)
1✔
927
            return split_string
1✔
928

929
        arguments_delta = self.visit(node_intrinsic_function.arguments)
1✔
930

931
        if not (
1✔
932
            is_nothing(arguments_delta.after)
933
            or isinstance(arguments_delta.after, list)
934
            and len(arguments_delta.after) == 2
935
        ):
UNCOV
936
            raise ValidationError(
×
937
                "Template error: every Fn::Split object requires two parameters, "
938
                "(1) a string delimiter and (2) a string to be split or a function that returns a string to be split."
939
            )
940

941
        delta = self._cached_apply(
1✔
942
            scope=node_intrinsic_function.scope,
943
            arguments_delta=arguments_delta,
944
            resolver=_compute_fn_split,
945
        )
946
        return delta
1✔
947

948
    def visit_node_intrinsic_function_fn_get_a_zs(
1✔
949
        self, node_intrinsic_function: NodeIntrinsicFunction
950
    ) -> PreprocEntityDelta:
951
        # TODO: add further support for schema validation
952

953
        def _compute_fn_get_a_zs(region) -> Any:
1✔
954
            if not isinstance(region, str):
1✔
UNCOV
955
                raise RuntimeError(f"Invalid region value for Fn::GetAZs: '{region}'")
×
956

957
            if not region:
1✔
958
                region = self._change_set.region_name
1✔
959

960
            account_id = self._change_set.account_id
1✔
961
            ec2_client = connect_to(aws_access_key_id=account_id, region_name=region).ec2
1✔
962
            try:
1✔
963
                get_availability_zones_result: DescribeAvailabilityZonesResult = (
1✔
964
                    ec2_client.describe_availability_zones()
965
                )
UNCOV
966
            except ClientError:
×
UNCOV
967
                raise RuntimeError(
×
968
                    "Could not describe zones availability whilst evaluating Fn::GetAZs"
969
                )
970
            availability_zones: AvailabilityZoneList = get_availability_zones_result[
1✔
971
                "AvailabilityZones"
972
            ]
973
            azs = [az["ZoneName"] for az in availability_zones]
1✔
974
            return azs
1✔
975

976
        arguments_delta = self.visit(node_intrinsic_function.arguments)
1✔
977
        delta = self._cached_apply(
1✔
978
            scope=node_intrinsic_function.scope,
979
            arguments_delta=arguments_delta,
980
            resolver=_compute_fn_get_a_zs,
981
        )
982
        return delta
1✔
983

984
    def visit_node_intrinsic_function_fn_base64(
1✔
985
        self, node_intrinsic_function: NodeIntrinsicFunction
986
    ) -> PreprocEntityDelta:
987
        # TODO: add further support for schema validation
988
        def _compute_fn_base_64(string) -> Any:
1✔
989
            if not isinstance(string, str):
1✔
UNCOV
990
                raise RuntimeError(f"Invalid valueToEncode for Fn::Base64: '{string}'")
×
991
            # Ported from v1:
992
            base64_string = to_str(base64.b64encode(to_bytes(string)))
1✔
993
            return base64_string
1✔
994

995
        arguments_delta = self.visit(node_intrinsic_function.arguments)
1✔
996
        delta = self._cached_apply(
1✔
997
            scope=node_intrinsic_function.scope,
998
            arguments_delta=arguments_delta,
999
            resolver=_compute_fn_base_64,
1000
        )
1001
        return delta
1✔
1002

1003
    def visit_node_intrinsic_function_fn_find_in_map(
1✔
1004
        self, node_intrinsic_function: NodeIntrinsicFunction
1005
    ) -> PreprocEntityDelta:
1006
        # TODO: add type checking/validation for result unit?
1007
        arguments_delta = self.visit(node_intrinsic_function.arguments)
1✔
1008
        before_arguments = arguments_delta.before
1✔
1009
        after_arguments = arguments_delta.after
1✔
1010
        before = Nothing
1✔
1011
        if before_arguments:
1✔
1012
            before_value_delta = self._resolve_mapping(*before_arguments)
1✔
1013
            before = before_value_delta.before
1✔
1014
        after = Nothing
1✔
1015
        if after_arguments:
1✔
1016
            after_value_delta = self._resolve_mapping(*after_arguments)
1✔
1017
            after = after_value_delta.after
1✔
1018
        return PreprocEntityDelta(before=before, after=after)
1✔
1019

1020
    def visit_node_mapping(self, node_mapping: NodeMapping) -> PreprocEntityDelta:
1✔
1021
        bindings_delta = self.visit(node_mapping.bindings)
1✔
1022
        return bindings_delta
1✔
1023

1024
    def visit_node_parameters(
1✔
1025
        self, node_parameters: NodeParameters
1026
    ) -> PreprocEntityDelta[dict[str, Any], dict[str, Any]]:
1027
        before_parameters = {}
1✔
1028
        after_parameters = {}
1✔
1029
        for parameter in node_parameters.parameters:
1✔
1030
            parameter_delta = self.visit(parameter)
1✔
1031
            parameter_before = parameter_delta.before
1✔
1032
            if not is_nothing(parameter_before):
1✔
1033
                before_parameters[parameter.name] = parameter_before
1✔
1034
            parameter_after = parameter_delta.after
1✔
1035
            if not is_nothing(parameter_after):
1✔
1036
                after_parameters[parameter.name] = parameter_after
1✔
1037
        return PreprocEntityDelta(before=before_parameters, after=after_parameters)
1✔
1038

1039
    def visit_node_parameter(self, node_parameter: NodeParameter) -> PreprocEntityDelta:
1✔
1040
        if not VALID_LOGICAL_RESOURCE_ID_RE.match(node_parameter.name):
1✔
1041
            raise ValidationError(
1✔
1042
                f"Template format error: Parameter name {node_parameter.name} is non alphanumeric."
1043
            )
1044
        dynamic_value = node_parameter.dynamic_value
1✔
1045
        dynamic_delta = self.visit(dynamic_value)
1✔
1046

1047
        default_value = node_parameter.default_value
1✔
1048
        default_delta = self.visit(default_value)
1✔
1049

1050
        before = dynamic_delta.before or default_delta.before
1✔
1051
        after = dynamic_delta.after or default_delta.after
1✔
1052

1053
        parameter_type = self.visit(node_parameter.type_)
1✔
1054

1055
        def _resolve_parameter_type(value: str, type_: str) -> Any:
1✔
1056
            match type_:
1✔
1057
                case "List<String>" | "CommaDelimitedList":
1✔
1058
                    return [item.strip() for item in value.split(",")]
1✔
1059
                case "Number":
1✔
1060
                    # TODO: validate the parameter type at template parse time (or whatever is in parity with AWS) so we know this cannot fail
1061
                    return to_number(value)
1✔
1062
            return value
1✔
1063

1064
        if not is_nothing(after):
1✔
1065
            after = _resolve_parameter_type(after, parameter_type.after)
1✔
1066

1067
        return PreprocEntityDelta(before=before, after=after)
1✔
1068

1069
    def visit_node_depends_on(self, node_depends_on: NodeDependsOn) -> PreprocEntityDelta:
1✔
1070
        array_identifiers_delta = self.visit(node_depends_on.depends_on)
1✔
1071
        return array_identifiers_delta
1✔
1072

1073
    def visit_node_condition(self, node_condition: NodeCondition) -> PreprocEntityDelta:
1✔
1074
        delta = self.visit(node_condition.body)
1✔
1075
        return delta
1✔
1076

1077
    def _resource_physical_resource_id_from(
1✔
1078
        self, logical_resource_id: str, resolved_resources: dict[str, ResolvedResource]
1079
    ) -> str | None:
1080
        # TODO: typing around resolved resources is needed and should be reflected here.
1081
        resolved_resource = resolved_resources.get(logical_resource_id, {})
1✔
1082
        if resolved_resource.get("ResourceStatus") not in {
1✔
1083
            ResourceStatus.CREATE_COMPLETE,
1084
            ResourceStatus.UPDATE_COMPLETE,
1085
        }:
1086
            return None
1✔
1087

1088
        physical_resource_id = resolved_resource.get("PhysicalResourceId")
1✔
1089
        if not isinstance(physical_resource_id, str):
1✔
UNCOV
1090
            raise RuntimeError(f"No PhysicalResourceId found for resource '{logical_resource_id}'")
×
1091
        return physical_resource_id
1✔
1092

1093
    def _before_resource_physical_id(self, resource_logical_id: str) -> str:
1✔
1094
        # TODO: typing around resolved resources is needed and should be reflected here.
1095
        return self._resource_physical_resource_id_from(
1✔
1096
            logical_resource_id=resource_logical_id,
1097
            resolved_resources=self._before_resolved_resources,
1098
        )
1099

1100
    def _after_resource_physical_id(self, resource_logical_id: str) -> str:
1✔
1101
        return self._before_resource_physical_id(resource_logical_id=resource_logical_id)
1✔
1102

1103
    def visit_node_intrinsic_function_ref(
1✔
1104
        self, node_intrinsic_function: NodeIntrinsicFunction
1105
    ) -> PreprocEntityDelta:
1106
        def _compute_fn_ref(logical_id: str) -> PreprocEntityDelta:
1✔
1107
            if logical_id == "AWS::NoValue":
1✔
1108
                return Nothing
1✔
1109

1110
            reference_delta: PreprocEntityDelta = self._resolve_reference(logical_id=logical_id)
1✔
1111
            if isinstance(before := reference_delta.before, PreprocResource):
1✔
1112
                reference_delta.before = before.physical_resource_id
1✔
1113
            if isinstance(after := reference_delta.after, PreprocResource):
1✔
1114
                reference_delta.after = after.physical_resource_id
1✔
1115
            return reference_delta
1✔
1116

1117
        arguments_delta = self.visit(node_intrinsic_function.arguments)
1✔
1118
        delta = self._cached_apply(
1✔
1119
            scope=node_intrinsic_function.scope,
1120
            arguments_delta=arguments_delta,
1121
            resolver=_compute_fn_ref,
1122
        )
1123
        return delta
1✔
1124

1125
    def visit_node_intrinsic_function_condition(
1✔
1126
        self, node_intrinsic_function: NodeIntrinsicFunction
1127
    ) -> PreprocEntityDelta:
1128
        arguments_delta = self.visit(node_intrinsic_function.arguments)
1✔
1129

1130
        def _delta_of_condition(name: str) -> PreprocEntityDelta:
1✔
1131
            node_condition = self._get_node_condition_if_exists(condition_name=name)
1✔
1132
            if is_nothing(node_condition):
1✔
UNCOV
1133
                raise RuntimeError(f"Undefined condition '{name}'")
×
1134
            condition_delta = self.visit(node_condition)
1✔
1135
            return condition_delta
1✔
1136

1137
        delta = self._cached_apply(
1✔
1138
            resolver=_delta_of_condition,
1139
            scope=node_intrinsic_function.scope,
1140
            arguments_delta=arguments_delta,
1141
        )
1142
        return delta
1✔
1143

1144
    def visit_node_array(self, node_array: NodeArray) -> PreprocEntityDelta:
1✔
1145
        node_change_type = node_array.change_type
1✔
1146
        before = [] if node_change_type != ChangeType.CREATED else Nothing
1✔
1147
        after = [] if node_change_type != ChangeType.REMOVED else Nothing
1✔
1148
        for change_set_entity in node_array.array:
1✔
1149
            delta: PreprocEntityDelta = self.visit(change_set_entity=change_set_entity)
1✔
1150
            delta_before = delta.before
1✔
1151
            delta_after = delta.after
1✔
1152
            if not is_nothing(before) and not is_nothing(delta_before):
1✔
1153
                before.append(delta_before)
1✔
1154
            if not is_nothing(after) and not is_nothing(delta_after):
1✔
1155
                after.append(delta_after)
1✔
1156
        return PreprocEntityDelta(before=before, after=after)
1✔
1157

1158
    def visit_node_properties(
1✔
1159
        self, node_properties: NodeProperties
1160
    ) -> PreprocEntityDelta[PreprocProperties, PreprocProperties]:
1161
        node_change_type = node_properties.change_type
1✔
1162
        before_bindings = {} if node_change_type != ChangeType.CREATED else Nothing
1✔
1163
        after_bindings = {} if node_change_type != ChangeType.REMOVED else Nothing
1✔
1164
        for node_property in node_properties.properties:
1✔
1165
            property_name = node_property.name
1✔
1166
            delta = self.visit(node_property)
1✔
1167
            delta_before = delta.before
1✔
1168
            delta_after = delta.after
1✔
1169
            if (
1✔
1170
                not is_nothing(before_bindings)
1171
                and not is_nothing(delta_before)
1172
                and delta_before is not None
1173
            ):
1174
                before_bindings[property_name] = delta_before
1✔
1175
            if (
1✔
1176
                not is_nothing(after_bindings)
1177
                and not is_nothing(delta_after)
1178
                and delta_after is not None
1179
            ):
1180
                after_bindings[property_name] = delta_after
1✔
1181
        before = Nothing
1✔
1182
        if not is_nothing(before_bindings):
1✔
1183
            before = PreprocProperties(properties=before_bindings)
1✔
1184
        after = Nothing
1✔
1185
        if not is_nothing(after_bindings):
1✔
1186
            after = PreprocProperties(properties=after_bindings)
1✔
1187
        return PreprocEntityDelta(before=before, after=after)
1✔
1188

1189
    def _resolve_resource_condition_reference(self, reference: TerminalValue) -> PreprocEntityDelta:
1✔
1190
        reference_delta = self.visit(reference)
1✔
1191
        before_reference = reference_delta.before
1✔
1192
        before = Nothing
1✔
1193
        if isinstance(before_reference, str):
1✔
1194
            before_delta = self._resolve_condition(logical_id=before_reference)
1✔
1195
            before = before_delta.before
1✔
1196
        after = Nothing
1✔
1197
        after_reference = reference_delta.after
1✔
1198
        if isinstance(after_reference, str):
1✔
1199
            after_delta = self._resolve_condition(logical_id=after_reference)
1✔
1200
            after = after_delta.after
1✔
1201
        return PreprocEntityDelta(before=before, after=after)
1✔
1202

1203
    def visit_node_resources(self, node_resources: NodeResources):
1✔
1204
        """
1205
        Skip resources where they conditionally evaluate to False
1206
        """
1207
        for node_resource in node_resources.resources:
1✔
1208
            if not is_nothing(node_resource.condition_reference):
1✔
1209
                condition_delta = self._resolve_resource_condition_reference(
1✔
1210
                    node_resource.condition_reference
1211
                )
1212
                condition_after = condition_delta.after
1✔
1213
                if condition_after is False:
1✔
1214
                    continue
1✔
1215
            self.visit(node_resource)
1✔
1216

1217
    def visit_node_resource(
1✔
1218
        self, node_resource: NodeResource
1219
    ) -> PreprocEntityDelta[PreprocResource, PreprocResource]:
1220
        if not VALID_LOGICAL_RESOURCE_ID_RE.match(node_resource.name):
1✔
1221
            raise ValidationError(
1✔
1222
                f"Template format error: Resource name {node_resource.name} is non alphanumeric."
1223
            )
1224
        change_type = node_resource.change_type
1✔
1225
        condition_before = Nothing
1✔
1226
        condition_after = Nothing
1✔
1227
        if not is_nothing(node_resource.condition_reference):
1✔
1228
            condition_delta = self._resolve_resource_condition_reference(
1✔
1229
                node_resource.condition_reference
1230
            )
1231
            condition_before = condition_delta.before
1✔
1232
            condition_after = condition_delta.after
1✔
1233

1234
        depends_on_before = Nothing
1✔
1235
        depends_on_after = Nothing
1✔
1236
        if not is_nothing(node_resource.depends_on):
1✔
1237
            depends_on_delta = self.visit(node_resource.depends_on)
1✔
1238
            depends_on_before = depends_on_delta.before
1✔
1239
            depends_on_after = depends_on_delta.after
1✔
1240

1241
        type_delta = self.visit(node_resource.type_)
1✔
1242
        properties_delta: PreprocEntityDelta[PreprocProperties, PreprocProperties] = self.visit(
1✔
1243
            node_resource.properties
1244
        )
1245

1246
        before = Nothing
1✔
1247
        after = Nothing
1✔
1248
        if change_type != ChangeType.CREATED and is_nothing(condition_before) or condition_before:
1✔
1249
            logical_resource_id = node_resource.name
1✔
1250
            before_physical_resource_id = self._before_resource_physical_id(
1✔
1251
                resource_logical_id=logical_resource_id
1252
            )
1253
            before = PreprocResource(
1✔
1254
                logical_id=logical_resource_id,
1255
                physical_resource_id=before_physical_resource_id,
1256
                condition=condition_before,
1257
                resource_type=type_delta.before,
1258
                properties=properties_delta.before,
1259
                depends_on=depends_on_before,
1260
                requires_replacement=False,
1261
            )
1262
        if change_type != ChangeType.REMOVED and is_nothing(condition_after) or condition_after:
1✔
1263
            logical_resource_id = node_resource.name
1✔
1264
            try:
1✔
1265
                after_physical_resource_id = self._after_resource_physical_id(
1✔
1266
                    resource_logical_id=logical_resource_id
1267
                )
UNCOV
1268
            except RuntimeError:
×
UNCOV
1269
                after_physical_resource_id = None
×
1270
            after = PreprocResource(
1✔
1271
                logical_id=logical_resource_id,
1272
                physical_resource_id=after_physical_resource_id,
1273
                condition=condition_after,
1274
                resource_type=type_delta.after,
1275
                properties=properties_delta.after,
1276
                depends_on=depends_on_after,
1277
                requires_replacement=node_resource.requires_replacement,
1278
            )
1279
        return PreprocEntityDelta(before=before, after=after)
1✔
1280

1281
    def visit_node_output(
1✔
1282
        self, node_output: NodeOutput
1283
    ) -> PreprocEntityDelta[PreprocOutput, PreprocOutput]:
1284
        change_type = node_output.change_type
1✔
1285
        value_delta = self.visit(node_output.value)
1✔
1286

1287
        condition_delta = Nothing
1✔
1288
        if not is_nothing(node_output.condition_reference):
1✔
1289
            condition_delta = self._resolve_resource_condition_reference(
1✔
1290
                node_output.condition_reference
1291
            )
1292
            condition_before = condition_delta.before
1✔
1293
            condition_after = condition_delta.after
1✔
1294
            if not condition_before and condition_after:
1✔
1295
                change_type = ChangeType.CREATED
1✔
1296
            elif condition_before and not condition_after:
1✔
1297
                change_type = ChangeType.REMOVED
1✔
1298

1299
        export_delta = Nothing
1✔
1300
        if not is_nothing(node_output.export):
1✔
1301
            export_delta = self.visit(node_output.export)
1✔
1302

1303
        before: Maybe[PreprocOutput] = Nothing
1✔
1304
        if change_type != ChangeType.CREATED:
1✔
1305
            before = PreprocOutput(
1✔
1306
                name=node_output.name,
1307
                value=value_delta.before,
1308
                export=export_delta.before if export_delta else None,
1309
                condition=condition_delta.before if condition_delta else None,
1310
            )
1311
        after: Maybe[PreprocOutput] = Nothing
1✔
1312
        if change_type != ChangeType.REMOVED:
1✔
1313
            after = PreprocOutput(
1✔
1314
                name=node_output.name,
1315
                value=value_delta.after,
1316
                export=export_delta.after if export_delta else None,
1317
                condition=condition_delta.after if condition_delta else None,
1318
            )
1319
        return PreprocEntityDelta(before=before, after=after)
1✔
1320

1321
    def visit_node_outputs(
1✔
1322
        self, node_outputs: NodeOutputs
1323
    ) -> PreprocEntityDelta[list[PreprocOutput], list[PreprocOutput]]:
1324
        before: list[PreprocOutput] = []
1✔
1325
        after: list[PreprocOutput] = []
1✔
1326
        for node_output in node_outputs.outputs:
1✔
1327
            if not is_nothing(node_output.condition_reference):
1✔
1328
                condition_delta = self._resolve_resource_condition_reference(
1✔
1329
                    node_output.condition_reference
1330
                )
1331
                condition_after = condition_delta.after
1✔
1332
                if condition_after is False:
1✔
1333
                    continue
1✔
1334

1335
            output_delta: PreprocEntityDelta[PreprocOutput, PreprocOutput] = self.visit(node_output)
1✔
1336
            output_before = output_delta.before
1✔
1337
            output_after = output_delta.after
1✔
1338
            if not is_nothing(output_before):
1✔
1339
                before.append(output_before)
1✔
1340
            if not is_nothing(output_after):
1✔
1341
                after.append(output_after)
1✔
1342
        return PreprocEntityDelta(before=before, after=after)
1✔
1343

1344
    def visit_node_intrinsic_function_fn_import_value(
1✔
1345
        self, node_intrinsic_function: NodeIntrinsicFunction
1346
    ) -> PreprocEntityDelta:
1347
        def _compute_fn_import_value(string) -> str:
1✔
1348
            if not isinstance(string, str):
1✔
UNCOV
1349
                raise RuntimeError(f"Invalid parameter for import: '{string}'")
×
1350

1351
            exports = exports_map(
1✔
1352
                account_id=self._change_set.account_id, region_name=self._change_set.region_name
1353
            )
1354

1355
            return exports.get(string, {}).get("Value") or Nothing
1✔
1356

1357
        arguments_delta = self.visit(node_intrinsic_function.arguments)
1✔
1358
        delta = self._cached_apply(
1✔
1359
            scope=node_intrinsic_function.scope,
1360
            arguments_delta=arguments_delta,
1361
            resolver=_compute_fn_import_value,
1362
        )
1363
        return delta
1✔
1364

1365
    def visit_node_intrinsic_function_fn_transform(
1✔
1366
        self, node_intrinsic_function: NodeIntrinsicFunction
1367
    ):
UNCOV
1368
        raise RuntimeError("Fn::Transform should have been handled by the Transformer")
×
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