103 |
103 |
def tearDown(self):
|
104 |
104 |
FR.json = vm_recv
|
105 |
105 |
|
106 |
|
@patch('%s.set_param' % rest_pkg)
|
107 |
|
@patch('%s.get' % rest_pkg, return_value=FR())
|
108 |
|
def test_servers_get(self, get, SP):
|
109 |
|
for args in product(
|
110 |
|
('', 'vm_id'),
|
111 |
|
('', 'cmd'),
|
112 |
|
(200, 204),
|
113 |
|
(None, '50m3-d473'),
|
114 |
|
({}, {'k': 'v'})):
|
115 |
|
(srv_id, command, success, changes_since, kwargs) = args
|
116 |
|
self.client.servers_get(*args[:4], **kwargs)
|
117 |
|
srv_str = '/%s' % srv_id if srv_id else ''
|
118 |
|
cmd_str = '/%s' % command if command else ''
|
119 |
|
self.assertEqual(get.mock_calls[-1], call(
|
120 |
|
'/servers%s%s' % (srv_str, cmd_str),
|
121 |
|
success=success,
|
122 |
|
**kwargs))
|
123 |
|
if changes_since:
|
124 |
|
self.assertEqual(
|
125 |
|
SP.mock_calls[-1],
|
126 |
|
call('changes-since', changes_since, changes_since))
|
127 |
|
|
128 |
106 |
@patch('%s.get' % rest_pkg, return_value=FR())
|
129 |
107 |
def test_networks_get(self, get):
|
130 |
108 |
for args in product(
|
... | ... | |
281 |
259 |
FR.status_code = 200
|
282 |
260 |
FR.json = vm_recv
|
283 |
261 |
|
284 |
|
@patch('%s.servers_get' % cyclades_pkg, return_value=FR())
|
285 |
|
def test_list_servers(self, SG):
|
286 |
|
FR.json = vm_list
|
287 |
|
for detail, since in ((0, 0), (True, 0), (0, 'd473'), (True, 'd473')):
|
288 |
|
r = self.client.list_servers(detail=detail, changes_since=since)
|
289 |
|
self.assertEqual(SG.mock_calls[-1], call(
|
290 |
|
command='detail' if detail else '',
|
291 |
|
changes_since=since))
|
292 |
|
expected = vm_list['servers']
|
293 |
|
for i, vm in enumerate(r):
|
294 |
|
self.assert_dicts_are_equal(vm, expected[i])
|
295 |
|
self.assertEqual(i + 1, len(expected))
|
296 |
|
|
297 |
|
@patch('%s.servers_post' % cyclades_pkg, return_value=FR())
|
|
262 |
@patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
|
298 |
263 |
def test_shutdown_server(self, SP):
|
299 |
264 |
vm_id = vm_recv['server']['id']
|
300 |
265 |
self.client.shutdown_server(vm_id)
|
301 |
266 |
SP.assert_called_once_with(
|
302 |
|
vm_id, 'action',
|
303 |
|
json_data=dict(shutdown=dict()), success=202)
|
|
267 |
vm_id, json_data=dict(shutdown=dict()), success=202)
|
304 |
268 |
|
305 |
|
@patch('%s.servers_post' % cyclades_pkg, return_value=FR())
|
|
269 |
@patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
|
306 |
270 |
def test_start_server(self, SP):
|
307 |
271 |
vm_id = vm_recv['server']['id']
|
308 |
272 |
self.client.start_server(vm_id)
|
309 |
273 |
SP.assert_called_once_with(
|
310 |
|
vm_id, 'action',
|
311 |
|
json_data=dict(start=dict()), success=202)
|
|
274 |
vm_id, json_data=dict(start=dict()), success=202)
|
312 |
275 |
|
313 |
|
@patch('%s.servers_post' % cyclades_pkg, return_value=FR())
|
|
276 |
@patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
|
314 |
277 |
def test_get_server_console(self, SP):
|
315 |
278 |
cnsl = dict(console=dict(info1='i1', info2='i2', info3='i3'))
|
316 |
279 |
FR.json = cnsl
|
317 |
280 |
vm_id = vm_recv['server']['id']
|
318 |
281 |
r = self.client.get_server_console(vm_id)
|
319 |
282 |
SP.assert_called_once_with(
|
320 |
|
vm_id, 'action',
|
321 |
|
json_data=dict(console=dict(type='vnc')), success=200)
|
|
283 |
vm_id, json_data=dict(console=dict(type='vnc')), success=200)
|
322 |
284 |
self.assert_dicts_are_equal(r, cnsl['console'])
|
323 |
285 |
|
324 |
286 |
def test_get_firewall_profile(self):
|
... | ... | |
338 |
300 |
self.client.get_firewall_profile,
|
339 |
301 |
vm_id)
|
340 |
302 |
|
341 |
|
@patch('%s.servers_post' % cyclades_pkg, return_value=FR())
|
|
303 |
@patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
|
342 |
304 |
def test_set_firewall_profile(self, SP):
|
343 |
305 |
vm_id = vm_recv['server']['id']
|
344 |
306 |
v = firewalls['attachments'][0]['firewallProfile']
|
345 |
307 |
self.client.set_firewall_profile(vm_id, v)
|
346 |
|
SP.assert_called_once_with(
|
347 |
|
vm_id, 'action',
|
348 |
|
json_data=dict(firewallProfile=dict(profile=v)), success=202)
|
349 |
|
|
350 |
|
@patch('%s.servers_get' % cyclades_pkg, return_value=FR())
|
351 |
|
def test_get_server_stats(self, SG):
|
352 |
|
vm_id = vm_recv['server']['id']
|
353 |
|
stats = dict(stat1='v1', stat2='v2', stat3='v3', stat4='v4')
|
354 |
|
FR.json = dict(stats=stats)
|
355 |
|
r = self.client.get_server_stats(vm_id)
|
356 |
|
SG.assert_called_once_with(vm_id, 'stats')
|
357 |
|
self.assert_dicts_are_equal(stats, r)
|
|
308 |
SP.assert_called_once_with(vm_id, json_data=dict(
|
|
309 |
firewallProfile=dict(profile=v)), success=202)
|
358 |
310 |
|
359 |
311 |
@patch('%s.networks_post' % cyclades_pkg, return_value=FR())
|
360 |
312 |
def test_create_network(self, NP):
|
... | ... | |
407 |
359 |
json_data=dict(remove=dict(attachment=nic_id)))
|
408 |
360 |
self.assertEqual(r, 1)
|
409 |
361 |
|
410 |
|
@patch('%s.servers_get' % cyclades_pkg, return_value=FR())
|
|
362 |
@patch('%s.servers_ips_get' % cyclades_pkg, return_value=FR())
|
411 |
363 |
def test_list_server_nics(self, SG):
|
412 |
364 |
vm_id = vm_recv['server']['id']
|
413 |
365 |
nics = dict(attachments=[dict(id='nic1'), dict(id='nic2')])
|
414 |
366 |
FR.json = nics
|
415 |
367 |
r = self.client.list_server_nics(vm_id)
|
416 |
|
SG.assert_called_once_with(vm_id, 'ips')
|
|
368 |
SG.assert_called_once_with(vm_id)
|
417 |
369 |
expected = nics['attachments']
|
418 |
370 |
for i in range(len(r)):
|
419 |
371 |
self.assert_dicts_are_equal(r[i], expected[i])
|
... | ... | |
544 |
496 |
self.assert_dicts_are_equal(r, FR.headers)
|
545 |
497 |
self.assertEqual(delete.mock_calls[-1], call(fip))
|
546 |
498 |
|
547 |
|
@patch('%s.servers_post' % cyclades_pkg, return_value=FR())
|
|
499 |
@patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
|
548 |
500 |
def test_attach_floating_ip(self, spost):
|
549 |
501 |
vmid, addr = 42, 'anIpAddress'
|
550 |
502 |
for err, args in {
|
... | ... | |
557 |
509 |
r = self.client.attach_floating_ip(vmid, addr)
|
558 |
510 |
self.assert_dicts_are_equal(r, FR.headers)
|
559 |
511 |
expected = dict(addFloatingIp=dict(address=addr))
|
560 |
|
self.assertEqual(
|
561 |
|
spost.mock_calls[-1], call(vmid, 'action', json_data=expected))
|
|
512 |
self.assertEqual(spost.mock_calls[-1], call(vmid, json_data=expected))
|
562 |
513 |
|
563 |
|
@patch('%s.servers_post' % cyclades_pkg, return_value=FR())
|
|
514 |
@patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
|
564 |
515 |
def test_detach_floating_ip(self, spost):
|
565 |
516 |
vmid, addr = 42, 'anIpAddress'
|
566 |
517 |
for err, args in {
|
... | ... | |
573 |
524 |
r = self.client.detach_floating_ip(vmid, addr)
|
574 |
525 |
self.assert_dicts_are_equal(r, FR.headers)
|
575 |
526 |
expected = dict(removeFloatingIp=dict(address=addr))
|
576 |
|
self.assertEqual(
|
577 |
|
spost.mock_calls[-1], call(vmid, 'action', json_data=expected))
|
|
527 |
self.assertEqual(spost.mock_calls[-1], call(vmid, json_data=expected))
|
578 |
528 |
|
579 |
529 |
|
580 |
530 |
if __name__ == '__main__':
|