code.grnet.gr Git - ganeti-local/blob - test/ganeti.opcodes_unittest.py

   1 #!/usr/bin/python
   2 #
   3
   4 # Copyright (C) 2010, 2011 Google Inc.
   5 #
   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.
  10 #
  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.
  15 #
  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
  19 # 02110-1301, USA.
  20
  21
  22 """Script for testing ganeti.backend"""
  23
  24 import os
  25 import sys
  26 import unittest
  27
  28 from ganeti import utils
  29 from ganeti import opcodes
  30 from ganeti import ht
  31 from ganeti import constants
  32 from ganeti import errors
  33 from ganeti import compat
  34
  35 import testutils
  36
  37
  38 class TestOpcodes(unittest.TestCase):
  39   def test(self):
  40     self.assertRaises(ValueError, opcodes.OpCode.LoadOpCode, None)
  41     self.assertRaises(ValueError, opcodes.OpCode.LoadOpCode, "")
  42     self.assertRaises(ValueError, opcodes.OpCode.LoadOpCode, {})
  43     self.assertRaises(ValueError, opcodes.OpCode.LoadOpCode, {"OP_ID": ""})
  44
  45     for cls in opcodes.OP_MAPPING.values():
  46       self.assert_(cls.OP_ID.startswith("OP_"))
  47       self.assert_(len(cls.OP_ID) > 3)
  48       self.assertEqual(cls.OP_ID, cls.OP_ID.upper())
  49       self.assertEqual(cls.OP_ID, opcodes._NameToId(cls.__name__))
  50       self.assertFalse(compat.any(cls.OP_ID.startswith(prefix)
  51                                   for prefix in opcodes._SUMMARY_PREFIX.keys()))
  52
  53       self.assertRaises(TypeError, cls, unsupported_parameter="some value")
  54
  55       args = [
  56         # No variables
  57         {},
  58
  59         # Variables supported by all opcodes
  60         {"dry_run": False, "debug_level": 0, },
  61
  62         # All variables
  63         dict([(name, False) for name in cls._all_slots()])
  64         ]
  65
  66       for i in args:
  67         op = cls(**i)
  68
  69         self.assertEqual(op.OP_ID, cls.OP_ID)
  70         self._checkSummary(op)
  71
  72         # Try a restore
  73         state = op.__getstate__()
  74         self.assert_(isinstance(state, dict))
  75
  76         restored = opcodes.OpCode.LoadOpCode(state)
  77         self.assert_(isinstance(restored, cls))
  78         self._checkSummary(restored)
  79
  80         for name in ["x_y_z", "hello_world"]:
  81           assert name not in cls._all_slots()
  82           for value in [None, True, False, [], "Hello World"]:
  83             self.assertRaises(AttributeError, setattr, op, name, value)
  84
  85   def _checkSummary(self, op):
  86     summary = op.Summary()
  87
  88     if hasattr(op, "OP_DSC_FIELD"):
  89       self.assert_(("OP_%s" % summary).startswith("%s(" % op.OP_ID))
  90       self.assert_(summary.endswith(")"))
  91     else:
  92       self.assertEqual("OP_%s" % summary, op.OP_ID)
  93
  94   def testSummary(self):
  95     class OpTest(opcodes.OpCode):
  96       OP_DSC_FIELD = "data"
  97       OP_PARAMS = [
  98         ("data", ht.NoDefault, ht.TString, None),
  99         ]
 100
 101     self.assertEqual(OpTest(data="").Summary(), "TEST()")
 102     self.assertEqual(OpTest(data="Hello World").Summary(),
 103                      "TEST(Hello World)")
 104     self.assertEqual(OpTest(data="node1.example.com").Summary(),
 105                      "TEST(node1.example.com)")
 106
 107   def testTinySummary(self):
 108     self.assertFalse(utils.FindDuplicates(opcodes._SUMMARY_PREFIX.values()))
 109     self.assertTrue(compat.all(prefix.endswith("_") and supplement.endswith("_")
 110                                for (prefix, supplement) in
 111                                  opcodes._SUMMARY_PREFIX.items()))
 112
 113     self.assertEqual(opcodes.OpClusterPostInit().TinySummary(), "C_POST_INIT")
 114     self.assertEqual(opcodes.OpNodeRemove().TinySummary(), "N_REMOVE")
 115     self.assertEqual(opcodes.OpInstanceMigrate().TinySummary(), "I_MIGRATE")
 116     self.assertEqual(opcodes.OpGroupQuery().TinySummary(), "G_QUERY")
 117     self.assertEqual(opcodes.OpTestJqueue().TinySummary(), "TEST_JQUEUE")
 118
 119   def testListSummary(self):
 120     class OpTest(opcodes.OpCode):
 121       OP_DSC_FIELD = "data"
 122       OP_PARAMS = [
 123         ("data", ht.NoDefault, ht.TList, None),
 124         ]
 125
 126     self.assertEqual(OpTest(data=["a", "b", "c"]).Summary(),
 127                      "TEST(a,b,c)")
 128     self.assertEqual(OpTest(data=["a", None, "c"]).Summary(),
 129                      "TEST(a,None,c)")
 130     self.assertEqual(OpTest(data=[1, 2, 3, 4]).Summary(), "TEST(1,2,3,4)")
 131
 132   def testOpId(self):
 133     self.assertFalse(utils.FindDuplicates(cls.OP_ID
 134                                           for cls in opcodes._GetOpList()))
 135     self.assertEqual(len(opcodes._GetOpList()), len(opcodes.OP_MAPPING))
 136
 137   def testParams(self):
 138     supported_by_all = set(["debug_level", "dry_run", "priority"])
 139
 140     self.assertTrue(opcodes.BaseOpCode not in opcodes.OP_MAPPING.values())
 141     self.assertTrue(opcodes.OpCode not in opcodes.OP_MAPPING.values())
 142
 143     for cls in opcodes.OP_MAPPING.values() + [opcodes.OpCode]:
 144       all_slots = cls._all_slots()
 145
 146       self.assertEqual(len(set(all_slots) & supported_by_all), 3,
 147                        msg=("Opcode %s doesn't support all base"
 148                             " parameters (%r)" % (cls.OP_ID, supported_by_all)))
 149
 150       # All opcodes must have OP_PARAMS
 151       self.assert_(hasattr(cls, "OP_PARAMS"),
 152                    msg="%s doesn't have OP_PARAMS" % cls.OP_ID)
 153
 154       param_names = [name for (name, _, _, _) in cls.GetAllParams()]
 155
 156       self.assertEqual(all_slots, param_names)
 157
 158       # Without inheritance
 159       self.assertEqual(cls.__slots__,
 160                        [name for (name, _, _, _) in cls.OP_PARAMS])
 161
 162       # This won't work if parameters are converted to a dictionary
 163       duplicates = utils.FindDuplicates(param_names)
 164       self.assertFalse(duplicates,
 165                        msg=("Found duplicate parameters %r in %s" %
 166                             (duplicates, cls.OP_ID)))
 167
 168       # Check parameter definitions
 169       for attr_name, aval, test, doc in cls.GetAllParams():
 170         self.assert_(attr_name)
 171         self.assert_(test is None or test is ht.NoType or callable(test),
 172                      msg=("Invalid type check for %s.%s" %
 173                           (cls.OP_ID, attr_name)))
 174         self.assertTrue(doc is None or isinstance(doc, basestring))
 175
 176         if callable(aval):
 177           self.assertFalse(callable(aval()),
 178                            msg="Default value returned by function is callable")
 179
 180       # If any parameter has documentation, all others need to have it as well
 181       has_doc = [doc is not None for (_, _, _, doc) in cls.OP_PARAMS]
 182       self.assertTrue(not compat.any(has_doc) or compat.all(has_doc),
 183                       msg="%s does not document all parameters" % cls)
 184
 185   def testValidateNoModification(self):
 186     class OpTest(opcodes.OpCode):
 187       OP_PARAMS = [
 188         ("nodef", ht.NoDefault, ht.TMaybeString, None),
 189         ("wdef", "default", ht.TMaybeString, None),
 190         ("number", 0, ht.TInt, None),
 191         ("notype", None, ht.NoType, None),
 192         ]
 193
 194     # Missing required parameter "nodef"
 195     op = OpTest()
 196     before = op.__getstate__()
 197     self.assertRaises(errors.OpPrereqError, op.Validate, False)
 198     self.assertFalse(hasattr(op, "nodef"))
 199     self.assertFalse(hasattr(op, "wdef"))
 200     self.assertFalse(hasattr(op, "number"))
 201     self.assertFalse(hasattr(op, "notype"))
 202     self.assertEqual(op.__getstate__(), before, msg="Opcode was modified")
 203
 204     # Required parameter "nodef" is provided
 205     op = OpTest(nodef="foo")
 206     before = op.__getstate__()
 207     op.Validate(False)
 208     self.assertEqual(op.__getstate__(), before, msg="Opcode was modified")
 209     self.assertEqual(op.nodef, "foo")
 210     self.assertFalse(hasattr(op, "wdef"))
 211     self.assertFalse(hasattr(op, "number"))
 212     self.assertFalse(hasattr(op, "notype"))
 213
 214     # Missing required parameter "nodef"
 215     op = OpTest(wdef="hello", number=999)
 216     before = op.__getstate__()
 217     self.assertRaises(errors.OpPrereqError, op.Validate, False)
 218     self.assertFalse(hasattr(op, "nodef"))
 219     self.assertFalse(hasattr(op, "notype"))
 220     self.assertEqual(op.__getstate__(), before, msg="Opcode was modified")
 221
 222     # Wrong type for "nodef"
 223     op = OpTest(nodef=987)
 224     before = op.__getstate__()
 225     self.assertRaises(errors.OpPrereqError, op.Validate, False)
 226     self.assertEqual(op.nodef, 987)
 227     self.assertFalse(hasattr(op, "notype"))
 228     self.assertEqual(op.__getstate__(), before, msg="Opcode was modified")
 229
 230     # Testing different types for "notype"
 231     op = OpTest(nodef="foo", notype=[1, 2, 3])
 232     before = op.__getstate__()
 233     op.Validate(False)
 234     self.assertEqual(op.nodef, "foo")
 235     self.assertEqual(op.notype, [1, 2, 3])
 236     self.assertEqual(op.__getstate__(), before, msg="Opcode was modified")
 237
 238     op = OpTest(nodef="foo", notype="Hello World")
 239     before = op.__getstate__()
 240     op.Validate(False)
 241     self.assertEqual(op.nodef, "foo")
 242     self.assertEqual(op.notype, "Hello World")
 243     self.assertEqual(op.__getstate__(), before, msg="Opcode was modified")
 244
 245   def testValidateSetDefaults(self):
 246     class OpTest(opcodes.OpCode):
 247       OP_PARAMS = [
 248         # Static default value
 249         ("value1", "default", ht.TMaybeString, None),
 250
 251         # Default value callback
 252         ("value2", lambda: "result", ht.TMaybeString, None),
 253         ]
 254
 255     op = OpTest()
 256     before = op.__getstate__()
 257     op.Validate(True)
 258     self.assertNotEqual(op.__getstate__(), before,
 259                         msg="Opcode was not modified")
 260     self.assertEqual(op.value1, "default")
 261     self.assertEqual(op.value2, "result")
 262     self.assert_(op.dry_run is None)
 263     self.assert_(op.debug_level is None)
 264     self.assertEqual(op.priority, constants.OP_PRIO_DEFAULT)
 265
 266     op = OpTest(value1="hello", value2="world", debug_level=123)
 267     before = op.__getstate__()
 268     op.Validate(True)
 269     self.assertNotEqual(op.__getstate__(), before,
 270                         msg="Opcode was not modified")
 271     self.assertEqual(op.value1, "hello")
 272     self.assertEqual(op.value2, "world")
 273     self.assertEqual(op.debug_level, 123)
 274
 275
 276 if __name__ == "__main__":
 277   testutils.GanetiTestProgram()