4 # Copyright (C) 2012 Google Inc.
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.
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.
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
22 """Remote API test utilities.
30 from ganeti import errors
31 from ganeti import opcodes
34 _URI_RE = re.compile(r"https://(?P<host>.*):(?P<port>\d+)(?P<path>/.*)")
37 class VerificationError(Exception):
38 """Dedicated error class for test utilities.
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
47 def _GetOpById(op_id):
48 """Tries to get an opcode class based on its C{OP_ID}.
52 return opcodes.OP_MAPPING[op_id]
54 raise VerificationError("Unknown opcode ID '%s'" % op_id)
57 def _HideInternalErrors(fn):
58 """Hides Ganeti-internal exceptions, see L{VerificationError}.
61 def wrapper(*args, **kwargs):
63 return fn(*args, **kwargs)
64 except errors.GenericError, err:
65 raise VerificationError("Unhandled Ganeti error: %s" % err)
71 def VerifyOpInput(op_id, data):
72 """Verifies opcode parameters according to their definition.
75 @param op_id: Opcode ID (C{OP_ID} attribute), e.g. C{OP_CLUSTER_VERIFY}
77 @param data: Opcode parameter values
78 @raise VerificationError: Parameter verification failed
81 op_cls = _GetOpById(op_id)
84 op = op_cls(**data) # pylint: disable=W0142
85 except TypeError, err:
86 raise VerificationError("Unable to create opcode instance: %s" % err)
90 except errors.OpPrereqError, err:
91 raise VerificationError("Parameter validation for opcode '%s' failed: %s" %
96 def VerifyOpResult(op_id, result):
97 """Verifies opcode results used in tests (e.g. in a mock).
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
105 resultcheck_fn = _GetOpById(op_id).OP_RESULT
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))
114 def _GetPathFromUri(uri):
115 """Gets the path and query from a URI.
118 match = _URI_RE.match(uri)
120 return match.groupdict()["path"]
129 def __init__(self, handler):
130 """Initialize this class
132 @param handler: Request handler instance
135 self._handler = handler
139 def setopt(self, opt, value):
140 self._opts[opt] = value
142 def getopt(self, opt):
143 return self._opts.get(opt)
145 def unsetopt(self, opt):
146 self._opts.pop(opt, None)
148 def getinfo(self, info):
149 return self._info[info]
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]
157 path = _GetPathFromUri(url)
158 (code, resp_body) = self._handler.FetchResponse(path, method, request_body)
160 self._info[pycurl.RESPONSE_CODE] = code
161 if resp_body is not None: