root / api / tests.py @ 25380811
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 |
from synnefo.logic import utils |
18 |
|
19 |
class APITestCase(TestCase): |
20 |
fixtures = ['api_test_data', ]
|
21 |
test_server_id = 1001
|
22 |
test_image_id = 1
|
23 |
test_flavor_id = 1
|
24 |
test_group_id = 1
|
25 |
test_wrong_server_id = 99999999
|
26 |
test_wrong_image_id = 99999999
|
27 |
test_wrong_flavor_id = 99999999
|
28 |
test_wrong_group_id = 99999999
|
29 |
#make the testing with these id's
|
30 |
|
31 |
def setUp(self): |
32 |
self.client = Client()
|
33 |
|
34 |
def test_api_version(self): |
35 |
""" check rackspace cloud servers API version
|
36 |
"""
|
37 |
response = self.client.get('/api/v1.0/') |
38 |
# Check that the response is 200 OK.
|
39 |
self.assertEqual(response.status_code, 200) |
40 |
api_version = json.loads(response.content)['version']
|
41 |
self.assertEqual(api_version['status'], 'CURRENT') |
42 |
self.assertEqual(api_version['wadl'], |
43 |
'http://docs.rackspacecloud.com/servers/api/v1.0/application.wadl')
|
44 |
self.assertEqual(api_version['docURL'], |
45 |
'http://docs.rackspacecloud.com/servers/api/v1.0/cs-devguide-20110112.pdf')
|
46 |
self.assertEqual(api_version['id'], 'v1.0') |
47 |
|
48 |
def test_server_list(self): |
49 |
""" test if the expected list of servers is returned by the API
|
50 |
"""
|
51 |
response = self.client.get('/api/v1.0/servers') |
52 |
vms_from_api = json.loads(response.content)['servers']
|
53 |
vms_from_db = VirtualMachine.objects.filter(deleted=False)
|
54 |
self.assertEqual(len(vms_from_api), len(vms_from_db)) |
55 |
self.assertTrue(response.status_code in [200, 203]) |
56 |
for vm_from_api in vms_from_api: |
57 |
vm_from_db = VirtualMachine.objects.get(id=vm_from_api['id'])
|
58 |
self.assertEqual(vm_from_api['id'], vm_from_db.id) |
59 |
self.assertEqual(vm_from_api['name'], vm_from_db.name) |
60 |
|
61 |
def test_server_details(self): |
62 |
""" test if the expected server is returned by the API
|
63 |
"""
|
64 |
response = self.client.get('/api/v1.0/servers/' + |
65 |
str(self.test_server_id)) |
66 |
vm_from_api = json.loads(response.content)['server']
|
67 |
vm_from_db = VirtualMachine.objects.get(id=self.test_server_id)
|
68 |
self.assertEqual(vm_from_api['flavorRef'], vm_from_db.flavor.id) |
69 |
self.assertEqual(vm_from_api['hostId'], vm_from_db.hostid) |
70 |
self.assertEqual(vm_from_api['id'], vm_from_db.id) |
71 |
self.assertEqual(vm_from_api['imageRef'], vm_from_db.flavor.id) |
72 |
self.assertEqual(vm_from_api['name'], vm_from_db.name) |
73 |
self.assertEqual(vm_from_api['status'], utils.get_rsapi_state(vm_from_db)) |
74 |
self.assertTrue(response.status_code in [200,203]) |
75 |
|
76 |
|
77 |
def test_servers_details(self): |
78 |
""" test if the servers details are returned by the API
|
79 |
"""
|
80 |
response = self.client.get('/api/v1.0/servers/detail') |
81 |
vms_from_db = VirtualMachine.objects.filter(deleted=False)
|
82 |
id_list = [vm.id for vm in vms_from_db] |
83 |
number = 0
|
84 |
for vm_id in id_list: |
85 |
vm_from_api = json.loads(response.content)['servers'][number]
|
86 |
vm_from_db = VirtualMachine.objects.get(id=vm_id) |
87 |
self.assertEqual(vm_from_api['flavorRef'], vm_from_db.flavor.id) |
88 |
self.assertEqual(vm_from_api['hostId'], vm_from_db.hostid) |
89 |
self.assertEqual(vm_from_api['id'], vm_from_db.id) |
90 |
self.assertEqual(vm_from_api['imageRef'], vm_from_db.flavor.id) |
91 |
self.assertEqual(vm_from_api['name'], vm_from_db.name) |
92 |
self.assertEqual(vm_from_api['status'], utils.get_rsapi_state(vm_from_db)) |
93 |
number += 1
|
94 |
vms_from_api = json.loads(response.content)['servers']
|
95 |
for vm_from_api in vms_from_api: |
96 |
vm_from_db = VirtualMachine.objects.get(id=vm_from_api['id'])
|
97 |
self.assertEqual(vm_from_api['flavorRef'], vm_from_db.flavor.id) |
98 |
self.assertEqual(vm_from_api['hostId'], vm_from_db.hostid) |
99 |
self.assertEqual(vm_from_api['id'], vm_from_db.id) |
100 |
self.assertEqual(vm_from_api['imageRef'], vm_from_db.flavor.id) |
101 |
self.assertEqual(vm_from_api['name'], vm_from_db.name) |
102 |
self.assertEqual(vm_from_api['status'], utils.get_rsapi_state(vm_from_db)) |
103 |
self.assertTrue(response.status_code in [200,203]) |
104 |
|
105 |
|
106 |
def test_wrong_server(self): |
107 |
""" test 404 response if server does not exist
|
108 |
"""
|
109 |
response = self.client.get('/api/v1.0/servers/' + |
110 |
str(self.test_wrong_server_id)) |
111 |
self.assertEqual(response.status_code, 404) |
112 |
|
113 |
def test_create_server_empty(self): |
114 |
""" test if the create server call returns a 400 badRequest if no
|
115 |
attributes are specified
|
116 |
"""
|
117 |
response = self.client.post('/api/v1.0/servers', {}) |
118 |
self.assertEqual(response.status_code, 400) |
119 |
|
120 |
def test_create_server(self): |
121 |
""" test if the create server call returns the expected response
|
122 |
if a valid request has been speficied
|
123 |
"""
|
124 |
request = { |
125 |
"server": {
|
126 |
"name": "new-server-test", |
127 |
"imageRef": 1, |
128 |
"flavorRef": 1, |
129 |
"metadata": {
|
130 |
"My Server Name": "Apache1" |
131 |
}, |
132 |
"personality": []
|
133 |
} |
134 |
} |
135 |
response = self.client.post('/api/v1.0/servers', |
136 |
json.dumps(request), |
137 |
content_type='application/json')
|
138 |
self.assertEqual(response.status_code, 202) |
139 |
#TODO: check response.content
|
140 |
#TODO: check create server with wrong options (eg non existing flavor)
|
141 |
|
142 |
def test_server_polling(self): |
143 |
""" test if the server polling works as expected
|
144 |
"""
|
145 |
response = self.client.get('/api/v1.0/servers/detail') |
146 |
vms_from_api_initial = json.loads(response.content)['servers']
|
147 |
then = datetime.datetime.now().isoformat().split('.')[0] |
148 |
|
149 |
#isoformat also gives miliseconds that are not needed
|
150 |
response = self.client.get('/api/v1.0/servers/detail?changes-since=%s' |
151 |
% then) |
152 |
self.assertEqual(len(response.content), 0) |
153 |
#no changes were made
|
154 |
|
155 |
#now create a machine. Then check if it is on the list
|
156 |
request = { |
157 |
"server": {
|
158 |
"name": "new-server-test", |
159 |
"imageRef": 1, |
160 |
"flavorRef": 1, |
161 |
"metadata": {
|
162 |
"My Server Name": "Apache1" |
163 |
}, |
164 |
"personality": []
|
165 |
} |
166 |
} |
167 |
response = self.client.post('/api/v1.0/servers', |
168 |
json.dumps(request), |
169 |
content_type='application/json')
|
170 |
self.assertEqual(response.status_code, 202) |
171 |
|
172 |
response = self.client.get('/api/v1.0/servers/detail?changes-since=%s' |
173 |
% then) |
174 |
vms_from_api_after = json.loads(response.content)['servers']
|
175 |
#make sure the newly created server is included on the updated list
|
176 |
self.assertEqual(len(vms_from_api_after), 1) |
177 |
|
178 |
def test_reboot_server(self): |
179 |
""" test if the specified server is rebooted
|
180 |
"""
|
181 |
request = { |
182 |
"reboot": '{"type" : "HARD"}' |
183 |
} |
184 |
response = self.client.post('/api/v1.0/servers/' + |
185 |
str(self.test_server_id) + '/action', |
186 |
json.dumps(request), |
187 |
content_type='application/json')
|
188 |
self.assertEqual(response.status_code, 202) |
189 |
#server id that does not exist
|
190 |
response = self.client.post('/api/v1.0/servers/' + |
191 |
str(self.test_wrong_server_id) + '/action', |
192 |
json.dumps(request), |
193 |
content_type='application/json')
|
194 |
self.assertEqual(response.status_code, 404) |
195 |
|
196 |
def test_shutdown_server(self): |
197 |
""" test if the specified server is shutdown
|
198 |
"""
|
199 |
request = { |
200 |
"shutdown": {"timeout": "5"} |
201 |
} |
202 |
response = self.client.post('/api/v1.0/servers/' + |
203 |
str(self.test_server_id) + '/action', |
204 |
json.dumps(request), |
205 |
content_type='application/json')
|
206 |
self.assertEqual(response.status_code, 202) |
207 |
#server id that does not exist
|
208 |
response = self.client.post('/api/v1.0/servers/' + |
209 |
str(self.test_wrong_server_id) + '/action', |
210 |
json.dumps(request), |
211 |
content_type='application/json')
|
212 |
self.assertEqual(response.status_code, 404) |
213 |
|
214 |
def test_start_server(self): |
215 |
""" test if the specified server is started
|
216 |
"""
|
217 |
request = { |
218 |
"start": {"type": "NORMAL"} |
219 |
} |
220 |
response = self.client.post('/api/v1.0/servers/' + |
221 |
str(self.test_server_id) + '/action', |
222 |
json.dumps(request), |
223 |
content_type='application/json')
|
224 |
self.assertEqual(response.status_code, 202) |
225 |
#server id that does not exist
|
226 |
response = self.client.post('/api/v1.0/servers/' + |
227 |
str(self.test_wrong_server_id) + '/action', |
228 |
json.dumps(request), |
229 |
content_type='application/json')
|
230 |
self.assertEqual(response.status_code, 404) |
231 |
|
232 |
def test_delete_server(self): |
233 |
""" test if the specified server is deleted
|
234 |
"""
|
235 |
response = self.client.delete('/api/v1.0/servers/' + |
236 |
str(self.test_server_id)) |
237 |
self.assertEqual(response.status_code, 202) |
238 |
#server id that does not exist
|
239 |
response = self.client.delete('/api/v1.0/servers/' + |
240 |
str(self.test_wrong_server_id)) |
241 |
self.assertEqual(response.status_code, 404) |
242 |
|
243 |
def test_flavor_list(self): |
244 |
""" test if the expected list of flavors is returned by the API
|
245 |
"""
|
246 |
response = self.client.get('/api/v1.0/flavors') |
247 |
flavors_from_api = json.loads(response.content)['flavors']
|
248 |
flavors_from_db = Flavor.objects.all() |
249 |
self.assertEqual(len(flavors_from_api), len(flavors_from_db)) |
250 |
self.assertTrue(response.status_code in [200, 203]) |
251 |
for flavor_from_api in flavors_from_api: |
252 |
flavor_from_db = Flavor.objects.get(id=flavor_from_api['id'])
|
253 |
self.assertEqual(flavor_from_api['id'], flavor_from_db.id) |
254 |
self.assertEqual(flavor_from_api['name'], flavor_from_db.name) |
255 |
|
256 |
def test_flavors_details(self): |
257 |
""" test if the flavors details are returned by the API
|
258 |
"""
|
259 |
response = self.client.get('/api/v1.0/flavors/detail') |
260 |
flavors_from_db = Flavor.objects.all() |
261 |
flavors_from_api = json.loads(response.content)['flavors']
|
262 |
|
263 |
# Assert that all flavors in the db appear inthe API call result
|
264 |
for i in range(0, len(flavors_from_db)): |
265 |
flavor_from_api = flavors_from_api[i] |
266 |
flavor_from_db = Flavor.objects.get(id=flavors_from_db[i].id) |
267 |
self.assertEqual(flavor_from_api['cpu'], flavor_from_db.cpu) |
268 |
self.assertEqual(flavor_from_api['id'], flavor_from_db.id) |
269 |
self.assertEqual(flavor_from_api['disk'], flavor_from_db.disk) |
270 |
self.assertEqual(flavor_from_api['name'], flavor_from_db.name) |
271 |
self.assertEqual(flavor_from_api['ram'], flavor_from_db.ram) |
272 |
|
273 |
# Assert that all flavors returned by the API also exist in the db
|
274 |
for flavor_from_api in flavors_from_api: |
275 |
flavor_from_db = Flavor.objects.get(id=flavor_from_api['id'])
|
276 |
self.assertEqual(flavor_from_api['cpu'], flavor_from_db.cpu) |
277 |
self.assertEqual(flavor_from_api['id'], flavor_from_db.id) |
278 |
self.assertEqual(flavor_from_api['disk'], flavor_from_db.disk) |
279 |
self.assertEqual(flavor_from_api['name'], flavor_from_db.name) |
280 |
self.assertEqual(flavor_from_api['ram'], flavor_from_db.ram) |
281 |
|
282 |
# Check if we have the right status_code
|
283 |
self.assertTrue(response.status_code in [200, 203]) |
284 |
|
285 |
def test_flavor_details(self): |
286 |
""" test if the expected flavor is returned by the API
|
287 |
"""
|
288 |
response = self.client.get('/api/v1.0/flavors/' + |
289 |
str(self.test_flavor_id)) |
290 |
flavor_from_api = json.loads(response.content)['flavor']
|
291 |
flavor_from_db = Flavor.objects.get(id=self.test_flavor_id)
|
292 |
self.assertEqual(flavor_from_api['cpu'], flavor_from_db.cpu) |
293 |
self.assertEqual(flavor_from_api['id'], flavor_from_db.id) |
294 |
self.assertEqual(flavor_from_api['disk'], flavor_from_db.disk) |
295 |
self.assertEqual(flavor_from_api['name'], flavor_from_db.name) |
296 |
self.assertEqual(flavor_from_api['ram'], flavor_from_db.ram) |
297 |
self.assertTrue(response.status_code in [200, 203]) |
298 |
|
299 |
def test_wrong_flavor(self): |
300 |
""" test 404 result when requesting a flavor that does not exist
|
301 |
"""
|
302 |
response = self.client.get('/api/v1.0/flavors/' + |
303 |
str(self.test_wrong_flavor_id)) |
304 |
self.assertTrue(response.status_code in [404, 503]) |
305 |
|
306 |
def test_image_list(self): |
307 |
""" test if the expected list of images is returned by the API
|
308 |
"""
|
309 |
response = self.client.get('/api/v1.0/images') |
310 |
images_from_api = json.loads(response.content)['images']
|
311 |
images_from_db = Image.objects.all() |
312 |
self.assertEqual(len(images_from_api), len(images_from_db)) |
313 |
self.assertTrue(response.status_code in [200, 203]) |
314 |
for image_from_api in images_from_api: |
315 |
image_from_db = Image.objects.get(id=image_from_api['id'])
|
316 |
self.assertEqual(image_from_api['id'], image_from_db.id) |
317 |
self.assertEqual(image_from_api['name'], image_from_db.name) |
318 |
|
319 |
def test_image_details(self): |
320 |
""" test if the expected image is returned by the API
|
321 |
"""
|
322 |
response = self.client.get('/api/v1.0/images/' + |
323 |
str(self.test_image_id)) |
324 |
image_from_api = json.loads(response.content)['image']
|
325 |
image_from_db = Image.objects.get(id=self.test_image_id)
|
326 |
self.assertEqual(image_from_api['name'], image_from_db.name) |
327 |
self.assertEqual(image_from_api['id'], image_from_db.id) |
328 |
self.assertEqual(image_from_api['serverId'], |
329 |
image_from_db.sourcevm and image_from_db.sourcevm.id or |
330 |
"")
|
331 |
self.assertEqual(image_from_api['size'], image_from_db.size) |
332 |
self.assertEqual(image_from_api['status'], image_from_db.state) |
333 |
self.assertEqual(image_from_api['metadata']['meta']['key'] |
334 |
['description'],
|
335 |
image_from_db.description) |
336 |
self.assertTrue(response.status_code in [200, 203]) |
337 |
|
338 |
def test_images_details(self): |
339 |
""" test if the images details are returned by the API
|
340 |
"""
|
341 |
response = self.client.get('/api/v1.0/images/detail') |
342 |
images_from_api = json.loads(response.content)['images']
|
343 |
images_from_db = Image.objects.all() |
344 |
for i in range(0, len(images_from_db)): |
345 |
image_from_db = Image.objects.get(id=images_from_db[i].id) |
346 |
image_from_api = images_from_api[i] |
347 |
self.assertEqual(image_from_api['name'], image_from_db.name) |
348 |
self.assertEqual(image_from_api['id'], image_from_db.id) |
349 |
self.assertEqual(image_from_api['serverId'], |
350 |
image_from_db.sourcevm and
|
351 |
image_from_db.sourcevm.id or "") |
352 |
self.assertEqual(image_from_api['size'], image_from_db.size) |
353 |
self.assertEqual(image_from_api['status'], image_from_db.state) |
354 |
self.assertEqual(image_from_api['metadata']['meta']['key'] |
355 |
['description'],
|
356 |
image_from_db.description) |
357 |
|
358 |
for image_from_api in images_from_api: |
359 |
image_from_db = Image.objects.get(id=image_from_api['id'])
|
360 |
self.assertEqual(image_from_api['name'], image_from_db.name) |
361 |
self.assertEqual(image_from_api['id'], image_from_db.id) |
362 |
self.assertEqual(image_from_api['serverId'], |
363 |
image_from_db.sourcevm and
|
364 |
image_from_db.sourcevm.id or "") |
365 |
self.assertEqual(image_from_api['size'], image_from_db.size) |
366 |
self.assertEqual(image_from_api['status'], image_from_db.state) |
367 |
self.assertEqual(image_from_api['metadata']['meta']['key'] |
368 |
['description'],
|
369 |
image_from_db.description) |
370 |
|
371 |
self.assertTrue(response.status_code in [200, 203]) |
372 |
|
373 |
def test_wrong_image(self): |
374 |
""" test 404 result if a non existent image is requested
|
375 |
"""
|
376 |
response = self.client.get('/api/v1.0/images/' + |
377 |
str(self.test_wrong_image_id)) |
378 |
self.assertEqual(response.status_code, 404) |
379 |
|
380 |
def test_server_metadata(self): |
381 |
""" test server's metadata (add, edit)
|
382 |
"""
|
383 |
request = { |
384 |
"metadata": {
|
385 |
"metadata_key": "name", |
386 |
"metadata_value": "a fancy name" |
387 |
} |
388 |
} |
389 |
response = self.client.put('/api/v1.0/servers' + |
390 |
str(self.test_server_id), |
391 |
json.dumps(request), |
392 |
content_type='application/json')
|
393 |
self.assertEqual(response.status_code, 404) |
394 |
#TODO: not working atm, due to problem with django piston and PUT
|
395 |
|
396 |
def test_vm_group_list(self): |
397 |
""" test if the expected list of groups is returned by the API
|
398 |
"""
|
399 |
response = self.client.get('/api/v1.0/groups') |
400 |
groups_from_api = json.loads(response.content)['groups']
|
401 |
groups_from_db = VirtualMachineGroup.objects.all() |
402 |
self.assertEqual(len(groups_from_api), len(groups_from_db)) |
403 |
self.assertTrue(response.status_code in [200, 203]) |
404 |
for group_from_api in groups_from_api: |
405 |
group_from_db = VirtualMachineGroup.objects.get( |
406 |
id=group_from_api['id']
|
407 |
) |
408 |
self.assertEqual(group_from_api['id'], group_from_db.id) |
409 |
self.assertEqual(group_from_api['name'], group_from_db.name) |
410 |
|
411 |
def test_vm_group_details(self): |
412 |
""" test if the expected virtual machine group is returned by the API
|
413 |
"""
|
414 |
response = self.client.get('/api/v1.0/groups/' + |
415 |
str(self.test_group_id)) |
416 |
group_from_api = json.loads(response.content)['group']
|
417 |
group_from_db = VirtualMachineGroup.objects.get(id=self.test_group_id)
|
418 |
self.assertEqual(group_from_api['name'], group_from_db.name) |
419 |
self.assertEqual(group_from_api['id'], group_from_db.id) |
420 |
self.assertEqual(group_from_api['server_id'], |
421 |
[machine.id |
422 |
for machine in group_from_db.machines.all()]) |
423 |
self.assertTrue(response.status_code in [200, 203]) |
424 |
|
425 |
def test_wrong_vm_group(self): |
426 |
""" test 404 result if a non existent VM group is requested
|
427 |
"""
|
428 |
response = self.client.get('/api/v1.0/groups/' + |
429 |
str(self.test_wrong_group_id)) |
430 |
self.assertEqual(response.status_code, 404) |
431 |
|
432 |
def test_groups_details(self): |
433 |
""" test if the groups details are returned by the API
|
434 |
"""
|
435 |
response = self.client.get('/api/v1.0/groups/detail') |
436 |
groups_from_api = json.loads(response.content)['groups']
|
437 |
groups_from_db = VirtualMachineGroup.objects.all() |
438 |
for i in range(0, len(groups_from_db)): |
439 |
group_from_db = VirtualMachineGroup.objects.get( |
440 |
id=groups_from_db[i].id) |
441 |
group_from_api = groups_from_api[i] |
442 |
self.assertEqual(group_from_api['name'], group_from_db.name) |
443 |
self.assertEqual(group_from_api['id'], group_from_db.id) |
444 |
self.assertEqual(group_from_api['server_id'], |
445 |
[machine.id |
446 |
for machine in group_from_db.machines.all()]) |
447 |
for group_from_api in groups_from_api: |
448 |
group_from_db = VirtualMachineGroup.objects.get( |
449 |
id=group_from_api['id'])
|
450 |
self.assertEqual(group_from_api['name'], group_from_db.name) |
451 |
self.assertEqual(group_from_api['id'], group_from_db.id) |
452 |
self.assertEqual(group_from_api['server_id'], |
453 |
[machine.id |
454 |
for machine in group_from_db.machines.all()]) |
455 |
self.assertTrue(response.status_code in [200, 203]) |