18363705737

Committed 08 Oct 2025 07:39PM UTC coverage: 86.912% (+0.02%) from 86.893%

Build # 18363705737

Build Type

push

github

Committed by

web-flow

Commit Message

Fix Event Bridge input transformation of booleans (#13236)

Coverage Stats

6 of 6 new or added lines in 1 file covered. (100.0%)

145 existing lines in 8 files now uncovered.

67991 of 78230 relevant lines covered (86.91%)

0.87 hits per line

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

95.1

/localstack-core/localstack/services/cloudformation/engine/v2/change_set_model_describer.py

from __future__ import annotations

import json
from typing import Final

import localstack.aws.api.cloudformation as cfn_api
from localstack.aws.api.cloudformation import Replacement
from localstack.services.cloudformation.engine.v2.change_set_model import (
    NodeIntrinsicFunction,
    NodeProperty,
    NodeResource,
    NodeResources,
    PropertiesKey,
    is_nothing,
)
from localstack.services.cloudformation.engine.v2.change_set_model_preproc import (
    ChangeSetModelPreproc,
    PreprocEntityDelta,
    PreprocProperties,
    PreprocResource,
)
from localstack.services.cloudformation.v2.entities import ChangeSet
from localstack.utils.numbers import is_number

CHANGESET_KNOWN_AFTER_APPLY: Final[str] = "{{changeSet:KNOWN_AFTER_APPLY}}"


class ChangeSetModelDescriber(ChangeSetModelPreproc):
    _include_property_values: Final[bool]
    _changes: Final[cfn_api.Changes]

    def __init__(
        self,
        change_set: ChangeSet,
        include_property_values: bool,
    ):
        super().__init__(change_set=change_set)
        self._include_property_values = include_property_values
        self._changes = []

    def get_changes(self) -> cfn_api.Changes:
        self._changes.clear()
        self.process()
        return self._changes

    def _setup_runtime_cache(self) -> None:
        # The describer can output {{changeSet:KNOWN_AFTER_APPLY}} values as not every field
        # is computable at describe time. Until a filtering logic or executor override logic
        # is available, the describer cannot benefit of previous evaluations to compute
        # change set resource changes.
        pass

    def _save_runtime_cache(self) -> None:
        # The describer can output {{changeSet:KNOWN_AFTER_APPLY}} values as not every field
        # is computable at describe time. Until a filtering logic or executor override logic
        # is available, there are no benefits in having the describer saving its runtime cache
        # for future changes chains.
        pass

    def _resolve_attribute(self, arguments: str | list[str], select_before: bool) -> str:
        if select_before:
            return super()._resolve_attribute(arguments=arguments, select_before=select_before)

        # Replicate AWS's limitations in describing change set's updated values.
        # Consideration: If we can properly compute the before and after value, why should we
        #                artificially limit the precision of our output to match AWS's?

        arguments_list: list[str]
        if isinstance(arguments, str):
            arguments_list = arguments.split(".")
        else:
            arguments_list = arguments
        logical_name_of_resource = arguments_list[0]
        attribute_name = arguments_list[1]

        node_resource = self._get_node_resource_for(
            resource_name=logical_name_of_resource,
            node_template=self._change_set.update_model.node_template,
        )
        node_property: NodeProperty | None = self._get_node_property_for(
            property_name=attribute_name, node_resource=node_resource
        )
        if node_property is not None:
            property_delta = self.visit(node_property)
            if property_delta.before == property_delta.after:
                value = property_delta.after
            else:
                value = CHANGESET_KNOWN_AFTER_APPLY
        else:
            try:
                value = self._after_deployed_property_value_of(
                    resource_logical_id=logical_name_of_resource,
                    property_name=attribute_name,
                )
            except RuntimeError:
                value = CHANGESET_KNOWN_AFTER_APPLY

        return value

    def visit_node_intrinsic_function(self, node_intrinsic_function: NodeIntrinsicFunction):
        """
        Intrinsic function results are always strings when referring to the describe output
        """
        # TODO: what about other places?
        # TODO: should this be put in the preproc?
        delta = super().visit_node_intrinsic_function(node_intrinsic_function)
        if is_number(delta.before):
            delta.before = str(delta.before)
        if is_number(delta.after):
            delta.after = str(delta.after)
        return delta

    def visit_node_intrinsic_function_fn_join(
        self, node_intrinsic_function: NodeIntrinsicFunction
    ) -> PreprocEntityDelta:
        delta_args = super().visit(node_intrinsic_function.arguments)
        if isinstance(delta_args.after, list) and CHANGESET_KNOWN_AFTER_APPLY in delta_args.after:
            delta_args.after = CHANGESET_KNOWN_AFTER_APPLY
            return delta_args

        delta = super().visit_node_intrinsic_function_fn_join(
            node_intrinsic_function=node_intrinsic_function
        )
        delta_before = delta.before
        if isinstance(delta_before, str) and CHANGESET_KNOWN_AFTER_APPLY in delta_before:
            delta.before = CHANGESET_KNOWN_AFTER_APPLY
        delta_after = delta.after
        if isinstance(delta_after, str) and CHANGESET_KNOWN_AFTER_APPLY in delta_after:
            delta.after = CHANGESET_KNOWN_AFTER_APPLY
        return delta

    def visit_node_intrinsic_function_fn_select(
        self, node_intrinsic_function: NodeIntrinsicFunction
    ):
        # TODO: should this not _ALWAYS_ return CHANGESET_KNOWN_AFTER_APPLY?
        arguments_delta = self.visit(node_intrinsic_function.arguments)
        delta = PreprocEntityDelta()
        if not is_nothing(arguments_delta.before):
            idx = arguments_delta.before[0]
            arr = arguments_delta.before[1]
            try:
                delta.before = arr[int(idx)]
            except Exception:
                delta.before = CHANGESET_KNOWN_AFTER_APPLY

        if not is_nothing(arguments_delta.after):
            idx = arguments_delta.after[0]
            arr = arguments_delta.after[1]
            try:
                delta.after = arr[int(idx)]
            except Exception:
                delta.after = CHANGESET_KNOWN_AFTER_APPLY

        return delta

    def _register_resource_change(
        self,
        logical_id: str,
        type_: str,
        physical_id: str | None,
        before_properties: PreprocProperties | None,
        after_properties: PreprocProperties | None,
        # TODO: remove default
        requires_replacement: bool = False,
    ) -> None:
        action = cfn_api.ChangeAction.Modify
        if before_properties is None:
            action = cfn_api.ChangeAction.Add
        elif after_properties is None:
            action = cfn_api.ChangeAction.Remove

        resource_change = cfn_api.ResourceChange()
        resource_change["Action"] = action
        resource_change["LogicalResourceId"] = logical_id
        resource_change["ResourceType"] = type_
        if physical_id:
            resource_change["PhysicalResourceId"] = physical_id
        if self._include_property_values:
            if before_properties is not None:
                before_context_properties = {PropertiesKey: before_properties.properties}
                before_context_properties_json_str = json.dumps(before_context_properties)
                resource_change["BeforeContext"] = before_context_properties_json_str

            if after_properties is not None:
                after_context_properties = {PropertiesKey: after_properties.properties}
                after_context_properties_json_str = json.dumps(after_context_properties)
                resource_change["AfterContext"] = after_context_properties_json_str

        if action == cfn_api.ChangeAction.Modify:
            # TODO: handle "Conditional" case
            resource_change["Replacement"] = (
                Replacement.True_ if requires_replacement else Replacement.False_
            )

        self._changes.append(
            cfn_api.Change(Type=cfn_api.ChangeType.Resource, ResourceChange=resource_change)
        )

    def _describe_resource_change(
        self, name: str, before: PreprocResource | None, after: PreprocResource | None
    ) -> None:
        if before == after:
            # unchanged: nothing to do.
            return
        if not is_nothing(before) and not is_nothing(after):
            # Case: change on same type.
            if before.resource_type == after.resource_type:
                # Register a Modified if changed.
                self._register_resource_change(
                    logical_id=name,
                    physical_id=before.physical_resource_id,
                    type_=before.resource_type,
                    before_properties=before.properties,
                    after_properties=after.properties,
                    requires_replacement=after.requires_replacement,
                )
            # Case: type migration.
            # TODO: Add test to assert that on type change the resources are replaced.
            else:
                # Register a Removed for the previous type.
                self._register_resource_change(
                    logical_id=name,
                    physical_id=before.physical_resource_id,
                    type_=before.resource_type,
                    before_properties=before.properties,
                    after_properties=None,
                )
                # Register a Create for the next type.
                self._register_resource_change(
                    logical_id=name,
                    physical_id=None,
                    type_=after.resource_type,
                    before_properties=None,
                    after_properties=after.properties,
                )
        elif not is_nothing(before):
            # Case: removal
            self._register_resource_change(
                logical_id=name,
                physical_id=before.physical_resource_id,
                type_=before.resource_type,
                before_properties=before.properties,
                after_properties=None,
            )
        elif not is_nothing(after):
            # Case: addition
            self._register_resource_change(
                logical_id=name,
                physical_id=None,
                type_=after.resource_type,
                before_properties=None,
                after_properties=after.properties,
            )

    def visit_node_resources(self, node_resources: NodeResources) -> None:
        for node_resource in node_resources.resources:
            delta_resource = self.visit(node_resource)
            self._describe_resource_change(
                name=node_resource.name, before=delta_resource.before, after=delta_resource.after
            )

    def visit_node_resource(
        self, node_resource: NodeResource
    ) -> PreprocEntityDelta[PreprocResource, PreprocResource]:
        delta = super().visit_node_resource(node_resource=node_resource)
        after_resource = delta.after
        if not is_nothing(after_resource) and after_resource.physical_resource_id is None:
            after_resource.physical_resource_id = CHANGESET_KNOWN_AFTER_APPLY
        return delta

    def visit_node_intrinsic_function_fn_import_value(
        self, node_intrinsic_function: NodeIntrinsicFunction
    ) -> PreprocEntityDelta:
        delta = super().visit_node_intrinsic_function_fn_import_value(
            node_intrinsic_function=node_intrinsic_function
        )
        after_value = delta.after
        if is_nothing(after_value) and self._include_property_values:
            # TODO find correct way to obtain parent resource
            resource_name = node_intrinsic_function.scope.split("/")[2]
            export_name = node_intrinsic_function.arguments.value

            self._change_set.status_reason = f"[WARN] --include-property-values option can return incomplete ChangeSet data because: ChangeSet creation failed for resource [{resource_name}] because: No export named {export_name}"
            delta.after = CHANGESET_KNOWN_AFTER_APPLY

        return delta

    def visit_node_intrinsic_function_fn_split(
        self, node_intrinsic_function: NodeIntrinsicFunction
    ) -> PreprocEntityDelta:
        delta = super().visit_node_intrinsic_function_fn_split(node_intrinsic_function)
        if isinstance(delta.after, list) and ":".join(delta.after) == CHANGESET_KNOWN_AFTER_APPLY:
            delta.after = [CHANGESET_KNOWN_AFTER_APPLY]
        return delta

1	from __future__ import annotations	1✔
2
3	import json	1✔
4	from typing import Final	1✔
5
6	import localstack.aws.api.cloudformation as cfn_api	1✔
7	from localstack.aws.api.cloudformation import Replacement	1✔
8	from localstack.services.cloudformation.engine.v2.change_set_model import (	1✔
9	NodeIntrinsicFunction,
10	NodeProperty,
11	NodeResource,
12	NodeResources,
13	PropertiesKey,
14	is_nothing,
15	)
16	from localstack.services.cloudformation.engine.v2.change_set_model_preproc import (	1✔
17	ChangeSetModelPreproc,
18	PreprocEntityDelta,
19	PreprocProperties,
20	PreprocResource,
21	)
22	from localstack.services.cloudformation.v2.entities import ChangeSet	1✔
23	from localstack.utils.numbers import is_number	1✔
24
25	CHANGESET_KNOWN_AFTER_APPLY: Final[str] = "{{changeSet:KNOWN_AFTER_APPLY}}"	1✔
26
27
28	class ChangeSetModelDescriber(ChangeSetModelPreproc):	1✔
29	_include_property_values: Final[bool]	1✔
30	_changes: Final[cfn_api.Changes]	1✔
31
32	def __init__(	1✔
33	self,
34	change_set: ChangeSet,
35	include_property_values: bool,
36	):
37	super().__init__(change_set=change_set)	1✔
38	self._include_property_values = include_property_values	1✔
39	self._changes = []	1✔
40
41	def get_changes(self) -> cfn_api.Changes:	1✔
42	self._changes.clear()	1✔
43	self.process()	1✔
44	return self._changes	1✔
45
46	def _setup_runtime_cache(self) -> None:	1✔
47	# The describer can output {{changeSet:KNOWN_AFTER_APPLY}} values as not every field
48	# is computable at describe time. Until a filtering logic or executor override logic
49	# is available, the describer cannot benefit of previous evaluations to compute
50	# change set resource changes.
51	pass	1✔
52
53	def _save_runtime_cache(self) -> None:	1✔
54	# The describer can output {{changeSet:KNOWN_AFTER_APPLY}} values as not every field
55	# is computable at describe time. Until a filtering logic or executor override logic
56	# is available, there are no benefits in having the describer saving its runtime cache
57	# for future changes chains.
58	pass	1✔
59
60	def _resolve_attribute(self, arguments: str \| list[str], select_before: bool) -> str:	1✔
61	if select_before:	1✔
62	return super()._resolve_attribute(arguments=arguments, select_before=select_before)	1✔
63
64	# Replicate AWS's limitations in describing change set's updated values.
65	# Consideration: If we can properly compute the before and after value, why should we
66	# artificially limit the precision of our output to match AWS's?
67
68	arguments_list: list[str]
69	if isinstance(arguments, str):	1✔
70	arguments_list = arguments.split(".")	1✔
71	else:
72	arguments_list = arguments	1✔
73	logical_name_of_resource = arguments_list[0]	1✔
74	attribute_name = arguments_list[1]	1✔
75
76	node_resource = self._get_node_resource_for(	1✔
77	resource_name=logical_name_of_resource,
78	node_template=self._change_set.update_model.node_template,
79	)
80	node_property: NodeProperty \| None = self._get_node_property_for(	1✔
81	property_name=attribute_name, node_resource=node_resource
82	)
83	if node_property is not None:	1✔
84	property_delta = self.visit(node_property)	1✔
85	if property_delta.before == property_delta.after:	1✔
86	value = property_delta.after	1✔
87	else:
88	value = CHANGESET_KNOWN_AFTER_APPLY	1✔
89	else:
90	try:	1✔
91	value = self._after_deployed_property_value_of(	1✔
92	resource_logical_id=logical_name_of_resource,
93	property_name=attribute_name,
94	)
95	except RuntimeError:	1✔
96	value = CHANGESET_KNOWN_AFTER_APPLY	1✔
97
98	return value	1✔
99
100	def visit_node_intrinsic_function(self, node_intrinsic_function: NodeIntrinsicFunction):	1✔
101	"""
102	Intrinsic function results are always strings when referring to the describe output
103	"""
104	# TODO: what about other places?
105	# TODO: should this be put in the preproc?
106	delta = super().visit_node_intrinsic_function(node_intrinsic_function)	1✔
107	if is_number(delta.before):	1✔
108	delta.before = str(delta.before)	1✔
109	if is_number(delta.after):	1✔
110	delta.after = str(delta.after)	1✔
111	return delta	1✔
112
113	def visit_node_intrinsic_function_fn_join(	1✔
114	self, node_intrinsic_function: NodeIntrinsicFunction
115	) -> PreprocEntityDelta:
116	delta_args = super().visit(node_intrinsic_function.arguments)	1✔
117	if isinstance(delta_args.after, list) and CHANGESET_KNOWN_AFTER_APPLY in delta_args.after:	1✔
UNCOV 118	delta_args.after = CHANGESET_KNOWN_AFTER_APPLY	×
UNCOV 119	return delta_args	×
120
121	delta = super().visit_node_intrinsic_function_fn_join(	1✔
122	node_intrinsic_function=node_intrinsic_function
123	)
124	delta_before = delta.before	1✔
125	if isinstance(delta_before, str) and CHANGESET_KNOWN_AFTER_APPLY in delta_before:	1✔
UNCOV 126	delta.before = CHANGESET_KNOWN_AFTER_APPLY	×
127	delta_after = delta.after	1✔
128	if isinstance(delta_after, str) and CHANGESET_KNOWN_AFTER_APPLY in delta_after:	1✔
129	delta.after = CHANGESET_KNOWN_AFTER_APPLY	1✔
130	return delta	1✔
131
132	def visit_node_intrinsic_function_fn_select(	1✔
133	self, node_intrinsic_function: NodeIntrinsicFunction
134	):
135	# TODO: should this not _ALWAYS_ return CHANGESET_KNOWN_AFTER_APPLY?
136	arguments_delta = self.visit(node_intrinsic_function.arguments)	1✔
137	delta = PreprocEntityDelta()	1✔
138	if not is_nothing(arguments_delta.before):	1✔
139	idx = arguments_delta.before[0]	1✔
140	arr = arguments_delta.before[1]	1✔
141	try:	1✔
142	delta.before = arr[int(idx)]	1✔
UNCOV 143	except Exception:	×
UNCOV 144	delta.before = CHANGESET_KNOWN_AFTER_APPLY	×
145
146	if not is_nothing(arguments_delta.after):	1✔
147	idx = arguments_delta.after[0]	1✔
148	arr = arguments_delta.after[1]	1✔
149	try:	1✔
150	delta.after = arr[int(idx)]	1✔
151	except Exception:	1✔
152	delta.after = CHANGESET_KNOWN_AFTER_APPLY	1✔
153
154	return delta	1✔
155
156	def _register_resource_change(	1✔
157	self,
158	logical_id: str,
159	type_: str,
160	physical_id: str \| None,
161	before_properties: PreprocProperties \| None,
162	after_properties: PreprocProperties \| None,
163	# TODO: remove default
164	requires_replacement: bool = False,
165	) -> None:
166	action = cfn_api.ChangeAction.Modify	1✔
167	if before_properties is None:	1✔
168	action = cfn_api.ChangeAction.Add	1✔
169	elif after_properties is None:	1✔
170	action = cfn_api.ChangeAction.Remove	1✔
171
172	resource_change = cfn_api.ResourceChange()	1✔
173	resource_change["Action"] = action	1✔
174	resource_change["LogicalResourceId"] = logical_id	1✔
175	resource_change["ResourceType"] = type_	1✔
176	if physical_id:	1✔
177	resource_change["PhysicalResourceId"] = physical_id	1✔
178	if self._include_property_values:	1✔
179	if before_properties is not None:	1✔
180	before_context_properties = {PropertiesKey: before_properties.properties}	1✔
181	before_context_properties_json_str = json.dumps(before_context_properties)	1✔
182	resource_change["BeforeContext"] = before_context_properties_json_str	1✔
183
184	if after_properties is not None:	1✔
185	after_context_properties = {PropertiesKey: after_properties.properties}	1✔
186	after_context_properties_json_str = json.dumps(after_context_properties)	1✔
187	resource_change["AfterContext"] = after_context_properties_json_str	1✔
188
189	if action == cfn_api.ChangeAction.Modify:	1✔
190	# TODO: handle "Conditional" case
191	resource_change["Replacement"] = (	1✔
192	Replacement.True_ if requires_replacement else Replacement.False_
193	)
194
195	self._changes.append(	1✔
196	cfn_api.Change(Type=cfn_api.ChangeType.Resource, ResourceChange=resource_change)
197	)
198
199	def _describe_resource_change(	1✔
200	self, name: str, before: PreprocResource \| None, after: PreprocResource \| None
201	) -> None:
202	if before == after:	1✔
203	# unchanged: nothing to do.
204	return	1✔
205	if not is_nothing(before) and not is_nothing(after):	1✔
206	# Case: change on same type.
207	if before.resource_type == after.resource_type:	1✔
208	# Register a Modified if changed.
209	self._register_resource_change(	1✔
210	logical_id=name,
211	physical_id=before.physical_resource_id,
212	type_=before.resource_type,
213	before_properties=before.properties,
214	after_properties=after.properties,
215	requires_replacement=after.requires_replacement,
216	)
217	# Case: type migration.
218	# TODO: Add test to assert that on type change the resources are replaced.
219	else:
220	# Register a Removed for the previous type.
UNCOV 221	self._register_resource_change(	×
222	logical_id=name,
223	physical_id=before.physical_resource_id,
224	type_=before.resource_type,
225	before_properties=before.properties,
226	after_properties=None,
227	)
228	# Register a Create for the next type.
UNCOV 229	self._register_resource_change(	×
230	logical_id=name,
231	physical_id=None,
232	type_=after.resource_type,
233	before_properties=None,
234	after_properties=after.properties,
235	)
236	elif not is_nothing(before):	1✔
237	# Case: removal
238	self._register_resource_change(	1✔
239	logical_id=name,
240	physical_id=before.physical_resource_id,
241	type_=before.resource_type,
242	before_properties=before.properties,
243	after_properties=None,
244	)
245	elif not is_nothing(after):	1✔
246	# Case: addition
247	self._register_resource_change(	1✔
248	logical_id=name,
249	physical_id=None,
250	type_=after.resource_type,
251	before_properties=None,
252	after_properties=after.properties,
253	)
254
255	def visit_node_resources(self, node_resources: NodeResources) -> None:	1✔
256	for node_resource in node_resources.resources:	1✔
257	delta_resource = self.visit(node_resource)	1✔
258	self._describe_resource_change(	1✔
259	name=node_resource.name, before=delta_resource.before, after=delta_resource.after
260	)
261
262	def visit_node_resource(	1✔
263	self, node_resource: NodeResource
264	) -> PreprocEntityDelta[PreprocResource, PreprocResource]:
265	delta = super().visit_node_resource(node_resource=node_resource)	1✔
266	after_resource = delta.after	1✔
267	if not is_nothing(after_resource) and after_resource.physical_resource_id is None:	1✔
268	after_resource.physical_resource_id = CHANGESET_KNOWN_AFTER_APPLY	1✔
269	return delta	1✔
270
271	def visit_node_intrinsic_function_fn_import_value(	1✔
272	self, node_intrinsic_function: NodeIntrinsicFunction
273	) -> PreprocEntityDelta:
274	delta = super().visit_node_intrinsic_function_fn_import_value(	1✔
275	node_intrinsic_function=node_intrinsic_function
276	)
277	after_value = delta.after	1✔
278	if is_nothing(after_value) and self._include_property_values:	1✔
279	# TODO find correct way to obtain parent resource
280	resource_name = node_intrinsic_function.scope.split("/")[2]	1✔
281	export_name = node_intrinsic_function.arguments.value	1✔
282
283	self._change_set.status_reason = f"[WARN] --include-property-values option can return incomplete ChangeSet data because: ChangeSet creation failed for resource [{resource_name}] because: No export named {export_name}"	1✔
284	delta.after = CHANGESET_KNOWN_AFTER_APPLY	1✔
285
286	return delta	1✔
287
288	def visit_node_intrinsic_function_fn_split(	1✔
289	self, node_intrinsic_function: NodeIntrinsicFunction
290	) -> PreprocEntityDelta:
291	delta = super().visit_node_intrinsic_function_fn_split(node_intrinsic_function)	1✔
292	if isinstance(delta.after, list) and ":".join(delta.after) == CHANGESET_KNOWN_AFTER_APPLY:	1✔
293	delta.after = [CHANGESET_KNOWN_AFTER_APPLY]	1✔
294	return delta	1✔

localstack / localstack / 18363705737

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