Revision 0651cda3 kamaki/clients/compute/test.py
b/kamaki/clients/compute/test.py | ||
---|---|---|
307 | 307 |
call('%s' % img_ref, '/meta/%s' % key)) |
308 | 308 |
self.assert_dicts_are_equal(img_recv['image'], r) |
309 | 309 |
|
310 |
""" |
|
311 |
@patch('%s.perform_request' % compute_pkg, return_value=FR()) |
|
312 |
def test_shutdown_server(self, PR): |
|
313 |
vm_id = vm_recv['server']['id'] |
|
314 |
FR.status_code = 202 |
|
315 |
self.client.shutdown_server(vm_id) |
|
316 |
self.assertEqual(self.client.http_client.url, self.url) |
|
317 |
self.assertEqual( |
|
318 |
self.client.http_client.path, |
|
319 |
'/servers/%s/action' % vm_id) |
|
320 |
self.assertEqual( |
|
321 |
PR.call_args[0], |
|
322 |
('post', '{"shutdown": {}}', {}, {})) |
|
323 |
|
|
324 |
@patch('%s.perform_request' % compute_pkg, return_value=FR()) |
|
325 |
def test_start_server(self, PR): |
|
326 |
vm_id = vm_recv['server']['id'] |
|
327 |
FR.status_code = 202 |
|
328 |
self.client.start_server(vm_id) |
|
329 |
self.assertEqual(self.client.http_client.url, self.url) |
|
330 |
self.assertEqual( |
|
331 |
self.client.http_client.path, |
|
332 |
'/servers/%s/action' % vm_id) |
|
333 |
self.assertEqual(PR.call_args[0], ('post', '{"start": {}}', {}, {})) |
|
334 |
|
|
335 |
@patch('%s.perform_request' % compute_pkg, return_value=FR()) |
|
336 |
def test_get_server_console(self, PR): |
|
337 |
cnsl = dict(console=dict(info1='i1', info2='i2', info3='i3')) |
|
338 |
FR.json = cnsl |
|
339 |
vm_id = vm_recv['server']['id'] |
|
340 |
r = self.client.get_server_console(vm_id) |
|
341 |
self.assertEqual(self.client.http_client.url, self.url) |
|
342 |
self.assertEqual( |
|
343 |
self.client.http_client.path, |
|
344 |
'/servers/%s/action' % vm_id) |
|
345 |
self.assert_dicts_are_equal(cnsl['console'], r) |
|
346 |
self.assertEqual( |
|
347 |
PR.call_args[0], |
|
348 |
('post', '{"console": {"type": "vnc"}}', {}, {})) |
|
349 |
|
|
350 |
def test_get_firewall_profile(self): |
|
351 |
vm_id = vm_recv['server']['id'] |
|
352 |
v = 'Some profile' |
|
353 |
ret = {'attachments': {'values': [{'firewallProfile': v, 1:1}]}} |
|
354 |
with patch.object( |
|
355 |
CycladesClient, |
|
356 |
'get_server_details', |
|
357 |
return_value=ret) as GSD: |
|
358 |
r = self.client.get_firewall_profile(vm_id) |
|
359 |
self.assertEqual(r, v) |
|
360 |
self.assertEqual(GSD.call_args[0], (vm_id,)) |
|
361 |
ret['attachments']['values'][0].pop('firewallProfile') |
|
362 |
self.assertRaises( |
|
363 |
ClientError, |
|
364 |
self.client.get_firewall_profile, |
|
365 |
vm_id) |
|
366 |
|
|
367 |
@patch('%s.perform_request' % compute_pkg, return_value=FR()) |
|
368 |
def test_set_firewall_profile(self, PR): |
|
369 |
vm_id = vm_recv['server']['id'] |
|
370 |
v = 'Some profile' |
|
371 |
FR.status_code = 202 |
|
372 |
self.client.set_firewall_profile(vm_id, v) |
|
373 |
self.assertEqual(self.client.http_client.url, self.url) |
|
374 |
self.assertEqual( |
|
375 |
self.client.http_client.path, |
|
376 |
'/servers/%s/action' % vm_id) |
|
377 |
self.assertEqual(PR.call_args[0], ( |
|
378 |
'post', |
|
379 |
'{"firewallProfile": {"profile": "%s"}}' % v, |
|
380 |
{}, |
|
381 |
{})) |
|
382 |
|
|
383 |
@patch('%s.perform_request' % compute_pkg, return_value=FR()) |
|
384 |
def test_get_server_stats(self, PR): |
|
385 |
vm_id = vm_recv['server']['id'] |
|
386 |
stats = dict(stat1='v1', stat2='v2', stat3='v3', stat4='v4') |
|
387 |
FR.json = dict(stats=stats) |
|
388 |
r = self.client.get_server_stats(vm_id) |
|
389 |
self.assertEqual(self.client.http_client.url, self.url) |
|
390 |
self.assertEqual( |
|
391 |
self.client.http_client.path, |
|
392 |
'/servers/%s/stats' % vm_id) |
|
393 |
self.assert_dicts_are_equal(stats, r) |
|
394 |
|
|
395 |
@patch('%s.perform_request' % compute_pkg, return_value=FR()) |
|
396 |
def test_create_network(self, PR): |
|
397 |
net_name = net_send['network']['name'] |
|
398 |
FR.json = net_recv |
|
399 |
FR.status_code = 202 |
|
400 |
full_args = dict( |
|
401 |
cidr='192.168.0.0/24', |
|
402 |
gateway='192.168.0.1', |
|
403 |
type='MAC_FILTERED', |
|
404 |
dhcp=True) |
|
405 |
test_args = dict(full_args) |
|
406 |
test_args.update(dict(empty=None, full=None)) |
|
407 |
for arg, val in test_args.items(): |
|
408 |
kwargs = {} if arg == 'empty' else full_args if ( |
|
409 |
arg == 'full') else {arg: val} |
|
410 |
r = self.client.create_network(net_name, **kwargs) |
|
411 |
self.assertEqual(self.client.http_client.url, self.url) |
|
412 |
self.assertEqual( |
|
413 |
self.client.http_client.path, |
|
414 |
'/networks') |
|
415 |
self.assert_dicts_are_equal(r, net_recv['network']) |
|
416 |
data = PR.call_args[0][1] |
|
417 |
expected = dict(network=dict(net_send['network'])) |
|
418 |
expected['network'].update(kwargs) |
|
419 |
self.assert_dicts_are_equal(loads(data), expected) |
|
420 |
|
|
421 |
@patch('%s.perform_request' % compute_pkg, return_value=FR()) |
|
422 |
def test_connect_server(self, PR): |
|
423 |
vm_id = vm_recv['server']['id'] |
|
424 |
net_id = net_recv['network']['id'] |
|
425 |
FR.status_code = 202 |
|
426 |
self.client.connect_server(vm_id, net_id) |
|
427 |
self.assertEqual(self.client.http_client.url, self.url) |
|
428 |
self.assertEqual( |
|
429 |
self.client.http_client.path, |
|
430 |
'/networks/%s/action' % net_id) |
|
431 |
self.assertEqual( |
|
432 |
PR.call_args[0], |
|
433 |
('post', '{"add": {"serverRef": %s}}' % vm_id, {}, {})) |
|
434 |
|
|
435 |
@patch('%s.networks_post' % compute_pkg, return_value=FR()) |
|
436 |
def test_disconnect_server(self, NP): |
|
437 |
vm_id = vm_recv['server']['id'] |
|
438 |
net_id = net_recv['network']['id'] |
|
439 |
nic_id = 'nic-%s-%s' % (net_id, vm_id) |
|
440 |
vm_nics = [ |
|
441 |
dict(id=nic_id, network_id=net_id), |
|
442 |
dict(id='another-nic-id', network_id='another-net-id'), |
|
443 |
dict(id=nic_id * 2, network_id=net_id * 2)] |
|
444 |
with patch.object( |
|
445 |
CycladesClient, |
|
446 |
'list_server_nics', |
|
447 |
return_value=vm_nics) as LSN: |
|
448 |
r = self.client.disconnect_server(vm_id, nic_id) |
|
449 |
self.assertEqual(r, 1) |
|
450 |
self.assertEqual(LSN.call_args[0], (vm_id,)) |
|
451 |
self.assertEqual(NP.call_args[0], (net_id, 'action')) |
|
452 |
self.assertEqual( |
|
453 |
NP.call_args[1], |
|
454 |
dict(json_data=dict(remove=dict(attachment=nic_id)))) |
|
455 |
|
|
456 |
@patch('%s.perform_request' % compute_pkg, return_value=FR()) |
|
457 |
def test_list_server_nics(self, PR): |
|
458 |
vm_id = vm_recv['server']['id'] |
|
459 |
nics = dict(addresses=dict(values=[dict(id='nic1'), dict(id='nic2')])) |
|
460 |
FR.json = nics |
|
461 |
r = self.client.list_server_nics(vm_id) |
|
462 |
self.assertEqual(self.client.http_client.url, self.url) |
|
463 |
self.assertEqual( |
|
464 |
self.client.http_client.path, |
|
465 |
'/servers/%s/ips' % vm_id) |
|
466 |
expected = nics['addresses']['values'] |
|
467 |
for i in range(len(r)): |
|
468 |
self.assert_dicts_are_equal(r[i], expected[i]) |
|
469 |
|
|
470 |
@patch('%s.perform_request' % compute_pkg, return_value=FR()) |
|
471 |
def test_list_networks(self, PR): |
|
472 |
FR.json = net_list |
|
473 |
r = self.client.list_networks() |
|
474 |
self.assertEqual(self.client.http_client.url, self.url) |
|
475 |
self.assertEqual(self.client.http_client.path, '/networks') |
|
476 |
expected = net_list['networks']['values'] |
|
477 |
for i in range(len(r)): |
|
478 |
self.assert_dicts_are_equal(expected[i], r[i]) |
|
479 |
self.client.list_networks(detail=True) |
|
480 |
self.assertEqual(self.client.http_client.url, self.url) |
|
481 |
self.assertEqual(self.client.http_client.path, '/networks/detail') |
|
482 |
|
|
483 |
@patch('%s.perform_request' % compute_pkg, return_value=FR()) |
|
484 |
def test_list_network_nics(self, PR): |
|
485 |
net_id = net_recv['network']['id'] |
|
486 |
FR.json = net_recv |
|
487 |
r = self.client.list_network_nics(net_id) |
|
488 |
self.assertEqual(self.client.http_client.url, self.url) |
|
489 |
self.assertEqual( |
|
490 |
self.client.http_client.path, |
|
491 |
'/networks/%s' % net_id) |
|
492 |
expected = net_recv['network']['attachments']['values'] |
|
493 |
for i in range(len(r)): |
|
494 |
self.assert_dicts_are_equal(r[i], expected[i]) |
|
495 |
|
|
496 |
@patch('%s.networks_post' % compute_pkg, return_value=FR()) |
|
497 |
def test_disconnect_network_nics(self, NP): |
|
498 |
net_id = net_recv['network']['id'] |
|
499 |
nics = ['nic1', 'nic2', 'nic3'] |
|
500 |
with patch.object( |
|
501 |
CycladesClient, |
|
502 |
'list_network_nics', |
|
503 |
return_value=nics) as lnn: |
|
504 |
self.client.disconnect_network_nics(net_id) |
|
505 |
lnn.assert_called_once_with(net_id) |
|
506 |
for i in range(len(nics)): |
|
507 |
expected = call(net_id, 'action', json_data=dict( |
|
508 |
remove=dict(attachment=nics[i]))) |
|
509 |
self.assertEqual(expected, NP.mock_calls[i]) |
|
510 |
|
|
511 |
@patch('%s.perform_request' % compute_pkg, return_value=FR()) |
|
512 |
def test_get_network_details(self, PR): |
|
513 |
FR.json = net_recv |
|
514 |
net_id = net_recv['network']['id'] |
|
515 |
r = self.client.get_network_details(net_id) |
|
516 |
self.assertEqual(self.client.http_client.url, self.url) |
|
517 |
self.assertEqual( |
|
518 |
self.client.http_client.path, |
|
519 |
'/networks/%s' % net_id) |
|
520 |
self.assert_dicts_are_equal(r, net_recv['network']) |
|
521 |
|
|
522 |
@patch('%s.perform_request' % compute_pkg, return_value=FR()) |
|
523 |
def test_update_network_name(self, PR): |
|
524 |
net_id = net_recv['network']['id'] |
|
525 |
new_name = '%s_new' % net_id |
|
526 |
FR.status_code = 204 |
|
527 |
self.client.update_network_name(net_id, new_name) |
|
528 |
self.assertEqual(self.client.http_client.url, self.url) |
|
529 |
self.assertEqual(self.client.http_client.path, '/networks/%s' % net_id) |
|
530 |
(method, data, a_headers, a_params) = PR.call_args[0] |
|
531 |
self.assert_dicts_are_equal( |
|
532 |
dict(network=dict(name=new_name)), |
|
533 |
loads(data)) |
|
534 |
|
|
535 |
@patch('%s.perform_request' % compute_pkg, return_value=FR()) |
|
536 |
def test_delete_server(self, PR): |
|
310 |
@patch('%s.servers_delete' % compute_pkg, return_value=FR()) |
|
311 |
def test_delete_server(self, SD): |
|
537 | 312 |
vm_id = vm_recv['server']['id'] |
538 |
FR.status_code = 204 |
|
539 | 313 |
self.client.delete_server(vm_id) |
540 |
self.assertEqual(self.client.http_client.url, self.url) |
|
541 |
self.assertEqual(self.client.http_client.path, '/servers/%s' % vm_id) |
|
314 |
self.assertEqual(SD.mock_calls[-1], call(vm_id)) |
|
542 | 315 |
|
316 |
""" |
|
543 | 317 |
@patch('%s.perform_request' % compute_pkg, return_value=FR()) |
544 | 318 |
def test_delete_image(self, PR): |
545 | 319 |
FR.status_code = 204 |
Also available in: Unified diff