Statistics
| Branch: | Tag: | Revision:

root / lib / rapi / testutils.py @ f90a1ab5

History | View | Annotate | Download (4 kB)

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)