"""Script for unittesting the RAPI client module"""
-import re
import unittest
import warnings
import pycurl
+from ganeti import opcodes
from ganeti import constants
from ganeti import http
from ganeti import serializer
from ganeti import utils
from ganeti import query
from ganeti import objects
+from ganeti import rapi
+import ganeti.rapi.testutils
from ganeti.rapi import connector
from ganeti.rapi import rlib2
from ganeti.rapi import client
import testutils
-_URI_RE = re.compile(r"https://(?P<host>.*):(?P<port>\d+)(?P<path>/.*)")
-
# List of resource handlers which aren't used by the RAPI client
_KNOWN_UNUSED = set([
rlib2.R_root,
_used_handlers = None
-def _GetPathFromUri(uri):
- """Gets the path and query from a URI.
-
- """
- match = _URI_RE.match(uri)
- if match:
- return match.groupdict()["path"]
- else:
- return None
-
-
-class FakeCurl:
- def __init__(self, rapi):
- self._rapi = rapi
- self._opts = {}
- self._info = {}
-
- def setopt(self, opt, value):
- self._opts[opt] = value
-
- def getopt(self, opt):
- return self._opts.get(opt)
-
- def unsetopt(self, opt):
- self._opts.pop(opt, None)
-
- def getinfo(self, info):
- return self._info[info]
-
- def perform(self):
- method = self._opts[pycurl.CUSTOMREQUEST]
- url = self._opts[pycurl.URL]
- request_body = self._opts[pycurl.POSTFIELDS]
- writefn = self._opts[pycurl.WRITEFUNCTION]
-
- path = _GetPathFromUri(url)
- (code, resp_body) = self._rapi.FetchResponse(path, method, request_body)
-
- self._info[pycurl.RESPONSE_CODE] = code
- if resp_body is not None:
- writefn(resp_body)
-
-
class RapiMock(object):
def __init__(self):
self._mapper = connector.Mapper()
def GetLastRequestData(self):
return self._last_req_data
- def FetchResponse(self, path, method, request_body):
+ def FetchResponse(self, path, method, headers, request_body):
self._last_req_data = request_body
try:
self.assertEqual(client.GANETI_RAPI_VERSION, constants.RAPI_VERSION)
self.assertEqual(client.HTTP_APP_JSON, http.HTTP_APP_JSON)
self.assertEqual(client._REQ_DATA_VERSION_FIELD, rlib2._REQ_DATA_VERSION)
- self.assertEqual(client._INST_CREATE_REQV1, rlib2._INST_CREATE_REQV1)
- self.assertEqual(client._INST_REINSTALL_REQV1, rlib2._INST_REINSTALL_REQV1)
- self.assertEqual(client._NODE_MIGRATE_REQV1, rlib2._NODE_MIGRATE_REQV1)
- self.assertEqual(client._NODE_EVAC_RES1, rlib2._NODE_EVAC_RES1)
- self.assertEqual(client._INST_NIC_PARAMS, constants.INIC_PARAMS)
self.assertEqual(client.JOB_STATUS_QUEUED, constants.JOB_STATUS_QUEUED)
self.assertEqual(client.JOB_STATUS_WAITING, constants.JOB_STATUS_WAITING)
self.assertEqual(client.JOB_STATUS_CANCELING,
self.assertEqual(client.JOB_STATUS_ALL, constants.JOB_STATUS_ALL)
# Node evacuation
- self.assertEqual(client.NODE_EVAC_PRI, constants.IALLOCATOR_NEVAC_PRI)
- self.assertEqual(client.NODE_EVAC_SEC, constants.IALLOCATOR_NEVAC_SEC)
- self.assertEqual(client.NODE_EVAC_ALL, constants.IALLOCATOR_NEVAC_ALL)
+ self.assertEqual(client.NODE_EVAC_PRI, constants.NODE_EVAC_PRI)
+ self.assertEqual(client.NODE_EVAC_SEC, constants.NODE_EVAC_SEC)
+ self.assertEqual(client.NODE_EVAC_ALL, constants.NODE_EVAC_ALL)
# Legacy name
self.assertEqual(client.JOB_STATUS_WAITLOCK, constants.JOB_STATUS_WAITING)
+ # RAPI feature strings
+ self.assertEqual(client._INST_CREATE_REQV1, rlib2._INST_CREATE_REQV1)
+ self.assertEqual(client.INST_CREATE_REQV1, rlib2._INST_CREATE_REQV1)
+ self.assertEqual(client._INST_REINSTALL_REQV1, rlib2._INST_REINSTALL_REQV1)
+ self.assertEqual(client.INST_REINSTALL_REQV1, rlib2._INST_REINSTALL_REQV1)
+ self.assertEqual(client._NODE_MIGRATE_REQV1, rlib2._NODE_MIGRATE_REQV1)
+ self.assertEqual(client.NODE_MIGRATE_REQV1, rlib2._NODE_MIGRATE_REQV1)
+ self.assertEqual(client._NODE_EVAC_RES1, rlib2._NODE_EVAC_RES1)
+ self.assertEqual(client.NODE_EVAC_RES1, rlib2._NODE_EVAC_RES1)
+
class RapiMockTest(unittest.TestCase):
def test(self):
rapi = RapiMock()
path = "/version"
- self.assertEqual((404, None), rapi.FetchResponse("/foo", "GET", None))
+ self.assertEqual((404, None), rapi.FetchResponse("/foo", "GET", None, None))
self.assertEqual((501, "Method not implemented"),
- rapi.FetchResponse("/version", "POST", None))
+ rapi.FetchResponse("/version", "POST", None, None))
rapi.AddResponse("2")
- code, response = rapi.FetchResponse("/version", "GET", None)
+ code, response = rapi.FetchResponse("/version", "GET", None, None)
self.assertEqual(200, code)
self.assertEqual("2", response)
self.failUnless(isinstance(rapi.GetLastHandler(), rlib2.R_version))
class TestExtendedConfig(unittest.TestCase):
def testAuth(self):
cl = client.GanetiRapiClient("master.example.com",
- username="user", password="pw",
- curl_factory=lambda: FakeCurl(RapiMock()))
+ username="user", password="pw",
+ curl_factory=lambda: rapi.testutils.FakeCurl(RapiMock()))
curl = cl._CreateCurl()
self.assertEqual(curl.getopt(pycurl.HTTPAUTH), pycurl.HTTPAUTH_BASIC)
verify_hostname=verify_hostname,
_pycurl_version_fn=pcverfn)
- curl_factory = lambda: FakeCurl(RapiMock())
+ curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())
cl = client.GanetiRapiClient("master.example.com",
curl_config_fn=cfgfn,
curl_factory=curl_factory)
def testNoCertVerify(self):
cfgfn = client.GenericCurlConfig()
- curl_factory = lambda: FakeCurl(RapiMock())
+ curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())
cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn,
curl_factory=curl_factory)
def testCertVerifyCurlBundle(self):
cfgfn = client.GenericCurlConfig(use_curl_cabundle=True)
- curl_factory = lambda: FakeCurl(RapiMock())
+ curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())
cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn,
curl_factory=curl_factory)
mycert = "/tmp/some/UNUSED/cert/file.pem"
cfgfn = client.GenericCurlConfig(cafile=mycert)
- curl_factory = lambda: FakeCurl(RapiMock())
+ curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())
cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn,
curl_factory=curl_factory)
cfgfn = client.GenericCurlConfig(capath=certdir,
_pycurl_version_fn=pcverfn)
- curl_factory = lambda: FakeCurl(RapiMock())
+ curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())
cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn,
curl_factory=curl_factory)
cfgfn = client.GenericCurlConfig(capath=certdir,
_pycurl_version_fn=pcverfn)
- curl_factory = lambda: FakeCurl(RapiMock())
+ curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())
cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn,
curl_factory=curl_factory)
cfgfn = client.GenericCurlConfig(capath=certdir,
_pycurl_version_fn=pcverfn)
- curl_factory = lambda: FakeCurl(RapiMock())
+ curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())
cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn,
curl_factory=curl_factory)
cfgfn = client.GenericCurlConfig(capath=certdir,
_pycurl_version_fn=pcverfn)
- curl_factory = lambda: FakeCurl(RapiMock())
+ curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())
cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn,
curl_factory=curl_factory)
cfgfn = client.GenericCurlConfig(connect_timeout=connect_timeout,
timeout=timeout)
- curl_factory = lambda: FakeCurl(RapiMock())
+ curl_factory = lambda: rapi.testutils.FakeCurl(RapiMock())
cl = client.GanetiRapiClient("master.example.com", curl_config_fn=cfgfn,
curl_factory=curl_factory)
testutils.GanetiTestCase.setUp(self)
self.rapi = RapiMock()
- self.curl = FakeCurl(self.rapi)
+ self.curl = rapi.testutils.FakeCurl(self.rapi)
self.client = client.GanetiRapiClient("master.example.com",
curl_factory=lambda: self.curl)
self.assertItems(["inst32"])
self.assertQuery("static", ["1"])
+ def testInstancesMultiAlloc(self):
+ response = {
+ constants.JOB_IDS_KEY: ["23423"],
+ opcodes.OpInstanceMultiAlloc.ALLOCATABLE_KEY: ["foobar"],
+ opcodes.OpInstanceMultiAlloc.FAILED_KEY: ["foobar2"],
+ }
+ self.rapi.AddResponse(serializer.DumpJson(response))
+ insts = [self.client.InstanceAllocation("create", "foobar",
+ "plain", [], []),
+ self.client.InstanceAllocation("create", "foobar2",
+ "drbd8", [{"size": 100}], [])]
+ resp = self.client.InstancesMultiAlloc(insts)
+ self.assertEqual(resp, response)
+ self.assertHandler(rlib2.R_2_instances_multi_alloc)
+
def testCreateInstanceOldVersion(self):
# The old request format, version 0, is no longer supported
self.rapi.AddResponse(None, code=404)
self.rapi.AddResponse(serializer.DumpJson([rlib2._NODE_EVAC_RES1]))
self.rapi.AddResponse("8888")
job_id = self.client.EvacuateNode("node-3", iallocator="hail", dry_run=True,
- mode=constants.IALLOCATOR_NEVAC_ALL,
+ mode=constants.NODE_EVAC_ALL,
early_release=True)
self.assertEqual(8888, job_id)
self.assertItems(["node-3"])