ganeti_web/tests/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
  21 class ResponseMap(object):
  22     """
  23     An object that encapsulates return values based on parameters given to the
  24     called method.
  25
  26     Return Map should be initialized with a list containing tuples all possible
  27     arg/kwarg combinations plus the result that should be sent for those args
  28     """
  29     def __init__(self, map):
  30         self.map = map
  31
  32     def __getitem__(self, key):
  33         for k, response in self.map:
  34             if key == k:
  35                 return response
  36
  37
  38 class CallProxy(object):
  39     """ Proxy for a method that will record calls to it.  To use this class
  40     monkey patch the original method using an instance of this class
  41
  42     setting the enabled flag can enable/disable whether the method is actually
  43     executed when it is called, or just recorded.
  44     """
  45     def __init__(self, func, enabled=True, response=None, **kwargs):
  46         """
  47         :parameters:
  48             func: function to proxy
  49             enabled: whether to call the wrapped function
  50             kwargs: kwargs passed to all calls.  may be overwritten by kwargs
  51                     passed to function
  52         """
  53         self.func = func
  54         self.calls = []
  55         self.enabled = enabled
  56         self.kwargs = kwargs
  57         self.response = response
  58         self.error = False
  59
  60         if func:
  61             self.matching_function = self.create_matching_function(func)
  62
  63     def assertCalled(self, testcase, *args, **kwargs):
  64         """
  65         Assertion function for checking if a callproxy was called
  66         """
  67         f = self.func
  68         calls = self.calls
  69         if args or kwargs:
  70             #detailed match
  71             for t in calls:
  72                 args_, kwargs_ = t
  73                 if args_==args and kwargs_==kwargs:
  74                     return t
  75             testcase.fail("exact call (%s) did not occur: %s" % (f, calls))
  76
  77         else:
  78             # simple match
  79             testcase.assert_(calls, "%s was not called: %s" % (f, calls))
  80             return calls[0]
  81
  82     def assertNotCalled(self, testcase, *args, **kwargs):
  83         """
  84         Assertion function for checking if callproxy was not called
  85         """
  86         f = self.func
  87         calls = self.calls
  88         if args or kwargs:
  89             #detailed match
  90             for t in calls:
  91                 args_, kwargs_ = t
  92                 if args_==args and kwargs_==kwargs:
  93                     testcase.fail("exact call (%s) was made: %s" % (f, calls))
  94         else:
  95             # simple match
  96             testcase.assertFalse(calls, '%s was called' % f)
  97
  98     def enable(self):
  99         self.enabled = True
 100
 101     def disable(self):
 102         self.enabled = False
 103
 104     def reset(self):
 105         self.calls = []
 106
 107     def __call__ (self, *args, **kwargs):
 108         if self.error:
 109             raise self.error
 110
 111         response = None
 112         kwargs_ = {}
 113         kwargs_.update(self.kwargs)
 114         kwargs_.update(kwargs)
 115         self.calls.append((args, kwargs_))
 116
 117         if self.enabled:
 118             response = self.func(*args, **kwargs_)
 119         elif self.func:
 120             # call matching call, this ensures the args are checked even when
 121             # the real function isn't actually called
 122             #
 123             # pass in the instance if it is set.  if this was a bound method
 124             # then it will fail without self passed.
 125             if self.func.im_self:
 126                 self.matching_function(self.func.im_self, *args, **kwargs)
 127             else:
 128                 self.matching_function(*args, **kwargs)
 129
 130         # return mandated response.  This may be a ResponseMap, so process
 131         # according to what type it is.
 132         if self.response:
 133             if isinstance(self.response, (ResponseMap,)):
 134                 return self.response[(args, kwargs)]
 135             return self.response
 136
 137         return response
 138
 139
 140     def create_matching_function(self, func):
 141         """
 142         constructs a function with a method signature that matches the
 143         function that is passed in.  The resulting function does not actually
 144         do anything.  It is only used for verifying arguments to the call match.
 145
 146         The function is constructed from a combination of properties from an
 147         inner function and the function passed in.
 148         """
 149         def base(): pass
 150
 151         base_code = base.func_code
 152         code = func.func_code
 153
 154         name = 'MATCHING_PROXY: %s' % func.__name__
 155
 156         new_code = types.CodeType( \
 157             code.co_argcount, \
 158             code.co_nlocals, \
 159             base_code.co_stacksize, \
 160             code.co_flags, \
 161             base_code.co_code, \
 162             base_code.co_consts, \
 163             base_code.co_names, \
 164             code.co_varnames, \
 165             base_code.co_filename, \
 166             name, \
 167             base_code.co_firstlineno, \
 168             base_code.co_lnotab)
 169
 170         return types.FunctionType(new_code, func.func_globals, \
 171                                   name, func.func_defaults)
 172
 173     @classmethod
 174     def patch(cls, instance, name, *args, **kwargs):
 175         """
 176         Helper function for patching a function on an instance.  useful since
 177         patching an instance requires a bit more work to ensure the proxy works
 178         correctly.
 179
 180         :parameters:
 181             * instance: instance to patch
 182             * name: name of function to
 183         """
 184         func = getattr(instance, name)
 185         proxy = CallProxy(func, *args, **kwargs)
 186         setattr(instance, name, proxy)