from ganeti import errors
from ganeti import opcodes
from ganeti import http
+from ganeti import server
+from ganeti import utils
+from ganeti import compat
+from ganeti import luxi
+from ganeti import rapi
+
+import ganeti.http.server # pylint: disable=W0611
+import ganeti.server.rapi
+import ganeti.rapi.client
_URI_RE = re.compile(r"https://(?P<host>.*):(?P<port>\d+)(?P<path>/.*)")
def wrapper(*args, **kwargs):
try:
return fn(*args, **kwargs)
- except errors.GenericError, err:
+ except (errors.GenericError, rapi.client.GanetiApiError), err:
raise VerificationError("Unhandled Ganeti error: %s" % err)
return wrapper
self._info[pycurl.RESPONSE_CODE] = code
if resp_body is not None:
writefn(resp_body)
+
+
+class _RapiMock:
+ """Mocking out the RAPI server parts.
+
+ """
+ def __init__(self, user_fn, luxi_client):
+ """Initialize this class.
+
+ @type user_fn: callable
+ @param user_fn: Function to authentication username
+ @param luxi_client: A LUXI client implementation
+
+ """
+ self.handler = \
+ server.rapi.RemoteApiHandler(user_fn, _client_cls=luxi_client)
+
+ def FetchResponse(self, path, method, headers, request_body):
+ """This is a callback method used to fetch a response.
+
+ This method is called by the FakeCurl.perform method
+
+ @type path: string
+ @param path: Requested path
+ @type method: string
+ @param method: HTTP method
+ @type request_body: string
+ @param request_body: Request body
+ @type headers: mimetools.Message
+ @param headers: Request headers
+ @return: Tuple containing status code and response body
+
+ """
+ req_msg = http.HttpMessage()
+ req_msg.start_line = \
+ http.HttpClientToServerStartLine(method, path, http.HTTP_1_0)
+ req_msg.headers = headers
+ req_msg.body = request_body
+
+ (_, _, _, resp_msg) = \
+ http.server.HttpResponder(self.handler)(lambda: (req_msg, None))
+
+ return (resp_msg.start_line.code, resp_msg.body)
+
+
+class _TestLuxiTransport:
+ """Mocked LUXI transport.
+
+ Raises L{errors.RapiTestResult} for all method calls, no matter the
+ arguments.
+
+ """
+ def __init__(self, record_fn, address, timeouts=None): # pylint: disable=W0613
+ """Initializes this class.
+
+ """
+ self._record_fn = record_fn
+
+ def Close(self):
+ pass
+
+ def Call(self, data):
+ """Calls LUXI method.
+
+ In this test class the method is not actually called, but added to a list
+ of called methods and then an exception (L{errors.RapiTestResult}) is
+ raised. There is no return value.
+
+ """
+ (method, _, _) = luxi.ParseRequest(data)
+
+ # Take a note of called method
+ self._record_fn(method)
+
+ # Everything went fine until here, so let's abort the test
+ raise errors.RapiTestResult
+
+
+class _LuxiCallRecorder:
+ """Records all called LUXI client methods.
+
+ """
+ def __init__(self):
+ """Initializes this class.
+
+ """
+ self._called = set()
+
+ def Record(self, name):
+ """Records a called function name.
+
+ """
+ self._called.add(name)
+
+ def CalledNames(self):
+ """Returns a list of called LUXI methods.
+
+ """
+ return self._called
+
+ def __call__(self):
+ """Creates an instrumented LUXI client.
+
+ The LUXI client will record all method calls (use L{CalledNames} to
+ retrieve them).
+
+ """
+ return luxi.Client(transport=compat.partial(_TestLuxiTransport,
+ self.Record))
+
+
+def _TestWrapper(fn, *args, **kwargs):
+ """Wrapper for ignoring L{errors.RapiTestResult}.
+
+ """
+ try:
+ return fn(*args, **kwargs)
+ except errors.RapiTestResult:
+ # Everything was fine up to the point of sending a LUXI request
+ return NotImplemented
+
+
+class InputTestClient:
+ """Test version of RAPI client.
+
+ Instances of this class can be used to test input arguments for RAPI client
+ calls. See L{rapi.client.GanetiRapiClient} for available methods and their
+ arguments. Functions can return C{NotImplemented} if all arguments are
+ acceptable, but a LUXI request would be necessary to provide an actual return
+ value. In case of an error, L{VerificationError} is raised.
+
+ @see: An example on how to use this class can be found in
+ C{doc/examples/rapi_testutils.py}
+
+ """
+ def __init__(self):
+ """Initializes this class.
+
+ """
+ username = utils.GenerateSecret()
+ password = utils.GenerateSecret()
+
+ def user_fn(wanted):
+ """Called to verify user credentials given in HTTP request.
+
+ """
+ assert username == wanted
+ return http.auth.PasswordFileUser(username, password,
+ [rapi.RAPI_ACCESS_WRITE])
+
+ self._lcr = _LuxiCallRecorder()
+
+ # Create a mock RAPI server
+ handler = _RapiMock(user_fn, self._lcr)
+
+ self._client = \
+ rapi.client.GanetiRapiClient("master.example.com",
+ username=username, password=password,
+ curl_factory=lambda: FakeCurl(handler))
+
+ def _GetLuxiCalls(self):
+ """Returns the names of all called LUXI client functions.
+
+ """
+ return self._lcr.CalledNames()
+
+ def __getattr__(self, name):
+ """Finds method by name.
+
+ The method is wrapped using L{_TestWrapper} to produce the actual test
+ result.
+
+ """
+ return _HideInternalErrors(compat.partial(_TestWrapper,
+ getattr(self._client, name)))
from ganeti import constants
from ganeti import errors
from ganeti import opcodes
+from ganeti import luxi
from ganeti import rapi
+from ganeti import utils
import ganeti.rapi.testutils
+import ganeti.rapi.client
import testutils
+KNOWN_UNUSED_LUXI = frozenset([
+ luxi.REQ_SUBMIT_MANY_JOBS,
+ luxi.REQ_ARCHIVE_JOB,
+ luxi.REQ_AUTOARCHIVE_JOBS,
+ luxi.REQ_QUERY_EXPORTS,
+ luxi.REQ_QUERY_CONFIG_VALUES,
+ luxi.REQ_QUERY_TAGS,
+ luxi.REQ_QUEUE_SET_DRAIN_FLAG,
+ luxi.REQ_SET_WATCHER_PAUSE,
+ ])
+
+
+# Global variable for storing used LUXI calls
+_used_luxi_calls = None
+
+
class TestHideInternalErrors(unittest.TestCase):
def test(self):
def inner():
vor(opcodes.OpTestDummy.OP_ID, None)
+class TestInputTestClient(unittest.TestCase):
+ def setUp(self):
+ self.cl = rapi.testutils.InputTestClient()
+
+ def tearDown(self):
+ _used_luxi_calls.update(self.cl._GetLuxiCalls())
+
+ def testGetInfo(self):
+ self.assertTrue(self.cl.GetInfo() is NotImplemented)
+
+ def testPrepareExport(self):
+ result = self.cl.PrepareExport("inst1.example.com",
+ constants.EXPORT_MODE_LOCAL)
+ self.assertTrue(result is NotImplemented)
+ self.assertRaises(rapi.client.GanetiApiError, self.cl.PrepareExport,
+ "inst1.example.com", "###invalid###")
+
+ def testGetJobs(self):
+ self.assertTrue(self.cl.GetJobs() is NotImplemented)
+
+ def testQuery(self):
+ result = self.cl.Query(constants.QR_NODE, ["name"])
+ self.assertTrue(result is NotImplemented)
+
+ def testQueryFields(self):
+ result = self.cl.QueryFields(constants.QR_INSTANCE)
+ self.assertTrue(result is NotImplemented)
+
+ def testCancelJob(self):
+ self.assertTrue(self.cl.CancelJob("1") is NotImplemented)
+
+ def testGetNodes(self):
+ self.assertTrue(self.cl.GetNodes() is NotImplemented)
+
+ def testGetInstances(self):
+ self.assertTrue(self.cl.GetInstances() is NotImplemented)
+
+ def testGetGroups(self):
+ self.assertTrue(self.cl.GetGroups() is NotImplemented)
+
+ def testWaitForJobChange(self):
+ result = self.cl.WaitForJobChange("1", ["id"], None, None)
+ self.assertTrue(result is NotImplemented)
+
+
+class CustomTestRunner(unittest.TextTestRunner):
+ def run(self, *args):
+ global _used_luxi_calls
+ assert _used_luxi_calls is None
+
+ diff = (KNOWN_UNUSED_LUXI - luxi.REQ_ALL)
+ assert not diff, "Non-existing LUXI calls listed as unused: %s" % diff
+
+ _used_luxi_calls = set()
+ try:
+ # Run actual tests
+ result = unittest.TextTestRunner.run(self, *args)
+
+ diff = _used_luxi_calls & KNOWN_UNUSED_LUXI
+ if diff:
+ raise AssertionError("LUXI methods marked as unused were called: %s" %
+ utils.CommaJoin(diff))
+
+ diff = (luxi.REQ_ALL - KNOWN_UNUSED_LUXI - _used_luxi_calls)
+ if diff:
+ raise AssertionError("The following LUXI methods were not used: %s" %
+ utils.CommaJoin(diff))
+ finally:
+ # Reset global variable
+ _used_luxi_calls = None
+
+ return result
+
+
if __name__ == "__main__":
- testutils.GanetiTestProgram()
+ testutils.GanetiTestProgram(testRunner=CustomTestRunner)
projects / ganeti-local / commitdiff