code.grnet.gr Git - ganeti-local/blob - lib/rapi/testutils.py

   1 #
   2 #
   3
   4 # Copyright (C) 2012 Google Inc.
   5 #
   6 # This program is free software; you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation; either version 2 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful, but
  12 # WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14 # General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program; if not, write to the Free Software
  18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  19 # 02110-1301, USA.
  20
  21
  22 """Remote API test utilities.
  23
  24 """
  25
  26 import logging
  27 import re
  28 import pycurl
  29
  30 from ganeti import errors
  31 from ganeti import opcodes
  32
  33
  34 _URI_RE = re.compile(r"https://(?P<host>.*):(?P<port>\d+)(?P<path>/.*)")
  35
  36
  37 class VerificationError(Exception):
  38   """Dedicated error class for test utilities.
  39
  40   This class is used to hide all of Ganeti's internal exception, so that
  41   external users of these utilities don't have to integrate Ganeti's exception
  42   hierarchy.
  43
  44   """
  45
  46
  47 def _GetOpById(op_id):
  48   """Tries to get an opcode class based on its C{OP_ID}.
  49
  50   """
  51   try:
  52     return opcodes.OP_MAPPING[op_id]
  53   except KeyError:
  54     raise VerificationError("Unknown opcode ID '%s'" % op_id)
  55
  56
  57 def _HideInternalErrors(fn):
  58   """Hides Ganeti-internal exceptions, see L{VerificationError}.
  59
  60   """
  61   def wrapper(*args, **kwargs):
  62     try:
  63       return fn(*args, **kwargs)
  64     except errors.GenericError, err:
  65       raise VerificationError("Unhandled Ganeti error: %s" % err)
  66
  67   return wrapper
  68
  69
  70 @_HideInternalErrors
  71 def VerifyOpInput(op_id, data):
  72   """Verifies opcode parameters according to their definition.
  73
  74   @type op_id: string
  75   @param op_id: Opcode ID (C{OP_ID} attribute), e.g. C{OP_CLUSTER_VERIFY}
  76   @type data: dict
  77   @param data: Opcode parameter values
  78   @raise VerificationError: Parameter verification failed
  79
  80   """
  81   op_cls = _GetOpById(op_id)
  82
  83   try:
  84     op = op_cls(**data) # pylint: disable=W0142
  85   except TypeError, err:
  86     raise VerificationError("Unable to create opcode instance: %s" % err)
  87
  88   try:
  89     op.Validate(False)
  90   except errors.OpPrereqError, err:
  91     raise VerificationError("Parameter validation for opcode '%s' failed: %s" %
  92                             (op_id, err))
  93
  94
  95 @_HideInternalErrors
  96 def VerifyOpResult(op_id, result):
  97   """Verifies opcode results used in tests (e.g. in a mock).
  98
  99   @type op_id: string
 100   @param op_id: Opcode ID (C{OP_ID} attribute), e.g. C{OP_CLUSTER_VERIFY}
 101   @param result: Mocked opcode result
 102   @raise VerificationError: Return value verification failed
 103
 104   """
 105   resultcheck_fn = _GetOpById(op_id).OP_RESULT
 106
 107   if not resultcheck_fn:
 108     logging.warning("Opcode '%s' has no result type definition", op_id)
 109   elif not resultcheck_fn(result):
 110     raise VerificationError("Given result does not match result description"
 111                             " for opcode '%s': %s" % (op_id, resultcheck_fn))
 112
 113
 114 def _GetPathFromUri(uri):
 115   """Gets the path and query from a URI.
 116
 117   """
 118   match = _URI_RE.match(uri)
 119   if match:
 120     return match.groupdict()["path"]
 121   else:
 122     return None
 123
 124
 125 class FakeCurl:
 126   """Fake cURL object.
 127
 128   """
 129   def __init__(self, handler):
 130     """Initialize this class
 131
 132     @param handler: Request handler instance
 133
 134     """
 135     self._handler = handler
 136     self._opts = {}
 137     self._info = {}
 138
 139   def setopt(self, opt, value):
 140     self._opts[opt] = value
 141
 142   def getopt(self, opt):
 143     return self._opts.get(opt)
 144
 145   def unsetopt(self, opt):
 146     self._opts.pop(opt, None)
 147
 148   def getinfo(self, info):
 149     return self._info[info]
 150
 151   def perform(self):
 152     method = self._opts[pycurl.CUSTOMREQUEST]
 153     url = self._opts[pycurl.URL]
 154     request_body = self._opts[pycurl.POSTFIELDS]
 155     writefn = self._opts[pycurl.WRITEFUNCTION]
 156
 157     path = _GetPathFromUri(url)
 158     (code, resp_body) = self._handler.FetchResponse(path, method, request_body)
 159
 160     self._info[pycurl.RESPONSE_CODE] = code
 161     if resp_body is not None:
 162       writefn(resp_body)