22048723723

Committed 13 Feb 2026 06:59PM UTC coverage: 87.006% (+0.1%) from 86.883%

Build # 22048723723

Build Type

push

github

Committed by

web-flow

Commit Message

CW Logs: Test suite for service internalization (#13692)

Run Details

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

928 existing lines in 33 files now uncovered.

69716 of 80128 relevant lines covered (87.01%)

0.87 hits per line

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

46.78

/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 lambda account_id, region_name, params, logical_resource_id, args, *kwargs: (	×
93	common.keys_to_lower(
94	obj=(params.get(key) if key else params), skip_children_of=skip_children_of
95	)
96	)
97
98
99	def merge_parameters(func1, func2):	1✔
UNCOV 100	return lambda account_id, region_name, properties, logical_resource_id, args, *kwargs: (	×
101	common.merge_dicts(
102	func1(account_id, region_name, properties, logical_resource_id, args, *kwargs),
103	func2(account_id, region_name, properties, logical_resource_id, args, *kwargs),
104	)
105	)
106
107
108	def str_or_none(o):	1✔
UNCOV 109	return o if o is None else json.dumps(o) if isinstance(o, (dict, list)) else str(o)	×
110
111
112	def params_dict_to_list(param_name, key_attr_name="Key", value_attr_name="Value", wrapper=None):	1✔
UNCOV 113	def do_replace(account_id: str, region_name: str, params, logical_resource_id, args, *kwargs):	×
UNCOV 114	result = []	×
UNCOV 115	for key, value in params.get(param_name, {}).items():	×
UNCOV 116	result.append({key_attr_name: key, value_attr_name: value})	×
UNCOV 117	if wrapper:	×
UNCOV 118	result = {wrapper: result}	×
119	return result	×
120
UNCOV 121	return do_replace	×
122
123
124	# TODO: remove
125	def params_select_attributes(*attrs):	1✔
126	def do_select(account_id: str, region_name: str, params, logical_resource_id, args, *kwargs):	×
127	result = {}	×
128	for attr in attrs:	×
129	if params.get(attr) is not None:	×
UNCOV 130	result[attr] = str_or_none(params.get(attr))	×
131	return result	×
132
UNCOV 133	return do_select	×
134
135
136	def param_json_to_str(name):	1✔
137	def _convert(account_id: str, region_name: str, params, logical_resource_id, args, *kwargs):	×
138	result = params.get(name)	×
139	if result:	×
140	result = json.dumps(result)	×
141	return result	×
142
143	return _convert	×
144
145
146	def lambda_select_params(*selected):	1✔
147	# TODO: remove and merge with function below
148	return select_parameters(*selected)	×
149
150
151	def select_parameters(*param_names):	1✔
UNCOV 152	return lambda account_id, region_name, properties, logical_resource_id, args, *kwargs: (	×
153	select_attributes(properties, param_names)
154	)
155
156
157	def is_none_or_empty_value(value):	1✔
158	return not value or value == PLACEHOLDER_AWS_NO_VALUE	×
159
160
161	def generate_default_name(stack_name: str, logical_resource_id: str):	1✔
162	random_id_part = short_uid()	×
UNCOV 163	resource_id_part = logical_resource_id[:24]	×
UNCOV 164	stack_name_part = stack_name[: 63 - 2 - (len(random_id_part) + len(resource_id_part))]	×
UNCOV 165	return f"{stack_name_part}-{resource_id_part}-{random_id_part}"	×
166
167
168	def generate_default_name_without_stack(logical_resource_id: str):	1✔
UNCOV 169	random_id_part = short_uid()	×
UNCOV 170	resource_id_part = logical_resource_id[: 63 - 1 - len(random_id_part)]	×
UNCOV 171	return f"{resource_id_part}-{random_id_part}"	×
172
173
174	# Utils for parameter conversion
175
176	# TODO: handling of multiple valid types
177	param_validation = re.compile(	1✔
178	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+)'>"
179	)
180
181
182	def get_nested(obj: dict, path: str):	1✔
183	parts = path.split(".")	1✔
184	result = obj	1✔
185	for p in parts[:-1]:	1✔
186	result = result.get(p, {})	1✔
187	return result.get(parts[-1])	1✔
188
189
190	def set_nested(obj: dict, path: str, value):	1✔
191	parts = path.split(".")	1✔
192	result = obj	1✔
193	for p in parts[:-1]:	1✔
194	result = result.get(p, {})	1✔
195	result[parts[-1]] = value	1✔
196
197
198	def fix_boto_parameters_based_on_report(original_params: dict, report: str) -> dict:	1✔
199	"""
200	Fix invalid type parameter validation errors in boto request parameters
201
202	:param original_params: original boto request parameters that lead to the parameter validation error
203	:param report: error report from botocore ParamValidator
204	:return: a copy of original_params with all values replaced by their correctly cast ones
205	"""
206	params = deepcopy(original_params)	1✔
207	for found in param_validation.findall(report):	1✔
208	param_name, value, wrong_class, valid_class = found	1✔
209	cast_class = getattr(builtins, valid_class)	1✔
210	old_value = get_nested(params, param_name)	1✔
211
212	if isinstance(cast_class, bool) and str(old_value).lower() in ["true", "false"]:	1✔
UNCOV 213	new_value = str(old_value).lower() == "true"	×
214	else:
215	new_value = cast_class(old_value)	1✔
216	set_nested(params, param_name, new_value)	1✔
217	return params	1✔
218
219
220	def fix_account_id_in_arns(params: dict, replacement_account_id: str) -> dict:	1✔
UNCOV 221	def fix_ids(o, **kwargs):	×
UNCOV 222	if isinstance(o, dict):	×
UNCOV 223	for k, v in o.items():	×
UNCOV 224	if is_string(v, exclude_binary=True):	×
UNCOV 225	o[k] = aws_stack.fix_account_id_in_arns(v, replacement=replacement_account_id)	×
UNCOV 226	elif is_string(o, exclude_binary=True):	×
UNCOV 227	o = aws_stack.fix_account_id_in_arns(o, replacement=replacement_account_id)	×
228	return o	×
229
UNCOV 230	result = recurse_object(params, fix_ids)	×
UNCOV 231	return result	×
232
233
234	def convert_data_types(type_conversions: dict[str, Callable], params: dict) -> dict:	1✔
235	"""Convert data types in the "params" object, with the type defs
236	specified in the 'types' attribute of "func_details"."""
237	attr_names = type_conversions.keys() or []	×
238
239	def cast(_obj, _type):	×
240	match _type:	×
241	case builtins.bool:	×
242	return _obj in ["True", "true", True]	×
243	case builtins.str:	×
UNCOV 244	if isinstance(_obj, bool):	×
245	return str(_obj).lower()	×
246	return str(_obj)	×
UNCOV 247	case builtins.int \| builtins.float:	×
UNCOV 248	return _type(_obj)	×
UNCOV 249	case _:	×
UNCOV 250	return _obj	×
251
252	def fix_types(o, **kwargs):	×
UNCOV 253	if isinstance(o, dict):	×
254	for k, v in o.items():	×
255	if k in attr_names:	×
256	o[k] = cast(v, type_conversions[k])	×
257	return o	×
258
259	result = recurse_object(params, fix_types)	×
260	return result	×
261
262
263	def log_not_available_message(resource_type: str, message: str):	1✔
264	LOG.warning(	1✔
265	"%s. To find out if %s is supported in LocalStack Pro, "
266	"please check out our docs at https://docs.localstack.cloud/user-guide/aws/cloudformation/#resources-pro--enterprise-edition",
267	message,
268	resource_type,
269	)
270
271
272	def dump_resource_as_json(resource: dict) -> str:	1✔
UNCOV 273	return str(run_safe(lambda: json.dumps(json_safe(resource))) or resource)	×
274
275
276	def get_action_name_for_resource_change(res_change: str) -> str:	1✔
277	return {"Add": "CREATE", "Remove": "DELETE", "Modify": "UPDATE"}.get(res_change)	1✔
278
279
280	def check_not_found_exception(e, resource_type, resource, resource_status=None):	1✔
281	# we expect this to be a "not found" exception
282	markers = [	1✔
283	"NoSuchBucket",
284	"ResourceNotFound",
285	"NoSuchEntity",
286	"NotFoundException",
287	"404",
288	"not found",
289	"not exist",
290	]
291
292	markers_hit = [m for m in markers if m in str(e)]	1✔
293	if not markers_hit:	1✔
294	LOG.warning(	1✔
295	"Unexpected error processing resource type %s: Exception: %s - %s - status: %s",
296	resource_type,
297	str(e),
298	resource,
299	resource_status,
300	)
301	if config.CFN_VERBOSE_ERRORS:	1✔
UNCOV 302	raise e	×
303	else:
304	return False	1✔
305
306	return True	1✔

localstack / localstack / 22048723723

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