From 6cef23cfaeff930583ed5370de17dfd2b917b9c0 Mon Sep 17 00:00:00 2001 From: Mayank Dhingra Date: Tue, 20 Oct 2026 12:39:58 -0400 Subject: [PATCH] Preserve special-method lookup through automatic proxies Resolve forwarded protocol hooks on the wrapped object type and preserve the descriptor binding Python uses for implicit special method calls. Instance attributes and custom attribute lookup should not replace the hooks used by iteration, awaiting, path conversion or descriptors. Signed-off-by: Mayank Dhingra --- docs/changes.rst | 9 + src/wrapt/proxies.py | 47 +++- tests/core/test_special_method_lookup.py | 279 +++++++++++++++++++++++ 3 files changed, 325 insertions(+), 10 deletions(-) create mode 100644 tests/core/test_special_method_lookup.py diff --git a/docs/changes.rst b/docs/changes.rst index 40e6cc0a..c42eb7fb 100644 --- a/docs/changes.rst +++ b/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, diff --git a/src/wrapt/proxies.py b/src/wrapt/proxies.py index 18c80e05..c5f5229f 100644 --- a/src/wrapt/proxies.py +++ b/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): diff --git a/tests/core/test_special_method_lookup.py b/tests/core/test_special_method_lookup.py new file mode 100644 index 00000000..6ee3e261 --- /dev/null +++ b/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)