Revision f90a1ab5

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)


  		












  

    
      b/test/ganeti.rapi.client_unittest.py
    
  






  22
  22
  
    
"""Script for unittesting the RAPI client module"""


  		





  23
  23
  
    

  		





  24
  24
  
    

  		





  25
  
  
    
import re


  		





  26
  25
  
    
import unittest


  		





  27
  26
  
    
import warnings


  		





  28
  27
  
    
import pycurl


  		





  ......




  33
  32
  
    
from ganeti import utils


  		





  34
  33
  
    
from ganeti import query


  		





  35
  34
  
    
from ganeti import objects


  		





  
  35
  
    
from ganeti import rapi


  		





  36
  36
  
    

  		





  
  37
  
    
import ganeti.rapi.testutils


  		





  37
  38
  
    
from ganeti.rapi import connector


  		





  38
  39
  
    
from ganeti.rapi import rlib2


  		





  39
  40
  
    
from ganeti.rapi import client


  		





  ......




  41
  42
  
    
import testutils


  		





  42
  43
  
    

  		





  43
  44
  
    

  		





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


  		





  45
  
  
    

  		





  46
  45
  
    
# List of resource handlers which aren't used by the RAPI client


  		





  47
  46
  
    
_KNOWN_UNUSED = set([


  		





  48
  47
  
    
  rlib2.R_root,


  		





  ......




  53
  52
  
    
_used_handlers = None


  		





  54
  53
  
    

  		





  55
  54
  
    

  		





  56
  
  
    
def _GetPathFromUri(uri):


  		





  57
  
  
    
  """Gets the path and query from a URI.


  		





  58
  
  
    

  		





  59
  
  
    
  """


  		





  60
  
  
    
  match = _URI_RE.match(uri)


  		





  61
  
  
    
  if match:


  		





  62
  
  
    
    return match.groupdict()["path"]


  		





  63
  
  
    
  else:


  		





  64
  
  
    
    return None


  		





  65
  
  
    

  		





  66
  
  
    

  		





  67
  
  
    
class FakeCurl:


  		





  68
  
  
    
  def __init__(self, rapi):


  		





  69
  
  
    
    self._rapi = rapi


  		





  70
  
  
    
    self._opts = {}


  		





  71
  
  
    
    self._info = {}


  		





  72
  
  
    

  		





  73
  
  
    
  def setopt(self, opt, value):


  		





  74
  
  
    
    self._opts[opt] = value


  		





  75
  
  
    

  		





  76
  
  
    
  def getopt(self, opt):


  		





  77
  
  
    
    return self._opts.get(opt)


  		





  78
  
  
    

  		





  79
  
  
    
  def unsetopt(self, opt):


  		





  80
  
  
    
    self._opts.pop(opt, None)


  		





  81
  
  
    

  		





  82
  
  
    
  def getinfo(self, info):


  		





  83
  
  
    
    return self._info[info]


  		





  84
  
  
    

  		





  85
  
  
    
  def perform(self):


  		





  86
  
  
    
    method = self._opts[pycurl.CUSTOMREQUEST]


  		





  87
  
  
    
    url = self._opts[pycurl.URL]


  		





  88
  
  
    
    request_body = self._opts[pycurl.POSTFIELDS]


  		





  89
  
  
    
    writefn = self._opts[pycurl.WRITEFUNCTION]


  		





  90
  
  
    

  		





  91
  
  
    
    path = _GetPathFromUri(url)


  		





  92
  
  
    
    (code, resp_body) = self._rapi.FetchResponse(path, method, request_body)


  		





  93
  
  
    

  		





  94
  
  
    
    self._info[pycurl.RESPONSE_CODE] = code


  		





  95
  
  
    
    if resp_body is not None:


  		





  96
  
  
    
      writefn(resp_body)


  		





  97
  
  
    

  		





  98
  
  
    

  		





  99
  55
  
    
class RapiMock(object):


  		





  100
  56
  
    
  def __init__(self):


  		





  101
  57
  
    
    self._mapper = connector.Mapper()


  		





  ......




  216
  172
  
    
class TestExtendedConfig(unittest.TestCase):


  		





  217
  173
  
    
  def testAuth(self):


  		





  218
  174
  
    
    cl = client.GanetiRapiClient("master.example.com",


  		





  219
  
  
    
                                 username="user", password="pw",


  		





  220
  
  
    
                                 curl_factory=lambda: FakeCurl(RapiMock()))


  		





  
  175
  
    
      username="user", password="pw",


  		





  
  176
  
    
      curl_factory=lambda: rapi.testutils.FakeCurl(RapiMock()))


  		





  221
  177
  
    

  		





  222
  178
  
    
    curl = cl._CreateCurl()


  		





  223
  179
  
    
    self.assertEqual(curl.getopt(pycurl.HTTPAUTH), pycurl.HTTPAUTH_BASIC)


  		





  ......




  254
  210
  
    
                                             verify_hostname=verify_hostname,


  		





  255
  211
  
    
                                             _pycurl_version_fn=pcverfn)


  		





  256
  212
  
    

  		





  257
  
  
    
            curl_factory = lambda: FakeCurl(RapiMock())


  		





  
  213
  
    
            curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())


  		





  258
  214
  
    
            cl = client.GanetiRapiClient("master.example.com",


  		





  259
  215
  
    
                                         curl_config_fn=cfgfn,


  		





  260
  216
  
    
                                         curl_factory=curl_factory)


  		





  ......




  271
  227
  
    
  def testNoCertVerify(self):


  		





  272
  228
  
    
    cfgfn = client.GenericCurlConfig()


  		





  273
  229
  
    

  		





  274
  
  
    
    curl_factory = lambda: FakeCurl(RapiMock())


  		





  
  230
  
    
    curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())


  		





  275
  231
  
    
    cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn,


  		





  276
  232
  
    
                                 curl_factory=curl_factory)


  		





  277
  233
  
    

  		





  ......




  283
  239
  
    
  def testCertVerifyCurlBundle(self):


  		





  284
  240
  
    
    cfgfn = client.GenericCurlConfig(use_curl_cabundle=True)


  		





  285
  241
  
    

  		





  286
  
  
    
    curl_factory = lambda: FakeCurl(RapiMock())


  		





  
  242
  
    
    curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())


  		





  287
  243
  
    
    cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn,


  		





  288
  244
  
    
                                 curl_factory=curl_factory)


  		





  289
  245
  
    

  		





  ......




  296
  252
  
    
    mycert = "/tmp/some/UNUSED/cert/file.pem"


  		





  297
  253
  
    
    cfgfn = client.GenericCurlConfig(cafile=mycert)


  		





  298
  254
  
    

  		





  299
  
  
    
    curl_factory = lambda: FakeCurl(RapiMock())


  		





  
  255
  
    
    curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())


  		





  300
  256
  
    
    cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn,


  		





  301
  257
  
    
                                 curl_factory=curl_factory)


  		





  302
  258
  
    

  		





  ......




  311
  267
  
    
    cfgfn = client.GenericCurlConfig(capath=certdir,


  		





  312
  268
  
    
                                     _pycurl_version_fn=pcverfn)


  		





  313
  269
  
    

  		





  314
  
  
    
    curl_factory = lambda: FakeCurl(RapiMock())


  		





  
  270
  
    
    curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())


  		





  315
  271
  
    
    cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn,


  		





  316
  272
  
    
                                 curl_factory=curl_factory)


  		





  317
  273
  
    

  		





  ......




  326
  282
  
    
    cfgfn = client.GenericCurlConfig(capath=certdir,


  		





  327
  283
  
    
                                     _pycurl_version_fn=pcverfn)


  		





  328
  284
  
    

  		





  329
  
  
    
    curl_factory = lambda: FakeCurl(RapiMock())


  		





  
  285
  
    
    curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())


  		





  330
  286
  
    
    cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn,


  		





  331
  287
  
    
                                 curl_factory=curl_factory)


  		





  332
  288
  
    

  		





  ......




  338
  294
  
    
    cfgfn = client.GenericCurlConfig(capath=certdir,


  		





  339
  295
  
    
                                     _pycurl_version_fn=pcverfn)


  		





  340
  296
  
    

  		





  341
  
  
    
    curl_factory = lambda: FakeCurl(RapiMock())


  		





  
  297
  
    
    curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())


  		





  342
  298
  
    
    cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn,


  		





  343
  299
  
    
                                 curl_factory=curl_factory)


  		





  344
  300
  
    

  		





  ......




  350
  306
  
    
    cfgfn = client.GenericCurlConfig(capath=certdir,


  		





  351
  307
  
    
                                     _pycurl_version_fn=pcverfn)


  		





  352
  308
  
    

  		





  353
  
  
    
    curl_factory = lambda: FakeCurl(RapiMock())


  		





  
  309
  
    
    curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())


  		





  354
  310
  
    
    cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn,


  		





  355
  311
  
    
                                 curl_factory=curl_factory)


  		





  356
  312
  
    

  		





  ......




  362
  318
  
    
        cfgfn = client.GenericCurlConfig(connect_timeout=connect_timeout,


  		





  363
  319
  
    
                                         timeout=timeout)


  		





  364
  320
  
    

  		





  365
  
  
    
        curl_factory = lambda: FakeCurl(RapiMock())


  		





  
  321
  
    
        curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())


  		





  366
  322
  
    
        cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn,


  		





  367
  323
  
    
                                     curl_factory=curl_factory)


  		





  368
  324
  
    

  		





  ......




  376
  332
  
    
    testutils.GanetiTestCase.setUp(self)


  		





  377
  333
  
    

  		





  378
  334
  
    
    self.rapi = RapiMock()


  		





  379
  
  
    
    self.curl = FakeCurl(self.rapi)


  		





  
  335
  
    
    self.curl = rapi.testutils.FakeCurl(self.rapi)


  		





  380
  336
  
    
    self.client = client.GanetiRapiClient("master.example.com",


  		





  381
  337
  
    
                                          curl_factory=lambda: self.curl)


  		





  382
  338
  
    

  		












Also available in:
  Unified diff