code.grnet.gr Git - kamaki/blob - kamaki/clients/cyclades/test.py

   1 # Copyright 2013 GRNET S.A. All rights reserved.
   2 #
   3 # Redistribution and use in source and binary forms, with or
   4 # without modification, are permitted provided that the following
   5 # conditions are met:
   6 #
   7 #   1. Redistributions of source code must retain the above
   8 #      copyright notice, this list of conditions and the following
   9 #      disclaimer.
  10 #
  11 #   2. Redistributions in binary form must reproduce the above
  12 #      copyright notice, this list of conditions and the following
  13 #      disclaimer in the documentation and/or other materials
  14 #      provided with the distribution.
  15 #
  16 # THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
  17 # OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  18 # WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  19 # PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
  20 # CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  21 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  22 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  23 # USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  24 # AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  25 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
  26 # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  27 # POSSIBILITY OF SUCH DAMAGE.
  28 #
  29 # The views and conclusions contained in the software and
  30 # documentation are those of the authors and should not be
  31 # interpreted as representing official policies, either expressed
  32 # or implied, of GRNET S.A.
  33
  34 from mock import patch, call
  35 from unittest import TestCase
  36 from itertools import product
  37 from json import dumps
  38
  39 from kamaki.clients import ClientError, cyclades
  40
  41 img_ref = "1m4g3-r3f3r3nc3"
  42 vm_name = "my new VM"
  43 fid = 42
  44 vm_recv = dict(server=dict(
  45     status="BUILD",
  46     updated="2013-03-01T10:04:00.637152+00:00",
  47     hostId="",
  48     name=vm_name,
  49     imageRef=img_ref,
  50     created="2013-03-01T10:04:00.087324+00:00",
  51     flavorRef=fid,
  52     adminPass="n0n3sh@11p@55",
  53     suspended=False,
  54     progress=0,
  55     id=31173,
  56     metadata=dict(os="debian", users="root")))
  57 vm_list = dict(servers=[
  58     dict(name='n1', id=1),
  59     dict(name='n2', id=2)])
  60 net_send = dict(network=dict(dhcp=False, name='someNet'))
  61 net_recv = dict(network=dict(
  62     status="PENDING",
  63     updated="2013-03-05T15:04:51.758780+00:00",
  64     name="someNet",
  65     created="2013-03-05T15:04:51.758728+00:00",
  66     cidr6=None,
  67     id="2130",
  68     gateway6=None,
  69     public=False,
  70     dhcp=False,
  71     cidr="192.168.1.0/24",
  72     type="MAC_FILTERED",
  73     gateway=None,
  74     attachments=[dict(name='att1'), dict(name='att2')]))
  75 net_list = dict(networks=[
  76     dict(id=1, name='n1'),
  77     dict(id=2, name='n2'),
  78     dict(id=3, name='n3')])
  79 firewalls = dict(attachments=[
  80     dict(firewallProfile='50m3_pr0f1L3', otherStuff='57uff')])
  81
  82
  83 class FR(object):
  84     """FR stands for Fake Response"""
  85     json = vm_recv
  86     headers = {}
  87     content = json
  88     status = None
  89     status_code = 200
  90
  91 rest_pkg = 'kamaki.clients.cyclades.CycladesRestClient'
  92 cyclades_pkg = 'kamaki.clients.cyclades.CycladesClient'
  93
  94
  95 class CycladesRestClient(TestCase):
  96
  97     """Set up a Cyclades thorough test"""
  98     def setUp(self):
  99         self.url = 'http://cyclades.example.com'
 100         self.token = 'cyc14d3s70k3n'
 101         self.client = cyclades.CycladesRestClient(self.url, self.token)
 102
 103     def tearDown(self):
 104         FR.json = vm_recv
 105
 106     @patch('%s.get' % rest_pkg, return_value=FR())
 107     def test_networks_get(self, get):
 108         for args in product(
 109                 ('', 'net_id'),
 110                 ('', 'cmd'),
 111                 (200, 204),
 112                 ({}, {'k': 'v'})):
 113             (srv_id, command, success, kwargs) = args
 114             self.client.networks_get(*args[:3], **kwargs)
 115             srv_str = '/%s' % srv_id if srv_id else ''
 116             cmd_str = '/%s' % command if command else ''
 117             self.assertEqual(get.mock_calls[-1], call(
 118                 '/networks%s%s' % (srv_str, cmd_str),
 119                 success=success,
 120                 **kwargs))
 121
 122     @patch('%s.delete' % rest_pkg, return_value=FR())
 123     def test_networks_delete(self, delete):
 124         for args in product(
 125                 ('', 'net_id'),
 126                 ('', 'cmd'),
 127                 (202, 204),
 128                 ({}, {'k': 'v'})):
 129             (srv_id, command, success, kwargs) = args
 130             self.client.networks_delete(*args[:3], **kwargs)
 131             srv_str = '/%s' % srv_id if srv_id else ''
 132             cmd_str = '/%s' % command if command else ''
 133             self.assertEqual(delete.mock_calls[-1], call(
 134                 '/networks%s%s' % (srv_str, cmd_str),
 135                 success=success,
 136                 **kwargs))
 137
 138     @patch('%s.set_header' % rest_pkg)
 139     @patch('%s.post' % rest_pkg, return_value=FR())
 140     def test_networks_post(self, post, SH):
 141         for args in product(
 142                 ('', 'net_id'),
 143                 ('', 'cmd'),
 144                 (None, [dict(json="data"), dict(data="json")]),
 145                 (202, 204),
 146                 ({}, {'k': 'v'})):
 147             (srv_id, command, json_data, success, kwargs) = args
 148             self.client.networks_post(*args[:4], **kwargs)
 149             vm_str = '/%s' % srv_id if srv_id else ''
 150             cmd_str = '/%s' % command if command else ''
 151             if json_data:
 152                 json_data = dumps(json_data)
 153                 self.assertEqual(SH.mock_calls[-2:], [
 154                     call('Content-Type', 'application/json'),
 155                     call('Content-Length', len(json_data))])
 156             self.assertEqual(post.mock_calls[-1], call(
 157                 '/networks%s%s' % (vm_str, cmd_str),
 158                 data=json_data, success=success,
 159                 **kwargs))
 160
 161     @patch('%s.set_header' % rest_pkg)
 162     @patch('%s.put' % rest_pkg, return_value=FR())
 163     def test_networks_put(self, put, SH):
 164         for args in product(
 165                 ('', 'net_id'),
 166                 ('', 'cmd'),
 167                 (None, [dict(json="data"), dict(data="json")]),
 168                 (202, 204),
 169                 ({}, {'k': 'v'})):
 170             (srv_id, command, json_data, success, kwargs) = args
 171             self.client.networks_put(*args[:4], **kwargs)
 172             vm_str = '/%s' % srv_id if srv_id else ''
 173             cmd_str = '/%s' % command if command else ''
 174             if json_data:
 175                 json_data = dumps(json_data)
 176                 self.assertEqual(SH.mock_calls[-2:], [
 177                     call('Content-Type', 'application/json'),
 178                     call('Content-Length', len(json_data))])
 179             self.assertEqual(put.mock_calls[-1], call(
 180                 '/networks%s%s' % (vm_str, cmd_str),
 181                 data=json_data, success=success,
 182                 **kwargs))
 183
 184     @patch('%s.get' % rest_pkg, return_value=FR())
 185     def test_floating_ip_pools_get(self, get):
 186         for args in product(
 187                 (200, 204),
 188                 ({}, {'k': 'v'})):
 189             success, kwargs = args
 190             r = self.client.floating_ip_pools_get(success, **kwargs)
 191             self.assertTrue(isinstance(r, FR))
 192             self.assertEqual(get.mock_calls[-1], call(
 193                 '/os-floating-ip-pools', success=success, **kwargs))
 194
 195     @patch('%s.get' % rest_pkg, return_value=FR())
 196     def test_floating_ips_get(self, get):
 197         for args in product(
 198                 ('fip', ''),
 199                 (200, 204),
 200                 ({}, {'k': 'v'})):
 201             fip, success, kwargs = args
 202             r = self.client.floating_ips_get(fip, success, **kwargs)
 203             self.assertTrue(isinstance(r, FR))
 204             expected = '' if not fip else '/%s' % fip
 205             self.assertEqual(get.mock_calls[-1], call(
 206                 '/os-floating-ips%s' % expected, success=success, **kwargs))
 207
 208     @patch('%s.set_header' % rest_pkg)
 209     @patch('%s.post' % rest_pkg, return_value=FR())
 210     def test_floating_ips_post(self, post, SH):
 211         for args in product(
 212                 (None, [dict(json="data"), dict(data="json")]),
 213                 ('fip', ''),
 214                 (202, 204),
 215                 ({}, {'k': 'v'})):
 216             json_data, fip, success, kwargs = args
 217             self.client.floating_ips_post(*args[:3], **kwargs)
 218             if json_data:
 219                 json_data = dumps(json_data)
 220                 self.assertEqual(SH.mock_calls[-2:], [
 221                     call('Content-Type', 'application/json'),
 222                     call('Content-Length', len(json_data))])
 223             expected = '' if not fip else '/%s' % fip
 224             self.assertEqual(post.mock_calls[-1], call(
 225                 '/os-floating-ips%s' % expected,
 226                 data=json_data, success=success,
 227                 **kwargs))
 228
 229     @patch('%s.delete' % rest_pkg, return_value=FR())
 230     def test_floating_ips_delete(self, delete):
 231         for args in product(
 232                 ('fip1', 'fip2'),
 233                 (200, 204),
 234                 ({}, {'k': 'v'})):
 235             fip, success, kwargs = args
 236             r = self.client.floating_ips_delete(fip, success, **kwargs)
 237             self.assertTrue(isinstance(r, FR))
 238             self.assertEqual(delete.mock_calls[-1], call(
 239                 '/os-floating-ips/%s' % fip, success=success, **kwargs))
 240
 241
 242 class CycladesNetworkClient(TestCase):
 243
 244     """Set up a ComputesRest thorough test"""
 245     def setUp(self):
 246         self.url = 'http://network.example.com'
 247         self.token = 'n2tw0rk70k3n'
 248         self.client = cyclades.CycladesNetworkClient(self.url, self.token)
 249
 250     def tearDown(self):
 251         FR.json = vm_recv
 252         del self.client
 253
 254     @patch('kamaki.clients.Client.get', return_value=FR)
 255     def test_list_networks(self, get):
 256         FR.json = dict(networks='ret val')
 257         for detail in (True, None):
 258             self.assertEqual(self.client.list_networks(detail), 'ret val')
 259             path = '/networks/detail' if detail else '/networks'
 260             self.assertEqual(get.mock_calls[-1], call(path, success=200))
 261
 262     @patch(
 263         'kamaki.clients.network.rest_api.NetworkRestClient.networks_post',
 264         return_value=FR())
 265     def test_create_network(self, networks_post):
 266         for name, shared in product((None, 'net name'), (None, True)):
 267             FR.json = dict(network='ret val')
 268             type = 'net type'
 269             self.assertEqual(
 270                 self.client.create_network(type, name=name, shared=shared),
 271                 'ret val')
 272             req = dict(type=type, admin_state_up=True)
 273             if name:
 274                 req['name'] = name
 275             if shared:
 276                 req['shared'] = shared
 277             expargs = dict(json_data=dict(network=req), success=201)
 278             self.assertEqual(networks_post.mock_calls[-1], call(**expargs))
 279
 280     @patch(
 281         'kamaki.clients.network.rest_api.NetworkRestClient.ports_post',
 282         return_value=FR)
 283     def test_create_port(self, ports_post):
 284         network_id, device_id, FR.json = 'netid', 'devid', dict(port='ret v')
 285         for name, sec_grp in product(('port name', None), ([1, 2, 3], None)):
 286             self.assertEqual(
 287                 self.client.create_port(
 288                     network_id, device_id,
 289                     name=name, security_groups=sec_grp),
 290                 'ret v')
 291             req = dict(network_id=network_id, device_id=device_id)
 292             if sec_grp:
 293                 req['security_groups'] = sec_grp
 294             if name:
 295                 req['name'] = name
 296             expargs = dict(json_data=dict(port=req), success=201)
 297             self.assertEqual(ports_post.mock_calls[-1], call(**expargs))
 298
 299
 300 class CycladesClient(TestCase):
 301
 302     def assert_dicts_are_equal(self, d1, d2):
 303         for k, v in d1.items():
 304             self.assertTrue(k in d2)
 305             if isinstance(v, dict):
 306                 self.assert_dicts_are_equal(v, d2[k])
 307             else:
 308                 self.assertEqual(unicode(v), unicode(d2[k]))
 309
 310     """Set up a Cyclades thorough test"""
 311     def setUp(self):
 312         self.url = 'http://cyclades.example.com'
 313         self.token = 'cyc14d3s70k3n'
 314         self.client = cyclades.CycladesClient(self.url, self.token)
 315
 316     def tearDown(self):
 317         FR.status_code = 200
 318         FR.json = vm_recv
 319
 320     @patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
 321     def test_shutdown_server(self, SP):
 322         vm_id = vm_recv['server']['id']
 323         self.client.shutdown_server(vm_id)
 324         SP.assert_called_once_with(
 325             vm_id, json_data=dict(shutdown=dict()), success=202)
 326
 327     @patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
 328     def test_start_server(self, SP):
 329         vm_id = vm_recv['server']['id']
 330         self.client.start_server(vm_id)
 331         SP.assert_called_once_with(
 332             vm_id, json_data=dict(start=dict()), success=202)
 333
 334     @patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
 335     def test_get_server_console(self, SP):
 336         cnsl = dict(console=dict(info1='i1', info2='i2', info3='i3'))
 337         FR.json = cnsl
 338         vm_id = vm_recv['server']['id']
 339         r = self.client.get_server_console(vm_id)
 340         SP.assert_called_once_with(
 341             vm_id, json_data=dict(console=dict(type='vnc')), success=200)
 342         self.assert_dicts_are_equal(r, cnsl['console'])
 343
 344     def test_get_firewall_profile(self):
 345         vm_id = vm_recv['server']['id']
 346         v = firewalls['attachments'][0]['firewallProfile']
 347         with patch.object(
 348                 cyclades.CycladesClient, 'get_server_details',
 349                 return_value=firewalls) as GSD:
 350             r = self.client.get_firewall_profile(vm_id)
 351             GSD.assert_called_once_with(vm_id)
 352             self.assertEqual(r, v)
 353         with patch.object(
 354                 cyclades.CycladesClient, 'get_server_details',
 355                 return_value=dict()):
 356             self.assertRaises(
 357                 ClientError,
 358                 self.client.get_firewall_profile,
 359                 vm_id)
 360
 361     @patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
 362     def test_set_firewall_profile(self, SP):
 363         vm_id = vm_recv['server']['id']
 364         v = firewalls['attachments'][0]['firewallProfile']
 365         self.client.set_firewall_profile(vm_id, v)
 366         SP.assert_called_once_with(vm_id, json_data=dict(
 367             firewallProfile=dict(profile=v)), success=202)
 368
 369     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 370     def test_create_network(self, NP):
 371         net_name = net_send['network']['name']
 372         FR.json = net_recv
 373         full_args = dict(
 374                 cidr='192.168.0.0/24',
 375                 gateway='192.168.0.1',
 376                 type='MAC_FILTERED',
 377                 dhcp=True)
 378         test_args = dict(full_args)
 379         test_args.update(dict(empty=None, full=None))
 380         net_exp = dict(dhcp=False, name=net_name, type='MAC_FILTERED')
 381         for arg, val in test_args.items():
 382             kwargs = {} if arg == 'empty' else full_args if (
 383                 arg == 'full') else {arg: val}
 384             expected = dict(network=dict(net_exp))
 385             expected['network'].update(kwargs)
 386             r = self.client.create_network(net_name, **kwargs)
 387             self.assertEqual(
 388                 NP.mock_calls[-1],
 389                 call(json_data=expected, success=202))
 390             self.assert_dicts_are_equal(r, net_recv['network'])
 391
 392     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 393     def test_connect_server(self, NP):
 394         vm_id = vm_recv['server']['id']
 395         net_id = net_recv['network']['id']
 396         self.client.connect_server(vm_id, net_id)
 397         NP.assert_called_once_with(
 398             net_id, 'action',
 399             json_data=dict(add=dict(serverRef=vm_id)))
 400
 401     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 402     def test_disconnect_server(self, NP):
 403         net_id, vm_id = net_recv['network']['id'], vm_recv['server']['id']
 404         nic_id = 'nic-%s-%s' % (net_id, vm_id)
 405         vm_nics = [
 406             dict(id=nic_id, network_id=net_id),
 407             dict(id='another-nic-id', network_id='another-net-id'),
 408             dict(id=nic_id * 2, network_id=net_id * 2)]
 409         with patch.object(
 410                 cyclades.CycladesClient,
 411                 'list_server_nics',
 412                 return_value=vm_nics) as LSN:
 413             r = self.client.disconnect_server(vm_id, nic_id)
 414             LSN.assert_called_once_with(vm_id)
 415             NP.assert_called_once_with(
 416                 net_id, 'action',
 417                 json_data=dict(remove=dict(attachment=nic_id)))
 418             self.assertEqual(r, 1)
 419
 420     @patch('%s.servers_ips_get' % cyclades_pkg, return_value=FR())
 421     def test_list_server_nics(self, SG):
 422         vm_id = vm_recv['server']['id']
 423         nics = dict(attachments=[dict(id='nic1'), dict(id='nic2')])
 424         FR.json = nics
 425         r = self.client.list_server_nics(vm_id)
 426         SG.assert_called_once_with(vm_id)
 427         expected = nics['attachments']
 428         for i in range(len(r)):
 429             self.assert_dicts_are_equal(r[i], expected[i])
 430         self.assertEqual(i + 1, len(r))
 431
 432     @patch('%s.networks_get' % cyclades_pkg, return_value=FR())
 433     def test_list_networks(self, NG):
 434         FR.json = net_list
 435         expected = net_list['networks']
 436         for detail in ('', 'detail'):
 437             r = self.client.list_networks(detail=True if detail else False)
 438             self.assertEqual(NG.mock_calls[-1], call(command=detail))
 439             for i, net in enumerate(expected):
 440                 self.assert_dicts_are_equal(r[i], net)
 441             self.assertEqual(i + 1, len(r))
 442
 443     @patch('%s.networks_get' % cyclades_pkg, return_value=FR())
 444     def test_list_network_nics(self, NG):
 445         net_id = net_recv['network']['id']
 446         FR.json = net_recv
 447         r = self.client.list_network_nics(net_id)
 448         NG.assert_called_once_with(network_id=net_id)
 449         expected = net_recv['network']['attachments']
 450         for i in range(len(r)):
 451             self.assert_dicts_are_equal(r[i], expected[i])
 452
 453     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 454     def test_disconnect_network_nics(self, NP):
 455         net_id = net_recv['network']['id']
 456         nics = ['nic1', 'nic2', 'nic3']
 457         with patch.object(
 458                 cyclades.CycladesClient,
 459                 'list_network_nics',
 460                 return_value=nics) as LNN:
 461             self.client.disconnect_network_nics(net_id)
 462             LNN.assert_called_once_with(net_id)
 463             for i in range(len(nics)):
 464                 expected = call(net_id, 'action', json_data=dict(
 465                     remove=dict(attachment=nics[i])))
 466                 self.assertEqual(expected, NP.mock_calls[i])
 467
 468     @patch('%s.networks_get' % cyclades_pkg, return_value=FR())
 469     def test_get_network_details(self, NG):
 470         FR.json = net_recv
 471         net_id = net_recv['network']['id']
 472         r = self.client.get_network_details(net_id)
 473         NG.assert_called_once_with(network_id=net_id)
 474         self.assert_dicts_are_equal(r, net_recv['network'])
 475
 476     @patch('%s.networks_put' % cyclades_pkg, return_value=FR())
 477     def test_update_network_name(self, NP):
 478         net_id = net_recv['network']['id']
 479         new_name = '%s_new' % net_id
 480         self.client.update_network_name(net_id, new_name)
 481         NP.assert_called_once_with(
 482             network_id=net_id,
 483             json_data=dict(network=dict(name=new_name)))
 484
 485     def test_delete_network(self):
 486         net_id = net_recv['network']['id']
 487         with patch.object(
 488                 cyclades.CycladesClient, 'networks_delete',
 489                 return_value=FR()) as ND:
 490             self.client.delete_network(net_id)
 491             ND.assert_called_once_with(net_id)
 492         with patch.object(
 493                 cyclades.CycladesClient, 'networks_delete',
 494                 side_effect=ClientError('A 421 Error', 421)):
 495             try:
 496                 self.client.delete_network(421)
 497             except ClientError as err:
 498                 self.assertEqual(err.status, 421)
 499                 self.assertEqual(err.details, [
 500                     'Network may be still connected to at least one server'])
 501
 502     @patch('%s.floating_ip_pools_get' % cyclades_pkg, return_value=FR())
 503     def test_get_floating_ip_pools(self, get):
 504         r = self.client.get_floating_ip_pools()
 505         self.assert_dicts_are_equal(r, FR.json)
 506         self.assertEqual(get.mock_calls[-1], call())
 507
 508     @patch('%s.floating_ips_get' % cyclades_pkg, return_value=FR())
 509     def test_get_floating_ips(self, get):
 510         r = self.client.get_floating_ips()
 511         self.assert_dicts_are_equal(r, FR.json)
 512         self.assertEqual(get.mock_calls[-1], call())
 513
 514     @patch('%s.floating_ips_post' % cyclades_pkg, return_value=FR())
 515     def test_alloc_floating_ip(self, post):
 516         FR.json = dict(floating_ip=dict(
 517             fixed_ip='fip',
 518             id=1,
 519             instance_id='lala',
 520             ip='102.0.0.1',
 521             pool='pisine'))
 522         for args in product(
 523                 (None, 'pisine'),
 524                 (None, 'Iwannanip')):
 525             r = self.client.alloc_floating_ip(*args)
 526             pool, address = args
 527             self.assert_dicts_are_equal(r, FR.json['floating_ip'])
 528             json_data = dict()
 529             if pool:
 530                 json_data['pool'] = pool
 531             if address:
 532                 json_data['address'] = address
 533             self.assertEqual(post.mock_calls[-1], call(json_data))
 534
 535     @patch('%s.floating_ips_get' % cyclades_pkg, return_value=FR())
 536     def test_get_floating_ip(self, get):
 537         FR.json = dict(floating_ip=dict(
 538             fixed_ip='fip',
 539             id=1,
 540             instance_id='lala',
 541             ip='102.0.0.1',
 542             pool='pisine'))
 543         self.assertRaises(AssertionError, self.client.get_floating_ip, None)
 544         fip = 'fip'
 545         r = self.client.get_floating_ip(fip)
 546         self.assert_dicts_are_equal(r, FR.json['floating_ip'])
 547         self.assertEqual(get.mock_calls[-1], call(fip))
 548
 549     @patch('%s.floating_ips_delete' % cyclades_pkg, return_value=FR())
 550     def test_delete_floating_ip(self, delete):
 551         self.assertRaises(AssertionError, self.client.delete_floating_ip, None)
 552         fip = 'fip'
 553         r = self.client.delete_floating_ip(fip)
 554         self.assert_dicts_are_equal(r, FR.headers)
 555         self.assertEqual(delete.mock_calls[-1], call(fip))
 556
 557     @patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
 558     def test_attach_floating_ip(self, spost):
 559         vmid, addr = 42, 'anIpAddress'
 560         for err, args in {
 561                 ValueError: ['not a server id', addr],
 562                 TypeError: [None, addr],
 563                 AssertionError: [vmid, None],
 564                 AssertionError: [vmid, '']}.items():
 565             self.assertRaises(
 566                 err, self.client.attach_floating_ip, *args)
 567         r = self.client.attach_floating_ip(vmid, addr)
 568         self.assert_dicts_are_equal(r, FR.headers)
 569         expected = dict(addFloatingIp=dict(address=addr))
 570         self.assertEqual(spost.mock_calls[-1], call(vmid, json_data=expected))
 571
 572     @patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
 573     def test_detach_floating_ip(self, spost):
 574         vmid, addr = 42, 'anIpAddress'
 575         for err, args in {
 576                 ValueError: ['not a server id', addr],
 577                 TypeError: [None, addr],
 578                 AssertionError: [vmid, None],
 579                 AssertionError: [vmid, '']}.items():
 580             self.assertRaises(
 581                 err, self.client.detach_floating_ip, *args)
 582         r = self.client.detach_floating_ip(vmid, addr)
 583         self.assert_dicts_are_equal(r, FR.headers)
 584         expected = dict(removeFloatingIp=dict(address=addr))
 585         self.assertEqual(spost.mock_calls[-1], call(vmid, json_data=expected))
 586
 587
 588 if __name__ == '__main__':
 589     from sys import argv
 590     from kamaki.clients.test import runTestCase
 591     not_found = True
 592     if not argv[1:] or argv[1] == 'CycladesClient':
 593         not_found = False
 594         runTestCase(CycladesNetworkClient, 'Cyclades Client', argv[2:])
 595     if not argv[1:] or argv[1] == 'CycladesNetworkClient':
 596         not_found = False
 597         runTestCase(CycladesNetworkClient, 'CycladesNetwork Client', argv[2:])
 598     if not argv[1:] or argv[1] == 'CycladesRestClient':
 599         not_found = False
 600         runTestCase(CycladesRestClient, 'CycladesRest Client', argv[2:])
 601     if not_found:
 602         print('TestCase %s not found' % argv[1])