17144436094

Committed 21 Aug 2025 11:28PM UTC coverage: 86.843% (-0.03%) from 86.876%

Build # 17144436094

Build Type

push

github

Committed by

web-flow

Commit Message

APIGW: internalize DeleteIntegrationResponse (#13046)

Coverage Stats

40 of 45 new or added lines in 1 file covered. (88.89%)

235 existing lines in 11 files now uncovered.

67068 of 77229 relevant lines covered (86.84%)

0.87 hits per line

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

92.63

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

import re
from typing import Any

from localstack.services.cloudformation.engine.v2.change_set_model import (
    Maybe,
    NodeIntrinsicFunction,
    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)

    def visit_node_intrinsic_function_fn_get_att(
        self, node_intrinsic_function: NodeIntrinsicFunction
    ) -> PreprocEntityDelta:
        arguments_delta = self.visit(node_intrinsic_function.arguments)
        before_arguments: Maybe[str | list[str]] = arguments_delta.before
        after_arguments: Maybe[str | list[str]] = arguments_delta.after

        before = self._before_cache.get(node_intrinsic_function.scope, Nothing)
        if is_nothing(before) and not is_nothing(before_arguments):
            before = ".".join(before_arguments)

        after = self._after_cache.get(node_intrinsic_function.scope, Nothing)
        if is_nothing(after) and not is_nothing(after_arguments):
            after = ".".join(after_arguments)

        return PreprocEntityDelta(before=before, after=after)

    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)

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

localstack / localstack / 17144436094

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