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]) |