Statistics
| Branch: | Tag: | Revision:

root / api / tests.py @ e646ebe5

History | View | Annotate | Download (21.1 kB)

1
#
2
# Unit Tests for api
3
#
4
# Provides automated tests for api module
5
#
6
# Copyright 2011 Greek Research and Technology Network
7
#
8

    
9
import datetime
10
from django.utils import simplejson as json
11
from django.test import TestCase
12
from django.test.client import Client
13
from synnefo.db.models import VirtualMachine, VirtualMachineGroup
14
from synnefo.db.models import Flavor, Image
15
from synnefo.api.tests_redux import APIReduxTestCase
16

    
17

    
18
class APITestCase(TestCase):
19
    fixtures = ['api_test_data', ]
20
    test_server_id = 1001
21
    test_image_id = 1
22
    test_flavor_id = 1
23
    test_group_id = 1
24
    test_wrong_server_id = 99999999
25
    test_wrong_image_id = 99999999
26
    test_wrong_flavor_id = 99999999
27
    test_wrong_group_id = 99999999
28
    #make the testing with these id's
29

    
30
    def setUp(self):
31
        self.client = Client()
32

    
33
    def test_api_version(self):
34
        """ check rackspace cloud servers API version
35
        """
36
        response = self.client.get('/api/v1.0/')
37
        # Check that the response is 200 OK.
38
        self.assertEqual(response.status_code, 200)
39
        api_version = json.loads(response.content)['version']
40
        self.assertEqual(api_version['status'], 'CURRENT')
41
        self.assertEqual(api_version['wadl'],
42
            'http://docs.rackspacecloud.com/servers/api/v1.0/application.wadl')
43
        self.assertEqual(api_version['docURL'],
44
    'http://docs.rackspacecloud.com/servers/api/v1.0/cs-devguide-20110112.pdf')
45
        self.assertEqual(api_version['id'], 'v1.0')
46

    
47
    def test_server_list(self):
48
        """ test if the expected list of servers is returned by the API
49
        """
50
        response = self.client.get('/api/v1.0/servers')
51
        vms_from_api = json.loads(response.content)['servers']
52
        vms_from_db = VirtualMachine.objects.filter(deleted=False)
53
        self.assertEqual(len(vms_from_api), len(vms_from_db))
54
        self.assertTrue(response.status_code in [200, 203])
55
        for vm_from_api in vms_from_api:
56
            vm_from_db = VirtualMachine.objects.get(id=vm_from_api['id'])
57
            self.assertEqual(vm_from_api['id'], vm_from_db.id)
58
            self.assertEqual(vm_from_api['name'], vm_from_db.name)
59

    
60
    def test_server_details(self):
61
        """ test if the expected server is returned by the API
62
        """
63
        response = self.client.get('/api/v1.0/servers/' +
64
                                   str(self.test_server_id))
65
        vm_from_api = json.loads(response.content)['server']
66
        vm_from_db = VirtualMachine.objects.get(id=self.test_server_id)
67
        self.assertEqual(vm_from_api['flavorRef'], vm_from_db.flavor.id)
68
        self.assertEqual(vm_from_api['hostId'], vm_from_db.hostid)
69
        self.assertEqual(vm_from_api['id'], vm_from_db.id)
70
        self.assertEqual(vm_from_api['imageRef'], vm_from_db.flavor.id)
71
        self.assertEqual(vm_from_api['name'], vm_from_db.name)
72
        self.assertEqual(vm_from_api['status'], vm_from_db.rsapi_state)
73
        self.assertTrue(response.status_code in [200, 203])
74

    
75
    def test_servers_details(self):
76
        """ test if the servers details are returned by the API
77
        """
78
        response = self.client.get('/api/v1.0/servers/detail')
79
        vms_from_db = VirtualMachine.objects.filter(deleted=False)
80
        id_list = [vm.id for vm in vms_from_db]
81
        number = 0
82
        for vm_id in id_list:
83
            vm_from_api = json.loads(response.content)['servers'][number]
84
            vm_from_db = VirtualMachine.objects.get(id=vm_id)
85
            self.assertEqual(vm_from_api['flavorRef'], vm_from_db.flavor.id)
86
            self.assertEqual(vm_from_api['hostId'], vm_from_db.hostid)
87
            self.assertEqual(vm_from_api['id'], vm_from_db.id)
88
            self.assertEqual(vm_from_api['imageRef'], vm_from_db.flavor.id)
89
            self.assertEqual(vm_from_api['name'], vm_from_db.name)
90
            self.assertEqual(vm_from_api['status'], vm_from_db.rsapi_state)
91
            number += 1
92
        vms_from_api = json.loads(response.content)['servers']
93
        for vm_from_api in vms_from_api:
94
            vm_from_db = VirtualMachine.objects.get(id=vm_from_api['id'])
95
            self.assertEqual(vm_from_api['flavorRef'], vm_from_db.flavor.id)
96
            self.assertEqual(vm_from_api['hostId'], vm_from_db.hostid)
97
            self.assertEqual(vm_from_api['id'], vm_from_db.id)
98
            self.assertEqual(vm_from_api['imageRef'], vm_from_db.flavor.id)
99
            self.assertEqual(vm_from_api['name'], vm_from_db.name)
100
            self.assertEqual(vm_from_api['status'], vm_from_db.rsapi_state)
101
        self.assertTrue(response.status_code in [200, 203])
102

    
103
    def test_wrong_server(self):
104
        """ test 404 response if server does not exist
105
        """
106
        response = self.client.get('/api/v1.0/servers/' +
107
                                   str(self.test_wrong_server_id))
108
        self.assertEqual(response.status_code, 404)
109

    
110
    def test_create_server_empty(self):
111
        """ test if the create server call returns a 400 badRequest if no
112
            attributes are specified
113
        """
114
        response = self.client.post('/api/v1.0/servers', {})
115
        self.assertEqual(response.status_code, 400)
116

    
117
    def test_create_server(self):
118
        """ test if the create server call returns the expected response
119
            if a valid request has been speficied
120
        """
121
        request = {
122
                    "server": {
123
                        "name": "new-server-test",
124
                        "imageRef": 1,
125
                        "flavorRef": 1,
126
                        "metadata": {
127
                            "My Server Name": "Apache1"
128
                        },
129
                        "personality": []
130
                    }
131
        }
132
        response = self.client.post('/api/v1.0/servers',
133
                                    json.dumps(request),
134
                                    content_type='application/json')
135
        self.assertEqual(response.status_code, 202)
136
        #TODO: check response.content
137
        #TODO: check create server with wrong options (eg non existing flavor)
138

    
139
    def test_server_polling(self):
140
        """ test if the server polling works as expected
141
        """
142
        response = self.client.get('/api/v1.0/servers/detail')
143
        vms_from_api_initial = json.loads(response.content)['servers']
144
        then = datetime.datetime.now().isoformat().split('.')[0]
145

    
146
        #isoformat also gives miliseconds that are not needed
147
        response = self.client.get('/api/v1.0/servers/detail?changes-since=%s'
148
                                   % then)
149
        self.assertEqual(len(response.content), 0)
150
        #no changes were made
151

    
152
        #now create a machine. Then check if it is on the list
153
        request = {
154
                    "server": {
155
                        "name": "new-server-test",
156
                        "imageRef": 1,
157
                        "flavorRef": 1,
158
                        "metadata": {
159
                            "My Server Name": "Apache1"
160
                        },
161
                        "personality": []
162
                    }
163
        }
164
        response = self.client.post('/api/v1.0/servers',
165
                                    json.dumps(request),
166
                                    content_type='application/json')
167
        self.assertEqual(response.status_code, 202)
168

    
169
        response = self.client.get('/api/v1.0/servers/detail?changes-since=%s'
170
                                   % then)
171
        vms_from_api_after = json.loads(response.content)['servers']
172
        #make sure the newly created server is included on the updated list
173
        self.assertEqual(len(vms_from_api_after), 1)
174

    
175
    def test_reboot_server(self):
176
        """ test if the specified server is rebooted
177
        """
178
        request = {
179
            "reboot": '{"type" : "HARD"}'
180
            }
181
        response = self.client.post('/api/v1.0/servers/' +
182
                                    str(self.test_server_id) + '/action',
183
                                    json.dumps(request),
184
                                    content_type='application/json')
185
        self.assertEqual(response.status_code, 202)
186
        #server id that does not exist
187
        response = self.client.post('/api/v1.0/servers/' +
188
                                    str(self.test_wrong_server_id) + '/action',
189
                                   json.dumps(request),
190
                                   content_type='application/json')
191
        self.assertEqual(response.status_code, 404)
192

    
193
    def test_shutdown_server(self):
194
        """ test if the specified server is shutdown
195
        """
196
        request = {
197
            "shutdown": {"timeout": "5"}
198
            }
199
        response = self.client.post('/api/v1.0/servers/' +
200
                                    str(self.test_server_id) + '/action',
201
                                    json.dumps(request),
202
                                    content_type='application/json')
203
        self.assertEqual(response.status_code, 202)
204
        #server id that does not exist
205
        response = self.client.post('/api/v1.0/servers/' +
206
                                    str(self.test_wrong_server_id) + '/action',
207
                                    json.dumps(request),
208
                                    content_type='application/json')
209
        self.assertEqual(response.status_code, 404)
210

    
211
    def test_start_server(self):
212
        """ test if the specified server is started
213
        """
214
        request = {
215
            "start": {"type": "NORMAL"}
216
            }
217
        response = self.client.post('/api/v1.0/servers/' +
218
                                    str(self.test_server_id) + '/action',
219
                                    json.dumps(request),
220
                                    content_type='application/json')
221
        self.assertEqual(response.status_code, 202)
222
        #server id that does not exist
223
        response = self.client.post('/api/v1.0/servers/' +
224
                                    str(self.test_wrong_server_id) + '/action',
225
                                    json.dumps(request),
226
                                    content_type='application/json')
227
        self.assertEqual(response.status_code, 404)
228

    
229
    def test_delete_server(self):
230
        """ test if the specified server is deleted
231
        """
232
        response = self.client.delete('/api/v1.0/servers/' +
233
                                      str(self.test_server_id))
234
        self.assertEqual(response.status_code, 202)
235
        #server id that does not exist
236
        response = self.client.delete('/api/v1.0/servers/' +
237
                                      str(self.test_wrong_server_id))
238
        self.assertEqual(response.status_code, 404)
239

    
240
    def test_flavor_list(self):
241
        """ test if the expected list of flavors is returned by the API
242
        """
243
        response = self.client.get('/api/v1.0/flavors')
244
        flavors_from_api = json.loads(response.content)['flavors']
245
        flavors_from_db = Flavor.objects.all()
246
        self.assertEqual(len(flavors_from_api), len(flavors_from_db))
247
        self.assertTrue(response.status_code in [200, 203])
248
        for flavor_from_api in flavors_from_api:
249
            flavor_from_db = Flavor.objects.get(id=flavor_from_api['id'])
250
            self.assertEqual(flavor_from_api['id'], flavor_from_db.id)
251
            self.assertEqual(flavor_from_api['name'], flavor_from_db.name)
252

    
253
    def test_flavors_details(self):
254
        """ test if the flavors details are returned by the API
255
        """
256
        response = self.client.get('/api/v1.0/flavors/detail')
257
        flavors_from_db = Flavor.objects.all()
258
        flavors_from_api = json.loads(response.content)['flavors']
259

    
260
        # Assert that all flavors in the db appear inthe API call result
261
        for i in range(0, len(flavors_from_db)):
262
            flavor_from_api = flavors_from_api[i]
263
            flavor_from_db = Flavor.objects.get(id=flavors_from_db[i].id)
264
            self.assertEqual(flavor_from_api['cpu'], flavor_from_db.cpu)
265
            self.assertEqual(flavor_from_api['id'], flavor_from_db.id)
266
            self.assertEqual(flavor_from_api['disk'], flavor_from_db.disk)
267
            self.assertEqual(flavor_from_api['name'], flavor_from_db.name)
268
            self.assertEqual(flavor_from_api['ram'], flavor_from_db.ram)
269

    
270
        # Assert that all flavors returned by the API also exist in the db
271
        for flavor_from_api in flavors_from_api:
272
            flavor_from_db = Flavor.objects.get(id=flavor_from_api['id'])
273
            self.assertEqual(flavor_from_api['cpu'], flavor_from_db.cpu)
274
            self.assertEqual(flavor_from_api['id'], flavor_from_db.id)
275
            self.assertEqual(flavor_from_api['disk'], flavor_from_db.disk)
276
            self.assertEqual(flavor_from_api['name'], flavor_from_db.name)
277
            self.assertEqual(flavor_from_api['ram'], flavor_from_db.ram)
278

    
279
        # Check if we have the right status_code
280
        self.assertTrue(response.status_code in [200, 203])
281

    
282
    def test_flavor_details(self):
283
        """ test if the expected flavor is returned by the API
284
        """
285
        response = self.client.get('/api/v1.0/flavors/' +
286
                                   str(self.test_flavor_id))
287
        flavor_from_api = json.loads(response.content)['flavor']
288
        flavor_from_db = Flavor.objects.get(id=self.test_flavor_id)
289
        self.assertEqual(flavor_from_api['cpu'], flavor_from_db.cpu)
290
        self.assertEqual(flavor_from_api['id'], flavor_from_db.id)
291
        self.assertEqual(flavor_from_api['disk'], flavor_from_db.disk)
292
        self.assertEqual(flavor_from_api['name'], flavor_from_db.name)
293
        self.assertEqual(flavor_from_api['ram'], flavor_from_db.ram)
294
        self.assertTrue(response.status_code in [200, 203])
295

    
296
    def test_wrong_flavor(self):
297
        """ test 404 result when requesting a flavor that does not exist
298
        """
299
        response = self.client.get('/api/v1.0/flavors/' +
300
                                   str(self.test_wrong_flavor_id))
301
        self.assertTrue(response.status_code in [404, 503])
302

    
303
    def test_image_list(self):
304
        """ test if the expected list of images is returned by the API
305
        """
306
        response = self.client.get('/api/v1.0/images')
307
        images_from_api = json.loads(response.content)['images']
308
        images_from_db = Image.objects.all()
309
        self.assertEqual(len(images_from_api), len(images_from_db))
310
        self.assertTrue(response.status_code in [200, 203])
311
        for image_from_api in images_from_api:
312
            image_from_db = Image.objects.get(id=image_from_api['id'])
313
            self.assertEqual(image_from_api['id'], image_from_db.id)
314
            self.assertEqual(image_from_api['name'], image_from_db.name)
315

    
316
    def test_image_details(self):
317
        """ test if the expected image is returned by the API
318
        """
319
        response = self.client.get('/api/v1.0/images/' +
320
                                   str(self.test_image_id))
321
        image_from_api = json.loads(response.content)['image']
322
        image_from_db = Image.objects.get(id=self.test_image_id)
323
        self.assertEqual(image_from_api['name'], image_from_db.name)
324
        self.assertEqual(image_from_api['id'], image_from_db.id)
325
        self.assertEqual(image_from_api['serverId'],
326
                        image_from_db.sourcevm and image_from_db.sourcevm.id or
327
                        "")
328
        self.assertEqual(image_from_api['size'], image_from_db.size)
329
        self.assertEqual(image_from_api['status'], image_from_db.state)
330
        self.assertEqual(image_from_api['metadata']['meta']['key']
331
                         ['description'],
332
                         image_from_db.description)
333
        self.assertTrue(response.status_code in [200, 203])
334

    
335
    def test_images_details(self):
336
        """ test if the images details are returned by the API
337
        """
338
        response = self.client.get('/api/v1.0/images/detail')
339
        images_from_api = json.loads(response.content)['images']
340
        images_from_db = Image.objects.all()
341
        for i in range(0, len(images_from_db)):
342
            image_from_db = Image.objects.get(id=images_from_db[i].id)
343
            image_from_api = images_from_api[i]
344
            self.assertEqual(image_from_api['name'], image_from_db.name)
345
            self.assertEqual(image_from_api['id'], image_from_db.id)
346
            self.assertEqual(image_from_api['serverId'],
347
                             image_from_db.sourcevm and
348
                             image_from_db.sourcevm.id or "")
349
            self.assertEqual(image_from_api['size'], image_from_db.size)
350
            self.assertEqual(image_from_api['status'], image_from_db.state)
351
            self.assertEqual(image_from_api['metadata']['meta']['key']
352
                             ['description'],
353
                             image_from_db.description)
354

    
355
        for image_from_api in images_from_api:
356
            image_from_db = Image.objects.get(id=image_from_api['id'])
357
            self.assertEqual(image_from_api['name'], image_from_db.name)
358
            self.assertEqual(image_from_api['id'], image_from_db.id)
359
            self.assertEqual(image_from_api['serverId'],
360
                             image_from_db.sourcevm and
361
                             image_from_db.sourcevm.id or "")
362
            self.assertEqual(image_from_api['size'], image_from_db.size)
363
            self.assertEqual(image_from_api['status'], image_from_db.state)
364
            self.assertEqual(image_from_api['metadata']['meta']['key']
365
                             ['description'],
366
                             image_from_db.description)
367

    
368
        self.assertTrue(response.status_code in [200, 203])
369

    
370
    def test_wrong_image(self):
371
        """ test 404 result if a non existent image is requested
372
        """
373
        response = self.client.get('/api/v1.0/images/' +
374
                                   str(self.test_wrong_image_id))
375
        self.assertEqual(response.status_code, 404)
376

    
377
    def test_server_metadata(self):
378
        """ test server's metadata (add, edit)
379
        """
380
        request = {
381
            "metadata": {
382
                "metadata_key": "name",
383
                "metadata_value": "a fancy name"
384
                }
385
            }
386
        response = self.client.put('/api/v1.0/servers' +
387
                                    str(self.test_server_id),
388
                                    json.dumps(request),
389
                                    content_type='application/json')
390
        self.assertEqual(response.status_code, 404)
391
        #TODO: not working atm, due to problem with django piston and PUT
392

    
393
    def test_vm_group_list(self):
394
        """ test if the expected list of groups is returned by the API
395
        """
396
        response = self.client.get('/api/v1.0/groups')
397
        groups_from_api = json.loads(response.content)['groups']
398
        groups_from_db = VirtualMachineGroup.objects.all()
399
        self.assertEqual(len(groups_from_api), len(groups_from_db))
400
        self.assertTrue(response.status_code in [200, 203])
401
        for group_from_api in groups_from_api:
402
            group_from_db = VirtualMachineGroup.objects.get(
403
                id=group_from_api['id']
404
                )
405
            self.assertEqual(group_from_api['id'], group_from_db.id)
406
            self.assertEqual(group_from_api['name'], group_from_db.name)
407

    
408
    def test_vm_group_details(self):
409
        """ test if the expected virtual machine group is returned by the API
410
        """
411
        response = self.client.get('/api/v1.0/groups/' +
412
                                   str(self.test_group_id))
413
        group_from_api = json.loads(response.content)['group']
414
        group_from_db = VirtualMachineGroup.objects.get(id=self.test_group_id)
415
        self.assertEqual(group_from_api['name'], group_from_db.name)
416
        self.assertEqual(group_from_api['id'], group_from_db.id)
417
        self.assertEqual(group_from_api['server_id'],
418
                         [machine.id
419
                          for machine in group_from_db.machines.all()])
420
        self.assertTrue(response.status_code in [200, 203])
421

    
422
    def test_wrong_vm_group(self):
423
        """ test 404 result if a non existent VM group is requested
424
        """
425
        response = self.client.get('/api/v1.0/groups/' +
426
                                   str(self.test_wrong_group_id))
427
        self.assertEqual(response.status_code, 404)
428

    
429
    def test_groups_details(self):
430
        """ test if the groups details are returned by the API
431
        """
432
        response = self.client.get('/api/v1.0/groups/detail')
433
        groups_from_api = json.loads(response.content)['groups']
434
        groups_from_db = VirtualMachineGroup.objects.all()
435
        for i in range(0, len(groups_from_db)):
436
            group_from_db = VirtualMachineGroup.objects.get(
437
                id=groups_from_db[i].id)
438
            group_from_api = groups_from_api[i]
439
            self.assertEqual(group_from_api['name'], group_from_db.name)
440
            self.assertEqual(group_from_api['id'], group_from_db.id)
441
            self.assertEqual(group_from_api['server_id'],
442
                             [machine.id
443
                              for machine in group_from_db.machines.all()])
444
        for group_from_api in groups_from_api:
445
            group_from_db = VirtualMachineGroup.objects.get(
446
                id=group_from_api['id'])
447
            self.assertEqual(group_from_api['name'], group_from_db.name)
448
            self.assertEqual(group_from_api['id'], group_from_db.id)
449
            self.assertEqual(group_from_api['server_id'],
450
                             [machine.id
451
                              for machine in group_from_db.machines.all()])
452
        self.assertTrue(response.status_code in [200, 203])