17784817164

Committed 16 Sep 2025 03:11PM UTC coverage: 86.879%. Remained the same

Build # 17784817164

Build Type

push

github

Committed by

localstack-bot

Commit Message

prepare next development iteration

Coverage Stats

67198 of 77347 relevant lines covered (86.88%)

0.87 hits per line

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

89.11

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

import re
from typing import Any

from botocore.exceptions import ParamValidationError

from localstack.services.cloudformation.engine.v2.change_set_model import (
    NodeIntrinsicFunction,
    NodeProperty,
    NodeResource,
    NodeTemplate,
    Nothing,
    is_nothing,
)
from localstack.services.cloudformation.engine.v2.change_set_model_preproc import (
    _PSEUDO_PARAMETERS,
    ChangeSetModelPreproc,
    PreprocEntityDelta,
    PreprocResource,
)


class ChangeSetModelValidator(ChangeSetModelPreproc):
    def validate(self):
        self.process()

    def visit_node_template(self, node_template: NodeTemplate):
        self.visit(node_template.mappings)
        self.visit(node_template.resources)
        self.visit(node_template.parameters)

    def visit_node_intrinsic_function_fn_get_att(
        self, node_intrinsic_function: NodeIntrinsicFunction
    ) -> PreprocEntityDelta:
        try:
            return super().visit_node_intrinsic_function_fn_get_att(node_intrinsic_function)
        except RuntimeError:
            return self.visit(node_intrinsic_function.arguments)

    def visit_node_intrinsic_function_fn_sub(
        self, node_intrinsic_function: NodeIntrinsicFunction
    ) -> PreprocEntityDelta:
        def _compute_sub(args: str | list[Any], select_before: bool) -> str:
            # TODO: add further schema validation.
            string_template: str
            sub_parameters: dict
            if isinstance(args, str):
                string_template = args
                sub_parameters = {}
            elif (
                isinstance(args, list)
                and len(args) == 2
                and isinstance(args[0], str)
                and isinstance(args[1], dict)
            ):
                string_template = args[0]
                sub_parameters = args[1]
            else:
                raise RuntimeError(
                    "Invalid arguments shape for Fn::Sub, expected a String "
                    f"or a Tuple of String and Map but got '{args}'"
                )
            sub_string = string_template
            template_variable_names = re.findall("\\${([^}]+)}", string_template)
            for template_variable_name in template_variable_names:
                template_variable_value = Nothing

                # Try to resolve the variable name as pseudo parameter.
                if template_variable_name in _PSEUDO_PARAMETERS:
                    template_variable_value = self._resolve_pseudo_parameter(
                        pseudo_parameter_name=template_variable_name
                    )

                # Try to resolve the variable name as an entry to the defined parameters.
                elif template_variable_name in sub_parameters:
                    template_variable_value = sub_parameters[template_variable_name]

                # Try to resolve the variable name as GetAtt.
                elif "." in template_variable_name:
                    try:
                        template_variable_value = self._resolve_attribute(
                            arguments=template_variable_name, select_before=select_before
                        )
                    except RuntimeError:
                        pass

                # Try to resolve the variable name as Ref.
                else:
                    try:
                        resource_delta = self._resolve_reference(logical_id=template_variable_name)
                        template_variable_value = (
                            resource_delta.before if select_before else resource_delta.after
                        )
                        if isinstance(template_variable_value, PreprocResource):
                            template_variable_value = template_variable_value.physical_resource_id
                    except RuntimeError:
                        pass

                if is_nothing(template_variable_value):
                    # override the base method just for this line to prevent accessing the
                    # resource properties since we are not deploying any resources
                    template_variable_value = ""

                if not isinstance(template_variable_value, str):
                    template_variable_value = str(template_variable_value)

                sub_string = sub_string.replace(
                    f"${{{template_variable_name}}}", template_variable_value
                )

            # FIXME: the following type reduction is ported from v1; however it appears as though such
            #        reduction is not performed by the engine, and certainly not at this depth given the
            #        lack of context. This section should be removed with Fn::Sub always retuning a string
            #        and the resource providers reviewed.
            account_id = self._change_set.account_id
            is_another_account_id = sub_string.isdigit() and len(sub_string) == len(account_id)
            if sub_string == account_id or is_another_account_id:
                result = sub_string
            elif sub_string.isdigit():
                result = int(sub_string)
            else:
                try:
                    result = float(sub_string)
                except ValueError:
                    result = sub_string
            return result

        arguments_delta = self.visit(node_intrinsic_function.arguments)
        arguments_before = arguments_delta.before
        arguments_after = arguments_delta.after
        before = self._before_cache.get(node_intrinsic_function.scope, Nothing)
        if is_nothing(before) and not is_nothing(arguments_before):
            before = _compute_sub(args=arguments_before, select_before=True)
        after = self._after_cache.get(node_intrinsic_function.scope, Nothing)
        if is_nothing(after) and not is_nothing(arguments_after):
            after = _compute_sub(args=arguments_after, select_before=False)
        return PreprocEntityDelta(before=before, after=after)

    def visit_node_intrinsic_function_fn_transform(
        self, node_intrinsic_function: NodeIntrinsicFunction
    ):
        # TODO Research this issue:
        # Function is already resolved in the template reaching this point
        # But transformation is still present in update model
        return self.visit(node_intrinsic_function.arguments)

    def visit_node_intrinsic_function_fn_split(
        self, node_intrinsic_function: NodeIntrinsicFunction
    ) -> PreprocEntityDelta:
        try:
            # If an argument is a Parameter it should be resolved, any other case, ignore it
            return super().visit_node_intrinsic_function_fn_split(node_intrinsic_function)
        except RuntimeError:
            return self.visit(node_intrinsic_function.arguments)

    def visit_node_intrinsic_function_fn_select(
        self, node_intrinsic_function: NodeIntrinsicFunction
    ) -> PreprocEntityDelta:
        try:
            # If an argument is a Parameter it should be resolved, any other case, ignore it
            return super().visit_node_intrinsic_function_fn_select(node_intrinsic_function)
        except RuntimeError:
            return self.visit(node_intrinsic_function.arguments)

    def visit_node_resource(self, node_resource: NodeResource) -> PreprocEntityDelta:
        try:
            if delta := super().visit_node_resource(node_resource):
                return delta
            return super().visit_node_properties(node_resource.properties)
        except RuntimeError:
            return super().visit_node_properties(node_resource.properties)

    def visit_node_property(self, node_property: NodeProperty) -> PreprocEntityDelta:
        try:
            return super().visit_node_property(node_property)
        except ParamValidationError:
            return self.visit(node_property.value)

    # ignore errors from dynamic replacements
    def _maybe_perform_dynamic_replacements(self, delta: PreprocEntityDelta) -> PreprocEntityDelta:
        try:
            return super()._maybe_perform_dynamic_replacements(delta)
        except Exception:
            return delta

1	import re	1✔
2	from typing import Any	1✔
3
4	from botocore.exceptions import ParamValidationError	1✔
5
6	from localstack.services.cloudformation.engine.v2.change_set_model import (	1✔
7	NodeIntrinsicFunction,
8	NodeProperty,
9	NodeResource,
10	NodeTemplate,
11	Nothing,
12	is_nothing,
13	)
14	from localstack.services.cloudformation.engine.v2.change_set_model_preproc import (	1✔
15	_PSEUDO_PARAMETERS,
16	ChangeSetModelPreproc,
17	PreprocEntityDelta,
18	PreprocResource,
19	)
20
21
22	class ChangeSetModelValidator(ChangeSetModelPreproc):	1✔
23	def validate(self):	1✔
24	self.process()	1✔
25
26	def visit_node_template(self, node_template: NodeTemplate):	1✔
27	self.visit(node_template.mappings)	1✔
28	self.visit(node_template.resources)	1✔
29	self.visit(node_template.parameters)	1✔
30
31	def visit_node_intrinsic_function_fn_get_att(	1✔
32	self, node_intrinsic_function: NodeIntrinsicFunction
33	) -> PreprocEntityDelta:
34	try:	1✔
35	return super().visit_node_intrinsic_function_fn_get_att(node_intrinsic_function)	1✔
36	except RuntimeError:	1✔
37	return self.visit(node_intrinsic_function.arguments)	1✔
38
39	def visit_node_intrinsic_function_fn_sub(	1✔
40	self, node_intrinsic_function: NodeIntrinsicFunction
41	) -> PreprocEntityDelta:
42	def _compute_sub(args: str \| list[Any], select_before: bool) -> str:	1✔
43	# TODO: add further schema validation.
44	string_template: str
45	sub_parameters: dict
46	if isinstance(args, str):	1✔
47	string_template = args	1✔
48	sub_parameters = {}	1✔
49	elif (	1✔
50	isinstance(args, list)
51	and len(args) == 2
52	and isinstance(args[0], str)
53	and isinstance(args[1], dict)
54	):
55	string_template = args[0]	1✔
56	sub_parameters = args[1]	1✔
57	else:
58	raise RuntimeError(	×
59	"Invalid arguments shape for Fn::Sub, expected a String "
60	f"or a Tuple of String and Map but got '{args}'"
61	)
62	sub_string = string_template	1✔
63	template_variable_names = re.findall("\\${([^}]+)}", string_template)	1✔
64	for template_variable_name in template_variable_names:	1✔
65	template_variable_value = Nothing	1✔
66
67	# Try to resolve the variable name as pseudo parameter.
68	if template_variable_name in _PSEUDO_PARAMETERS:	1✔
69	template_variable_value = self._resolve_pseudo_parameter(	1✔
70	pseudo_parameter_name=template_variable_name
71	)
72
73	# Try to resolve the variable name as an entry to the defined parameters.
74	elif template_variable_name in sub_parameters:	1✔
75	template_variable_value = sub_parameters[template_variable_name]	1✔
76
77	# Try to resolve the variable name as GetAtt.
78	elif "." in template_variable_name:	1✔
79	try:	1✔
80	template_variable_value = self._resolve_attribute(	1✔
81	arguments=template_variable_name, select_before=select_before
82	)
83	except RuntimeError:	1✔
84	pass	1✔
85
86	# Try to resolve the variable name as Ref.
87	else:
88	try:	1✔
89	resource_delta = self._resolve_reference(logical_id=template_variable_name)	1✔
90	template_variable_value = (	1✔
91	resource_delta.before if select_before else resource_delta.after
92	)
93	if isinstance(template_variable_value, PreprocResource):	1✔
94	template_variable_value = template_variable_value.physical_resource_id	1✔
95	except RuntimeError:	×
96	pass	×
97
98	if is_nothing(template_variable_value):	1✔
99	# override the base method just for this line to prevent accessing the
100	# resource properties since we are not deploying any resources
101	template_variable_value = ""	1✔
102
103	if not isinstance(template_variable_value, str):	1✔
104	template_variable_value = str(template_variable_value)	1✔
105
106	sub_string = sub_string.replace(	1✔
107	f"${{{template_variable_name}}}", template_variable_value
108	)
109
110	# FIXME: the following type reduction is ported from v1; however it appears as though such
111	# reduction is not performed by the engine, and certainly not at this depth given the
112	# lack of context. This section should be removed with Fn::Sub always retuning a string
113	# and the resource providers reviewed.
114	account_id = self._change_set.account_id	1✔
115	is_another_account_id = sub_string.isdigit() and len(sub_string) == len(account_id)	1✔
116	if sub_string == account_id or is_another_account_id:	1✔
117	result = sub_string	1✔
118	elif sub_string.isdigit():	1✔
119	result = int(sub_string)	1✔
120	else:
121	try:	1✔
122	result = float(sub_string)	1✔
123	except ValueError:	1✔
124	result = sub_string	1✔
125	return result	1✔
126
127	arguments_delta = self.visit(node_intrinsic_function.arguments)	1✔
128	arguments_before = arguments_delta.before	1✔
129	arguments_after = arguments_delta.after	1✔
130	before = self._before_cache.get(node_intrinsic_function.scope, Nothing)	1✔
131	if is_nothing(before) and not is_nothing(arguments_before):	1✔
132	before = _compute_sub(args=arguments_before, select_before=True)	1✔
133	after = self._after_cache.get(node_intrinsic_function.scope, Nothing)	1✔
134	if is_nothing(after) and not is_nothing(arguments_after):	1✔
135	after = _compute_sub(args=arguments_after, select_before=False)	1✔
136	return PreprocEntityDelta(before=before, after=after)	1✔
137
138	def visit_node_intrinsic_function_fn_transform(	1✔
139	self, node_intrinsic_function: NodeIntrinsicFunction
140	):
141	# TODO Research this issue:
142	# Function is already resolved in the template reaching this point
143	# But transformation is still present in update model
144	return self.visit(node_intrinsic_function.arguments)	×
145
146	def visit_node_intrinsic_function_fn_split(	1✔
147	self, node_intrinsic_function: NodeIntrinsicFunction
148	) -> PreprocEntityDelta:
149	try:	1✔
150	# If an argument is a Parameter it should be resolved, any other case, ignore it
151	return super().visit_node_intrinsic_function_fn_split(node_intrinsic_function)	1✔
152	except RuntimeError:	1✔
153	return self.visit(node_intrinsic_function.arguments)	1✔
154
155	def visit_node_intrinsic_function_fn_select(	1✔
156	self, node_intrinsic_function: NodeIntrinsicFunction
157	) -> PreprocEntityDelta:
158	try:	1✔
159	# If an argument is a Parameter it should be resolved, any other case, ignore it
160	return super().visit_node_intrinsic_function_fn_select(node_intrinsic_function)	1✔
161	except RuntimeError:	1✔
162	return self.visit(node_intrinsic_function.arguments)	1✔
163
164	def visit_node_resource(self, node_resource: NodeResource) -> PreprocEntityDelta:	1✔
165	try:	1✔
166	if delta := super().visit_node_resource(node_resource):	1✔
167	return delta	1✔
168	return super().visit_node_properties(node_resource.properties)	×
169	except RuntimeError:	×
170	return super().visit_node_properties(node_resource.properties)	×
171
172	def visit_node_property(self, node_property: NodeProperty) -> PreprocEntityDelta:	1✔
173	try:	1✔
174	return super().visit_node_property(node_property)	1✔
175	except ParamValidationError:	×
176	return self.visit(node_property.value)	×
177
178	# ignore errors from dynamic replacements
179	def _maybe_perform_dynamic_replacements(self, delta: PreprocEntityDelta) -> PreprocEntityDelta:	1✔
180	try:	1✔
181	return super()._maybe_perform_dynamic_replacements(delta)	1✔
182	except Exception:	×
183	return delta	×

localstack / localstack / 17784817164

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