4 # Copyright (C) 2010 Google Inc.
6 # This program is free software; you can redistribute it and/or modify
7 # it under the terms of the GNU General Public License as published by
8 # the Free Software Foundation; either version 2 of the License, or
9 # (at your option) any later version.
11 # This program is distributed in the hope that it will be useful, but
12 # WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 # General Public License for more details.
16 # You should have received a copy of the GNU General Public License
17 # along with this program; if not, write to the Free Software
18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
22 """Script for testing ganeti.backend"""
28 from ganeti import utils
29 from ganeti import opcodes
35 class TestOpcodes(unittest.TestCase):
37 self.assertRaises(ValueError, opcodes.OpCode.LoadOpCode, None)
38 self.assertRaises(ValueError, opcodes.OpCode.LoadOpCode, "")
39 self.assertRaises(ValueError, opcodes.OpCode.LoadOpCode, {})
40 self.assertRaises(ValueError, opcodes.OpCode.LoadOpCode, {"OP_ID": ""})
42 for cls in opcodes.OP_MAPPING.values():
43 self.assert_(cls.OP_ID.startswith("OP_"))
44 self.assertEqual(cls.OP_ID, cls.OP_ID.upper())
46 self.assertRaises(TypeError, cls, unsupported_parameter="some value")
52 # Variables supported by all opcodes
53 {"dry_run": False, "debug_level": 0, },
56 dict([(name, False) for name in cls._all_slots()])
62 self.assertEqual(op.OP_ID, cls.OP_ID)
63 self._checkSummary(op)
66 state = op.__getstate__()
67 self.assert_(isinstance(state, dict))
69 restored = opcodes.OpCode.LoadOpCode(state)
70 self.assert_(isinstance(restored, cls))
71 self._checkSummary(restored)
73 for name in ["x_y_z", "hello_world"]:
74 assert name not in cls._all_slots()
75 for value in [None, True, False, [], "Hello World"]:
76 self.assertRaises(AttributeError, setattr, op, name, value)
78 def _checkSummary(self, op):
79 summary = op.Summary()
81 if hasattr(op, "OP_DSC_FIELD"):
82 self.assert_(("OP_%s" % summary).startswith("%s(" % op.OP_ID))
83 self.assert_(summary.endswith(")"))
85 self.assertEqual("OP_%s" % summary, op.OP_ID)
88 supported_by_all = set(["debug_level", "dry_run", "priority"])
90 self.assert_(opcodes.BaseOpCode not in opcodes.OP_MAPPING.values())
91 self.assert_(opcodes.OpCode in opcodes.OP_MAPPING.values())
93 for cls in opcodes.OP_MAPPING.values():
94 all_slots = cls._all_slots()
96 self.assertEqual(len(set(all_slots) & supported_by_all), 3,
97 msg=("Opcode %s doesn't support all base"
98 " parameters (%r)" % (cls.OP_ID, supported_by_all)))
100 # All opcodes must have OP_PARAMS
101 self.assert_(hasattr(cls, "OP_PARAMS"),
102 msg="%s doesn't have OP_PARAMS" % cls.OP_ID)
104 param_names = [name for (name, _, _) in cls.GetAllParams()]
106 self.assertEqual(all_slots, param_names)
108 # Without inheritance
109 self.assertEqual(cls.__slots__, [name for (name, _, _) in cls.OP_PARAMS])
111 # This won't work if parameters are converted to a dictionary
112 duplicates = utils.FindDuplicates(param_names)
113 self.assertFalse(duplicates,
114 msg=("Found duplicate parameters %r in %s" %
115 (duplicates, cls.OP_ID)))
117 # Check parameter definitions
118 for attr_name, aval, test in cls.GetAllParams():
119 self.assert_(attr_name)
120 self.assert_(test is None or test is ht.NoType or callable(test),
121 msg=("Invalid type check for %s.%s" %
122 (cls.OP_ID, attr_name)))
125 self.assertFalse(callable(aval()),
126 msg="Default value returned by function is callable")
129 if __name__ == "__main__":
130 testutils.GanetiTestProgram()