ganeti_web/util/proxy/call_proxy.py

   1 # Copyright (C) 2010 Oregon State University et al.
   2 #
   3 # This program is free software; you can redistribute it and/or
   4 # modify it under the terms of the GNU General Public License
   5 # as published by the Free Software Foundation; either version 2
   6 # of the License, or (at your option) any later version.
   7 #
   8 # This program is distributed in the hope that it will be useful,
   9 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11 # GNU General Public License for more details.
  12 #
  13 # You should have received a copy of the GNU General Public License
  14 # along with this program; if not, write to the Free Software
  15 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301,
  16 # USA.
  17
  18
  19 import types
  20 from response_map import ResponseMap
  21
  22
  23 class CallProxy(object):
  24     """ Proxy for a method that will record calls to it.  To use this class
  25     monkey patch the original method using an instance of this class
  26
  27     setting the enabled flag can enable/disable whether the method is actually
  28     executed when it is called, or just recorded.
  29     """
  30     def __init__(self, func, enabled=True, response=None, **kwargs):
  31         """
  32         :parameters:
  33             func: function to proxy
  34             enabled: whether to call the wrapped function
  35             kwargs: kwargs passed to all calls.  may be overwritten by kwargs
  36                     passed to function
  37         """
  38         self.func = func
  39         self.calls = []
  40         self.enabled = enabled
  41         self.kwargs = kwargs
  42         self.response = response
  43         self.error = False
  44
  45         if func:
  46             self.matching_function = self.create_matching_function(func)
  47
  48     def assertCalled(self, testcase, *args, **kwargs):
  49         """
  50         Assertion function for checking if a callproxy was called
  51         """
  52         f = self.func
  53         calls = self.calls
  54         if args or kwargs:
  55             #detailed match
  56             for t in calls:
  57                 args_, kwargs_ = t
  58                 if args_ == args and kwargs_ == kwargs:
  59                     return t
  60             testcase.fail("exact call (%s) did not occur: %s" % (f, calls))
  61
  62         else:
  63             # simple match
  64             testcase.assert_(calls, "%s was not called: %s" % (f, calls))
  65             return calls[0]
  66
  67     def assertNotCalled(self, testcase, *args, **kwargs):
  68         """
  69         Assertion function for checking if callproxy was not called
  70         """
  71         f = self.func
  72         calls = self.calls
  73         if args or kwargs:
  74             #detailed match
  75             for t in calls:
  76                 args_, kwargs_ = t
  77                 if args_ == args and kwargs_ == kwargs:
  78                     testcase.fail("exact call (%s) was made: %s" % (f, calls))
  79         else:
  80             # simple match
  81             testcase.assertFalse(calls, '%s was called' % f)
  82
  83     def enable(self):
  84         self.enabled = True
  85
  86     def disable(self):
  87         self.enabled = False
  88
  89     def reset(self):
  90         self.calls = []
  91
  92     def __call__(self, *args, **kwargs):
  93         if self.error:
  94             raise self.error
  95
  96         response = None
  97         kwargs_ = {}
  98         kwargs_.update(self.kwargs)
  99         kwargs_.update(kwargs)
 100         self.calls.append((args, kwargs_))
 101
 102         if self.enabled:
 103             response = self.func(*args, **kwargs_)
 104         elif self.func:
 105             # call matching call, this ensures the args are checked even when
 106             # the real function isn't actually called
 107             #
 108             # pass in the instance if it is set.  if this was a bound method
 109             # then it will fail without self passed.
 110             if self.func.im_self:
 111                 self.matching_function(self.func.im_self, *args, **kwargs)
 112             else:
 113                 self.matching_function(*args, **kwargs)
 114
 115         # return mandated response.  This may be a ResponseMap, so process
 116         # according to what type it is.
 117         if self.response:
 118             if isinstance(self.response, (ResponseMap,)):
 119                 return self.response[(args, kwargs)]
 120             return self.response
 121
 122         return response
 123
 124     def create_matching_function(self, func):
 125         """
 126         constructs a function with a method signature that matches the
 127         function that is passed in.  The resulting function does not actually
 128         do anything.  It is only used for verifying arguments
 129         to the call match.
 130
 131         The function is constructed from a combination of properties from an
 132         inner function and the function passed in.
 133         """
 134         def base():
 135             pass
 136
 137         base_code = base.func_code
 138         code = func.func_code
 139
 140         name = 'MATCHING_PROXY: %s' % func.__name__
 141
 142         new_code = types.CodeType(
 143             code.co_argcount,
 144             code.co_nlocals,
 145             base_code.co_stacksize,
 146             code.co_flags,
 147             base_code.co_code,
 148             base_code.co_consts,
 149             base_code.co_names,
 150             code.co_varnames,
 151             base_code.co_filename,
 152             name,
 153             base_code.co_firstlineno,
 154             base_code.co_lnotab)
 155
 156         return types.FunctionType(new_code, func.func_globals,
 157                                   name, func.func_defaults)
 158
 159     @classmethod
 160     def patch(cls, instance, name, *args, **kwargs):
 161         """
 162         Helper function for patching a function on an instance.  useful since
 163         patching an instance requires a bit more work to ensure the proxy works
 164         correctly.
 165
 166         :parameters:
 167             * instance: instance to patch
 168             * name: name of function to
 169         """
 170         func = getattr(instance, name)
 171         proxy = CallProxy(func, *args, **kwargs)
 172         setattr(instance, name, proxy)