Preserve native special-method lookup in automatic proxies

docs/changes.rst

@@ -6,6 +6,15 @@ Version 2.5.2
 
 **Bugs Fixed**
 
+* Special methods forwarded by ``wrapt.AutoObjectProxy`` and
+  ``wrapt.LazyObjectProxy`` were retrieved from the wrapped instance
+  rather than its type. Instance attributes shadowing those methods,
+  or a custom ``__getattribute__()``, could therefore change iteration,
+  awaiting, path conversion, and descriptor operations through the proxy
+  even though Python ignored them on the wrapped object. Forwarding now
+  looks up the methods on the actual type and preserves Python's
+  descriptor binding rules, including static and class methods.
+
 * The special methods added to the class generated for an
   ``wrapt.AutoObjectProxy``, and adjusted when ``__wrapped__`` was
   reassigned, were determined by inspecting the wrapped object itself,

src/wrapt/proxies.py

@@ -6,6 +6,30 @@
 from .__wrapt__ import BaseObjectProxy
 from .synchronization import synchronized
 
+# Read the real class dictionaries and MRO even when a metaclass overrides
+# attribute access or exposes a different value under either name.
+_TYPE_DICT = type.__dict__["__dict__"]
+_TYPE_MRO = type.__dict__["__mro__"]
+
+
+def _get_type_attribute(cls, name):
+    for base in _TYPE_MRO.__get__(cls):
+        namespace = _TYPE_DICT.__get__(base)
+        if name in namespace:
+            return namespace[name]
+    raise AttributeError(name)
+
+
+def _get_special_method(instance, name):
+    owner = type(instance)
+    attribute = _get_type_attribute(owner, name)
+    try:
+        getter = _get_type_attribute(type(attribute), "__get__")
+    except AttributeError:
+        return attribute
+    return getter(attribute, instance, owner)
+
+
 # Define ObjectProxy which for compatibility adds `__iter__()` support which
 # has been removed from `BaseObjectProxy`.
 
@@ -44,43 +68,46 @@ def __wrapper_iter__(self):
 
 
 def __wrapper_next__(self):
-    return self.__wrapped__.__next__()
+    return _get_special_method(self.__wrapped__, "__next__")()
 
 
 def __wrapper_aiter__(self):
-    return self.__wrapped__.__aiter__()
+    return _get_special_method(self.__wrapped__, "__aiter__")()
 
 
 async def __wrapper_anext__(self):
-    return await self.__wrapped__.__anext__()
+    return await _get_special_method(self.__wrapped__, "__anext__")()
 
 
 def __wrapper_length_hint__(self):
-    return self.__wrapped__.__length_hint__()
+    return _get_special_method(self.__wrapped__, "__length_hint__")()
 
 
 def __wrapper_fspath__(self):
-    return self.__wrapped__.__fspath__()
+    return _get_special_method(self.__wrapped__, "__fspath__")()
 
 
 def __wrapper_await__(self):
-    return (yield from self.__wrapped__.__await__())
+    return (yield from _get_special_method(self.__wrapped__, "__await__")())
 
 
 def __wrapper_get__(self, instance, owner=None):
-    return self.__wrapped__.__get__(instance, owner)
+    # The descriptor hook itself is called with its descriptor explicitly,
+    # without first binding the hook, as for CPython's tp_descr_get slot.
+    wrapped = self.__wrapped__
+    return _get_type_attribute(type(wrapped), "__get__")(wrapped, instance, owner)
 
 
 def __wrapper_set__(self, instance, value):
-    return self.__wrapped__.__set__(instance, value)
+    return _get_special_method(self.__wrapped__, "__set__")(instance, value)
 
 
 def __wrapper_delete__(self, instance):
-    return self.__wrapped__.__delete__(instance)
+    return _get_special_method(self.__wrapped__, "__delete__")(instance)
 
 
 def __wrapper_set_name__(self, owner, name):
-    return self.__wrapped__.__set_name__(owner, name)
+    return _get_special_method(self.__wrapped__, "__set_name__")(owner, name)
 
 
 class AutoObjectProxy(BaseObjectProxy):

tests/core/test_special_method_lookup.py

@@ -0,0 +1,279 @@
+import asyncio
+import operator
+import os
+import unittest
+
+import wrapt
+
+
+def shadow(*args, **kwargs):
+    raise AssertionError("instance attribute must not implement the protocol")
+
+
+class SpecialMethodLookupTests(unittest.TestCase):
+    def test_iterator_and_length_hint_ignore_instance_attributes(self):
+        class Iterator:
+            def __init__(self):
+                self.values = iter([1, 2])
+
+            def __iter__(self):
+                return self
+
+            def __next__(self):
+                return next(self.values)
+
+            def __length_hint__(self):
+                return operator.length_hint(self.values)
+
+        target = Iterator()
+        target.__next__ = shadow
+        target.__length_hint__ = shadow
+        proxy = wrapt.AutoObjectProxy(target)
+        self.assertEqual(operator.length_hint(target), 2)
+        self.assertEqual(operator.length_hint(proxy), 2)
+        self.assertEqual(next(target), 1)
+        self.assertEqual(next(proxy), 2)
+        with self.assertRaises(StopIteration):
+            next(proxy)
+
+    def test_path_ignores_instance_getattribute(self):
+        class Path:
+            def __getattribute__(self, name):
+                if name == "__fspath__":
+                    return shadow
+                return super().__getattribute__(name)
+
+            def __fspath__(self):
+                return "original-path"
+
+        target = Path()
+        self.assertEqual(os.fspath(target), "original-path")
+        self.assertEqual(os.fspath(wrapt.AutoObjectProxy(target)), os.fspath(target))
+        lazy = wrapt.LazyObjectProxy(lambda: target, interface=Path)
+        self.assertEqual(os.fspath(lazy), os.fspath(target))
+
+    def test_async_iteration_ignores_instance_attributes(self):
+        class Iterator:
+            def __init__(self):
+                self.values = iter([1, 2])
+                self.__aiter__ = shadow
+                self.__anext__ = shadow
+
+            def __aiter__(self):
+                return self
+
+            async def __anext__(self):
+                try:
+                    return next(self.values)
+                except StopIteration:
+                    raise StopAsyncIteration from None
+
+        async def collect(value):
+            return [item async for item in value]
+
+        self.assertEqual(asyncio.run(collect(Iterator())), [1, 2])
+        self.assertEqual(
+            asyncio.run(collect(wrapt.AutoObjectProxy(Iterator()))), [1, 2]
+        )
+
+    def test_await_ignores_instance_attributes(self):
+        class Awaitable:
+            def __await__(self):
+                yield
+                return 42
+
+        async def wait(value):
+            return await value
+
+        target = Awaitable()
+        target.__await__ = shadow
+        self.assertEqual(asyncio.run(wait(target)), 42)
+        self.assertEqual(asyncio.run(wait(wrapt.AutoObjectProxy(target))), 42)
+
+    def test_descriptor_hooks_ignore_instance_attributes(self):
+        class Descriptor:
+            def __init__(self):
+                self.events = []
+                self.value = 1
+                self.__get__ = shadow
+                self.__set__ = shadow
+                self.__delete__ = shadow
+                self.__set_name__ = shadow
+
+            def __set_name__(self, owner, name):
+                self.events.append((owner.__name__, name))
+
+            def __get__(self, instance, owner):
+                return self.value
+
+            def __set__(self, instance, value):
+                self.value = value
+
+            def __delete__(self, instance):
+                self.value = None
+
+        for proxied in (False, True):
+            with self.subTest(proxied=proxied):
+                target = Descriptor()
+                descriptor = wrapt.AutoObjectProxy(target) if proxied else target
+
+                class Owner:
+                    field = descriptor
+
+                self.assertEqual(target.events, [("Owner", "field")])
+                self.assertEqual(Owner.field, 1)
+                owner = Owner()
+                self.assertEqual(owner.field, 1)
+                owner.field = 2
+                self.assertEqual(owner.field, 2)
+                del owner.field
+                self.assertIsNone(owner.field)
+
+    def test_static_and_class_methods_ignore_metaclass_getattribute(self):
+        checking = [False]
+
+        class Meta(type):
+            def __getattribute__(cls, name):
+                if checking[0] and name in ("__dict__", "__mro__", "__next__"):
+                    raise AssertionError("metaclass lookup must not run")
+                return super().__getattribute__(name)
+
+        class StaticIterator(metaclass=Meta):
+            def __iter__(self):
+                return self
+
+            @staticmethod
+            def __next__():
+                return "static"
+
+        class ClassIterator(metaclass=Meta):
+            value = "class"
+
+            def __iter__(self):
+                return self
+
+            @classmethod
+            def __next__(cls):
+                return cls.value
+
+        for cls in (StaticIterator, ClassIterator):
+            with self.subTest(cls=cls.__name__):
+                checking[0] = False
+                target = cls()
+                proxy = wrapt.AutoObjectProxy(target)
+                checking[0] = True
+                self.assertEqual(next(proxy), next(target))
+
+    def test_custom_descriptor_binds_to_target(self):
+        class Method:
+            def __get__(self, instance, owner):
+                return lambda: (instance.value, owner.__name__)
+
+        class Iterator:
+            value = 42
+            __next__ = Method()
+
+        target = Iterator()
+        self.assertEqual(next(wrapt.AutoObjectProxy(target)), next(target))
+
+    def test_callable_special_method_without_descriptor(self):
+        class Next:
+            def __call__(self):
+                return 42
+
+        class Iterator:
+            __next__ = Next()
+
+        target = Iterator()
+        self.assertEqual(next(wrapt.AutoObjectProxy(target)), next(target))
+
+    def test_binding_attribute_error_propagates(self):
+        class Method:
+            def __get__(self, instance, owner):
+                raise AttributeError("binding failed")
+
+        class Iterator:
+            __next__ = Method()
+
+        for target in (Iterator(), wrapt.AutoObjectProxy(Iterator())):
+            with self.assertRaisesRegex(AttributeError, "binding failed"):
+                next(target)
+
+    def test_descriptor_classmethod_get_stays_noncallable(self):
+        class Descriptor:
+            @classmethod
+            def __get__(cls, instance, owner):
+                return 42
+
+        for proxied in (False, True):
+            with self.subTest(proxied=proxied):
+                target = Descriptor()
+
+                class Owner:
+                    field = wrapt.AutoObjectProxy(target) if proxied else target
+
+                with self.assertRaises(TypeError):
+                    Owner.field
+
+    def test_descriptor_staticmethod_get_receives_descriptor(self):
+        class Descriptor:
+            value = 42
+
+            @staticmethod
+            def __get__(descriptor, instance, owner):
+                return descriptor.value
+
+        for proxied in (False, True):
+            with self.subTest(proxied=proxied):
+                target = Descriptor()
+
+                class Owner:
+                    field = wrapt.AutoObjectProxy(target) if proxied else target
+
+                # Raw staticmethod objects became callable in Python 3.10.
+                if callable(Descriptor.__dict__["__get__"]):
+                    self.assertEqual(Owner.field, 42)
+                else:
+                    with self.assertRaises(TypeError):
+                        Owner.field
+
+    def test_special_method_staticmethod_get_receives_descriptor(self):
+        class Method:
+            @staticmethod
+            def __get__(descriptor, instance, owner):
+                return lambda: instance.value
+
+        class Iterator:
+            value = 42
+            __next__ = Method()
+
+        target = Iterator()
+        proxy = wrapt.AutoObjectProxy(target)
+        if callable(Method.__dict__["__get__"]):
+            self.assertEqual(next(proxy), next(target))
+        else:
+            for value in (target, proxy):
+                with self.assertRaises(TypeError):
+                    next(value)
+
+    def test_replacement_target_keeps_special_lookup(self):
+        class Path:
+            def __init__(self, value):
+                self.value = value
+                self.__fspath__ = shadow
+
+            def __fspath__(self):
+                return self.value
+
+        proxy = wrapt.AutoObjectProxy(Path("first"))
+        self.assertEqual(os.fspath(proxy), "first")
+        proxy.__wrapped__ = Path("second")
+        self.assertEqual(os.fspath(proxy), "second")
+
+    def test_none_method_stays_noncallable(self):
+        class Iterator:
+            __next__ = None
+
+        for target in (Iterator(), wrapt.AutoObjectProxy(Iterator())):
+            with self.assertRaises(TypeError):
+                next(target)