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

   1 #!/usr/bin/python
   2 #
   3
   4 # Copyright (C) 2010 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 unittesting the RAPI rlib2 module
  23
  24 """
  25
  26
  27 import unittest
  28 import tempfile
  29
  30 from ganeti import constants
  31 from ganeti import opcodes
  32 from ganeti import compat
  33 from ganeti import http
  34
  35 from ganeti.rapi import rlib2
  36
  37 import testutils
  38
  39
  40 class TestParseInstanceCreateRequestVersion1(testutils.GanetiTestCase):
  41   def setUp(self):
  42     testutils.GanetiTestCase.setUp(self)
  43
  44     self.Parse = rlib2._ParseInstanceCreateRequestVersion1
  45
  46   def test(self):
  47     disk_variants = [
  48       # No disks
  49       [],
  50
  51       # Two disks
  52       [{"size": 5, }, {"size": 100, }],
  53
  54       # Disk with mode
  55       [{"size": 123, "mode": constants.DISK_RDWR, }],
  56
  57       # With unknown setting
  58       [{"size": 123, "unknown": 999 }],
  59       ]
  60
  61     nic_variants = [
  62       # No NIC
  63       [],
  64
  65       # Three NICs
  66       [{}, {}, {}],
  67
  68       # Two NICs
  69       [
  70         { "ip": "192.0.2.6", "mode": constants.NIC_MODE_ROUTED,
  71           "mac": "01:23:45:67:68:9A",
  72         },
  73         { "mode": constants.NIC_MODE_BRIDGED, "link": "n0", "bridge": "br1", },
  74       ],
  75
  76       # Unknown settings
  77       [{ "unknown": 999, }, { "foobar": "Hello World", }],
  78       ]
  79
  80     beparam_variants = [
  81       None,
  82       {},
  83       { constants.BE_VCPUS: 2, },
  84       { constants.BE_MEMORY: 123, },
  85       { constants.BE_VCPUS: 2,
  86         constants.BE_MEMORY: 1024,
  87         constants.BE_AUTO_BALANCE: True, }
  88       ]
  89
  90     hvparam_variants = [
  91       None,
  92       { constants.HV_BOOT_ORDER: "anc", },
  93       { constants.HV_KERNEL_PATH: "/boot/fookernel",
  94         constants.HV_ROOT_PATH: "/dev/hda1", },
  95       ]
  96
  97     for mode in [constants.INSTANCE_CREATE, constants.INSTANCE_IMPORT]:
  98       for nics in nic_variants:
  99         for disk_template in constants.DISK_TEMPLATES:
 100           for disks in disk_variants:
 101             for beparams in beparam_variants:
 102               for hvparams in hvparam_variants:
 103                 data = {
 104                   "name": "inst1.example.com",
 105                   "hypervisor": constants.HT_FAKE,
 106                   "disks": disks,
 107                   "nics": nics,
 108                   "mode": mode,
 109                   "disk_template": disk_template,
 110                   "os": "debootstrap",
 111                   }
 112
 113                 if beparams is not None:
 114                   data["beparams"] = beparams
 115
 116                 if hvparams is not None:
 117                   data["hvparams"] = hvparams
 118
 119                 for dry_run in [False, True]:
 120                   op = self.Parse(data, dry_run)
 121                   self.assert_(isinstance(op, opcodes.OpCreateInstance))
 122                   self.assertEqual(op.mode, mode)
 123                   self.assertEqual(op.disk_template, disk_template)
 124                   self.assertEqual(op.dry_run, dry_run)
 125                   self.assertEqual(len(op.disks), len(disks))
 126                   self.assertEqual(len(op.nics), len(nics))
 127
 128                   for opdisk, disk in zip(op.disks, disks):
 129                     for key in constants.IDISK_PARAMS:
 130                       self.assertEqual(opdisk.get(key), disk.get(key))
 131                     self.assertFalse("unknown" in opdisk)
 132
 133                   for opnic, nic in zip(op.nics, nics):
 134                     for key in constants.INIC_PARAMS:
 135                       self.assertEqual(opnic.get(key), nic.get(key))
 136                     self.assertFalse("unknown" in opnic)
 137                     self.assertFalse("foobar" in opnic)
 138
 139                   if beparams is None:
 140                     self.assertEqualValues(op.beparams, {})
 141                   else:
 142                     self.assertEqualValues(op.beparams, beparams)
 143
 144                   if hvparams is None:
 145                     self.assertEqualValues(op.hvparams, {})
 146                   else:
 147                     self.assertEqualValues(op.hvparams, hvparams)
 148
 149   def testErrors(self):
 150     # Test all required fields
 151     reqfields = {
 152       "name": "inst1.example.com",
 153       "disks": [],
 154       "nics": [],
 155       "mode": constants.INSTANCE_CREATE,
 156       "disk_template": constants.DT_PLAIN,
 157       "os": "debootstrap",
 158       }
 159
 160     for name in reqfields.keys():
 161       self.assertRaises(http.HttpBadRequest, self.Parse,
 162                         dict(i for i in reqfields.iteritems() if i[0] != name),
 163                         False)
 164
 165     # Invalid disks and nics
 166     for field in ["disks", "nics"]:
 167       invalid_values = [None, 1, "", {}, [1, 2, 3], ["hda1", "hda2"]]
 168
 169       if field == "disks":
 170         invalid_values.append([
 171           # Disks without size
 172           {},
 173           { "mode": constants.DISK_RDWR, },
 174           ])
 175
 176       for invvalue in invalid_values:
 177         data = reqfields.copy()
 178         data[field] = invvalue
 179         self.assertRaises(http.HttpBadRequest, self.Parse, data, False)
 180
 181
 182 class TestParseExportInstanceRequest(testutils.GanetiTestCase):
 183   def setUp(self):
 184     testutils.GanetiTestCase.setUp(self)
 185
 186     self.Parse = rlib2._ParseExportInstanceRequest
 187
 188   def test(self):
 189     name = "instmoo"
 190     data = {
 191       "mode": constants.EXPORT_MODE_REMOTE,
 192       "destination": [(1, 2, 3), (99, 99, 99)],
 193       "shutdown": True,
 194       "remove_instance": True,
 195       "x509_key_name": ("name", "hash"),
 196       "destination_x509_ca": ("x", "y", "z"),
 197       }
 198     op = self.Parse(name, data)
 199     self.assert_(isinstance(op, opcodes.OpExportInstance))
 200     self.assertEqual(op.instance_name, name)
 201     self.assertEqual(op.mode, constants.EXPORT_MODE_REMOTE)
 202     self.assertEqual(op.shutdown, True)
 203     self.assertEqual(op.remove_instance, True)
 204     self.assertEqualValues(op.x509_key_name, ("name", "hash"))
 205     self.assertEqualValues(op.destination_x509_ca, ("x", "y", "z"))
 206
 207   def testDefaults(self):
 208     name = "inst1"
 209     data = {
 210       "destination": "node2",
 211       "shutdown": False,
 212       }
 213     op = self.Parse(name, data)
 214     self.assert_(isinstance(op, opcodes.OpExportInstance))
 215     self.assertEqual(op.instance_name, name)
 216     self.assertEqual(op.mode, constants.EXPORT_MODE_LOCAL)
 217     self.assertEqual(op.remove_instance, False)
 218
 219   def testErrors(self):
 220     self.assertRaises(http.HttpBadRequest, self.Parse, "err1",
 221                       { "remove_instance": "True", })
 222     self.assertRaises(http.HttpBadRequest, self.Parse, "err1",
 223                       { "remove_instance": "False", })
 224
 225
 226 class TestParseMigrateInstanceRequest(testutils.GanetiTestCase):
 227   def setUp(self):
 228     testutils.GanetiTestCase.setUp(self)
 229
 230     self.Parse = rlib2._ParseMigrateInstanceRequest
 231
 232   def test(self):
 233     name = "instYooho6ek"
 234
 235     for cleanup in [False, True]:
 236       for mode in constants.HT_MIGRATION_MODES:
 237         data = {
 238           "cleanup": cleanup,
 239           "mode": mode,
 240           }
 241         op = self.Parse(name, data)
 242         self.assert_(isinstance(op, opcodes.OpMigrateInstance))
 243         self.assertEqual(op.instance_name, name)
 244         self.assertEqual(op.mode, mode)
 245         self.assertEqual(op.cleanup, cleanup)
 246
 247   def testDefaults(self):
 248     name = "instnohZeex0"
 249
 250     op = self.Parse(name, {})
 251     self.assert_(isinstance(op, opcodes.OpMigrateInstance))
 252     self.assertEqual(op.instance_name, name)
 253     self.assertEqual(op.mode, None)
 254     self.assertFalse(op.cleanup)
 255
 256
 257 class TestParseRenameInstanceRequest(testutils.GanetiTestCase):
 258   def setUp(self):
 259     testutils.GanetiTestCase.setUp(self)
 260
 261     self.Parse = rlib2._ParseRenameInstanceRequest
 262
 263   def test(self):
 264     name = "instij0eeph7"
 265
 266     for new_name in ["ua0aiyoo", "fai3ongi"]:
 267       for ip_check in [False, True]:
 268         for name_check in [False, True]:
 269           data = {
 270             "new_name": new_name,
 271             "ip_check": ip_check,
 272             "name_check": name_check,
 273             }
 274
 275           op = self.Parse(name, data)
 276           self.assert_(isinstance(op, opcodes.OpRenameInstance))
 277           self.assertEqual(op.instance_name, name)
 278           self.assertEqual(op.new_name, new_name)
 279           self.assertEqual(op.ip_check, ip_check)
 280           self.assertEqual(op.name_check, name_check)
 281
 282   def testDefaults(self):
 283     name = "instahchie3t"
 284
 285     for new_name in ["thag9mek", "quees7oh"]:
 286       data = {
 287         "new_name": new_name,
 288         }
 289
 290       op = self.Parse(name, data)
 291       self.assert_(isinstance(op, opcodes.OpRenameInstance))
 292       self.assertEqual(op.instance_name, name)
 293       self.assertEqual(op.new_name, new_name)
 294       self.assert_(op.ip_check)
 295       self.assert_(op.name_check)
 296
 297
 298 class TestParseModifyInstanceRequest(testutils.GanetiTestCase):
 299   def setUp(self):
 300     testutils.GanetiTestCase.setUp(self)
 301
 302     self.Parse = rlib2._ParseModifyInstanceRequest
 303
 304   def test(self):
 305     name = "instush8gah"
 306
 307     test_disks = [
 308       [],
 309       [(1, { constants.IDISK_MODE: constants.DISK_RDWR, })],
 310       ]
 311
 312     for osparams in [{}, { "some": "value", "other": "Hello World", }]:
 313       for hvparams in [{}, { constants.HV_KERNEL_PATH: "/some/kernel", }]:
 314         for beparams in [{}, { constants.BE_MEMORY: 128, }]:
 315           for force in [False, True]:
 316             for nics in [[], [(0, { constants.INIC_IP: "192.0.2.1", })]]:
 317               for disks in test_disks:
 318                 for disk_template in constants.DISK_TEMPLATES:
 319                   data = {
 320                     "osparams": osparams,
 321                     "hvparams": hvparams,
 322                     "beparams": beparams,
 323                     "nics": nics,
 324                     "disks": disks,
 325                     "force": force,
 326                     "disk_template": disk_template,
 327                     }
 328
 329                   op = self.Parse(name, data)
 330                   self.assert_(isinstance(op, opcodes.OpSetInstanceParams))
 331                   self.assertEqual(op.instance_name, name)
 332                   self.assertEqual(op.hvparams, hvparams)
 333                   self.assertEqual(op.beparams, beparams)
 334                   self.assertEqual(op.osparams, osparams)
 335                   self.assertEqual(op.force, force)
 336                   self.assertEqual(op.nics, nics)
 337                   self.assertEqual(op.disks, disks)
 338                   self.assertEqual(op.disk_template, disk_template)
 339                   self.assert_(op.remote_node is None)
 340                   self.assert_(op.os_name is None)
 341                   self.assertFalse(op.force_variant)
 342
 343   def testDefaults(self):
 344     name = "instir8aish31"
 345
 346     op = self.Parse(name, {})
 347     self.assert_(isinstance(op, opcodes.OpSetInstanceParams))
 348     self.assertEqual(op.instance_name, name)
 349     self.assertEqual(op.hvparams, {})
 350     self.assertEqual(op.beparams, {})
 351     self.assertEqual(op.osparams, {})
 352     self.assertFalse(op.force)
 353     self.assertEqual(op.nics, [])
 354     self.assertEqual(op.disks, [])
 355     self.assert_(op.disk_template is None)
 356     self.assert_(op.remote_node is None)
 357     self.assert_(op.os_name is None)
 358     self.assertFalse(op.force_variant)
 359
 360
 361 class TestParseInstanceReinstallRequest(testutils.GanetiTestCase):
 362   def setUp(self):
 363     testutils.GanetiTestCase.setUp(self)
 364
 365     self.Parse = rlib2._ParseInstanceReinstallRequest
 366
 367   def _Check(self, ops, name):
 368     expcls = [
 369       opcodes.OpShutdownInstance,
 370       opcodes.OpReinstallInstance,
 371       opcodes.OpStartupInstance,
 372       ]
 373
 374     self.assert_(compat.all(isinstance(op, exp)
 375                             for op, exp in zip(ops, expcls)))
 376     self.assert_(compat.all(op.instance_name == name for op in ops))
 377
 378   def test(self):
 379     name = "shoo0tihohma"
 380
 381     ops = self.Parse(name, {"os": "sys1", "start": True,})
 382     self.assertEqual(len(ops), 3)
 383     self._Check(ops, name)
 384     self.assertEqual(ops[1].os_type, "sys1")
 385     self.assertFalse(ops[1].osparams)
 386
 387     ops = self.Parse(name, {"os": "sys2", "start": False,})
 388     self.assertEqual(len(ops), 2)
 389     self._Check(ops, name)
 390     self.assertEqual(ops[1].os_type, "sys2")
 391
 392     osparams = {
 393       "reformat": "1",
 394       }
 395     ops = self.Parse(name, {"os": "sys4035", "start": True,
 396                             "osparams": osparams,})
 397     self.assertEqual(len(ops), 3)
 398     self._Check(ops, name)
 399     self.assertEqual(ops[1].os_type, "sys4035")
 400     self.assertEqual(ops[1].osparams, osparams)
 401
 402   def testDefaults(self):
 403     name = "noolee0g"
 404
 405     ops = self.Parse(name, {"os": "linux1"})
 406     self.assertEqual(len(ops), 3)
 407     self._Check(ops, name)
 408     self.assertEqual(ops[1].os_type, "linux1")
 409     self.assertFalse(ops[1].osparams)
 410
 411
 412 if __name__ == '__main__':
 413   testutils.GanetiTestProgram()