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": "1.2.3.4", "mode": constants.NIC_MODE_ROUTED, },
  71         { "mode": constants.NIC_MODE_BRIDGED, "link": "n0", "bridge": "br1", },
  72       ],
  73
  74       # Unknown settings
  75       [{ "unknown": 999, }, { "foobar": "Hello World", }],
  76       ]
  77
  78     beparam_variants = [
  79       None,
  80       {},
  81       { constants.BE_VCPUS: 2, },
  82       { constants.BE_MEMORY: 123, },
  83       { constants.BE_VCPUS: 2,
  84         constants.BE_MEMORY: 1024,
  85         constants.BE_AUTO_BALANCE: True, }
  86       ]
  87
  88     hvparam_variants = [
  89       None,
  90       { constants.HV_BOOT_ORDER: "anc", },
  91       { constants.HV_KERNEL_PATH: "/boot/fookernel",
  92         constants.HV_ROOT_PATH: "/dev/hda1", },
  93       ]
  94
  95     for mode in [constants.INSTANCE_CREATE, constants.INSTANCE_IMPORT]:
  96       for nics in nic_variants:
  97         for disk_template in constants.DISK_TEMPLATES:
  98           for disks in disk_variants:
  99             for beparams in beparam_variants:
 100               for hvparams in hvparam_variants:
 101                 data = {
 102                   "name": "inst1.example.com",
 103                   "hypervisor": constants.HT_FAKE,
 104                   "disks": disks,
 105                   "nics": nics,
 106                   "mode": mode,
 107                   "disk_template": disk_template,
 108                   }
 109
 110                 if beparams is not None:
 111                   data["beparams"] = beparams
 112
 113                 if hvparams is not None:
 114                   data["hvparams"] = hvparams
 115
 116                 for dry_run in [False, True]:
 117                   op = self.Parse(data, dry_run)
 118                   self.assert_(isinstance(op, opcodes.OpCreateInstance))
 119                   self.assertEqual(op.mode, mode)
 120                   self.assertEqual(op.disk_template, disk_template)
 121                   self.assertEqual(op.dry_run, dry_run)
 122                   self.assertEqual(len(op.disks), len(disks))
 123                   self.assertEqual(len(op.nics), len(nics))
 124
 125                   self.assert_(compat.all(opdisk.get("size") ==
 126                                           disk.get("size") and
 127                                           opdisk.get("mode") ==
 128                                           disk.get("mode") and
 129                                           "unknown" not in opdisk
 130                                           for opdisk, disk in zip(op.disks,
 131                                                                   disks)))
 132
 133                   self.assert_(compat.all(opnic.get("size") ==
 134                                           nic.get("size") and
 135                                           opnic.get("mode") ==
 136                                           nic.get("mode") and
 137                                           "unknown" not in opnic and
 138                                           "foobar" not in opnic
 139                                           for opnic, nic in zip(op.nics, nics)))
 140
 141                   if beparams is None:
 142                     self.assertEqualValues(op.beparams, {})
 143                   else:
 144                     self.assertEqualValues(op.beparams, beparams)
 145
 146                   if hvparams is None:
 147                     self.assertEqualValues(op.hvparams, {})
 148                   else:
 149                     self.assertEqualValues(op.hvparams, hvparams)
 150
 151   def testErrors(self):
 152     # Test all required fields
 153     reqfields = {
 154       "name": "inst1.example.com",
 155       "disks": [],
 156       "nics": [],
 157       "mode": constants.INSTANCE_CREATE,
 158       "disk_template": constants.DT_PLAIN
 159       }
 160
 161     for name in reqfields.keys():
 162       self.assertRaises(http.HttpBadRequest, self.Parse,
 163                         dict(i for i in reqfields.iteritems() if i[0] != name),
 164                         False)
 165
 166     # Invalid disks and nics
 167     for field in ["disks", "nics"]:
 168       invalid_values = [None, 1, "", {}, [1, 2, 3], ["hda1", "hda2"]]
 169
 170       if field == "disks":
 171         invalid_values.append([
 172           # Disks without size
 173           {},
 174           { "mode": constants.DISK_RDWR, },
 175           ])
 176
 177       for invvalue in invalid_values:
 178         data = reqfields.copy()
 179         data[field] = invvalue
 180         self.assertRaises(http.HttpBadRequest, self.Parse, data, False)
 181
 182
 183 class TestParseExportInstanceRequest(testutils.GanetiTestCase):
 184   def setUp(self):
 185     testutils.GanetiTestCase.setUp(self)
 186
 187     self.Parse = rlib2._ParseExportInstanceRequest
 188
 189   def test(self):
 190     name = "instmoo"
 191     data = {
 192       "mode": constants.EXPORT_MODE_REMOTE,
 193       "destination": [(1, 2, 3), (99, 99, 99)],
 194       "shutdown": True,
 195       "remove_instance": True,
 196       "x509_key_name": ("name", "hash"),
 197       "destination_x509_ca": ("x", "y", "z"),
 198       }
 199     op = self.Parse(name, data)
 200     self.assert_(isinstance(op, opcodes.OpExportInstance))
 201     self.assertEqual(op.instance_name, name)
 202     self.assertEqual(op.mode, constants.EXPORT_MODE_REMOTE)
 203     self.assertEqual(op.shutdown, True)
 204     self.assertEqual(op.remove_instance, True)
 205     self.assertEqualValues(op.x509_key_name, ("name", "hash"))
 206     self.assertEqualValues(op.destination_x509_ca, ("x", "y", "z"))
 207
 208   def testDefaults(self):
 209     name = "inst1"
 210     data = {
 211       "destination": "node2",
 212       "shutdown": False,
 213       }
 214     op = self.Parse(name, data)
 215     self.assert_(isinstance(op, opcodes.OpExportInstance))
 216     self.assertEqual(op.instance_name, name)
 217     self.assertEqual(op.mode, constants.EXPORT_MODE_LOCAL)
 218     self.assertEqual(op.remove_instance, False)
 219
 220   def testErrors(self):
 221     self.assertRaises(http.HttpBadRequest, self.Parse, "err1",
 222                       { "remove_instance": "True", })
 223     self.assertRaises(http.HttpBadRequest, self.Parse, "err1",
 224                       { "remove_instance": "False", })
 225
 226
 227 if __name__ == '__main__':
 228   testutils.GanetiTestProgram()