Statistics
| Branch: | Tag: | Revision:

root / test / ganeti.rapi.rlib2_unittest.py @ 3882937a

History | View | Annotate | Download (11.2 kB)

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
if __name__ == '__main__':
362
  testutils.GanetiTestProgram()