Revision 039e3e61 snf-cyclades-app/synnefo/api/tests.py
b/snf-cyclades-app/synnefo/api/tests.py | ||
---|---|---|
36 | 36 |
from django.utils import simplejson as json |
37 | 37 |
from django.test import TestCase |
38 | 38 |
|
39 |
from mock import patch, Mock
|
|
39 |
from mock import patch |
|
40 | 40 |
from contextlib import contextmanager |
41 | 41 |
from functools import wraps |
42 | 42 |
|
43 | 43 |
from synnefo.db.models import * |
44 | 44 |
from synnefo.db import models_factory as mfactory |
45 |
from synnefo.logic.utils import get_rsapi_state |
|
46 | 45 |
|
47 | 46 |
from synnefo.api import faults |
48 | 47 |
|
49 | 48 |
|
49 |
|
|
50 | 50 |
@contextmanager |
51 | 51 |
def astakos_user(user): |
52 | 52 |
""" |
... | ... | |
108 | 108 |
self.assertFault(response, 404, 'itemNotFound') |
109 | 109 |
|
110 | 110 |
|
111 |
class FlavorAPITest(BaseAPITest): |
|
112 |
|
|
113 |
def setUp(self): |
|
114 |
self.flavor1 = mfactory.FlavorFactory() |
|
115 |
self.flavor2 = mfactory.FlavorFactory(deleted=True) |
|
116 |
self.flavor3 = mfactory.FlavorFactory() |
|
117 |
|
|
118 |
def test_flavor_list(self): |
|
119 |
"""Test if the expected list of flavors is returned by.""" |
|
120 |
response = self.get('/api/v1.1/flavors') |
|
121 |
self.assertSuccess(response) |
|
122 |
|
|
123 |
flavors_from_api = json.loads(response.content)['flavors']['values'] |
|
124 |
flavors_from_db = Flavor.objects.filter(deleted=False) |
|
125 |
self.assertEqual(len(flavors_from_api), len(flavors_from_db)) |
|
126 |
for flavor_from_api in flavors_from_api: |
|
127 |
flavor_from_db = Flavor.objects.get(id=flavor_from_api['id']) |
|
128 |
self.assertEqual(flavor_from_api['id'], flavor_from_db.id) |
|
129 |
self.assertEqual(flavor_from_api['name'], flavor_from_db.name) |
|
130 |
|
|
131 |
def test_flavors_details(self): |
|
132 |
"""Test if the flavors details are returned.""" |
|
133 |
response = self.get('/api/v1.1/flavors/detail') |
|
134 |
self.assertSuccess(response) |
|
135 |
|
|
136 |
flavors_from_db = Flavor.objects.filter(deleted=False) |
|
137 |
flavors_from_api = json.loads(response.content)['flavors']['values'] |
|
138 |
|
|
139 |
# Assert that all flavors in the db appear inthe API call result |
|
140 |
for i in range(0, len(flavors_from_db)): |
|
141 |
flavor_from_api = flavors_from_api[i] |
|
142 |
flavor_from_db = Flavor.objects.get(id=flavors_from_db[i].id) |
|
143 |
self.assertEqual(flavor_from_api['cpu'], flavor_from_db.cpu) |
|
144 |
self.assertEqual(flavor_from_api['id'], flavor_from_db.id) |
|
145 |
self.assertEqual(flavor_from_api['disk'], flavor_from_db.disk) |
|
146 |
self.assertEqual(flavor_from_api['name'], flavor_from_db.name) |
|
147 |
self.assertEqual(flavor_from_api['ram'], flavor_from_db.ram) |
|
148 |
|
|
149 |
# Assert that all flavors returned by the API also exist in the db |
|
150 |
for flavor_from_api in flavors_from_api: |
|
151 |
flavor_from_db = Flavor.objects.get(id=flavor_from_api['id']) |
|
152 |
self.assertEqual(flavor_from_api['cpu'], flavor_from_db.cpu) |
|
153 |
self.assertEqual(flavor_from_api['id'], flavor_from_db.id) |
|
154 |
self.assertEqual(flavor_from_api['disk'], flavor_from_db.disk) |
|
155 |
self.assertEqual(flavor_from_api['name'], flavor_from_db.name) |
|
156 |
self.assertEqual(flavor_from_api['ram'], flavor_from_db.ram) |
|
157 |
|
|
158 |
def test_flavor_details(self): |
|
159 |
"""Test if the expected flavor is returned.""" |
|
160 |
flavor = self.flavor3 |
|
161 |
|
|
162 |
response = self.get('/api/v1.1/flavors/%d' % flavor.id) |
|
163 |
self.assertSuccess(response) |
|
164 |
|
|
165 |
flavor_from_api = json.loads(response.content)['flavor'] |
|
166 |
flavor_from_db = Flavor.objects.get(id=flavor.id) |
|
167 |
self.assertEqual(flavor_from_api['cpu'], flavor_from_db.cpu) |
|
168 |
self.assertEqual(flavor_from_api['id'], flavor_from_db.id) |
|
169 |
self.assertEqual(flavor_from_api['disk'], flavor_from_db.disk) |
|
170 |
self.assertEqual(flavor_from_api['name'], flavor_from_db.name) |
|
171 |
self.assertEqual(flavor_from_api['ram'], flavor_from_db.ram) |
|
172 |
|
|
173 |
def test_deleted_flavor_details(self): |
|
174 |
"""Test that API returns details for deleted flavors""" |
|
175 |
flavor = self.flavor2 |
|
176 |
response = self.get('/api/v1.1/flavors/%d' % flavor.id) |
|
177 |
self.assertSuccess(response) |
|
178 |
flavor_from_api = json.loads(response.content)['flavor'] |
|
179 |
self.assertEquals(flavor_from_api['name'], flavor.name) |
|
180 |
|
|
181 |
def test_deleted_flavors_list(self): |
|
182 |
"""Test that deleted flavors do not appear to flavors list""" |
|
183 |
response = self.get('/api/v1.1/flavors') |
|
184 |
self.assertSuccess(response) |
|
185 |
flavors_from_api = json.loads(response.content)['flavors']['values'] |
|
186 |
self.assertEqual(len(flavors_from_api), 2) |
|
187 |
|
|
188 |
def test_deleted_flavors_details(self): |
|
189 |
"""Test that deleted flavors do not appear to flavors detail list""" |
|
190 |
mfactory.FlavorFactory(deleted=True) |
|
191 |
response = self.get('/api/v1.1/flavors/detail') |
|
192 |
self.assertSuccess(response) |
|
193 |
flavors_from_api = json.loads(response.content)['flavors']['values'] |
|
194 |
self.assertEqual(len(flavors_from_api), 2) |
|
195 |
|
|
196 |
def test_wrong_flavor(self): |
|
197 |
"""Test 404 result when requesting a flavor that does not exist.""" |
|
198 |
|
|
199 |
response = self.get('/api/v1.1/flavors/%d' % 22) |
|
200 |
self.assertItemNotFound(response) |
|
201 |
|
|
202 |
|
|
203 |
class ServerAPITest(BaseAPITest): |
|
204 |
def setUp(self): |
|
205 |
self.user1 = 'user1' |
|
206 |
self.user2 = 'user2' |
|
207 |
self.vm1 = mfactory.VirtualMachineFactory(userid=self.user1) |
|
208 |
self.vm2 = mfactory.VirtualMachineFactory(userid=self.user2) |
|
209 |
self.vm3 = mfactory.VirtualMachineFactory(deleted=True, |
|
210 |
userid=self.user1) |
|
211 |
self.vm4 = mfactory.VirtualMachineFactory(userid=self.user2) |
|
212 |
|
|
213 |
def test_server_list_1(self): |
|
214 |
"""Test if the expected list of servers is returned.""" |
|
215 |
response = self.get('/api/v1.1/servers') |
|
216 |
self.assertSuccess(response) |
|
217 |
servers = json.loads(response.content)['servers']['values'] |
|
218 |
self.assertEqual(servers, []) |
|
219 |
|
|
220 |
def test_server_list_2(self): |
|
221 |
"""Test if the expected list of servers is returned.""" |
|
222 |
response = self.get('/api/v1.1/servers', self.user1) |
|
223 |
self.assertSuccess(response) |
|
224 |
servers = json.loads(response.content)['servers']['values'] |
|
225 |
db_server = self.vm1 |
|
226 |
self.assertEqual(servers, [{'name': db_server.name, |
|
227 |
'id': db_server.id}]) |
|
228 |
|
|
229 |
def test_server_list_detail(self): |
|
230 |
"""Test if the servers list details are returned.""" |
|
231 |
user = self.user2 |
|
232 |
user_vms = {self.vm2.id: self.vm2, |
|
233 |
self.vm4.id: self.vm4} |
|
234 |
|
|
235 |
response = self.get('/api/v1.1/servers/detail', user) |
|
236 |
servers = json.loads(response.content)['servers']['values'] |
|
237 |
self.assertEqual(len(servers), len(user_vms)) |
|
238 |
for api_vm in servers: |
|
239 |
db_vm = user_vms[api_vm['id']] |
|
240 |
self.assertEqual(api_vm['flavorRef'], db_vm.flavor.id) |
|
241 |
self.assertEqual(api_vm['hostId'], db_vm.hostid) |
|
242 |
self.assertEqual(api_vm['id'], db_vm.id) |
|
243 |
self.assertEqual(api_vm['imageRef'], db_vm.imageid) |
|
244 |
self.assertEqual(api_vm['name'], db_vm.name) |
|
245 |
self.assertEqual(api_vm['status'], get_rsapi_state(db_vm)) |
|
246 |
self.assertSuccess(response) |
|
247 |
|
|
248 |
def test_server_detail(self): |
|
249 |
"""Test if a server details are returned.""" |
|
250 |
db_vm = self.vm2 |
|
251 |
user = self.vm2.userid |
|
252 |
net = mfactory.NetworkFactory() |
|
253 |
nic = mfactory.NetworkInterfaceFactory(machine=self.vm2, network=net) |
|
254 |
|
|
255 |
db_vm_meta = mfactory.VirtualMachineMetadataFactory(vm=db_vm) |
|
256 |
|
|
257 |
response = self.get('/api/v1.1/servers/%d' % db_vm.id, user) |
|
258 |
server = json.loads(response.content)['server'] |
|
259 |
|
|
260 |
self.assertEqual(server['flavorRef'], db_vm.flavor.id) |
|
261 |
self.assertEqual(server['hostId'], db_vm.hostid) |
|
262 |
self.assertEqual(server['id'], db_vm.id) |
|
263 |
self.assertEqual(server['imageRef'], db_vm.imageid) |
|
264 |
self.assertEqual(server['name'], db_vm.name) |
|
265 |
self.assertEqual(server['status'], get_rsapi_state(db_vm)) |
|
266 |
api_nic = server['attachments']['values'][0] |
|
267 |
self.assertEqual(api_nic['network_id'], str(net.id)) |
|
268 |
self.assertEqual(api_nic['mac_address'], nic.mac) |
|
269 |
self.assertEqual(api_nic['firewallProfile'], nic.firewall_profile) |
|
270 |
self.assertEqual(api_nic['ipv4'], nic.ipv4) |
|
271 |
self.assertEqual(api_nic['ipv6'], nic.ipv6) |
|
272 |
self.assertEqual(api_nic['id'], 'nic-%s-%s' % (db_vm.id, nic.index)) |
|
273 |
|
|
274 |
metadata = server['metadata']['values'] |
|
275 |
self.assertEqual(len(metadata), 1) |
|
276 |
self.assertEqual(metadata[db_vm_meta.meta_key], db_vm_meta.meta_value) |
|
277 |
self.assertSuccess(response) |
|
278 |
|
|
279 |
def test_noauthorized(self): |
|
280 |
"""Test 404 for detail of other user vm""" |
|
281 |
db_vm = self.vm2 |
|
282 |
|
|
283 |
response = self.get('/api/v1.1/servers/%d' % db_vm.id, 'wrong_user') |
|
284 |
self.assertItemNotFound(response) |
|
285 |
|
|
286 |
def test_wrong_server(self): |
|
287 |
"""Test 404 response if server does not exist.""" |
|
288 |
response = self.get('/api/v1.1/servers/%d' % 5000) |
|
289 |
self.assertItemNotFound(response) |
|
290 |
|
|
291 |
def test_create_server_empty(self): |
|
292 |
"""Test if the create server call returns a 400 badRequest if |
|
293 |
no attributes are specified.""" |
|
294 |
|
|
295 |
response = self.post('/api/v1.1/servers', params={}) |
|
296 |
self.assertBadRequest(response) |
|
297 |
|
|
298 |
def test_rename_server(self): |
|
299 |
vm = self.vm2 |
|
300 |
request = {'server': {'name': 'new_name'}} |
|
301 |
response = self.put('/api/v1.1/servers/%d' % vm.id, vm.userid, |
|
302 |
json.dumps(request), 'json') |
|
303 |
self.assertSuccess(response) |
|
304 |
self.assertEqual(VirtualMachine.objects.get(id=vm.id).name, "new_name") |
|
305 |
|
|
306 |
|
|
307 |
@patch('synnefo.api.util.get_image') |
|
308 |
@patch('synnefo.logic.rapi_pool.GanetiRapiClient') |
|
309 |
class ServerCreateAPITest(BaseAPITest): |
|
310 |
def test_create_server(self, mrapi, mimage): |
|
311 |
"""Test if the create server call returns the expected response |
|
312 |
if a valid request has been speficied.""" |
|
313 |
mimage.return_value = {'location': 'pithos://foo', |
|
314 |
'disk_format': 'diskdump'} |
|
315 |
mrapi().CreateInstance.return_value = 12 |
|
316 |
flavor = mfactory.FlavorFactory() |
|
317 |
# Create public network and backend |
|
318 |
network = mfactory.NetworkFactory(public=True) |
|
319 |
backend = mfactory.BackendFactory() |
|
320 |
mfactory.BackendNetworkFactory(network=network, backend=backend) |
|
321 |
|
|
322 |
request = { |
|
323 |
"server": { |
|
324 |
"name": "new-server-test", |
|
325 |
"userid": "test_user", |
|
326 |
"imageRef": 1, |
|
327 |
"flavorRef": flavor.id, |
|
328 |
"metadata": { |
|
329 |
"My Server Name": "Apache1" |
|
330 |
}, |
|
331 |
"personality": [] |
|
332 |
} |
|
333 |
} |
|
334 |
response = self.post('/api/v1.1/servers', 'test_user', |
|
335 |
json.dumps(request), 'json') |
|
336 |
self.assertEqual(response.status_code, 202) |
|
337 |
mrapi().CreateInstance.assert_called_once() |
|
338 |
|
|
339 |
api_server = json.loads(response.content)['server'] |
|
340 |
self.assertEqual(api_server['status'], "BUILD") |
|
341 |
self.assertEqual(api_server['progress'], 0) |
|
342 |
self.assertEqual(api_server['metadata']['values'], |
|
343 |
{"My Server Name": "Apache1"}) |
|
344 |
self.assertTrue('adminPass' in api_server) |
|
345 |
|
|
346 |
db_vm = VirtualMachine.objects.get(userid='test_user') |
|
347 |
self.assertEqual(api_server['name'], db_vm.name) |
|
348 |
self.assertEqual(api_server['status'], db_vm.operstate) |
|
349 |
|
|
350 |
def test_create_server_no_flavor(self, mrapi, mimage): |
|
351 |
request = { |
|
352 |
"server": { |
|
353 |
"name": "new-server-test", |
|
354 |
"userid": "test_user", |
|
355 |
"imageRef": 1, |
|
356 |
"flavorRef": 42, |
|
357 |
"metadata": { |
|
358 |
"My Server Name": "Apache1" |
|
359 |
}, |
|
360 |
"personality": [] |
|
361 |
} |
|
362 |
} |
|
363 |
response = self.post('/api/v1.1/servers', 'test_user', |
|
364 |
json.dumps(request), 'json') |
|
365 |
self.assertItemNotFound(response) |
|
366 |
|
|
367 |
|
|
368 |
@patch('synnefo.logic.rapi_pool.GanetiRapiClient') |
|
369 |
class ServerDestroyAPITest(BaseAPITest): |
|
370 |
def test_delete_server(self, mrapi): |
|
371 |
vm = mfactory.VirtualMachineFactory() |
|
372 |
response = self.delete('/api/v1.1/servers/%d' % vm.id, vm.userid) |
|
373 |
self.assertEqual(response.status_code, 204) |
|
374 |
mrapi().DeleteInstance.assert_called_once() |
|
375 |
|
|
376 |
def test_non_existing_delete_server(self, mrapi): |
|
377 |
vm = mfactory.VirtualMachineFactory() |
|
378 |
response = self.delete('/api/v1.1/servers/%d' % 42, vm.userid) |
|
379 |
self.assertItemNotFound(response) |
|
380 |
self.assertFalse(mrapi.mock_calls) |
|
381 |
|
|
382 |
|
|
383 |
@patch('synnefo.api.util.get_image') |
|
384 |
@patch('synnefo.logic.rapi_pool.GanetiRapiClient') |
|
385 |
class ServerActionAPITest(BaseAPITest): |
|
386 |
def test_actions(self, mrapi, mimage): |
|
387 |
actions = ['start', 'shutdown', 'reboot'] |
|
388 |
vm = mfactory.VirtualMachineFactory() |
|
389 |
vm.operstate = "STOPPED" |
|
390 |
vm.save() |
|
391 |
for action in actions: |
|
392 |
val = {'type': 'HARD'} if action == 'reboot' else {} |
|
393 |
request = {action: val} |
|
394 |
response = self.post('/api/v1.1/servers/%d/action' % vm.id, |
|
395 |
vm.userid, json.dumps(request), 'json') |
|
396 |
self.assertEqual(response.status_code, 202) |
|
397 |
if action == 'shutdown': |
|
398 |
self.assertEqual(VirtualMachine.objects.get(id=vm.id).action, |
|
399 |
"STOP") |
|
400 |
else: |
|
401 |
self.assertEqual(VirtualMachine.objects.get(id=vm.id).action, |
|
402 |
action.upper()) |
|
403 |
|
|
404 |
def test_action_in_building_vm(self, mrapi, mimage): |
|
405 |
"""Test building in progress""" |
|
406 |
vm = mfactory.VirtualMachineFactory() |
|
407 |
request = {'start': '{}'} |
|
408 |
response = self.post('/api/v1.1/servers/%d/action' % vm.id, |
|
409 |
vm.userid, json.dumps(request), 'json') |
|
410 |
self.assertEqual(response.status_code, 409) |
|
411 |
self.assertFalse(mrapi.mock_calls) |
|
412 |
|
|
413 |
def test_destroy_build_vm(self, mrapi, mimage): |
|
414 |
"""Test building in progress""" |
|
415 |
vm = mfactory.VirtualMachineFactory() |
|
416 |
response = self.delete('/api/v1.1/servers/%d' % vm.id, |
|
417 |
vm.userid) |
|
418 |
self.assertSuccess(response) |
|
419 |
mrapi().RemoveInstance.assert_called_once() |
|
420 |
|
|
421 |
def test_firewall(self, mrapi, mimage): |
|
422 |
vm = mfactory.VirtualMachineFactory() |
|
423 |
vm.operstate = "STOPPED" |
|
424 |
vm.save() |
|
425 |
request = {'firewallProfile': {'profile': 'PROTECTED'}} |
|
426 |
response = self.post('/api/v1.1/servers/%d/action' % vm.id, |
|
427 |
vm.userid, json.dumps(request), 'json') |
|
428 |
self.assertEqual(response.status_code, 202) |
|
429 |
mrapi().ModifyInstance.assert_called_once() |
|
430 |
|
|
431 |
def test_unsupported_firewall(self, mrapi, mimage): |
|
432 |
vm = mfactory.VirtualMachineFactory() |
|
433 |
vm.operstate = "STOPPED" |
|
434 |
vm.save() |
|
435 |
request = {'firewallProfile': {'profile': 'FOO'}} |
|
436 |
response = self.post('/api/v1.1/servers/%d/action' % vm.id, |
|
437 |
vm.userid, json.dumps(request), 'json') |
|
438 |
self.assertBadRequest(response) |
|
439 |
self.assertFalse(mrapi.mock_calls) |
|
440 |
|
|
441 |
|
|
442 |
class ServerMetadataAPITest(BaseAPITest): |
|
443 |
def setUp(self): |
|
444 |
self.vm = mfactory.VirtualMachineFactory() |
|
445 |
self.metadata = mfactory.VirtualMachineMetadataFactory(vm=self.vm) |
|
446 |
|
|
447 |
def test_get_metadata(self): |
|
448 |
vm = self.vm |
|
449 |
create_meta = lambda: mfactory.VirtualMachineMetadataFactory(vm=vm) |
|
450 |
metadata = [create_meta(), create_meta(), create_meta()] |
|
451 |
response = self.get('/api/v1.1/servers/%d/meta' % vm.id, vm.userid) |
|
452 |
self.assertTrue(response.status_code in [200, 203]) |
|
453 |
api_metadata = json.loads(response.content)['metadata']['values'] |
|
454 |
self.assertEqual(len(api_metadata), len(metadata) + 1) |
|
455 |
for db_m in metadata: |
|
456 |
self.assertEqual(api_metadata[db_m.meta_key], db_m.meta_value) |
|
457 |
|
|
458 |
request = {'metadata': |
|
459 |
{'foo': 'bar'}, |
|
460 |
metadata[0].meta_key: 'bar2' |
|
461 |
} |
|
462 |
response = self.post('/api/v1.1/servers/%d/meta' % vm.id, vm.userid, |
|
463 |
json.dumps(request), 'json') |
|
464 |
metadata2 = VirtualMachineMetadata.objects.filter(vm=vm) |
|
465 |
response = self.get('/api/v1.1/servers/%d/meta' % vm.id, vm.userid) |
|
466 |
self.assertTrue(response.status_code in [200, 203]) |
|
467 |
api_metadata2 = json.loads(response.content)['metadata']['values'] |
|
468 |
self.assertTrue('foo' in api_metadata2.keys()) |
|
469 |
self.assertTrue(api_metadata2[metadata[0].meta_key], 'bar2') |
|
470 |
self.assertEqual(len(api_metadata2), len(metadata2)) |
|
471 |
for db_m in metadata2: |
|
472 |
self.assertEqual(api_metadata2[db_m.meta_key], db_m.meta_value) |
|
473 |
|
|
474 |
# Create new meta |
|
475 |
request = {'meta': {'foo2': 'bar2'}} |
|
476 |
response = self.put('/api/v1.1/servers/%d/meta/foo2' % vm.id, |
|
477 |
vm.userid, json.dumps(request), 'json') |
|
478 |
|
|
479 |
# Get the new meta |
|
480 |
response = self.get('/api/v1.1/servers/%d/meta/foo2' % vm.id, |
|
481 |
vm.userid) |
|
482 |
meta = json.loads(response.content)['meta'] |
|
483 |
self.assertEqual(meta['foo2'], 'bar2') |
|
484 |
|
|
485 |
# Delete the new meta |
|
486 |
response = self.delete('/api/v1.1/servers/%d/meta/foo2' % vm.id, |
|
487 |
vm.userid) |
|
488 |
self.assertEqual(response.status_code, 204) |
|
489 |
|
|
490 |
# Try to get the deleted meta: should raise 404 |
|
491 |
response = self.get('/api/v1.1/servers/%d/meta/foo2' % vm.id, |
|
492 |
vm.userid) |
|
493 |
self.assertEqual(response.status_code, 404) |
|
494 |
|
|
495 |
def test_invalid_metadata(self): |
|
496 |
vm = self.vm |
|
497 |
response = self.post('/api/v1.1/servers/%d/meta' % vm.id, vm.userid) |
|
498 |
self.assertBadRequest(response) |
|
499 |
self.assertEqual(len(vm.metadata.all()), 1) |
|
500 |
|
|
501 |
def test_invalid_metadata_server(self): |
|
502 |
response = self.post('/api/v1.1/servers/42/meta', 'user') |
|
503 |
self.assertItemNotFound(response) |
|
504 |
|
|
505 |
def test_get_meta_invalid_key(self): |
|
506 |
vm = self.vm |
|
507 |
response = self.get('/api/v1.1/servers/%d/meta/foo2' % vm.id, |
|
508 |
vm.userid) |
|
509 |
self.assertItemNotFound(response) |
|
510 |
|
|
511 |
|
|
512 |
@patch('synnefo.logic.rapi_pool.GanetiRapiClient') |
|
513 |
class NetworkAPITest(BaseAPITest): |
|
514 |
def setUp(self): |
|
515 |
self.mac_prefixes = mfactory.MacPrefixPoolTableFactory() |
|
516 |
self.bridges = mfactory.BridgePoolTableFactory() |
|
517 |
self.user = 'dummy-user' |
|
518 |
self.net1 = mfactory.NetworkFactory(userid=self.user) |
|
519 |
self.net2 = mfactory.NetworkFactory(userid=self.user) |
|
520 |
|
|
521 |
def assertNetworksEqual(self, db_net, api_net, detail=False): |
|
522 |
self.assertEqual(str(db_net.id), api_net["id"]) |
|
523 |
self.assertEqual(db_net.name, api_net['name']) |
|
524 |
if detail: |
|
525 |
self.assertEqual(db_net.state, api_net['status']) |
|
526 |
self.assertEqual(db_net.flavor, api_net['type']) |
|
527 |
self.assertEqual(db_net.subnet, api_net['cidr']) |
|
528 |
self.assertEqual(db_net.subnet6, api_net['cidr6']) |
|
529 |
self.assertEqual(db_net.gateway, api_net['gateway']) |
|
530 |
self.assertEqual(db_net.gateway6, api_net['gateway6']) |
|
531 |
self.assertEqual(db_net.dhcp, api_net['dhcp']) |
|
532 |
|
|
533 |
def test_create_network_1(self, mrapi): |
|
534 |
request = { |
|
535 |
'network': {'name': 'foo'} |
|
536 |
} |
|
537 |
response = self.post('/api/v1.1/networks/', 'user1', |
|
538 |
json.dumps(request), 'json') |
|
539 |
self.assertEqual(response.status_code, 202) |
|
540 |
db_networks = Network.objects.filter(userid='user1') |
|
541 |
self.assertEqual(len(db_networks), 1) |
|
542 |
db_net = db_networks[0] |
|
543 |
api_net = json.loads(response.content)['network'] |
|
544 |
self.assertNetworksEqual(db_net, api_net) |
|
545 |
mrapi.CreateNetwork.assert_called() |
|
546 |
mrapi.ConnectNetwork.assert_called() |
|
547 |
|
|
548 |
def test_invalid_data_1(self, mrapi): |
|
549 |
"""Test invalid flavor""" |
|
550 |
request = { |
|
551 |
'network': {'name': 'foo', 'type': 'LoLo'} |
|
552 |
} |
|
553 |
response = self.post('/api/v1.1/networks/', 'user1', |
|
554 |
json.dumps(request), 'json') |
|
555 |
self.assertBadRequest(response) |
|
556 |
self.assertEqual(len(Network.objects.filter(userid='user1')), 0) |
|
557 |
|
|
558 |
def test_invalid_data_2(self, mrapi): |
|
559 |
"""Test invalid subnet""" |
|
560 |
request = { |
|
561 |
'network': {'name': 'foo', 'cidr': '10.0.0.0/8'} |
|
562 |
} |
|
563 |
response = self.post('/api/v1.1/networks/', 'user1', |
|
564 |
json.dumps(request), 'json') |
|
565 |
self.assertFault(response, 413, "overLimit") |
|
566 |
|
|
567 |
def test_invalid_data_3(self, mrapi): |
|
568 |
"""Test unauthorized to create public network""" |
|
569 |
request = { |
|
570 |
'network': {'name': 'foo', 'public': True} |
|
571 |
} |
|
572 |
response = self.post('/api/v1.1/networks/', 'user1', |
|
573 |
json.dumps(request), 'json') |
|
574 |
self.assertFault(response, 403, "forbidden") |
|
575 |
|
|
576 |
def test_list_networks(self, mrapi): |
|
577 |
mfactory.NetworkFactory(userid=self.user, deleted=True) |
|
578 |
response = self.get('/api/v1.1/networks/', self.user) |
|
579 |
self.assertSuccess(response) |
|
580 |
db_nets = Network.objects.filter(userid=self.user, deleted=False) |
|
581 |
api_nets = json.loads(response.content)["networks"]["values"] |
|
582 |
self.assertEqual(len(db_nets), len(api_nets)) |
|
583 |
for api_net in api_nets: |
|
584 |
net_id = api_net['id'] |
|
585 |
self.assertNetworksEqual(Network.objects.get(id=net_id), api_net) |
|
586 |
|
|
587 |
def test_list_networks_detail(self, mrapi): |
|
588 |
mfactory.NetworkFactory(userid=self.user, deleted=True) |
|
589 |
response = self.get('/api/v1.1/networks/detail', self.user) |
|
590 |
self.assertSuccess(response) |
|
591 |
db_nets = Network.objects.filter(userid=self.user, deleted=False) |
|
592 |
api_nets = json.loads(response.content)["networks"]["values"] |
|
593 |
self.assertEqual(len(db_nets), len(api_nets)) |
|
594 |
for api_net in api_nets: |
|
595 |
net_id = api_net['id'] |
|
596 |
self.assertNetworksEqual(Network.objects.get(id=net_id), api_net, |
|
597 |
detail=True) |
|
598 |
|
|
599 |
def test_network_details_1(self, mrapi): |
|
600 |
response = self.get('/api/v1.1/networks/%d' % self.net1.id, |
|
601 |
self.net1.userid) |
|
602 |
self.assertSuccess(response) |
|
603 |
api_net = json.loads(response.content)["network"] |
|
604 |
self.assertNetworksEqual(self.net1, api_net, detail=True) |
|
605 |
|
|
606 |
def test_invalid_network(self, mrapi): |
|
607 |
response = self.get('/api/v1.1/networks/%d' % 42, |
|
608 |
self.net1.userid) |
|
609 |
self.assertItemNotFound(response) |
|
610 |
|
|
611 |
def test_rename_network(self, mrapi): |
|
612 |
request = {'network': {'name': "new_name"}} |
|
613 |
response = self.put('/api/v1.1/networks/%d' % self.net2.id, |
|
614 |
self.net2.userid, json.dumps(request), 'json') |
|
615 |
self.assertEqual(response.status_code, 204) |
|
616 |
self.assertEqual(Network.objects.get(id=self.net2.id).name, "new_name") |
|
617 |
|
|
618 |
def test_rename_public_network(self, mrapi): |
|
619 |
net = mfactory.NetworkFactory(public=True) |
|
620 |
request = {'network': {'name': "new_name"}} |
|
621 |
response = self.put('/api/v1.1/networks/%d' % net.id, |
|
622 |
self.net2.userid, json.dumps(request), 'json') |
|
623 |
self.assertFault(response, 403, 'forbidden') |
|
624 |
|
|
625 |
def test_delete_network(self, mrapi): |
|
626 |
response = self.delete('/api/v1.1/networks/%d' % self.net2.id, |
|
627 |
self.net2.userid) |
|
628 |
self.assertEqual(response.status_code, 204) |
|
629 |
net = Network.objects.get(id=self.net2.id, userid=self.net2.userid) |
|
630 |
self.assertEqual(net.action, 'DESTROY') |
|
631 |
mrapi.DeleteNetwork.assert_called() |
|
632 |
|
|
633 |
def test_delete_public_network(self, mrapi): |
|
634 |
net = mfactory.NetworkFactory(public=True) |
|
635 |
response = self.delete('/api/v1.1/networks/%d' % net.id, |
|
636 |
self.net2.userid) |
|
637 |
self.assertFault(response, 403, 'forbidden') |
|
638 |
self.assertFalse(mrapi.mock_calls) |
|
639 |
|
|
640 |
def test_delete_network_in_use(self, mrapi): |
|
641 |
net = mfactory.NetworkFactory(deleted=False) |
|
642 |
vm = mfactory.VirtualMachineFactory() |
|
643 |
mfactory.NetworkInterfaceFactory(machine=vm, network=net) |
|
644 |
response = self.delete('/api/v1.1/networks/%d' % net.id, |
|
645 |
net.userid) |
|
646 |
self.assertFault(response, 421, 'networkInUse') |
|
647 |
self.assertFalse(mrapi.mock_calls) |
|
648 |
|
|
649 |
def test_add_nic(self, mrapi): |
|
650 |
user = 'userr' |
|
651 |
vm = mfactory.VirtualMachineFactory(name='yo', userid=user) |
|
652 |
net = mfactory.NetworkFactory(state='ACTIVE', userid=user) |
|
653 |
request = {'add': {'serverRef': vm.id}} |
|
654 |
response = self.post('/api/v1.1/networks/%d/action' % net.id, |
|
655 |
net.userid, json.dumps(request), 'json') |
|
656 |
self.assertEqual(response.status_code, 202) |
|
657 |
|
|
658 |
def test_add_nic_to_public_network(self, mrapi): |
|
659 |
user = 'userr' |
|
660 |
vm = mfactory.VirtualMachineFactory(name='yo', userid=user) |
|
661 |
net = mfactory.NetworkFactory(state='ACTIVE', userid=user, public=True) |
|
662 |
request = {'add': {'serverRef': vm.id}} |
|
663 |
response = self.post('/api/v1.1/networks/%d/action' % net.id, |
|
664 |
net.userid, json.dumps(request), 'json') |
|
665 |
self.assertFault(response, 403, 'forbidden') |
|
666 |
|
|
667 |
def test_add_nic_malformed(self, mrapi): |
|
668 |
user = 'userr' |
|
669 |
vm = mfactory.VirtualMachineFactory(name='yo', userid=user) |
|
670 |
net = mfactory.NetworkFactory(state='ACTIVE', userid=user) |
|
671 |
request = {'add': {'serveRef': vm.id}} |
|
672 |
response = self.post('/api/v1.1/networks/%d/action' % net.id, |
|
673 |
net.userid, json.dumps(request), 'json') |
|
674 |
self.assertBadRequest(response) |
|
675 |
|
|
676 |
def test_add_nic_not_active(self, mrapi): |
|
677 |
"""Test connecting VM to non-active network""" |
|
678 |
user = 'dummy' |
|
679 |
vm = mfactory.VirtualMachineFactory(name='yo', userid=user) |
|
680 |
net = mfactory.NetworkFactory(state='PENDING', subnet='10.0.0.0/31', |
|
681 |
userid=user) |
|
682 |
request = {'add': {'serveRef': vm.id}} |
|
683 |
response = self.post('/api/v1.1/networks/%d/action' % net.id, |
|
684 |
net.userid, json.dumps(request), 'json') |
|
685 |
# Test that returns BuildInProgress |
|
686 |
self.assertEqual(response.status_code, 409) |
|
687 |
|
|
688 |
def test_add_nic_full_network(self, mrapi): |
|
689 |
"""Test connecting VM to a full network""" |
|
690 |
user = 'userr' |
|
691 |
vm = mfactory.VirtualMachineFactory(name='yo', userid=user) |
|
692 |
net = mfactory.NetworkFactory(state='ACTIVE', subnet='10.0.0.0/30', |
|
693 |
userid=user, dhcp=True) |
|
694 |
pool = net.get_pool() |
|
695 |
while not pool.empty(): |
|
696 |
pool.get() |
|
697 |
pool.save() |
|
698 |
pool = net.get_pool() |
|
699 |
self.assertTrue(pool.empty()) |
|
700 |
request = {'add': {'serverRef': vm.id}} |
|
701 |
response = self.post('/api/v1.1/networks/%d/action' % net.id, |
|
702 |
net.userid, json.dumps(request), 'json') |
|
703 |
# Test that returns OverLimit |
|
704 |
self.assertEqual(response.status_code, 413) |
|
705 |
|
|
706 |
def test_remove_nic(self, mrapi): |
|
707 |
user = 'userr' |
|
708 |
vm = mfactory.VirtualMachineFactory(name='yo', userid=user) |
|
709 |
net = mfactory.NetworkFactory(state='ACTIVE', userid=user) |
|
710 |
nic = mfactory.NetworkInterfaceFactory(machine=vm, network=net) |
|
711 |
request = {'remove': {'attachment': 'nic-%s-%s' % (vm.id, nic.index)}} |
|
712 |
response = self.post('/api/v1.1/networks/%d/action' % net.id, |
|
713 |
net.userid, json.dumps(request), 'json') |
|
714 |
self.assertEqual(response.status_code, 202) |
|
715 |
self.assertTrue(NetworkInterface.objects.get(id=nic.id).dirty) |
|
716 |
# Remove dirty nic |
|
717 |
response = self.post('/api/v1.1/networks/%d/action' % net.id, |
|
718 |
net.userid, json.dumps(request), 'json') |
|
719 |
self.assertFault(response, 409, 'buildInProgress') |
|
720 |
|
|
721 |
def test_remove_nic_malformed(self, mrapi): |
|
722 |
user = 'userr' |
|
723 |
vm = mfactory.VirtualMachineFactory(name='yo', userid=user) |
|
724 |
net = mfactory.NetworkFactory(state='ACTIVE', userid=user) |
|
725 |
nic = mfactory.NetworkInterfaceFactory(machine=vm, network=net) |
|
726 |
request = {'remove': |
|
727 |
{'att234achment': 'nic-%s-%s' % (vm.id, nic.index)} |
|
728 |
} |
|
729 |
response = self.post('/api/v1.1/networks/%d/action' % net.id, |
|
730 |
net.userid, json.dumps(request), 'json') |
|
731 |
self.assertBadRequest(response) |
|
732 |
|
|
733 |
def test_remove_nic_malformed_2(self, mrapi): |
|
734 |
user = 'userr' |
|
735 |
vm = mfactory.VirtualMachineFactory(name='yo', userid=user) |
|
736 |
net = mfactory.NetworkFactory(state='ACTIVE', userid=user) |
|
737 |
request = {'remove': |
|
738 |
{'attachment': 'nic-%s' % vm.id} |
|
739 |
} |
|
740 |
response = self.post('/api/v1.1/networks/%d/action' % net.id, |
|
741 |
net.userid, json.dumps(request), 'json') |
|
742 |
self.assertBadRequest(response) |
|
743 |
|
|
744 |
|
|
745 |
class ServerVNCConsole(BaseAPITest): |
|
746 |
|
|
747 |
def test_not_active_server(self): |
|
748 |
"""Test console req for not ACTIVE server returns badRequest""" |
|
749 |
vm = mfactory.VirtualMachineFactory() |
|
750 |
data = json.dumps({'console': {'type': 'vnc'}}) |
|
751 |
response = self.post('/api/v1.1/servers/%d/action' % vm.id, |
|
752 |
vm.userid, data, 'json') |
|
753 |
self.assertBadRequest(response) |
|
754 |
|
|
755 |
def test_active_server(self): |
|
756 |
"""Test console req for ACTIVE server""" |
|
757 |
vm = mfactory.VirtualMachineFactory() |
|
758 |
vm.operstate = 'STARTED' |
|
759 |
vm.save() |
|
760 |
|
|
761 |
data = json.dumps({'console': {'type': 'vnc'}}) |
|
762 |
response = self.post('/api/v1.1/servers/%d/action' % vm.id, |
|
763 |
vm.userid, data, 'json') |
|
764 |
self.assertEqual(response.status_code, 200) |
|
765 |
reply = json.loads(response.content) |
|
766 |
self.assertEqual(reply.keys(), ['console']) |
|
767 |
console = reply['console'] |
|
768 |
self.assertEqual(console['type'], 'vnc') |
|
769 |
self.assertEqual(set(console.keys()), |
|
770 |
set(['type', 'host', 'port', 'password'])) |
|
771 |
|
|
772 |
def test_wrong_console_type(self): |
|
773 |
"""Test console req for ACTIVE server""" |
|
774 |
vm = mfactory.VirtualMachineFactory() |
|
775 |
vm.operstate = 'STARTED' |
|
776 |
vm.save() |
|
777 |
|
|
778 |
data = json.dumps({'console': {'type': 'foo'}}) |
|
779 |
response = self.post('/api/v1.1/servers/%d/action' % vm.id, |
|
780 |
vm.userid, data, 'json') |
|
781 |
self.assertBadRequest(response) |
|
782 |
|
|
783 |
def assert_backend_closed(func): |
|
784 |
@wraps(func) |
|
785 |
def wrapper(self, backend): |
|
786 |
result = func(self, backend) |
|
787 |
if backend.called is True: |
|
788 |
backend.return_value.close.assert_called_once_with() |
|
789 |
return result |
|
790 |
return wrapper |
|
791 |
|
|
792 |
|
|
793 |
@patch('synnefo.api.images.ImageBackend') |
|
794 |
class ImageAPITest(BaseAPITest): |
|
795 |
@assert_backend_closed |
|
796 |
def test_create_image(self, mimage): |
|
797 |
"""Test that create image is not implemented""" |
|
798 |
response = self.post('/api/v1.1/images/', 'user', json.dumps(''), |
|
799 |
'json') |
|
800 |
self.assertEqual(response.status_code, 503) |
|
801 |
|
|
802 |
@assert_backend_closed |
|
803 |
def test_list_images(self, mimage): |
|
804 |
"""Test that expected list of images is returned""" |
|
805 |
images = [{'id': 1, 'name': 'image-1'}, |
|
806 |
{'id': 2, 'name': 'image-2'}, |
|
807 |
{'id': 3, 'name': 'image-3'}] |
|
808 |
mimage().list.return_value = images |
|
809 |
response = self.get('/api/v1.1/images/', 'user') |
|
810 |
self.assertSuccess(response) |
|
811 |
api_images = json.loads(response.content)['images']['values'] |
|
812 |
self.assertEqual(images, api_images) |
|
813 |
|
|
814 |
@assert_backend_closed |
|
815 |
def test_list_images_detail(self, mimage): |
|
816 |
images = [{'id': 1, |
|
817 |
'name': 'image-1', |
|
818 |
'status':'available', |
|
819 |
'created_at': '2012-11-26 11:52:54', |
|
820 |
'updated_at': '2012-12-26 11:52:54', |
|
821 |
'deleted_at': '', |
|
822 |
'properties': {'foo':'bar'}}, |
|
823 |
{'id': 2, |
|
824 |
'name': 'image-2', |
|
825 |
'status': 'deleted', |
|
826 |
'created_at': '2012-11-26 11:52:54', |
|
827 |
'updated_at': '2012-12-26 11:52:54', |
|
828 |
'deleted_at': '2012-12-27 11:52:54', |
|
829 |
'properties': ''}, |
|
830 |
{'id': 3, |
|
831 |
'name': 'image-3', |
|
832 |
'status': 'available', |
|
833 |
'created_at': '2012-11-26 11:52:54', |
|
834 |
'deleted_at': '', |
|
835 |
'updated_at': '2012-12-26 11:52:54', |
|
836 |
'properties': ''}] |
|
837 |
result_images = [ |
|
838 |
{'id': 1, |
|
839 |
'name': 'image-1', |
|
840 |
'status':'ACTIVE', |
|
841 |
'progress': 100, |
|
842 |
'created': '2012-11-26T11:52:54+00:00', |
|
843 |
'updated': '2012-12-26T11:52:54+00:00', |
|
844 |
'metadata': {'values': {'foo':'bar'}}}, |
|
845 |
{'id': 2, |
|
846 |
'name': 'image-2', |
|
847 |
'status': 'DELETED', |
|
848 |
'progress': 0, |
|
849 |
'created': '2012-11-26T11:52:54+00:00', |
|
850 |
'updated': '2012-12-26T11:52:54+00:00'}, |
|
851 |
{'id': 3, |
|
852 |
'name': 'image-3', |
|
853 |
'status': 'ACTIVE', |
|
854 |
'progress': 100, |
|
855 |
'created': '2012-11-26T11:52:54+00:00', |
|
856 |
'updated': '2012-12-26T11:52:54+00:00'}] |
|
857 |
mimage().list.return_value = images |
|
858 |
response = self.get('/api/v1.1/images/detail', 'user') |
|
859 |
self.assertSuccess(response) |
|
860 |
api_images = json.loads(response.content)['images']['values'] |
|
861 |
self.assertEqual(len(result_images), len(api_images)) |
|
862 |
self.assertEqual(result_images, api_images) |
|
863 |
|
|
864 |
@assert_backend_closed |
|
865 |
def test_get_image_details(self, mimage): |
|
866 |
image = {'id': 42, |
|
867 |
'name': 'image-1', |
|
868 |
'status': 'available', |
|
869 |
'created_at': '2012-11-26 11:52:54', |
|
870 |
'updated_at': '2012-12-26 11:52:54', |
|
871 |
'deleted_at': '', |
|
872 |
'properties': {'foo': 'bar'}} |
|
873 |
result_image = \ |
|
874 |
{'id': 42, |
|
875 |
'name': 'image-1', |
|
876 |
'status': 'ACTIVE', |
|
877 |
'progress': 100, |
|
878 |
'created': '2012-11-26T11:52:54+00:00', |
|
879 |
'updated': '2012-12-26T11:52:54+00:00', |
|
880 |
'metadata': {'values': {'foo': 'bar'}}} |
|
881 |
with patch('synnefo.api.util.get_image') as m: |
|
882 |
m.return_value = image |
|
883 |
response = self.get('/api/v1.1/images/42', 'user') |
|
884 |
self.assertSuccess(response) |
|
885 |
api_image = json.loads(response.content)['image'] |
|
886 |
self.assertEqual(api_image, result_image) |
|
887 |
|
|
888 |
@assert_backend_closed |
|
889 |
def test_invalid_image(self, mimage): |
|
890 |
with patch('synnefo.api.util.get_image') as m: |
|
891 |
m.side_effect = faults.ItemNotFound('Image not found') |
|
892 |
response = self.get('/api/v1.1/images/42', 'user') |
|
893 |
self.assertItemNotFound(response) |
|
894 |
|
|
895 |
def test_delete_image(self, mimage): |
|
896 |
# TODO |
|
897 |
pass |
|
898 |
|
|
899 |
|
|
900 |
@patch('synnefo.api.util.ImageBackend') |
|
901 |
class ImageMetadataAPITest(BaseAPITest): |
|
902 |
def setUp(self): |
|
903 |
self.image = {'id': 42, |
|
904 |
'name': 'image-1', |
|
905 |
'status': 'available', |
|
906 |
'created_at': '2012-11-26 11:52:54', |
|
907 |
'updated_at': '2012-12-26 11:52:54', |
|
908 |
'deleted_at': '', |
|
909 |
'properties': {'foo': 'bar', 'foo2': 'bar2'}} |
|
910 |
self.result_image = \ |
|
911 |
{'id': 42, |
|
912 |
'name': 'image-1', |
|
913 |
'status': 'ACTIVE', |
|
914 |
'progress': 100, |
|
915 |
'created': '2012-11-26T11:52:54+00:00', |
|
916 |
'updated': '2012-12-26T11:52:54+00:00', |
|
917 |
'metadata': {'values': {'foo': 'bar'}}} |
|
918 |
|
|
919 |
@assert_backend_closed |
|
920 |
def test_list_metadata(self, backend): |
|
921 |
backend.return_value.get_image.return_value = self.image |
|
922 |
response = self.get('/api/v1.1/images/42/meta', 'user') |
|
923 |
self.assertSuccess(response) |
|
924 |
meta = json.loads(response.content)['metadata']['values'] |
|
925 |
self.assertEqual(meta, self.image['properties']) |
|
926 |
|
|
927 |
@assert_backend_closed |
|
928 |
def test_get_metadata(self, backend): |
|
929 |
backend.return_value.get_image.return_value = self.image |
|
930 |
response = self.get('/api/v1.1/images/42/meta/foo', 'user') |
|
931 |
self.assertSuccess(response) |
|
932 |
meta = json.loads(response.content)['meta'] |
|
933 |
self.assertEqual(meta['foo'], 'bar') |
|
934 |
|
|
935 |
@assert_backend_closed |
|
936 |
def test_get_invalid_metadata(self, backend): |
|
937 |
backend.return_value.get_image.return_value = self.image |
|
938 |
response = self.get('/api/v1.1/images/42/meta/not_found', 'user') |
|
939 |
self.assertItemNotFound(response) |
|
940 |
|
|
941 |
@assert_backend_closed |
|
942 |
def test_delete_metadata_item(self, backend): |
|
943 |
backend.return_value.get_image.return_value = self.image |
|
944 |
with patch("synnefo.api.images.ImageBackend") as m: |
|
945 |
response = self.delete('/api/v1.1/images/42/meta/foo', 'user') |
|
946 |
self.assertEqual(response.status_code, 204) |
|
947 |
m.return_value.update.assert_called_once_with('42', |
|
948 |
{'properties': {'foo2': 'bar2'}}) |
|
949 |
|
|
950 |
@assert_backend_closed |
|
951 |
def test_create_metadata_item(self, backend): |
|
952 |
backend.return_value.get_image.return_value = self.image |
|
953 |
with patch("synnefo.api.images.ImageBackend") as m: |
|
954 |
request = {'meta': {'foo3': 'bar3'}} |
|
955 |
response = self.put('/api/v1.1/images/42/meta/foo3', 'user', |
|
956 |
json.dumps(request), 'json') |
|
957 |
self.assertEqual(response.status_code, 201) |
|
958 |
m.return_value.update.assert_called_once_with('42', |
|
959 |
{'properties': |
|
960 |
{'foo': 'bar', 'foo2': 'bar2', 'foo3': 'bar3'}}) |
|
961 |
|
|
962 |
@assert_backend_closed |
|
963 |
def test_create_metadata_malformed_1(self, backend): |
|
964 |
backend.return_value.get_image.return_value = self.image |
|
965 |
with patch("synnefo.api.images.ImageBackend"): |
|
966 |
request = {'met': {'foo3': 'bar3'}} |
|
967 |
response = self.put('/api/v1.1/images/42/meta/foo3', 'user', |
|
968 |
json.dumps(request), 'json') |
|
969 |
self.assertBadRequest(response) |
|
970 |
|
|
971 |
@assert_backend_closed |
|
972 |
def test_create_metadata_malformed_2(self, backend): |
|
973 |
backend.return_value.get_image.return_value = self.image |
|
974 |
with patch("synnefo.api.images.ImageBackend"): |
|
975 |
request = {'meta': [('foo3', 'bar3')]} |
|
976 |
response = self.put('/api/v1.1/images/42/meta/foo3', 'user', |
|
977 |
json.dumps(request), 'json') |
|
978 |
self.assertBadRequest(response) |
|
979 |
|
|
980 |
@assert_backend_closed |
|
981 |
def test_create_metadata_malformed_3(self, backend): |
|
982 |
backend.return_value.get_image.return_value = self.image |
|
983 |
with patch("synnefo.api.images.ImageBackend"): |
|
984 |
request = {'met': {'foo3': 'bar3', 'foo4': 'bar4'}} |
|
985 |
response = self.put('/api/v1.1/images/42/meta/foo3', 'user', |
|
986 |
json.dumps(request), 'json') |
|
987 |
self.assertBadRequest(response) |
|
988 |
|
|
989 |
@assert_backend_closed |
|
990 |
def test_create_metadata_malformed_4(self, backend): |
|
991 |
backend.return_value.get_image.return_value = self.image |
|
992 |
with patch("synnefo.api.images.ImageBackend"): |
|
993 |
request = {'met': {'foo3': 'bar3'}} |
|
994 |
response = self.put('/api/v1.1/images/42/meta/foo4', 'user', |
|
995 |
json.dumps(request), 'json') |
|
996 |
self.assertBadRequest(response) |
|
997 |
|
|
998 |
@assert_backend_closed |
|
999 |
def test_update_metadata_item(self, backend): |
|
1000 |
backend.return_value.get_image.return_value = self.image |
|
1001 |
with patch("synnefo.api.images.ImageBackend") as m: |
|
1002 |
request = {'metadata': {'foo': 'bar_new', 'foo4': 'bar4'}} |
|
1003 |
response = self.post('/api/v1.1/images/42/meta', 'user', |
|
1004 |
json.dumps(request), 'json') |
|
1005 |
self.assertEqual(response.status_code, 201) |
|
1006 |
m.return_value.update.assert_called_once_with('42', |
|
1007 |
{'properties': |
|
1008 |
{'foo': 'bar_new', 'foo2': 'bar2', 'foo4': 'bar4'} |
|
1009 |
}) |
|
1010 |
|
|
1011 |
@assert_backend_closed |
|
1012 |
def test_update_metadata_malformed(self, backend): |
|
1013 |
backend.return_value.get_image.return_value = self.image |
|
1014 |
with patch("synnefo.api.images.ImageBackend"): |
|
1015 |
request = {'meta': {'foo': 'bar_new', 'foo4': 'bar4'}} |
|
1016 |
response = self.post('/api/v1.1/images/42/meta', 'user', |
|
1017 |
json.dumps(request), 'json') |
|
1018 |
self.assertBadRequest(response) |
|
1019 |
|
|
1020 |
|
|
1021 | 111 |
class APITest(TestCase): |
1022 | 112 |
def test_api_version(self): |
1023 | 113 |
"""Check API version.""" |
... | ... | |
1027 | 117 |
api_version = json.loads(response.content)['version'] |
1028 | 118 |
self.assertEqual(api_version['id'], 'v1.1') |
1029 | 119 |
self.assertEqual(api_version['status'], 'CURRENT') |
120 |
|
|
121 |
|
|
122 |
# Import TestCases |
|
123 |
from synnefo.api.test.servers import * |
|
124 |
from synnefo.api.test.networks import * |
|
125 |
from synnefo.api.test.flavors import * |
|
126 |
from synnefo.api.test.images import * |
Also available in: Unified diff