diff --git a/kgb/signature.py b/kgb/signature.py index 646af9b5ac64b29574d5af7d0d22462aae02cd51..b6b6dd5454e8bf294d092f568b275adbbaed0446 100644 --- a/kgb/signature.py +++ b/kgb/signature.py @@ -5,8 +5,10 @@ from __future__ import unicode_literals import inspect import logging import sys +import types from kgb.errors import InternalKGBError +from kgb.utils import get_defined_attr_value logger = logging.getLogger('kgb') @@ -25,6 +27,9 @@ class _UnsetArg(object): return '_UNSET_ARG' +_UNSET_ARG = _UnsetArg() + + class BaseFunctionSig(object): """Base class for a function signature introspector. @@ -35,6 +40,43 @@ class BaseFunctionSig(object): How this is all done depends entirely on the version of Python. Subclasses must implement all this logic. + + Version Changed: + 5.0: + Added support for the following attributes: + + * :py:attr:`defined_func`. + * :py:attr:`has_getter`. + * :py:attr:`has_setter`. + * :py:attr:`is_slippery`. + + Attributes: + defined_func (callable or object): + The actual function (or wrapping object) that's defined on + somewhere in the owner's class hierarchy (or the function itself if + this is a standalone function). This may differ from + :py:attr:`func`. + + has_getter (bool): + Whether this signature represents a descriptor with a ``__get__`` + method. + + has_setter (bool): + Whether this signature represents a descriptor with a ``__set__`` + method. + + is_slippery (bool): + Whether this represents a slippery function. This is a method on + a class that returns a different function every time its attribute + is accessed on an instance. + + This occurs when a method decorator is used that wraps a function + on access and returns the wrapper function, but does not cache the + wrapper function. These are returned as standard functions and + not methods. + + Slippery functions can only be detected when an explicit owner is + provided. """ #: The signature represents a standard function. @@ -50,17 +92,18 @@ class BaseFunctionSig(object): #: Unbound methods are standard methods on a class. TYPE_UNBOUND_METHOD = 2 - def __init__(self, func, owner=None): + def __init__(self, func, owner=_UNSET_ARG): """Initialize the signature. Subclasses must override this to parse function types/ownership and - available arguments. + available arguments. They must call :py:meth:`finalize_state` once + they've calculated all signature state. Args: func (callable): The function to use for the signature. - owner (type, optional): + owner (type or object, optional): The owning class, as provided when spying on the function. This is not stored directly (as it may be invalid), but can be used for informative purposes for subclasses. @@ -82,6 +125,9 @@ class BaseFunctionSig(object): self.kwarg_names = [] self.args_param_name = [] self.kwargs_param_name = [] + self.is_slippery = False + self.has_getter = False + self.has_setter = False def is_compatible_with(self, other_sig): """Return whether two function signatures are compatible. @@ -202,6 +248,34 @@ class BaseFunctionSig(object): """ raise NotImplementedError + def finalize_state(self): + """Finalize the state for the signature. + + This will set any remaining values for the signature based on the + calculations already performed by the subclasses. This must be called + at the end of a subclass's :py:meth:`__init__`. + """ + if self.owner is None: + self.defined_func = self.func + else: + try: + self.defined_func = get_defined_attr_value(self.owner, + self.func_name) + except AttributeError: + # This was a dynamically-injected function. We won't find it + # in the class hierarchy. Use the provided function instead. + self.defined_func = self.func + + if not isinstance(self.defined_func, (types.FunctionType, + types.MethodType, + classmethod, + staticmethod)): + if hasattr(self.defined_func, '__get__'): + self.has_getter = True + + if hasattr(self.defined_func, '__set__'): + self.has_setter = True + class FunctionSigPy2(BaseFunctionSig): """Function signature introspector for Python 2. @@ -217,7 +291,7 @@ class FunctionSigPy2(BaseFunctionSig): FUNC_NAME_ATTR = 'func_name' METHOD_SELF_ATTR = 'im_self' - def __init__(self, func, owner=None): + def __init__(self, func, owner=_UNSET_ARG): """Initialize the signature. Subclasses must override this to parse function types/ownership and @@ -229,13 +303,23 @@ class FunctionSigPy2(BaseFunctionSig): owner (type, optional): The owning class, as provided when spying on the function. - The value is ignored for Python 2. + The value is ignored for methods on which an owner can be + calculated, favoring the calculated value instead. """ super(FunctionSigPy2, self).__init__(func=func, owner=owner) + func_name = self.func_name + # Figure out the owner and method type. if inspect.ismethod(func): + # This is a bound or unbound method. If it's unbound, and an + # owner is not specified, we're going to need to warn the user, + # since things are going to break on Python 3. + # + # Otherwise, we're going to determine the bound vs. unbound type + # and use the owner specified by the method. (The provided owner + # will be validated in FunctionSpy.) method_owner = func.im_self if method_owner is None: @@ -251,6 +335,30 @@ class FunctionSigPy2(BaseFunctionSig): else: self.func_type = self.TYPE_BOUND_METHOD self.owner = method_owner + elif owner is not _UNSET_ARG: + # This is a standard function, but an owner (as an instance) has + # been provided. Find out if the owner has this function (either + # the actual instance or one with the same name and bytecode), + # and if so, treat this as a bound method. + # + # This is necessary when trying to spy on a decorated method that + # generates functions dynamically (using something like + # functools.wraps). We call these "slippery functions." A + # real-world example is something like Stripe's + # stripe.Customer.delete() method, which is a different function + # every time you call it. + owner_func = getattr(owner, func_name, None) + + if (owner_func is not None and + (owner_func is func or + owner_func.func_code is func.func_code)): + if inspect.isclass(owner): + self.func_type = self.TYPE_UNBOUND_METHOD + else: + self.func_type = self.TYPE_BOUND_METHOD + + self.owner = owner + self.is_slippery = owner_func is not func # Load information on the arguments. argspec = inspect.getargspec(func) @@ -272,6 +380,8 @@ class FunctionSigPy2(BaseFunctionSig): self.kwargs_param_name = argspec.keywords self._defaults = argspec.defaults + self.finalize_state() + def format_forward_call_arg(self, arg_name): """Return a string used to reference an argument in a forwarding call. @@ -324,7 +434,7 @@ class FunctionSigPy3(BaseFunctionSig): FUNC_NAME_ATTR = '__name__' METHOD_SELF_ATTR = '__self__' - def __init__(self, func, owner=None): + def __init__(self, func, owner=_UNSET_ARG): """Initialize the signature. Subclasses must override this to parse function types/ownership and @@ -336,7 +446,7 @@ class FunctionSigPy3(BaseFunctionSig): owner (type, optional): The owning class, as provided when spying on the function. - This is used only when spying on unbound methods. + This is used only when spying on unbound or slippery methods. """ super(FunctionSigPy3, self).__init__(func=func, owner=owner) @@ -348,6 +458,8 @@ class FunctionSigPy3(BaseFunctionSig): 'function.' % sys.version_info[:2]) + func_name = self.func_name + # Figure out the owner and method type. # # Python 3 does not officially have unbound methods. Methods on @@ -372,7 +484,12 @@ class FunctionSigPy3(BaseFunctionSig): elif '.' in func.__qualname__: if owner is not _UNSET_ARG: self.owner = owner - self.func_type = self.TYPE_UNBOUND_METHOD + self.is_slippery = getattr(owner, func_name) is not func + + if owner is _UNSET_ARG or inspect.isclass(owner): + self.func_type = self.TYPE_UNBOUND_METHOD + else: + self.func_type = self.TYPE_BOUND_METHOD elif '' in func.__qualname__: # We can only assume this is a function. It might not be. self.func_type = self.TYPE_FUNCTION @@ -439,6 +556,8 @@ class FunctionSigPy3(BaseFunctionSig): self.kwarg_names = kwargs self._sig = sig + self.finalize_state() + def format_forward_call_arg(self, arg_name): """Return a string used to reference an argument in a forwarding call. @@ -500,6 +619,3 @@ elif sys.version_info[0] == 3: FunctionSig = FunctionSigPy3 else: raise Exception('Unsupported Python version') - - -_UNSET_ARG = _UnsetArg() diff --git a/kgb/spies.py b/kgb/spies.py index fce6f8e6b6674b278df86fb0853707f980127781..e6c8f58214d6d0e1d545fe50fb86e624117276b3 100644 --- a/kgb/spies.py +++ b/kgb/spies.py @@ -9,6 +9,7 @@ from kgb.errors import (ExistingSpyError, InternalKGBError) from kgb.pycompat import iteritems, iterkeys, pyver, text_type from kgb.signature import FunctionSig, _UNSET_ARG +from kgb.utils import is_attr_defined_on_ancestor class SpyCall(object): @@ -188,6 +189,11 @@ class FunctionSpy(object): what it returned, and adding methods and state onto the function for callers to access in order to get those results. + Version Added: + 5.0: + Added support for specifying an instance in ``owner`` when spying + on bound methods using decorators that return plain functions. + Args: agency (kgb.agency.SpyAgency): The spy agency that manages this spy. @@ -203,6 +209,17 @@ class FunctionSpy(object): invoked. If ``False``, no function will be called. This is ignored if ``call_fake`` is provided. + + owner (type or object, optional): + The owner of the function or method. + + If spying on an unbound method, this **must** be set to the + class that owns it. + + If spying on a bound method that identifies as a plain + function (which may happen if the method is decorated and + dynamically returns a new function on access), this should + be the instance of the object you're spying on. """ # Check the parameters passed to make sure that invalid data wasn't # provided. @@ -244,13 +261,28 @@ class FunctionSpy(object): 'The owner passed does not match the actual owner of ' 'the bound method.') + if (self._sig.is_slippery and + self.func_type == self.TYPE_UNBOUND_METHOD): + raise ValueError('Unable to spies on unbound slippery methods ' + '(those that return a new function on each ' + 'attribute access). Please spy on an instance ' + 'instead.') + if (self.owner is not None and (not inspect.isclass(self.owner) or - any( - inspect.isclass(parent_cls) and - hasattr(parent_cls, self.func_name) - for parent_cls in self.owner.__bases__ - ))): + self._sig.is_slippery or + is_attr_defined_on_ancestor(self.owner, self.func_name))): + # We need to store the original attribute value for the function, + # as defined in the class that owns it. That may be the provided + # or calculated owner, or a parent of it. + # + # This is needed because the function provided may not actually be + # what's defined on the class. What's defined might be a decorator + # that returns a function, and it might not even be the same + # function each time it's accessed. + self._owner_func_attr_value = \ + self.owner.__dict__.get(self.func_name) + # Construct a replacement function for this method, and # re-assign it to the owner. We do this in order to prevent # two spies on the same method on two separate instances @@ -268,6 +300,8 @@ class FunctionSpy(object): types.MethodType(real_func, self.owner)) else: self._set_method(self.owner, self.func_name, real_func) + else: + self._owner_func_attr_value = self.orig_func # If call_fake was provided, check that it's valid and has a # compatible function signature. @@ -543,7 +577,8 @@ class FunctionSpy(object): setattr(self._real_func, FunctionSig.FUNC_CODE_ATTR, self._old_code) if self.owner is not None: - self._set_method(self.owner, self.func_name, self.orig_func) + self._set_method(self.owner, self.func_name, + self._owner_func_attr_value) if unregister: self.agency.spies.remove(self) @@ -799,7 +834,21 @@ class FunctionSpy(object): result = None else: try: - result = self.func(*args, **kwargs) + if self._sig.has_getter: + # This isn't a standard function. It's a descriptor with + # a __get__() method. We need to fetch the value it + # returns. + result = self._sig.defined_func.__get__(self.owner) + + if self._sig.is_slippery: + # Since we know this represents a slippery function, + # we need to take the function from the descriptor's + # result and call it. + result = result(*args, **kwargs) + else: + # This is a typical function/method. We can call it + # directly. + result = self.func(*args, **kwargs) except Exception as e: call.exception = e raise @@ -882,11 +931,17 @@ class FunctionSpy(object): def _set_method(self, owner, name, method): """Set a new method on an object. - This will set the method using a standard :py:func:`setattr` if - working on a class, or using :py:meth:`object.__setattr__` if working - on an instance (in order to avoid triggering a subclass-defined version - of :py:meth:`~object.__setattr__`, which might lose or override our - spy). + This will set the method (or delete the attribute for one if setting + ``None``). + + If setting on a class, this will use a standard + :py:func:`setattr`/:py:func:`delattr`. + + If setting on an instance, this will use a standard + :py:meth:`object.__setattr__`/:py:meth:`object.__delattr__` (in order + to avoid triggering a subclass-defined version of + :py:meth:`~object.__setattr__`/:py:meth:`~object.__delattr__`, which + might lose or override our spy). Args: owner (type or object): @@ -896,17 +951,28 @@ class FunctionSpy(object): The name of the attribute to set for the method. method (types.MethodType): - The method to set. + The method to set (or ``None`` to delete). """ if inspect.isclass(owner): - setattr(owner, name, method) - else: + if method is None: + delattr(owner, name) + else: + setattr(owner, name, method) + elif method is None: try: - object.__setattr__(owner, name, method) + object.__delattr__(owner, name) except TypeError as e: - if str(e) == "can't apply this __setattr__ to instance object": + if str(e) == "can't apply this __delattr__ to instance object": # This is likely Python 2.6, or early 2.7, where we can't - # run object.__setattr__ on old-style classes. We have to + # run object.__delattr__ on old-style classes. We have to # fall back to modifying __dict__. It's not ideal but # doable. + del owner.__dict__[name] + else: + try: + object.__setattr__(owner, name, method) + except TypeError as e: + if str(e) == "can't apply this __setattr__ to instance object": + # Similarly as above, we have to default to dict + # manipulation on this version of Python. owner.__dict__[name] = method diff --git a/kgb/tests/test_function_spy.py b/kgb/tests/test_function_spy.py index e9a8b3a63aba637f61afe2d51d390f2dd1bb733a..a15aa0aa7c9b1718656f5f25f30d91f377d67ad7 100644 --- a/kgb/tests/test_function_spy.py +++ b/kgb/tests/test_function_spy.py @@ -1,5 +1,6 @@ from __future__ import unicode_literals +import functools import inspect import re import types @@ -70,6 +71,33 @@ class AdderSubclass(AdderObject): pass +class slippery_func(object): + count = 0 + + def __init__(self): + pass + + def __call__(self, method): + self.method = method + return self + + def __get__(self, obj, objtype=None): + @functools.wraps(self.method) + def _wrapper(*args, **kwargs): + return self.method(obj, _wrapper.counter) + + _wrapper.counter = self.count + type(self).count += 1 + + return _wrapper + + +class SlipperyFuncObject(object): + @slippery_func() + def my_func(self, value): + return value + + class FunctionSpyTests(TestCase): """Test cases for kgb.spies.FunctionSpy.""" @@ -221,6 +249,59 @@ class FunctionSpyTests(TestCase): 'This function has no owner, but an owner was ' 'passed to spy_on().') + def test_init_with_slippery_bound_method(self): + """Testing FunctionSpy construction with new slippery function + generated dynamically from accessing a spied-on bound method from + instance + """ + obj = SlipperyFuncObject() + + # Verify that we're producing a different function on each access. + func1 = obj.my_func + func2 = obj.my_func + self.assertIsInstance(func1, types.FunctionType) + self.assertIsInstance(func2, types.FunctionType) + self.assertIsNot(func1, func2) + + spy = self.agency.spy_on(obj.my_func, + owner=obj, + call_original=True) + + self.assertTrue(hasattr(obj.my_func, 'spy')) + self.assertIs(obj.my_func.spy, spy) + + self.assertEqual(spy.func_name, 'my_func') + self.assertEqual(spy.func_type, spy.TYPE_BOUND_METHOD) + self.assertIsInstance(obj.my_func, types.MethodType) + + slippery_func.count = 0 + func1 = obj.my_func + func2 = obj.my_func + self.assertIs(func1, func2) + + # Since the functions are wrapped, we'll be getting a new value each + # time we invoke them (and not when accessing the attributes as we + # did above). + self.assertEqual(func1(), 0) + self.assertEqual(func1(), 1) + self.assertEqual(func2(), 2) + self.assertEqual(len(obj.my_func.calls), 3) + + def test_init_with_slippery_unbound_method(self): + """Testing FunctionSpy construction with new slippery function + generated dynamically from accessing a spied-on unbound method from + instance + """ + with self.assertRaises(ValueError) as cm: + self.agency.spy_on(SlipperyFuncObject.my_func, + owner=SlipperyFuncObject) + + self.assertEqual( + text_type(cm.exception), + 'Unable to spies on unbound slippery methods (those that return ' + 'a new function on each attribute access). Please spy on an ' + 'instance instead.') + def test_construction_with_classmethod_on_parent(self): """Testing FunctionSpy construction with classmethod from parent of class @@ -1269,6 +1350,36 @@ class FunctionSpyTests(TestCase): self.assertEqual(MyObject.foo.__dict__, obj_func_dict) self.assertEqual(MyParent.foo.__dict__, parent_func_dict) + def test_unspy_with_slippery_bound_method(self): + """Testing FunctionSpy.unspy with slippery function generated by + spied-on bound method + """ + obj = SlipperyFuncObject() + spy = self.agency.spy_on(obj.my_func, owner=obj) + spy.unspy() + + self.assertFalse(hasattr(obj.my_func, 'spy')) + self.assertNotIn('my_func', obj.__dict__) + + # Make sure the old behavior has reverted. + slippery_func.count = 0 + + func1 = obj.my_func + func2 = obj.my_func + self.assertIsNot(func1, func2) + + # These will have already had their values set, so we should get + # stable results again. + self.assertEqual(func1(), 0) + self.assertEqual(func1(), 0) + self.assertEqual(func2(), 1) + self.assertEqual(func2(), 1) + + # These will trigger new counter increments when re-generating the + # function in the decorator. + self.assertEqual(obj.my_func(), 2) + self.assertEqual(obj.my_func(), 3) + def test_called_with(self): """Testing FunctionSpy.called_with""" obj = MathClass() diff --git a/kgb/utils.py b/kgb/utils.py new file mode 100644 index 0000000000000000000000000000000000000000..d6b550619a5f86ef825260683fb6f020a2a7b1ea --- /dev/null +++ b/kgb/utils.py @@ -0,0 +1,71 @@ +"""Common utility functions used by KGB.""" + +from __future__ import unicode_literals + +import inspect + + +def get_defined_attr_value(owner, name, ancestors_only=False): + """Return a value as defined in a class, instance, or ancestor. + + This will look for the real definition, and not the definition returned + when accessing the attribute or instantiating a class. This will bypass any + descriptors and return the actual definition from the instance or class + that defines it. + + Args: + owner (type or object): + The owner of the attribute. + + name (unicode): + The name of the attribute. + + ancestors_only (bool, optional): + Whether to only look in ancestors of ``owner``, and not in + ``owner`` itself. + + Returns: + object: + The attribute value. + + Raises: + AttributeError: + The attribute could not be found. + """ + if not ancestors_only: + d = owner.__dict__ + + if name in d: + return d[name] + + if not inspect.isclass(owner): + # NOTE: It's important we use __class__ and not type(). They are not + # synonymous. The latter will not return the class for an + # instance for old-style classes. + return get_defined_attr_value(owner.__class__, name) + + for parent_cls in owner.__bases__: + try: + return get_defined_attr_value(parent_cls, name) + except AttributeError: + pass + + raise AttributeError + + +def is_attr_defined_on_ancestor(cls, name): + """Return whether an attribute is defined on an ancestor of a class. + + Args: + name (unicode): + The name of the attribute. + + Returns: + bool: + ``True`` if an ancestor defined the attribute. ``False`` if it did not. + """ + try: + get_defined_attr_value(cls, name, ancestors_only=True) + return True + except AttributeError: + return False