20565403496

Committed 29 Dec 2025 05:11AM UTC coverage: 84.103% (-2.8%) from 86.921%

Build # 20565403496

Build Type

Pull #13567

github

Committed by

web-flow

Commit Message

Merge 4816837a5 into 2417384aa

Pull Request Pull Request #13567: Update ASF APIs

Run Details

67166 of 79862 relevant lines covered (84.1%)

0.84 hits per line

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

46.2

/localstack-core/localstack/services/cloudformation/deployment_utils.py

import builtins
import json
import logging
import re
from collections.abc import Callable
from copy import deepcopy

from localstack import config
from localstack.utils import common
from localstack.utils.aws import aws_stack
from localstack.utils.common import select_attributes, short_uid
from localstack.utils.functions import run_safe
from localstack.utils.json import json_safe
from localstack.utils.objects import recurse_object
from localstack.utils.strings import is_string

# placeholders
PLACEHOLDER_AWS_NO_VALUE = "__aws_no_value__"

LOG = logging.getLogger(__name__)


def dump_json_params(param_func=None, *param_names):
    def replace(account_id: str, region_name: str, params, logical_resource_id, *args, **kwargs):
        result = (
            param_func(account_id, region_name, params, logical_resource_id, *args, **kwargs)
            if param_func
            else params
        )
        for name in param_names:
            if isinstance(result.get(name), (dict, list)):
                # Fix for https://github.com/localstack/localstack/issues/2022
                # Convert any date instances to date strings, etc, Version: "2012-10-17"
                param_value = common.json_safe(result[name])
                result[name] = json.dumps(param_value)
        return result

    return replace


# TODO: remove
def param_defaults(param_func, defaults):
    def replace(
        account_id: str,
        region_name: str,
        properties: dict,
        logical_resource_id: str,
        *args,
        **kwargs,
    ):
        result = param_func(
            account_id, region_name, properties, logical_resource_id, *args, **kwargs
        )
        for key, value in defaults.items():
            if result.get(key) in ["", None]:
                result[key] = value
        return result

    return replace


def remove_none_values(params):
    """Remove None values and AWS::NoValue placeholders (recursively) in the given object."""

    def remove_nones(o, **kwargs):
        if isinstance(o, dict):
            for k, v in dict(o).items():
                if v in [None, PLACEHOLDER_AWS_NO_VALUE]:
                    o.pop(k)
        if isinstance(o, list):
            common.run_safe(o.remove, None)
            common.run_safe(o.remove, PLACEHOLDER_AWS_NO_VALUE)
        return o

    result = common.recurse_object(params, remove_nones)
    return result


def params_list_to_dict(param_name, key_attr_name="Key", value_attr_name="Value"):
    def do_replace(account_id: str, region_name: str, params, logical_resource_id, *args, **kwargs):
        result = {}
        for entry in params.get(param_name, []):
            key = entry[key_attr_name]
            value = entry[value_attr_name]
            result[key] = value
        return result

    return do_replace


def lambda_keys_to_lower(key=None, skip_children_of: list[str] = None):
    return (
        lambda account_id,
        region_name,
        params,
        logical_resource_id,
        *args,
        **kwargs: common.keys_to_lower(
            obj=(params.get(key) if key else params), skip_children_of=skip_children_of
        )
    )


def merge_parameters(func1, func2):
    return (
        lambda account_id,
        region_name,
        properties,
        logical_resource_id,
        *args,
        **kwargs: common.merge_dicts(
            func1(account_id, region_name, properties, logical_resource_id, *args, **kwargs),
            func2(account_id, region_name, properties, logical_resource_id, *args, **kwargs),
        )
    )


def str_or_none(o):
    return o if o is None else json.dumps(o) if isinstance(o, (dict, list)) else str(o)


def params_dict_to_list(param_name, key_attr_name="Key", value_attr_name="Value", wrapper=None):
    def do_replace(account_id: str, region_name: str, params, logical_resource_id, *args, **kwargs):
        result = []
        for key, value in params.get(param_name, {}).items():
            result.append({key_attr_name: key, value_attr_name: value})
        if wrapper:
            result = {wrapper: result}
        return result

    return do_replace


# TODO: remove
def params_select_attributes(*attrs):
    def do_select(account_id: str, region_name: str, params, logical_resource_id, *args, **kwargs):
        result = {}
        for attr in attrs:
            if params.get(attr) is not None:
                result[attr] = str_or_none(params.get(attr))
        return result

    return do_select


def param_json_to_str(name):
    def _convert(account_id: str, region_name: str, params, logical_resource_id, *args, **kwargs):
        result = params.get(name)
        if result:
            result = json.dumps(result)
        return result

    return _convert


def lambda_select_params(*selected):
    # TODO: remove and merge with function below
    return select_parameters(*selected)


def select_parameters(*param_names):
    return (
        lambda account_id,
        region_name,
        properties,
        logical_resource_id,
        *args,
        **kwargs: select_attributes(properties, param_names)
    )


def is_none_or_empty_value(value):
    return not value or value == PLACEHOLDER_AWS_NO_VALUE


def generate_default_name(stack_name: str, logical_resource_id: str):
    random_id_part = short_uid()
    resource_id_part = logical_resource_id[:24]
    stack_name_part = stack_name[: 63 - 2 - (len(random_id_part) + len(resource_id_part))]
    return f"{stack_name_part}-{resource_id_part}-{random_id_part}"


def generate_default_name_without_stack(logical_resource_id: str):
    random_id_part = short_uid()
    resource_id_part = logical_resource_id[: 63 - 1 - len(random_id_part)]
    return f"{resource_id_part}-{random_id_part}"


# Utils for parameter conversion

# TODO: handling of multiple valid types
param_validation = re.compile(
    r"Invalid type for parameter (?P<param>[\w.]+), value: (?P<value>\w+), type: <class '(?P<wrong_class>\w+)'>, valid types: <class '(?P<valid_class>\w+)'>"
)


def get_nested(obj: dict, path: str):
    parts = path.split(".")
    result = obj
    for p in parts[:-1]:
        result = result.get(p, {})
    return result.get(parts[-1])


def set_nested(obj: dict, path: str, value):
    parts = path.split(".")
    result = obj
    for p in parts[:-1]:
        result = result.get(p, {})
    result[parts[-1]] = value


def fix_boto_parameters_based_on_report(original_params: dict, report: str) -> dict:
    """
    Fix invalid type parameter validation errors in boto request parameters

    :param original_params: original boto request parameters that lead to the parameter validation error
    :param report: error report from botocore ParamValidator
    :return: a copy of original_params with all values replaced by their correctly cast ones
    """
    params = deepcopy(original_params)
    for found in param_validation.findall(report):
        param_name, value, wrong_class, valid_class = found
        cast_class = getattr(builtins, valid_class)
        old_value = get_nested(params, param_name)

        if isinstance(cast_class, bool) and str(old_value).lower() in ["true", "false"]:
            new_value = str(old_value).lower() == "true"
        else:
            new_value = cast_class(old_value)
        set_nested(params, param_name, new_value)
    return params


def fix_account_id_in_arns(params: dict, replacement_account_id: str) -> dict:
    def fix_ids(o, **kwargs):
        if isinstance(o, dict):
            for k, v in o.items():
                if is_string(v, exclude_binary=True):
                    o[k] = aws_stack.fix_account_id_in_arns(v, replacement=replacement_account_id)
        elif is_string(o, exclude_binary=True):
            o = aws_stack.fix_account_id_in_arns(o, replacement=replacement_account_id)
        return o

    result = recurse_object(params, fix_ids)
    return result


def convert_data_types(type_conversions: dict[str, Callable], params: dict) -> dict:
    """Convert data types in the "params" object, with the type defs
    specified in the 'types' attribute of "func_details"."""
    attr_names = type_conversions.keys() or []

    def cast(_obj, _type):
        match _type:
            case builtins.bool:
                return _obj in ["True", "true", True]
            case builtins.str:
                if isinstance(_obj, bool):
                    return str(_obj).lower()
                return str(_obj)
            case builtins.int | builtins.float:
                return _type(_obj)
            case _:
                return _obj

    def fix_types(o, **kwargs):
        if isinstance(o, dict):
            for k, v in o.items():
                if k in attr_names:
                    o[k] = cast(v, type_conversions[k])
        return o

    result = recurse_object(params, fix_types)
    return result


def log_not_available_message(resource_type: str, message: str):
    LOG.warning(
        "%s. To find out if %s is supported in LocalStack Pro, "
        "please check out our docs at https://docs.localstack.cloud/user-guide/aws/cloudformation/#resources-pro--enterprise-edition",
        message,
        resource_type,
    )


def dump_resource_as_json(resource: dict) -> str:
    return str(run_safe(lambda: json.dumps(json_safe(resource))) or resource)


def get_action_name_for_resource_change(res_change: str) -> str:
    return {"Add": "CREATE", "Remove": "DELETE", "Modify": "UPDATE"}.get(res_change)


def check_not_found_exception(e, resource_type, resource, resource_status=None):
    # we expect this to be a "not found" exception
    markers = [
        "NoSuchBucket",
        "ResourceNotFound",
        "NoSuchEntity",
        "NotFoundException",
        "404",
        "not found",
        "not exist",
    ]

    markers_hit = [m for m in markers if m in str(e)]
    if not markers_hit:
        LOG.warning(
            "Unexpected error processing resource type %s: Exception: %s - %s - status: %s",
            resource_type,
            str(e),
            resource,
            resource_status,
        )
        if config.CFN_VERBOSE_ERRORS:
            raise e
        else:
            return False

    return True

1	import builtins	1✔
2	import json	1✔
3	import logging	1✔
4	import re	1✔
5	from collections.abc import Callable	1✔
6	from copy import deepcopy	1✔
7
8	from localstack import config	1✔
9	from localstack.utils import common	1✔
10	from localstack.utils.aws import aws_stack	1✔
11	from localstack.utils.common import select_attributes, short_uid	1✔
12	from localstack.utils.functions import run_safe	1✔
13	from localstack.utils.json import json_safe	1✔
14	from localstack.utils.objects import recurse_object	1✔
15	from localstack.utils.strings import is_string	1✔
16
17	# placeholders
18	PLACEHOLDER_AWS_NO_VALUE = "__aws_no_value__"	1✔
19
20	LOG = logging.getLogger(__name__)	1✔
21
22
23	def dump_json_params(param_func=None, *param_names):	1✔
24	def replace(account_id: str, region_name: str, params, logical_resource_id, args, *kwargs):	×
25	result = (	×
26	param_func(account_id, region_name, params, logical_resource_id, args, *kwargs)
27	if param_func
28	else params
29	)
30	for name in param_names:	×
31	if isinstance(result.get(name), (dict, list)):	×
32	# Fix for https://github.com/localstack/localstack/issues/2022
33	# Convert any date instances to date strings, etc, Version: "2012-10-17"
34	param_value = common.json_safe(result[name])	×
35	result[name] = json.dumps(param_value)	×
36	return result	×
37
38	return replace	×
39
40
41	# TODO: remove
42	def param_defaults(param_func, defaults):	1✔
43	def replace(	×
44	account_id: str,
45	region_name: str,
46	properties: dict,
47	logical_resource_id: str,
48	*args,
49	**kwargs,
50	):
51	result = param_func(	×
52	account_id, region_name, properties, logical_resource_id, args, *kwargs
53	)
54	for key, value in defaults.items():	×
55	if result.get(key) in ["", None]:	×
56	result[key] = value	×
57	return result	×
58
59	return replace	×
60
61
62	def remove_none_values(params):	1✔
63	"""Remove None values and AWS::NoValue placeholders (recursively) in the given object."""
64
65	def remove_nones(o, **kwargs):	1✔
66	if isinstance(o, dict):	1✔
67	for k, v in dict(o).items():	1✔
68	if v in [None, PLACEHOLDER_AWS_NO_VALUE]:	1✔
69	o.pop(k)	1✔
70	if isinstance(o, list):	1✔
71	common.run_safe(o.remove, None)	1✔
72	common.run_safe(o.remove, PLACEHOLDER_AWS_NO_VALUE)	1✔
73	return o	1✔
74
75	result = common.recurse_object(params, remove_nones)	1✔
76	return result	1✔
77
78
79	def params_list_to_dict(param_name, key_attr_name="Key", value_attr_name="Value"):	1✔
80	def do_replace(account_id: str, region_name: str, params, logical_resource_id, args, *kwargs):	×
81	result = {}	×
82	for entry in params.get(param_name, []):	×
83	key = entry[key_attr_name]	×
84	value = entry[value_attr_name]	×
85	result[key] = value	×
86	return result	×
87
88	return do_replace	×
89
90
91	def lambda_keys_to_lower(key=None, skip_children_of: list[str] = None):	1✔
92	return (	×
93	lambda account_id,
94	region_name,
95	params,
96	logical_resource_id,
97	*args,
98	**kwargs: common.keys_to_lower(
99	obj=(params.get(key) if key else params), skip_children_of=skip_children_of
100	)
101	)
102
103
104	def merge_parameters(func1, func2):	1✔
105	return (	×
106	lambda account_id,
107	region_name,
108	properties,
109	logical_resource_id,
110	*args,
111	**kwargs: common.merge_dicts(
112	func1(account_id, region_name, properties, logical_resource_id, args, *kwargs),
113	func2(account_id, region_name, properties, logical_resource_id, args, *kwargs),
114	)
115	)
116
117
118	def str_or_none(o):	1✔
119	return o if o is None else json.dumps(o) if isinstance(o, (dict, list)) else str(o)	×
120
121
122	def params_dict_to_list(param_name, key_attr_name="Key", value_attr_name="Value", wrapper=None):	1✔
123	def do_replace(account_id: str, region_name: str, params, logical_resource_id, args, *kwargs):	×
124	result = []	×
125	for key, value in params.get(param_name, {}).items():	×
126	result.append({key_attr_name: key, value_attr_name: value})	×
127	if wrapper:	×
128	result = {wrapper: result}	×
129	return result	×
130
131	return do_replace	×
132
133
134	# TODO: remove
135	def params_select_attributes(*attrs):	1✔
136	def do_select(account_id: str, region_name: str, params, logical_resource_id, args, *kwargs):	×
137	result = {}	×
138	for attr in attrs:	×
139	if params.get(attr) is not None:	×
140	result[attr] = str_or_none(params.get(attr))	×
141	return result	×
142
143	return do_select	×
144
145
146	def param_json_to_str(name):	1✔
147	def _convert(account_id: str, region_name: str, params, logical_resource_id, args, *kwargs):	×
148	result = params.get(name)	×
149	if result:	×
150	result = json.dumps(result)	×
151	return result	×
152
153	return _convert	×
154
155
156	def lambda_select_params(*selected):	1✔
157	# TODO: remove and merge with function below
158	return select_parameters(*selected)	×
159
160
161	def select_parameters(*param_names):	1✔
162	return (	×
163	lambda account_id,
164	region_name,
165	properties,
166	logical_resource_id,
167	*args,
168	**kwargs: select_attributes(properties, param_names)
169	)
170
171
172	def is_none_or_empty_value(value):	1✔
173	return not value or value == PLACEHOLDER_AWS_NO_VALUE	×
174
175
176	def generate_default_name(stack_name: str, logical_resource_id: str):	1✔
177	random_id_part = short_uid()	×
178	resource_id_part = logical_resource_id[:24]	×
179	stack_name_part = stack_name[: 63 - 2 - (len(random_id_part) + len(resource_id_part))]	×
180	return f"{stack_name_part}-{resource_id_part}-{random_id_part}"	×
181
182
183	def generate_default_name_without_stack(logical_resource_id: str):	1✔
184	random_id_part = short_uid()	×
185	resource_id_part = logical_resource_id[: 63 - 1 - len(random_id_part)]	×
186	return f"{resource_id_part}-{random_id_part}"	×
187
188
189	# Utils for parameter conversion
190
191	# TODO: handling of multiple valid types
192	param_validation = re.compile(	1✔
193	r"Invalid type for parameter (?P<param>[\w.]+), value: (?P<value>\w+), type: <class '(?P<wrong_class>\w+)'>, valid types: <class '(?P<valid_class>\w+)'>"
194	)
195
196
197	def get_nested(obj: dict, path: str):	1✔
198	parts = path.split(".")	1✔
199	result = obj	1✔
200	for p in parts[:-1]:	1✔
201	result = result.get(p, {})	1✔
202	return result.get(parts[-1])	1✔
203
204
205	def set_nested(obj: dict, path: str, value):	1✔
206	parts = path.split(".")	1✔
207	result = obj	1✔
208	for p in parts[:-1]:	1✔
209	result = result.get(p, {})	1✔
210	result[parts[-1]] = value	1✔
211
212
213	def fix_boto_parameters_based_on_report(original_params: dict, report: str) -> dict:	1✔
214	"""
215	Fix invalid type parameter validation errors in boto request parameters
216
217	:param original_params: original boto request parameters that lead to the parameter validation error
218	:param report: error report from botocore ParamValidator
219	:return: a copy of original_params with all values replaced by their correctly cast ones
220	"""
221	params = deepcopy(original_params)	1✔
222	for found in param_validation.findall(report):	1✔
223	param_name, value, wrong_class, valid_class = found	1✔
224	cast_class = getattr(builtins, valid_class)	1✔
225	old_value = get_nested(params, param_name)	1✔
226
227	if isinstance(cast_class, bool) and str(old_value).lower() in ["true", "false"]:	1✔
228	new_value = str(old_value).lower() == "true"	×
229	else:
230	new_value = cast_class(old_value)	1✔
231	set_nested(params, param_name, new_value)	1✔
232	return params	1✔
233
234
235	def fix_account_id_in_arns(params: dict, replacement_account_id: str) -> dict:	1✔
236	def fix_ids(o, **kwargs):	×
237	if isinstance(o, dict):	×
238	for k, v in o.items():	×
239	if is_string(v, exclude_binary=True):	×
240	o[k] = aws_stack.fix_account_id_in_arns(v, replacement=replacement_account_id)	×
241	elif is_string(o, exclude_binary=True):	×
242	o = aws_stack.fix_account_id_in_arns(o, replacement=replacement_account_id)	×
243	return o	×
244
245	result = recurse_object(params, fix_ids)	×
246	return result	×
247
248
249	def convert_data_types(type_conversions: dict[str, Callable], params: dict) -> dict:	1✔
250	"""Convert data types in the "params" object, with the type defs
251	specified in the 'types' attribute of "func_details"."""
252	attr_names = type_conversions.keys() or []	×
253
254	def cast(_obj, _type):	×
255	match _type:	×
256	case builtins.bool:	×
257	return _obj in ["True", "true", True]	×
258	case builtins.str:	×
259	if isinstance(_obj, bool):	×
260	return str(_obj).lower()	×
261	return str(_obj)	×
262	case builtins.int \| builtins.float:	×
263	return _type(_obj)	×
264	case _:	×
265	return _obj	×
266
267	def fix_types(o, **kwargs):	×
268	if isinstance(o, dict):	×
269	for k, v in o.items():	×
270	if k in attr_names:	×
271	o[k] = cast(v, type_conversions[k])	×
272	return o	×
273
274	result = recurse_object(params, fix_types)	×
275	return result	×
276
277
278	def log_not_available_message(resource_type: str, message: str):	1✔
279	LOG.warning(	1✔
280	"%s. To find out if %s is supported in LocalStack Pro, "
281	"please check out our docs at https://docs.localstack.cloud/user-guide/aws/cloudformation/#resources-pro--enterprise-edition",
282	message,
283	resource_type,
284	)
285
286
287	def dump_resource_as_json(resource: dict) -> str:	1✔
288	return str(run_safe(lambda: json.dumps(json_safe(resource))) or resource)	×
289
290
291	def get_action_name_for_resource_change(res_change: str) -> str:	1✔
292	return {"Add": "CREATE", "Remove": "DELETE", "Modify": "UPDATE"}.get(res_change)	×
293
294
295	def check_not_found_exception(e, resource_type, resource, resource_status=None):	1✔
296	# we expect this to be a "not found" exception
297	markers = [	1✔
298	"NoSuchBucket",
299	"ResourceNotFound",
300	"NoSuchEntity",
301	"NotFoundException",
302	"404",
303	"not found",
304	"not exist",
305	]
306
307	markers_hit = [m for m in markers if m in str(e)]	1✔
308	if not markers_hit:	1✔
309	LOG.warning(	1✔
310	"Unexpected error processing resource type %s: Exception: %s - %s - status: %s",
311	resource_type,
312	str(e),
313	resource,
314	resource_status,
315	)
316	if config.CFN_VERBOSE_ERRORS:	1✔
317	raise e	×
318	else:
319	return False	1✔
320
321	return True	1✔

localstack / localstack / 20565403496

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