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