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

90.98

/localstack-core/localstack/utils/patch.py

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


def get_defining_object(method):
    """Returns either the class or the module that defines the given function/method."""
    # adapted from https://stackoverflow.com/a/25959545/804840
    if inspect.ismethod(method):
        return method.__self__

    if inspect.isfunction(method):
        class_name = method.__qualname__.split(".<locals>", 1)[0].rsplit(".", 1)[0]
        try:
            # method is not bound but referenced by a class, like MyClass.mymethod
            cls = getattr(inspect.getmodule(method), class_name)
        except AttributeError:
            cls = method.__globals__.get(class_name)

        if isinstance(cls, type):
            return cls

    # method is a module-level function
    return inspect.getmodule(method)


def to_metadata_string(obj: Any) -> str:
    """
    Creates a string that helps humans understand where the given object comes from. Examples::

      to_metadata_string(func_thread.run) == "function(localstack.utils.threads:FuncThread.run)"

    :param obj: a class, module, method, function or object
    :return: a string representing the objects origin
    """
    if inspect.isclass(obj):
        return f"class({obj.__module__}:{obj.__name__})"
    if inspect.ismodule(obj):
        return f"module({obj.__name__})"
    if inspect.ismethod(obj):
        return f"method({obj.__module__}:{obj.__qualname__})"
    if inspect.isfunction(obj):
        # TODO: distinguish bound method
        return f"function({obj.__module__}:{obj.__qualname__})"
    if isinstance(obj, object):
        return f"object({obj.__module__}:{obj.__class__.__name__})"
    return str(obj)


def create_patch_proxy(target: Callable, new: Callable):
    """
    Creates a proxy that calls `new` but passes as first argument the target.
    """

    @functools.wraps(target)
    def proxy(*args, **kwargs):
        if _is_bound_method:
            # bound object "self" is passed as first argument if this is a bound method
            args = args[1:]
        return new(target, *args, **kwargs)

    # keep track of the real proxy subject (i.e., the new function that is used as patch)
    proxy.__subject__ = new

    _is_bound_method = inspect.ismethod(target)
    if _is_bound_method:
        proxy = types.MethodType(proxy, target.__self__)

    return proxy


class Patch:
    applied_patches: list["Patch"] = []
    """Bookkeeping for patches that are applied. You can use this to debug patches. For instance,
    you could write something like::

      for patch in Patch.applied_patches:
          print(patch)

    Which will output in a human readable format information about the currently active patches.
    """

    obj: Any
    name: str
    new: Any

    def __init__(self, obj: Any, name: str, new: Any) -> None:
        super().__init__()
        self.obj = obj
        self.name = name
        try:
            self.old = getattr(self.obj, name)
        except AttributeError:
            self.old = None
        self.new = new
        self.is_applied = False

    def apply(self):
        if self.is_applied:
            return

        if self.old and self.name == "__getattr__":
            raise Exception("You can't patch class types implementing __getattr__")
        if not self.old and self.name != "__getattr__":
            raise AttributeError(f"`{self.obj.__name__}` object has no attribute `{self.name}`")
        setattr(self.obj, self.name, self.new)
        Patch.applied_patches.append(self)
        self.is_applied = True

    def undo(self):
        if not self.is_applied:
            return

        # If we added a method to a class type, we don't have a self.old. We just delete __getattr__
        setattr(self.obj, self.name, self.old) if self.old else delattr(self.obj, self.name)
        Patch.applied_patches.remove(self)
        self.is_applied = False

    def __enter__(self):
        self.apply()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.undo()
        return self

    @staticmethod
    def extend_class(target: type, fn: Callable):
        def _getattr(obj, name):
            if name != fn.__name__:
                raise AttributeError(f"`{target.__name__}` object has no attribute `{name}`")

            return functools.partial(fn, obj)

        return Patch(target, "__getattr__", _getattr)

    @staticmethod
    def function(target: Callable, fn: Callable, pass_target: bool = True):
        obj = get_defining_object(target)
        name = target.__name__

        is_class_instance = not inspect.isclass(obj) and not inspect.ismodule(obj)
        if is_class_instance:
            # special case: If the defining object is not a class, but a class instance,
            # then we need to bind the patch function to the target object. Also, we need
            # to ensure that the final patched method has the same name as the original
            # method on the defining object (required for restoring objects with patched
            # methods from persistence, to avoid AttributeError).
            fn.__name__ = name
            fn = types.MethodType(fn, obj)

        if pass_target:
            new = create_patch_proxy(target, fn)
        else:
            new = fn

        return Patch(obj, name, new)

    def __str__(self):
        try:
            # try to unwrap the original underlying function that is used as patch (basically undoes what
            # ``create_patch_proxy`` does)
            new = self.new.__subject__
        except AttributeError:
            new = self.new

        old = self.old
        return f"Patch({to_metadata_string(old)} -> {to_metadata_string(new)}, applied={self.is_applied})"


class Patches:
    patches: list[Patch]

    def __init__(self, patches: list[Patch] = None) -> None:
        super().__init__()

        self.patches = []
        if patches:
            self.patches.extend(patches)

    def apply(self):
        for p in self.patches:
            p.apply()

    def undo(self):
        for p in self.patches:
            p.undo()

    def __enter__(self):
        self.apply()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.undo()

    def add(self, patch: Patch):
        self.patches.append(patch)

    def function(self, target: Callable, fn: Callable, pass_target: bool = True):
        self.add(Patch.function(target, fn, pass_target))


def patch(target, pass_target=True):
    """
    Function decorator to create a patch via Patch.function and immediately apply it.

    Example::

        def echo(string):
            return "echo " + string

        @patch(target=echo)
        def echo_uppercase(target, string):
            return target(string).upper()

        echo("foo")
        # will print "ECHO FOO"

        echo_uppercase.patch.undo()
        echo("foo")
        # will print "echo foo"

    When you are patching classes, with ``pass_target=True``, the unbound function will be passed as the first
    argument before ``self``.

    For example::

        @patch(target=MyEchoer.do_echo, pass_target=True)
        def my_patch(fn, self, *args):
            return fn(self, *args)

        @patch(target=MyEchoer.do_echo, pass_target=False)
        def my_patch(self, *args):
            ...

    This decorator can also patch a class type with a new method.

    For example:
        @patch(target=MyEchoer)
        def new_echo(self, *args):
            ...

    :param target: the function or method to patch
    :param pass_target: whether to pass the target to the patching function as first parameter
    :returns: the same function, but with a patch created
    """

    @functools.wraps(target)
    def wrapper(fn):
        fn.patch = (
            Patch.extend_class(target, fn)
            if inspect.isclass(target)
            else Patch.function(target, fn, pass_target=pass_target)
        )
        fn.patch.apply()
        return fn

    return wrapper

1	import functools	1✔
2	import inspect	1✔
3	import types	1✔
4	from collections.abc import Callable	1✔
5	from typing import Any	1✔
6
7
8	def get_defining_object(method):	1✔
9	"""Returns either the class or the module that defines the given function/method."""
10	# adapted from https://stackoverflow.com/a/25959545/804840
11	if inspect.ismethod(method):	1✔
12	return method.__self__	1✔
13
14	if inspect.isfunction(method):	1✔
15	class_name = method.__qualname__.split(".<locals>", 1)[0].rsplit(".", 1)[0]	1✔
16	try:	1✔
17	# method is not bound but referenced by a class, like MyClass.mymethod
18	cls = getattr(inspect.getmodule(method), class_name)	1✔
19	except AttributeError:	×
UNCOV 20	cls = method.__globals__.get(class_name)	×
21
22	if isinstance(cls, type):	1✔
23	return cls	1✔
24
25	# method is a module-level function
26	return inspect.getmodule(method)	1✔
27
28
29	def to_metadata_string(obj: Any) -> str:	1✔
30	"""
31	Creates a string that helps humans understand where the given object comes from. Examples::
32
33	to_metadata_string(func_thread.run) == "function(localstack.utils.threads:FuncThread.run)"
34
35	:param obj: a class, module, method, function or object
36	:return: a string representing the objects origin
37	"""
38	if inspect.isclass(obj):	1✔
39	return f"class({obj.__module__}:{obj.__name__})"	1✔
40	if inspect.ismodule(obj):	1✔
UNCOV 41	return f"module({obj.__name__})"	×
42	if inspect.ismethod(obj):	1✔
UNCOV 43	return f"method({obj.__module__}:{obj.__qualname__})"	×
44	if inspect.isfunction(obj):	1✔
45	# TODO: distinguish bound method
46	return f"function({obj.__module__}:{obj.__qualname__})"	1✔
47	if isinstance(obj, object):	×
48	return f"object({obj.__module__}:{obj.__class__.__name__})"	×
UNCOV 49	return str(obj)	×
50
51
52	def create_patch_proxy(target: Callable, new: Callable):	1✔
53	"""
54	Creates a proxy that calls `new` but passes as first argument the target.
55	"""
56
57	@functools.wraps(target)	1✔
58	def proxy(args, *kwargs):	1✔
59	if _is_bound_method:	1✔
60	# bound object "self" is passed as first argument if this is a bound method
61	args = args[1:]	1✔
62	return new(target, args, *kwargs)	1✔
63
64	# keep track of the real proxy subject (i.e., the new function that is used as patch)
65	proxy.__subject__ = new	1✔
66
67	_is_bound_method = inspect.ismethod(target)	1✔
68	if _is_bound_method:	1✔
69	proxy = types.MethodType(proxy, target.__self__)	1✔
70
71	return proxy	1✔
72
73
74	class Patch:	1✔
75	applied_patches: list["Patch"] = []	1✔
76	"""Bookkeeping for patches that are applied. You can use this to debug patches. For instance,	1✔
77	you could write something like::
78
79	for patch in Patch.applied_patches:
80	print(patch)
81
82	Which will output in a human readable format information about the currently active patches.
83	"""
84
85	obj: Any	1✔
86	name: str	1✔
87	new: Any	1✔
88
89	def __init__(self, obj: Any, name: str, new: Any) -> None:	1✔
90	super().__init__()	1✔
91	self.obj = obj	1✔
92	self.name = name	1✔
93	try:	1✔
94	self.old = getattr(self.obj, name)	1✔
95	except AttributeError:	1✔
96	self.old = None	1✔
97	self.new = new	1✔
98	self.is_applied = False	1✔
99
100	def apply(self):	1✔
101	if self.is_applied:	1✔
102	return	1✔
103
104	if self.old and self.name == "__getattr__":	1✔
UNCOV 105	raise Exception("You can't patch class types implementing __getattr__")	×
106	if not self.old and self.name != "__getattr__":	1✔
107	raise AttributeError(f"`{self.obj.__name__}` object has no attribute `{self.name}`")	1✔
108	setattr(self.obj, self.name, self.new)	1✔
109	Patch.applied_patches.append(self)	1✔
110	self.is_applied = True	1✔
111
112	def undo(self):	1✔
113	if not self.is_applied:	1✔
UNCOV 114	return	×
115
116	# If we added a method to a class type, we don't have a self.old. We just delete __getattr__
117	setattr(self.obj, self.name, self.old) if self.old else delattr(self.obj, self.name)	1✔
118	Patch.applied_patches.remove(self)	1✔
119	self.is_applied = False	1✔
120
121	def __enter__(self):	1✔
122	self.apply()	1✔
123	return self	1✔
124
125	def __exit__(self, exc_type, exc_val, exc_tb):	1✔
126	self.undo()	1✔
127	return self	1✔
128
129	@staticmethod	1✔
130	def extend_class(target: type, fn: Callable):	1✔
131	def _getattr(obj, name):	1✔
132	if name != fn.__name__:	1✔
UNCOV 133	raise AttributeError(f"`{target.__name__}` object has no attribute `{name}`")	×
134
135	return functools.partial(fn, obj)	1✔
136
137	return Patch(target, "__getattr__", _getattr)	1✔
138
139	@staticmethod	1✔
140	def function(target: Callable, fn: Callable, pass_target: bool = True):	1✔
141	obj = get_defining_object(target)	1✔
142	name = target.__name__	1✔
143
144	is_class_instance = not inspect.isclass(obj) and not inspect.ismodule(obj)	1✔
145	if is_class_instance:	1✔
146	# special case: If the defining object is not a class, but a class instance,
147	# then we need to bind the patch function to the target object. Also, we need
148	# to ensure that the final patched method has the same name as the original
149	# method on the defining object (required for restoring objects with patched
150	# methods from persistence, to avoid AttributeError).
151	fn.__name__ = name	1✔
152	fn = types.MethodType(fn, obj)	1✔
153
154	if pass_target:	1✔
155	new = create_patch_proxy(target, fn)	1✔
156	else:
157	new = fn	1✔
158
159	return Patch(obj, name, new)	1✔
160
161	def __str__(self):	1✔
162	try:	1✔
163	# try to unwrap the original underlying function that is used as patch (basically undoes what
164	# ``create_patch_proxy`` does)
165	new = self.new.__subject__	1✔
166	except AttributeError:	1✔
167	new = self.new	1✔
168
169	old = self.old	1✔
170	return f"Patch({to_metadata_string(old)} -> {to_metadata_string(new)}, applied={self.is_applied})"	1✔
171
172
173	class Patches:	1✔
174	patches: list[Patch]	1✔
175
176	def __init__(self, patches: list[Patch] = None) -> None:	1✔
177	super().__init__()	1✔
178
179	self.patches = []	1✔
180	if patches:	1✔
181	self.patches.extend(patches)	1✔
182
183	def apply(self):	1✔
184	for p in self.patches:	1✔
185	p.apply()	1✔
186
187	def undo(self):	1✔
188	for p in self.patches:	1✔
189	p.undo()	1✔
190
191	def __enter__(self):	1✔
192	self.apply()	1✔
193	return self	1✔
194
195	def __exit__(self, exc_type, exc_val, exc_tb):	1✔
196	self.undo()	1✔
197
198	def add(self, patch: Patch):	1✔
UNCOV 199	self.patches.append(patch)	×
200
201	def function(self, target: Callable, fn: Callable, pass_target: bool = True):	1✔
UNCOV 202	self.add(Patch.function(target, fn, pass_target))	×
203
204
205	def patch(target, pass_target=True):	1✔
206	"""
207	Function decorator to create a patch via Patch.function and immediately apply it.
208
209	Example::
210
211	def echo(string):
212	return "echo " + string
213
214	@patch(target=echo)
215	def echo_uppercase(target, string):
216	return target(string).upper()
217
218	echo("foo")
219	# will print "ECHO FOO"
220
221	echo_uppercase.patch.undo()
222	echo("foo")
223	# will print "echo foo"
224
225	When you are patching classes, with ``pass_target=True``, the unbound function will be passed as the first
226	argument before ``self``.
227
228	For example::
229
230	@patch(target=MyEchoer.do_echo, pass_target=True)
231	def my_patch(fn, self, *args):
232	return fn(self, *args)
233
234	@patch(target=MyEchoer.do_echo, pass_target=False)
235	def my_patch(self, *args):
236	...
237
238	This decorator can also patch a class type with a new method.
239
240	For example:
241	@patch(target=MyEchoer)
242	def new_echo(self, *args):
243	...
244
245	:param target: the function or method to patch
246	:param pass_target: whether to pass the target to the patching function as first parameter
247	:returns: the same function, but with a patch created
248	"""
249
250	@functools.wraps(target)	1✔
251	def wrapper(fn):	1✔
252	fn.patch = (	1✔
253	Patch.extend_class(target, fn)
254	if inspect.isclass(target)
255	else Patch.function(target, fn, pass_target=pass_target)
256	)
257	fn.patch.apply()	1✔
258	return fn	1✔
259
260	return wrapper	1✔

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