#!/usr/bin/python
#
-# Copyright (C) 2010 Google Inc.
+# Copyright (C) 2010, 2012 Google Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
from ganeti import query
from ganeti import luxi
from ganeti import errors
+from ganeti import rapi
from ganeti.rapi import rlib2
+from ganeti.rapi import baserlib
+from ganeti.rapi import connector
import testutils
class _FakeClient:
- def __init__(self):
+ def __init__(self, address=None):
self._jobs = []
def GetNextSubmittedJob(self):
def GetNextClient(self):
return self._clients.pop(0)
- def __call__(self):
- cl = self._client_cls()
+ def __call__(self, address=None):
+ cl = self._client_cls(address=address)
self._clients.append(cl)
return cl
class TestClientConnectError(unittest.TestCase):
@staticmethod
- def _FailingClient():
+ def _FailingClient(address=None):
raise luxi.NoMasterError("test")
def test(self):
class TestJobSubmitError(unittest.TestCase):
class _SubmitErrorClient:
+ def __init__(self, address=None):
+ pass
+
@staticmethod
def SubmitJob(ops):
raise errors.JobQueueFull("test")
None,
{},
{ constants.BE_VCPUS: 2, },
- { constants.BE_MEMORY: 123, },
+ { constants.BE_MAXMEM: 200, },
+ { constants.BE_MEMORY: 256, },
{ constants.BE_VCPUS: 2,
- constants.BE_MEMORY: 1024,
- constants.BE_AUTO_BALANCE: True, }
+ constants.BE_MAXMEM: 1024,
+ constants.BE_MINMEM: 1024,
+ constants.BE_AUTO_BALANCE: True,
+ constants.BE_ALWAYS_FAILOVER: True, }
]
hvparam_variants = [
for osparams in [{}, { "some": "value", "other": "Hello World", }]:
for hvparams in [{}, { constants.HV_KERNEL_PATH: "/some/kernel", }]:
- for beparams in [{}, { constants.BE_MEMORY: 128, }]:
+ for beparams in [{}, { constants.BE_MAXMEM: 128, }]:
for force in [False, True]:
for nics in [[], [(0, { constants.INIC_IP: "192.0.2.1", })]]:
for disks in test_disks:
self.assertFalse(hasattr(op, "disks"))
self.assertRaises(IndexError, cl.GetNextSubmittedJob)
+ def testNoDisks(self):
+ clfactory = _FakeClientFactory(_FakeClient)
+
+ handler = _CreateHandler(rlib2.R_2_instances_name_replace_disks,
+ ["inst20661"], {}, {}, clfactory)
+ self.assertRaises(http.HttpBadRequest, handler.POST)
+
+ for disks in [None, "", {}]:
+ handler = _CreateHandler(rlib2.R_2_instances_name_replace_disks,
+ ["inst20661"], {}, {
+ "disks": disks,
+ }, clfactory)
+ self.assertRaises(http.HttpBadRequest, handler.POST)
+
def testWrong(self):
clfactory = _FakeClientFactory(_FakeClient)
class TestClusterInfo(unittest.TestCase):
class _ClusterInfoClient:
- def __init__(self):
+ def __init__(self, address=None):
self.cluster_info = None
def QueryClusterInfo(self):
self.assertEqual(result, cl.cluster_info)
-if __name__ == '__main__':
+class TestInstancesMultiAlloc(unittest.TestCase):
+ def testInstanceUpdate(self):
+ clfactory = _FakeClientFactory(_FakeClient)
+ data = {
+ "instances": [{
+ "instance_name": "bar",
+ "mode": "create",
+ }, {
+ "instance_name": "foo",
+ "mode": "create",
+ }],
+ }
+ handler = _CreateHandler(rlib2.R_2_instances_multi_alloc, [], {}, data,
+ clfactory)
+ (body, _) = handler.GetPostOpInput()
+ self.assertTrue(compat.all([inst["OP_ID"] == handler.POST_OPCODE.OP_ID
+ for inst in body["instances"]]))
+
+
+class TestPermissions(unittest.TestCase):
+ def testEquality(self):
+ self.assertEqual(rlib2.R_2_query.GET_ACCESS, rlib2.R_2_query.PUT_ACCESS)
+ self.assertEqual(rlib2.R_2_query.GET_ACCESS,
+ rlib2.R_2_instances_name_console.GET_ACCESS)
+
+ def testMethodAccess(self):
+ for handler in connector.CONNECTOR.values():
+ for method in baserlib._SUPPORTED_METHODS:
+ access = getattr(handler, "%s_ACCESS" % method)
+ self.assertFalse(set(access) - rapi.RAPI_ACCESS_ALL,
+ msg=("Handler '%s' uses unknown access options for"
+ " method %s" % (handler, method)))
+ self.assertTrue(rapi.RAPI_ACCESS_READ not in access or
+ rapi.RAPI_ACCESS_WRITE in access,
+ msg=("Handler '%s' gives query, but not write access"
+ " for method %s (the latter includes query and"
+ " should therefore be given as well)" %
+ (handler, method)))
+
+
+if __name__ == "__main__":
testutils.GanetiTestProgram()