#!/usr/bin/python
#
-# Copyright (C) 2010, 2011 Google Inc.
+# Copyright (C) 2010, 2011, 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
import testutils
+#: Unless an opcode is included in the following list it must have a result
+#: check of some sort
+MISSING_RESULT_CHECK = frozenset([
+ opcodes.OpTestAllocator,
+ opcodes.OpTestDelay,
+ opcodes.OpTestDummy,
+ opcodes.OpTestJqueue,
+ ])
+
+
class TestOpcodes(unittest.TestCase):
def test(self):
self.assertRaises(ValueError, opcodes.OpCode.LoadOpCode, None)
self.assertEqual(cls.OP_ID, opcodes._NameToId(cls.__name__))
self.assertFalse(compat.any(cls.OP_ID.startswith(prefix)
for prefix in opcodes._SUMMARY_PREFIX.keys()))
- self.assertTrue(cls.OP_RESULT is None or callable(cls.OP_RESULT))
+ if cls in MISSING_RESULT_CHECK:
+ self.assertTrue(cls.OP_RESULT is None,
+ msg=("%s is listed to not have a result check" %
+ cls.OP_ID))
+ else:
+ self.assertTrue(callable(cls.OP_RESULT),
+ msg=("%s should have a result check" % cls.OP_ID))
self.assertRaises(TypeError, cls, unsupported_parameter="some value")
{"dry_run": False, "debug_level": 0, },
# All variables
- dict([(name, False) for name in cls._all_slots()])
+ dict([(name, []) for name in cls.GetAllSlots()])
]
for i in args:
self._checkSummary(restored)
for name in ["x_y_z", "hello_world"]:
- assert name not in cls._all_slots()
+ assert name not in cls.GetAllSlots()
for value in [None, True, False, [], "Hello World"]:
self.assertRaises(AttributeError, setattr, op, name, value)
self.assertTrue(opcodes.OpCode not in opcodes.OP_MAPPING.values())
for cls in opcodes.OP_MAPPING.values() + [opcodes.OpCode]:
- all_slots = cls._all_slots()
+ all_slots = cls.GetAllSlots()
self.assertEqual(len(set(all_slots) & supported_by_all), 3,
msg=("Opcode %s doesn't support all base"
self.assertTrue(doc is None or isinstance(doc, basestring))
if callable(aval):
- self.assertFalse(callable(aval()),
- msg="Default value returned by function is callable")
+ default_value = aval()
+ self.assertFalse(callable(default_value),
+ msg=("Default value of %s.%s returned by function"
+ " is callable" % (cls.OP_ID, attr_name)))
+ else:
+ self.assertFalse(isinstance(aval, (list, dict, set)),
+ msg=("Default value of %s.%s is mutable (%s)" %
+ (cls.OP_ID, attr_name, repr(aval))))
+
+ default_value = aval
+
+ if aval is not ht.NoDefault and test is not ht.NoType:
+ self.assertTrue(test(default_value),
+ msg=("Default value of %s.%s does not verify" %
+ (cls.OP_ID, attr_name)))
# If any parameter has documentation, all others need to have it as well
has_doc = [doc is not None for (_, _, _, doc) in cls.OP_PARAMS]
self.assertEqual(op.value2, "world")
self.assertEqual(op.debug_level, 123)
+ def testOpInstanceMultiAlloc(self):
+ inst = dict([(name, []) for name in opcodes.OpInstanceCreate.GetAllSlots()])
+ inst_op = opcodes.OpInstanceCreate(**inst)
+ inst_state = inst_op.__getstate__()
+
+ multialloc = opcodes.OpInstanceMultiAlloc(instances=[inst_op])
+ state = multialloc.__getstate__()
+ self.assertEquals(state["instances"], [inst_state])
+ loaded_multialloc = opcodes.OpCode.LoadOpCode(state)
+ (loaded_inst,) = loaded_multialloc.instances
+ self.assertNotEquals(loaded_inst, inst_op)
+ self.assertEquals(loaded_inst.__getstate__(), inst_state)
+
class TestOpcodeDepends(unittest.TestCase):
def test(self):
self.assertFalse(check_norelative(i))
+class TestResultChecks(unittest.TestCase):
+ def testJobIdList(self):
+ for i in [[], [(False, "error")], [(False, "")],
+ [(True, 123), (True, "999")]]:
+ self.assertTrue(opcodes.TJobIdList(i))
+
+ for i in ["", [("x", 1)], [[], []], [[False, "", None], [True, 123]]]:
+ self.assertFalse(opcodes.TJobIdList(i))
+
+ def testJobIdListOnly(self):
+ self.assertTrue(opcodes.TJobIdListOnly({
+ constants.JOB_IDS_KEY: [],
+ }))
+ self.assertTrue(opcodes.TJobIdListOnly({
+ constants.JOB_IDS_KEY: [(True, "9282")],
+ }))
+
+ self.assertFalse(opcodes.TJobIdListOnly({
+ "x": None,
+ }))
+ self.assertFalse(opcodes.TJobIdListOnly({
+ constants.JOB_IDS_KEY: [],
+ "x": None,
+ }))
+ self.assertFalse(opcodes.TJobIdListOnly({
+ constants.JOB_IDS_KEY: [("foo", "bar")],
+ }))
+ self.assertFalse(opcodes.TJobIdListOnly({
+ constants.JOB_IDS_KEY: [("one", "two", "three")],
+ }))
+
+
+class TestClusterOsList(unittest.TestCase):
+ def test(self):
+ good = [
+ None,
+ [],
+ [(constants.DDM_ADD, "dos"),
+ (constants.DDM_REMOVE, "linux")],
+ ]
+
+ for i in good:
+ self.assertTrue(opcodes._TestClusterOsList(i))
+
+ wrong = ["", 0, "xy", ["Hello World"], object(),
+ [("foo", "bar")],
+ [("", "")],
+ [[constants.DDM_ADD]],
+ [(constants.DDM_ADD, "")],
+ [(constants.DDM_REMOVE, "")],
+ [(constants.DDM_ADD, None)],
+ [(constants.DDM_REMOVE, None)],
+ ]
+
+ for i in wrong:
+ self.assertFalse(opcodes._TestClusterOsList(i))
+
+
+class TestOpInstanceSetParams(unittest.TestCase):
+ def _GenericTests(self, fn):
+ self.assertTrue(fn([]))
+ self.assertTrue(fn([(constants.DDM_ADD, {})]))
+ self.assertTrue(fn([(constants.DDM_REMOVE, {})]))
+ for i in [0, 1, 2, 3, 9, 10, 1024]:
+ self.assertTrue(fn([(i, {})]))
+
+ self.assertFalse(fn(None))
+ self.assertFalse(fn({}))
+ self.assertFalse(fn(""))
+ self.assertFalse(fn(0))
+ self.assertFalse(fn([(-100, {})]))
+ self.assertFalse(fn([(constants.DDM_ADD, 2, 3)]))
+ self.assertFalse(fn([[constants.DDM_ADD]]))
+
+ def testNicModifications(self):
+ fn = opcodes.OpInstanceSetParams.TestNicModifications
+ self._GenericTests(fn)
+
+ for param in constants.INIC_PARAMS:
+ self.assertTrue(fn([[constants.DDM_ADD, {param: None}]]))
+ self.assertTrue(fn([[constants.DDM_ADD, {param: param}]]))
+
+ def testDiskModifications(self):
+ fn = opcodes.OpInstanceSetParams.TestDiskModifications
+ self._GenericTests(fn)
+
+ for param in constants.IDISK_PARAMS:
+ self.assertTrue(fn([[constants.DDM_ADD, {param: 0}]]))
+ self.assertTrue(fn([[constants.DDM_ADD, {param: param}]]))
+
+
if __name__ == "__main__":
testutils.GanetiTestProgram()
projects / ganeti-local / blobdiff