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

86.23

/localstack-core/localstack/utils/objects.py

import functools
import re
import threading
from collections.abc import Callable
from typing import Any, Generic, TypeVar

from .collections import ensure_list
from .strings import first_char_to_lower, first_char_to_upper

ComplexType = list | dict | object

_T = TypeVar("_T")


class Value(Generic[_T]):
    """
    Simple value container.
    """

    value: _T | None

    def __init__(self, value: _T = None) -> None:
        self.value = value

    def clear(self):
        self.value = None

    def set(self, value: _T):
        self.value = value

    def is_set(self) -> bool:
        return self.value is not None

    def get(self) -> _T | None:
        return self.value

    def __bool__(self):
        return True if self.value else False


class ArbitraryAccessObj:
    """Dummy object that can be arbitrarily accessed - any attributes, as a callable, item assignment, ..."""

    def __init__(self, name=None):
        self.name = name

    def __getattr__(self, name, *args, **kwargs):
        return ArbitraryAccessObj(name)

    def __call__(self, *args, **kwargs):
        if self.name in ["items", "keys", "values"] and not args and not kwargs:
            return []
        return ArbitraryAccessObj()

    def __getitem__(self, *args, **kwargs):
        return ArbitraryAccessObj()

    def __setitem__(self, *args, **kwargs):
        return ArbitraryAccessObj()


class Mock:
    """Dummy class that can be used for mocking custom attributes."""

    pass


class ObjectIdHashComparator:
    """Simple wrapper class that allows us to create a hashset using the object id(..) as the entries' hash value"""

    def __init__(self, obj):
        self.obj = obj
        self._hash = id(obj)

    def __hash__(self):
        return self._hash

    def __eq__(self, other):
        # assumption here is that we're comparing only against ObjectIdHash instances!
        return self.obj == other.obj


class SubtypesInstanceManager:
    """Simple instance manager base class that scans the subclasses of a base type for concrete named
    implementations, and lazily creates and returns (singleton) instances on demand."""

    _instances: dict[str, "SubtypesInstanceManager"]

    @classmethod
    def get(cls, subtype_name: str, raise_if_missing: bool = True):
        instances = cls.instances()
        base_type = cls.get_base_type()
        instance = instances.get(subtype_name)
        if instance is None:
            # lazily load subtype instance (required if new plugins are dynamically loaded at runtime)
            for clazz in get_all_subclasses(base_type):
                impl_name = clazz.impl_name()
                if impl_name not in instances and subtype_name == impl_name:
                    instances[impl_name] = clazz()
            instance = instances.get(subtype_name)
        if not instance and raise_if_missing:
            raise NotImplementedError(
                f"Unable to find implementation named '{subtype_name}' for base type {base_type}"
            )
        return instance

    @classmethod
    def instances(cls) -> dict[str, "SubtypesInstanceManager"]:
        base_type = cls.get_base_type()
        if not hasattr(base_type, "_instances"):
            base_type._instances = {}
        return base_type._instances

    @staticmethod
    def impl_name() -> str:
        """Name of this concrete subtype - to be implemented by subclasses."""
        raise NotImplementedError

    @classmethod
    def get_base_type(cls) -> type:
        """Get the base class for which instances are being managed - can be customized by subtypes."""
        return cls


# this requires that all subclasses have been imported before(!)
def get_all_subclasses(clazz: type) -> set[type]:
    """Recursively get all subclasses of the given class."""
    result = set()
    subs = clazz.__subclasses__()
    for sub in subs:
        result.add(sub)
        result.update(get_all_subclasses(sub))
    return result


def fully_qualified_class_name(klass: type) -> str:
    return f"{klass.__module__}.{klass.__name__}"


def not_none_or(value: Any | None, alternative: Any) -> Any:
    """Return 'value' if it is not None, or 'alternative' otherwise."""
    return value if value is not None else alternative


def recurse_object(obj: ComplexType, func: Callable, path: str = "") -> ComplexType:
    """Recursively apply `func` to `obj` (might be a list, dict, or other object)."""
    obj = func(obj, path=path)
    if isinstance(obj, list):
        for i in range(len(obj)):
            tmp_path = f"{path or '.'}[{i}]"
            obj[i] = recurse_object(obj[i], func, tmp_path)
    elif isinstance(obj, dict):
        for k, v in obj.items():
            tmp_path = f"{f'{path}.' if path else ''}{k}"
            obj[k] = recurse_object(v, func, tmp_path)
    return obj


def keys_to(
    obj: ComplexType, op: Callable[[str], str], skip_children_of: list[str] = None
) -> ComplexType:
    """Recursively changes all dict keys to apply op. Skip children
    of any elements whose names are contained in skip_children_of (e.g., ['Tags'])"""
    skip_children_of = ensure_list(skip_children_of or [])

    def fix_keys(o, path="", **kwargs):
        if any(re.match(rf"(^|.*\.){k}($|[.\[].*)", path) for k in skip_children_of):
            return o
        if isinstance(o, dict):
            for k, v in dict(o).items():
                o.pop(k)
                o[op(k)] = v
        return o

    result = recurse_object(obj, fix_keys)
    return result


def keys_to_lower(obj: ComplexType, skip_children_of: list[str] = None) -> ComplexType:
    return keys_to(obj, first_char_to_lower, skip_children_of)


def keys_to_upper(obj: ComplexType, skip_children_of: list[str] = None) -> ComplexType:
    return keys_to(obj, first_char_to_upper, skip_children_of)


def singleton_factory(factory: Callable[[], _T]) -> Callable[[], _T]:
    """
    Decorator for methods that create a particular value once and then return the same value in a thread safe way.

    :param factory: the method to decorate
    :return: a threadsafe singleton factory
    """
    lock = threading.RLock()
    instance: Value[_T] = Value()

    @functools.wraps(factory)
    def _singleton_factory() -> _T:
        if instance.is_set():
            return instance.get()

        with lock:
            if not instance:
                instance.set(factory())

            return instance.get()

    _singleton_factory.clear = instance.clear

    return _singleton_factory


def get_value_from_path(data, path):
    keys = path.split(".")
    try:
        result = functools.reduce(dict.__getitem__, keys, data)
        return result
    except KeyError:
        # Handle the case where the path is not valid for the given dictionary
        return None


def set_value_at_path(data, path, new_value):
    keys = path.split(".")
    last_key = keys[-1]

    # Traverse the dictionary to the second-to-last level
    nested_dict = functools.reduce(dict.__getitem__, keys[:-1], data)

    try:
        # Set the new value
        nested_dict[last_key] = new_value
    except KeyError:
        # Handle the case where the path is not valid for the given dictionary
        raise ValueError(f"Invalid path: {path}")

1	import functools	1✔
2	import re	1✔
3	import threading	1✔
4	from collections.abc import Callable	1✔
5	from typing import Any, Generic, TypeVar	1✔
6
7	from .collections import ensure_list	1✔
8	from .strings import first_char_to_lower, first_char_to_upper	1✔
9
10	ComplexType = list \| dict \| object	1✔
11
12	_T = TypeVar("_T")	1✔
13
14
15	class Value(Generic[_T]):	1✔
16	"""
17	Simple value container.
18	"""
19
20	value: _T \| None	1✔
21
22	def __init__(self, value: _T = None) -> None:	1✔
23	self.value = value	1✔
24
25	def clear(self):	1✔
26	self.value = None	1✔
27
28	def set(self, value: _T):	1✔
29	self.value = value	1✔
30
31	def is_set(self) -> bool:	1✔
32	return self.value is not None	1✔
33
34	def get(self) -> _T \| None:	1✔
35	return self.value	1✔
36
37	def __bool__(self):	1✔
38	return True if self.value else False	1✔
39
40
41	class ArbitraryAccessObj:	1✔
42	"""Dummy object that can be arbitrarily accessed - any attributes, as a callable, item assignment, ..."""
43
44	def __init__(self, name=None):	1✔
UNCOV 45	self.name = name	×
46
47	def __getattr__(self, name, args, *kwargs):	1✔
UNCOV 48	return ArbitraryAccessObj(name)	×
49
50	def __call__(self, args, *kwargs):	1✔
51	if self.name in ["items", "keys", "values"] and not args and not kwargs:	×
52	return []	×
UNCOV 53	return ArbitraryAccessObj()	×
54
55	def __getitem__(self, args, *kwargs):	1✔
UNCOV 56	return ArbitraryAccessObj()	×
57
58	def __setitem__(self, args, *kwargs):	1✔
UNCOV 59	return ArbitraryAccessObj()	×
60
61
62	class Mock:	1✔
63	"""Dummy class that can be used for mocking custom attributes."""
64
65	pass	1✔
66
67
68	class ObjectIdHashComparator:	1✔
69	"""Simple wrapper class that allows us to create a hashset using the object id(..) as the entries' hash value"""
70
71	def __init__(self, obj):	1✔
72	self.obj = obj	×
UNCOV 73	self._hash = id(obj)	×
74
75	def __hash__(self):	1✔
UNCOV 76	return self._hash	×
77
78	def __eq__(self, other):	1✔
79	# assumption here is that we're comparing only against ObjectIdHash instances!
UNCOV 80	return self.obj == other.obj	×
81
82
83	class SubtypesInstanceManager:	1✔
84	"""Simple instance manager base class that scans the subclasses of a base type for concrete named
85	implementations, and lazily creates and returns (singleton) instances on demand."""
86
87	_instances: dict[str, "SubtypesInstanceManager"]	1✔
88
89	@classmethod	1✔
90	def get(cls, subtype_name: str, raise_if_missing: bool = True):	1✔
91	instances = cls.instances()	1✔
92	base_type = cls.get_base_type()	1✔
93	instance = instances.get(subtype_name)	1✔
94	if instance is None:	1✔
95	# lazily load subtype instance (required if new plugins are dynamically loaded at runtime)
96	for clazz in get_all_subclasses(base_type):	1✔
97	impl_name = clazz.impl_name()	1✔
98	if impl_name not in instances and subtype_name == impl_name:	1✔
99	instances[impl_name] = clazz()	1✔
100	instance = instances.get(subtype_name)	1✔
101	if not instance and raise_if_missing:	1✔
102	raise NotImplementedError(
103	f"Unable to find implementation named '{subtype_name}' for base type {base_type}"
104	)
105	return instance	1✔
106
107	@classmethod	1✔
108	def instances(cls) -> dict[str, "SubtypesInstanceManager"]:	1✔
109	base_type = cls.get_base_type()	1✔
110	if not hasattr(base_type, "_instances"):	1✔
111	base_type._instances = {}	1✔
112	return base_type._instances	1✔
113
114	@staticmethod	1✔
115	def impl_name() -> str:	1✔
116	"""Name of this concrete subtype - to be implemented by subclasses."""
117	raise NotImplementedError
118
119	@classmethod	1✔
120	def get_base_type(cls) -> type:	1✔
121	"""Get the base class for which instances are being managed - can be customized by subtypes."""
122	return cls	1✔
123
124
125	# this requires that all subclasses have been imported before(!)
126	def get_all_subclasses(clazz: type) -> set[type]:	1✔
127	"""Recursively get all subclasses of the given class."""
128	result = set()	1✔
129	subs = clazz.__subclasses__()	1✔
130	for sub in subs:	1✔
131	result.add(sub)	1✔
132	result.update(get_all_subclasses(sub))	1✔
133	return result	1✔
134
135
136	def fully_qualified_class_name(klass: type) -> str:	1✔
137	return f"{klass.__module__}.{klass.__name__}"	1✔
138
139
140	def not_none_or(value: Any \| None, alternative: Any) -> Any:	1✔
141	"""Return 'value' if it is not None, or 'alternative' otherwise."""
142	return value if value is not None else alternative	1✔
143
144
145	def recurse_object(obj: ComplexType, func: Callable, path: str = "") -> ComplexType:	1✔
146	"""Recursively apply `func` to `obj` (might be a list, dict, or other object)."""
147	obj = func(obj, path=path)	1✔
148	if isinstance(obj, list):	1✔
149	for i in range(len(obj)):	1✔
150	tmp_path = f"{path or '.'}[{i}]"	1✔
151	obj[i] = recurse_object(obj[i], func, tmp_path)	1✔
152	elif isinstance(obj, dict):	1✔
153	for k, v in obj.items():	1✔
154	tmp_path = f"{f'{path}.' if path else ''}{k}"	1✔
155	obj[k] = recurse_object(v, func, tmp_path)	1✔
156	return obj	1✔
157
158
159	def keys_to(	1✔
160	obj: ComplexType, op: Callable[[str], str], skip_children_of: list[str] = None
161	) -> ComplexType:
162	"""Recursively changes all dict keys to apply op. Skip children
163	of any elements whose names are contained in skip_children_of (e.g., ['Tags'])"""
164	skip_children_of = ensure_list(skip_children_of or [])	1✔
165
166	def fix_keys(o, path="", **kwargs):	1✔
167	if any(re.match(rf"(^\|.\.){k}($\|[.\[].)", path) for k in skip_children_of):	1✔
168	return o	1✔
169	if isinstance(o, dict):	1✔
170	for k, v in dict(o).items():	1✔
171	o.pop(k)	1✔
172	o[op(k)] = v	1✔
173	return o	1✔
174
175	result = recurse_object(obj, fix_keys)	1✔
176	return result	1✔
177
178
179	def keys_to_lower(obj: ComplexType, skip_children_of: list[str] = None) -> ComplexType:	1✔
180	return keys_to(obj, first_char_to_lower, skip_children_of)	1✔
181
182
183	def keys_to_upper(obj: ComplexType, skip_children_of: list[str] = None) -> ComplexType:	1✔
UNCOV 184	return keys_to(obj, first_char_to_upper, skip_children_of)	×
185
186
187	def singleton_factory(factory: Callable[[], _T]) -> Callable[[], _T]:	1✔
188	"""
189	Decorator for methods that create a particular value once and then return the same value in a thread safe way.
190
191	:param factory: the method to decorate
192	:return: a threadsafe singleton factory
193	"""
194	lock = threading.RLock()	1✔
195	instance: Value[_T] = Value()	1✔
196
197	@functools.wraps(factory)	1✔
198	def _singleton_factory() -> _T:	1✔
199	if instance.is_set():	1✔
200	return instance.get()	1✔
201
202	with lock:	1✔
203	if not instance:	1✔
204	instance.set(factory())	1✔
205
206	return instance.get()	1✔
207
208	_singleton_factory.clear = instance.clear	1✔
209
210	return _singleton_factory	1✔
211
212
213	def get_value_from_path(data, path):	1✔
214	keys = path.split(".")	1✔
215	try:	1✔
216	result = functools.reduce(dict.__getitem__, keys, data)	1✔
217	return result	1✔
218	except KeyError:	1✔
219	# Handle the case where the path is not valid for the given dictionary
220	return None	1✔
221
222
223	def set_value_at_path(data, path, new_value):	1✔
224	keys = path.split(".")	×
UNCOV 225	last_key = keys[-1]	×
226
227	# Traverse the dictionary to the second-to-last level
UNCOV 228	nested_dict = functools.reduce(dict.__getitem__, keys[:-1], data)	×
229
UNCOV 230	try:	×
231	# Set the new value
232	nested_dict[last_key] = new_value	×
UNCOV 233	except KeyError:	×
234	# Handle the case where the path is not valid for the given dictionary
UNCOV 235	raise ValueError(f"Invalid path: {path}")	×

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