18505123992

Committed 14 Oct 2025 05:30PM UTC coverage: 86.888% (-0.01%) from 86.899%

Build # 18505123992

Build Type

push

github

Committed by

web-flow

Commit Message

S3: fix `aws-global` validation in CreateBucket (#13250)

Run Details

10 of 10 new or added lines in 4 files covered. (100.0%)

831 existing lines in 40 files now uncovered.

68028 of 78294 relevant lines covered (86.89%)

0.87 hits per line

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

88.24

/localstack-core/localstack/utils/functions.py

"""Higher-order functional tools."""

import functools
import inspect
import logging
from collections.abc import Callable
from typing import Any

LOG = logging.getLogger(__name__)


def run_safe(_python_lambda, *args, _default=None, **kwargs):
    print_error = kwargs.get("print_error", False)
    try:
        return _python_lambda(*args, **kwargs)
    except Exception as e:
        if print_error:
            LOG.warning("Unable to execute function: %s", e)
        return _default


def call_safe(
    func: Callable, args: tuple = None, kwargs: dict = None, exception_message: str = None
) -> Any | None:
    """
    Call the given function with the given arguments, and if it fails, log the given exception_message.
    If logging.DEBUG is set for the logger, then we also log the traceback.

    :param func: function to call
    :param args: arguments to pass
    :param kwargs: keyword arguments to pass
    :param exception_message: message to log on exception
    :return: whatever the func returns
    """
    if exception_message is None:
        exception_message = f"error calling function {func.__name__}"
    if args is None:
        args = ()
    if kwargs is None:
        kwargs = {}

    try:
        return func(*args, **kwargs)
    except Exception as e:
        if LOG.isEnabledFor(logging.DEBUG):
            LOG.exception(exception_message)
        else:
            LOG.warning("%s: %s", exception_message, e)


def prevent_stack_overflow(match_parameters=False):
    """Function decorator to protect a function from stack overflows -
    raises an exception if a (potential) infinite recursion is detected."""

    def _decorator(wrapped):
        @functools.wraps(wrapped)
        def func(*args, **kwargs):
            def _matches(frame):
                if frame.function != wrapped.__name__:
                    return False
                frame = frame.frame

                if not match_parameters:
                    return False

                # construct dict of arguments this stack frame has been called with
                prev_call_args = {
                    frame.f_code.co_varnames[i]: frame.f_locals[frame.f_code.co_varnames[i]]
                    for i in range(frame.f_code.co_argcount)
                }

                # construct dict of arguments the original function has been called with
                sig = inspect.signature(wrapped)
                this_call_args = dict(zip(sig.parameters.keys(), args, strict=False))
                this_call_args.update(kwargs)

                return prev_call_args == this_call_args

            matching_frames = [frame[2] for frame in inspect.stack(context=1) if _matches(frame)]
            if matching_frames:
                raise RecursionError("(Potential) infinite recursion detected")
            return wrapped(*args, **kwargs)

        return func

    return _decorator


def empty_context_manager():
    import contextlib

    return contextlib.nullcontext()

1	"""Higher-order functional tools."""
2
3	import functools	1✔
4	import inspect	1✔
5	import logging	1✔
6	from collections.abc import Callable	1✔
7	from typing import Any	1✔
8
9	LOG = logging.getLogger(__name__)	1✔
10
11
12	def run_safe(_python_lambda, args, _default=None, *kwargs):	1✔
13	print_error = kwargs.get("print_error", False)	1✔
14	try:	1✔
15	return _python_lambda(args, *kwargs)	1✔
16	except Exception as e:	1✔
17	if print_error:	1✔
UNCOV 18	LOG.warning("Unable to execute function: %s", e)	×
19	return _default	1✔
20
21
22	def call_safe(	1✔
23	func: Callable, args: tuple = None, kwargs: dict = None, exception_message: str = None
24	) -> Any \| None:
25	"""
26	Call the given function with the given arguments, and if it fails, log the given exception_message.
27	If logging.DEBUG is set for the logger, then we also log the traceback.
28
29	:param func: function to call
30	:param args: arguments to pass
31	:param kwargs: keyword arguments to pass
32	:param exception_message: message to log on exception
33	:return: whatever the func returns
34	"""
35	if exception_message is None:	1✔
36	exception_message = f"error calling function {func.__name__}"	1✔
37	if args is None:	1✔
38	args = ()	1✔
39	if kwargs is None:	1✔
40	kwargs = {}	1✔
41
42	try:	1✔
43	return func(args, *kwargs)	1✔
44	except Exception as e:	1✔
45	if LOG.isEnabledFor(logging.DEBUG):	1✔
46	LOG.exception(exception_message)	1✔
47	else:
UNCOV 48	LOG.warning("%s: %s", exception_message, e)	×
49
50
51	def prevent_stack_overflow(match_parameters=False):	1✔
52	"""Function decorator to protect a function from stack overflows -
53	raises an exception if a (potential) infinite recursion is detected."""
54
55	def _decorator(wrapped):	1✔
56	@functools.wraps(wrapped)	1✔
57	def func(args, *kwargs):	1✔
58	def _matches(frame):	1✔
59	if frame.function != wrapped.__name__:	1✔
60	return False	1✔
61	frame = frame.frame	1✔
62
63	if not match_parameters:	1✔
UNCOV 64	return False	×
65
66	# construct dict of arguments this stack frame has been called with
67	prev_call_args = {	1✔
68	frame.f_code.co_varnames[i]: frame.f_locals[frame.f_code.co_varnames[i]]
69	for i in range(frame.f_code.co_argcount)
70	}
71
72	# construct dict of arguments the original function has been called with
73	sig = inspect.signature(wrapped)	1✔
74	this_call_args = dict(zip(sig.parameters.keys(), args, strict=False))	1✔
75	this_call_args.update(kwargs)	1✔
76
77	return prev_call_args == this_call_args	1✔
78
79	matching_frames = [frame[2] for frame in inspect.stack(context=1) if _matches(frame)]	1✔
80	if matching_frames:	1✔
UNCOV 81	raise RecursionError("(Potential) infinite recursion detected")	×
82	return wrapped(args, *kwargs)	1✔
83
84	return func	1✔
85
86	return _decorator	1✔
87
88
89	def empty_context_manager():	1✔
UNCOV 90	import contextlib	×
91
UNCOV 92	return contextlib.nullcontext()	×

localstack / localstack / 18505123992

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