20449761985

Committed 22 Dec 2025 09:22PM UTC coverage: 86.912% (-0.008%) from 86.92%

Build # 20449761985

Build Type

push

github

Committed by

web-flow

Commit Message

APIGW: improve store typing (#13552)

Run Details

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

130 existing lines in 7 files now uncovered.

70016 of 80560 relevant lines covered (86.91%)

0.87 hits per line

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

89.42

/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,
)
from localstack.services.cloudformation.engine.validations import ValidationError


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

localstack / localstack / 20449761985

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