Synnefo » snf-ganeti » ganeti-local

from qa_utils import (AssertEqual, AssertNotEqual, AssertIn, AssertMatch,

                      StartSSH)

class OpenerFactory:

  """A factory singleton to construct urllib opener chain.

  This is needed because qa_config is not initialized yet at module load time

  """

  _opener = None

  @classmethod

  def Opener(cls):

    """Construct the opener if not yet done.

    """

    if not cls._opener:

      # Create opener which doesn't try to look for proxies and does auth

      master = qa_config.GetMasterNode()

      host = master["primary"]

      port = qa_config.get("rapi-port", default=constants.DEFAULT_RAPI_PORT)

      passman = urllib2.HTTPPasswordMgrWithDefaultRealm()

      passman.add_password(None, 'https://%s:%s' % (host, port),

                           qa_config.get("rapi-user", default=""),

                           qa_config.get("rapi-pass", default=""))

      authhandler = urllib2.HTTPBasicAuthHandler(passman)

      cls._opener = urllib2.build_opener(urllib2.ProxyHandler({}), authhandler)

    return cls._opener

class RapiRequest(urllib2.Request):

  """This class supports other methods beside GET/POST.

  """

  def __init__(self, url, data=None, headers={}, origin_req_host=None,

               unverifiable=False, method="GET"):

    urllib2.Request.__init__(self, url, data, headers, origin_req_host,

                             unverifiable)

    self._method = method

  def get_method(self):

    return self._method

  for uri, verify, method in uris:

    req = RapiRequest(url, method=method)

    response = OpenerFactory.Opener().open(req)

    ("/version", constants.RAPI_VERSION, 'GET'),

    ("/", None, 'GET'),

    ("/2/info", _VerifyInfo, 'GET'),

    ("/2/tags", None, 'GET'),

    ("/2/nodes", _VerifyNodes, 'GET'),

    ("/2/nodes?bulk=1", _VerifyNodesBulk, 'GET'),

    ("/2/instances", [], 'GET'),

    ("/2/instances?bulk=1", [], 'GET'),

    ("/2/os", None, 'GET'),

  def _VerifyReturnsJob(data):

    AssertMatch(data, r'^\d+$')

    ("/2/instances/%s" % instance["name"], _VerifyInstance, 'GET'),

    ("/2/instances", _VerifyInstancesList, 'GET'),

    ("/2/instances?bulk=1", _VerifyInstancesBulk, 'GET'),

    ("/2/instances/%s/activate-disks" % instance["name"], _VerifyReturnsJob, 'PUT'),

    ("/2/instances/%s/deactivate-disks" % instance["name"], _VerifyReturnsJob, 'PUT'),

    ("/2/nodes/%s" % node["primary"], _VerifyNode, 'GET'),

    ("/2/nodes", _VerifyNodesList, 'GET'),

    ("/2/nodes?bulk=1", _VerifyNodesBulk, 'GET'),

    (uri, _VerifyTags, 'GET'),

import re

def AssertMatch(string, pattern):

  """Raises an error when string doesn't match regexp pattern.

  """

  if not re.match(pattern, string):

    raise qa_error.Error("%r doesn't match /%r/" % (string, pattern))