#!/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
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")
{ constants.BE_VCPUS: 2,
constants.BE_MAXMEM: 1024,
constants.BE_MINMEM: 1024,
- constants.BE_AUTO_BALANCE: True, }
+ constants.BE_AUTO_BALANCE: True,
+ constants.BE_ALWAYS_FAILOVER: True, }
]
hvparam_variants = [
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"]]))
+
+
+if __name__ == "__main__":
testutils.GanetiTestProgram()