17086927072

Committed 19 Aug 2025 10:02PM UTC coverage: 86.889% (+0.01%) from 86.875%

Build # 17086927072

Build Type

push

github

Committed by

web-flow

Commit Message

APIGW: fix TestInvokeMethod path logic (#13030)

Coverage Stats

4 of 23 new or added lines in 1 file covered. (17.39%)

264 existing lines in 17 files now uncovered.

67018 of 77131 relevant lines covered (86.89%)

0.87 hits per line

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

90.91

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

import functools
import inspect
import types
from typing import Any, Callable


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

localstack / localstack / 17086927072

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