#!/usr/bin/python
#
-# Copyright (C) 2008 Google Inc.
+# Copyright (C) 2008, 2011 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
nonexist_filename = os.path.join(self.tmpdir, "does-not-exist")
(errcode, msg) = cmdlib._VerifyCertificate(nonexist_filename)
- self.assertEqual(errcode, cmdlib.LUClusterVerify.ETYPE_ERROR)
+ self.assertEqual(errcode, cmdlib.LUClusterVerifyConfig.ETYPE_ERROR)
# Try to load non-certificate file
invalid_cert = self._TestDataFilename("bdev-net.txt")
(errcode, msg) = cmdlib._VerifyCertificate(invalid_cert)
- self.assertEqual(errcode, cmdlib.LUClusterVerify.ETYPE_ERROR)
+ self.assertEqual(errcode, cmdlib.LUClusterVerifyConfig.ETYPE_ERROR)
class TestOpcodeParams(testutils.GanetiTestCase):
self.cfg = mocks.FakeConfig()
self.op = opcode
- class TestOpcode(opcodes.OpCode):
- OP_ID = "OP_TEST"
- OP_PARAMS = [
- ("iallocator", None, ht.NoType),
- ("node", None, ht.NoType),
+ class OpTest(opcodes.OpCode):
+ OP_PARAMS = [
+ ("iallocator", None, ht.NoType, None),
+ ("node", None, ht.NoType, None),
]
default_iallocator = mocks.FakeConfig().GetDefaultIAllocator()
other_iallocator = default_iallocator + "_not"
- op = TestOpcode()
+ op = OpTest()
lu = TestLU(op)
c_i = lambda: cmdlib._CheckIAllocatorOrNode(lu, "iallocator", "node")
self.assertRaises(errors.OpPrereqError, c_i)
-class TestLUTestJobqueue(unittest.TestCase):
+class TestLUTestJqueue(unittest.TestCase):
def test(self):
- self.assert_(cmdlib.LUTestJobqueue._CLIENT_CONNECT_TIMEOUT <
+ self.assert_(cmdlib.LUTestJqueue._CLIENT_CONNECT_TIMEOUT <
(luxi.WFJC_TIMEOUT * 0.75),
msg=("Client timeout too high, might not notice bugs"
" in WaitForJobChange"))
class TestLUQuery(unittest.TestCase):
def test(self):
self.assertEqual(sorted(cmdlib._QUERY_IMPL.keys()),
- sorted(constants.QR_OP_QUERY))
+ sorted(constants.QR_VIA_OP))
- assert constants.QR_NODE in constants.QR_OP_QUERY
- assert constants.QR_INSTANCE in constants.QR_OP_QUERY
+ assert constants.QR_NODE in constants.QR_VIA_OP
+ assert constants.QR_INSTANCE in constants.QR_VIA_OP
- for i in constants.QR_OP_QUERY:
+ for i in constants.QR_VIA_OP:
self.assert_(cmdlib._GetQueryImplementation(i))
self.assertRaises(errors.OpPrereqError, cmdlib._GetQueryImplementation, "")
"xyz")
-class TestLUAssignGroupNodes(unittest.TestCase):
+class TestLUGroupAssignNodes(unittest.TestCase):
def testCheckAssignmentForSplitInstances(self):
node_data = dict((name, objects.Node(name=name, group=group))
# Test first with the existing state.
(new, prev) = \
- cmdlib.LUAssignGroupNodes.CheckAssignmentForSplitInstances([],
+ cmdlib.LUGroupAssignNodes.CheckAssignmentForSplitInstances([],
node_data,
instance_data)
# And now some changes.
(new, prev) = \
- cmdlib.LUAssignGroupNodes.CheckAssignmentForSplitInstances([("n1b",
+ cmdlib.LUGroupAssignNodes.CheckAssignmentForSplitInstances([("n1b",
"g3")],
node_data,
instance_data)