Revision f90a1ab5 lib/rapi/testutils.py

b/lib/rapi/testutils.py
24 24
"""
25 25

  
26 26
import logging
27
import re
28
import pycurl
27 29

  
28 30
from ganeti import errors
29 31
from ganeti import opcodes
30 32

  
31 33

  
34
_URI_RE = re.compile(r"https://(?P<host>.*):(?P<port>\d+)(?P<path>/.*)")
35

  
36

  
32 37
class VerificationError(Exception):
33 38
  """Dedicated error class for test utilities.
34 39

  
......
104 109
  elif not resultcheck_fn(result):
105 110
    raise VerificationError("Given result does not match result description"
106 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)

Also available in: Unified diff