Revision d9492490

b/lib/rapi/testutils.py
25 25 

  		





  26
  26
  
    
import logging


  		





  27
  27
  
    
import re


  		





  
  28
  
    
import mimetools


  		





  
  29
  
    
import base64


  		





  28
  30
  
    
import pycurl


  		





  
  31
  
    
from cStringIO import StringIO


  		





  29
  32
  
    

  		





  30
  33
  
    
from ganeti import errors


  		





  31
  34
  
    
from ganeti import opcodes


  		





  
  35
  
    
from ganeti import http


  		





  32
  36
  
    

  		





  33
  37
  
    

  		





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


  		





  ......




  154
  158
  
    
    request_body = self._opts[pycurl.POSTFIELDS]


  		





  155
  159
  
    
    writefn = self._opts[pycurl.WRITEFUNCTION]


  		





  156
  160
  
    

  		





  
  161
  
    
    if pycurl.HTTPHEADER in self._opts:


  		





  
  162
  
    
      baseheaders = "\n".join(self._opts[pycurl.HTTPHEADER])


  		





  
  163
  
    
    else:


  		





  
  164
  
    
      baseheaders = ""


  		





  
  165
  
    

  		





  
  166
  
    
    headers = mimetools.Message(StringIO(baseheaders), 0)


  		





  
  167
  
    

  		





  
  168
  
    
    if request_body:


  		





  
  169
  
    
      headers[http.HTTP_CONTENT_LENGTH] = str(len(request_body))


  		





  
  170
  
    

  		





  
  171
  
    
    if self._opts.get(pycurl.HTTPAUTH, 0) & pycurl.HTTPAUTH_BASIC:


  		





  
  172
  
    
      try:


  		





  
  173
  
    
        userpwd = self._opts[pycurl.USERPWD]


  		





  
  174
  
    
      except KeyError:


  		





  
  175
  
    
        raise errors.ProgrammerError("Basic authentication requires username"


  		





  
  176
  
    
                                     " and password")


  		





  
  177
  
    

  		





  
  178
  
    
      headers[http.HTTP_AUTHORIZATION] = \


  		





  
  179
  
    
        "%s %s" % (http.auth.HTTP_BASIC_AUTH, base64.b64encode(userpwd))


  		





  
  180
  
    

  		





  157
  181
  
    
    path = _GetPathFromUri(url)


  		





  158
  
  
    
    (code, resp_body) = self._handler.FetchResponse(path, method, request_body)


  		





  
  182
  
    
    (code, resp_body) = \


  		





  
  183
  
    
      self._handler.FetchResponse(path, method, headers, request_body)


  		





  159
  184
  
    

  		





  160
  185
  
    
    self._info[pycurl.RESPONSE_CODE] = code


  		





  161
  186
  
    
    if resp_body is not None:


  		












  

    
      b/test/ganeti.rapi.client_unittest.py
    
  






  74
  74
  
    
  def GetLastRequestData(self):


  		





  75
  75
  
    
    return self._last_req_data


  		





  76
  76
  
    

  		





  77
  
  
    
  def FetchResponse(self, path, method, request_body):


  		





  
  77
  
    
  def FetchResponse(self, path, method, headers, request_body):


  		





  78
  78
  
    
    self._last_req_data = request_body


  		





  79
  79
  
    

  		





  80
  80
  
    
    try:


  		





  ......




  139
  139
  
    
  def test(self):


  		





  140
  140
  
    
    rapi = RapiMock()


  		





  141
  141
  
    
    path = "/version"


  		





  142
  
  
    
    self.assertEqual((404, None), rapi.FetchResponse("/foo", "GET", None))


  		





  
  142
  
    
    self.assertEqual((404, None), rapi.FetchResponse("/foo", "GET", None, None))


  		





  143
  143
  
    
    self.assertEqual((501, "Method not implemented"),


  		





  144
  
  
    
                     rapi.FetchResponse("/version", "POST", None))


  		





  
  144
  
    
                     rapi.FetchResponse("/version", "POST", None, None))


  		





  145
  145
  
    
    rapi.AddResponse("2")


  		





  146
  
  
    
    code, response = rapi.FetchResponse("/version", "GET", None)


  		





  
  146
  
    
    code, response = rapi.FetchResponse("/version", "GET", None, None)


  		





  147
  147
  
    
    self.assertEqual(200, code)


  		





  148
  148
  
    
    self.assertEqual("2", response)


  		





  149
  149
  
    
    self.failUnless(isinstance(rapi.GetLastHandler(), rlib2.R_version))


  		












Also available in:
  Unified diff