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, fixed_ips in product(
 286                 ('port name', None),
 287                 ([1, 2, 3], None),
 288                 (
 289                     dict(subnet_id='sid', ip_address='ipa'),
 290                     dict(subnet_id='sid'), dict(ip_address='ipa'),
 291                     None)):
 292
 293             if fixed_ips:
 294                 diff = set(['subnet_id', 'ip_address']).difference(fixed_ips)
 295                 if diff:
 296                     self.assertRaises(
 297                         ValueError, self.client.create_port,
 298                         network_id, device_id,
 299                         name=name,
 300                         security_groups=sec_grp,
 301                         fixed_ips=fixed_ips)
 302                     continue
 303             self.assertEqual(
 304                 self.client.create_port(
 305                     network_id, device_id,
 306                     name=name, security_groups=sec_grp, fixed_ips=fixed_ips),
 307                 'ret v')
 308             req = dict(network_id=network_id, device_id=device_id)
 309             if sec_grp:
 310                 req['security_groups'] = sec_grp
 311             if name:
 312                 req['name'] = name
 313             if fixed_ips:
 314                 req['fixed_ips'] = fixed_ips
 315             expargs = dict(json_data=dict(port=req), success=201)
 316             self.assertEqual(ports_post.mock_calls[-1], call(**expargs))
 317
 318
 319 class CycladesClient(TestCase):
 320
 321     def assert_dicts_are_equal(self, d1, d2):
 322         for k, v in d1.items():
 323             self.assertTrue(k in d2)
 324             if isinstance(v, dict):
 325                 self.assert_dicts_are_equal(v, d2[k])
 326             else:
 327                 self.assertEqual(unicode(v), unicode(d2[k]))
 328
 329     """Set up a Cyclades thorough test"""
 330     def setUp(self):
 331         self.url = 'http://cyclades.example.com'
 332         self.token = 'cyc14d3s70k3n'
 333         self.client = cyclades.CycladesClient(self.url, self.token)
 334
 335     def tearDown(self):
 336         FR.status_code = 200
 337         FR.json = vm_recv
 338
 339     @patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
 340     def test_shutdown_server(self, SP):
 341         vm_id = vm_recv['server']['id']
 342         self.client.shutdown_server(vm_id)
 343         SP.assert_called_once_with(
 344             vm_id, json_data=dict(shutdown=dict()), success=202)
 345
 346     @patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
 347     def test_start_server(self, SP):
 348         vm_id = vm_recv['server']['id']
 349         self.client.start_server(vm_id)
 350         SP.assert_called_once_with(
 351             vm_id, json_data=dict(start=dict()), success=202)
 352
 353     @patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
 354     def test_get_server_console(self, SP):
 355         cnsl = dict(console=dict(info1='i1', info2='i2', info3='i3'))
 356         FR.json = cnsl
 357         vm_id = vm_recv['server']['id']
 358         r = self.client.get_server_console(vm_id)
 359         SP.assert_called_once_with(
 360             vm_id, json_data=dict(console=dict(type='vnc')), success=200)
 361         self.assert_dicts_are_equal(r, cnsl['console'])
 362
 363     def test_get_firewall_profile(self):
 364         vm_id = vm_recv['server']['id']
 365         v = firewalls['attachments'][0]['firewallProfile']
 366         with patch.object(
 367                 cyclades.CycladesClient, 'get_server_details',
 368                 return_value=firewalls) as GSD:
 369             r = self.client.get_firewall_profile(vm_id)
 370             GSD.assert_called_once_with(vm_id)
 371             self.assertEqual(r, v)
 372         with patch.object(
 373                 cyclades.CycladesClient, 'get_server_details',
 374                 return_value=dict()):
 375             self.assertRaises(
 376                 ClientError,
 377                 self.client.get_firewall_profile,
 378                 vm_id)
 379
 380     @patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
 381     def test_set_firewall_profile(self, SP):
 382         vm_id = vm_recv['server']['id']
 383         v = firewalls['attachments'][0]['firewallProfile']
 384         self.client.set_firewall_profile(vm_id, v)
 385         SP.assert_called_once_with(vm_id, json_data=dict(
 386             firewallProfile=dict(profile=v)), success=202)
 387
 388     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 389     def test_create_network(self, NP):
 390         net_name = net_send['network']['name']
 391         FR.json = net_recv
 392         full_args = dict(
 393                 cidr='192.168.0.0/24',
 394                 gateway='192.168.0.1',
 395                 type='MAC_FILTERED',
 396                 dhcp=True)
 397         test_args = dict(full_args)
 398         test_args.update(dict(empty=None, full=None))
 399         net_exp = dict(dhcp=False, name=net_name, type='MAC_FILTERED')
 400         for arg, val in test_args.items():
 401             kwargs = {} if arg == 'empty' else full_args if (
 402                 arg == 'full') else {arg: val}
 403             expected = dict(network=dict(net_exp))
 404             expected['network'].update(kwargs)
 405             r = self.client.create_network(net_name, **kwargs)
 406             self.assertEqual(
 407                 NP.mock_calls[-1],
 408                 call(json_data=expected, success=202))
 409             self.assert_dicts_are_equal(r, net_recv['network'])
 410
 411     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 412     def test_connect_server(self, NP):
 413         vm_id = vm_recv['server']['id']
 414         net_id = net_recv['network']['id']
 415         self.client.connect_server(vm_id, net_id)
 416         NP.assert_called_once_with(
 417             net_id, 'action',
 418             json_data=dict(add=dict(serverRef=vm_id)))
 419
 420     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 421     def test_disconnect_server(self, NP):
 422         net_id, vm_id = net_recv['network']['id'], vm_recv['server']['id']
 423         nic_id = 'nic-%s-%s' % (net_id, vm_id)
 424         vm_nics = [
 425             dict(id=nic_id, network_id=net_id),
 426             dict(id='another-nic-id', network_id='another-net-id'),
 427             dict(id=nic_id * 2, network_id=net_id * 2)]
 428         with patch.object(
 429                 cyclades.CycladesClient,
 430                 'list_server_nics',
 431                 return_value=vm_nics) as LSN:
 432             r = self.client.disconnect_server(vm_id, nic_id)
 433             LSN.assert_called_once_with(vm_id)
 434             NP.assert_called_once_with(
 435                 net_id, 'action',
 436                 json_data=dict(remove=dict(attachment=nic_id)))
 437             self.assertEqual(r, 1)
 438
 439     @patch('%s.servers_ips_get' % cyclades_pkg, return_value=FR())
 440     def test_list_server_nics(self, SG):
 441         vm_id = vm_recv['server']['id']
 442         nics = dict(attachments=[dict(id='nic1'), dict(id='nic2')])
 443         FR.json = nics
 444         r = self.client.list_server_nics(vm_id)
 445         SG.assert_called_once_with(vm_id)
 446         expected = nics['attachments']
 447         for i in range(len(r)):
 448             self.assert_dicts_are_equal(r[i], expected[i])
 449         self.assertEqual(i + 1, len(r))
 450
 451     @patch('%s.networks_get' % cyclades_pkg, return_value=FR())
 452     def test_list_networks(self, NG):
 453         FR.json = net_list
 454         expected = net_list['networks']
 455         for detail in ('', 'detail'):
 456             r = self.client.list_networks(detail=True if detail else False)
 457             self.assertEqual(NG.mock_calls[-1], call(command=detail))
 458             for i, net in enumerate(expected):
 459                 self.assert_dicts_are_equal(r[i], net)
 460             self.assertEqual(i + 1, len(r))
 461
 462     @patch('%s.networks_get' % cyclades_pkg, return_value=FR())
 463     def test_list_network_nics(self, NG):
 464         net_id = net_recv['network']['id']
 465         FR.json = net_recv
 466         r = self.client.list_network_nics(net_id)
 467         NG.assert_called_once_with(network_id=net_id)
 468         expected = net_recv['network']['attachments']
 469         for i in range(len(r)):
 470             self.assert_dicts_are_equal(r[i], expected[i])
 471
 472     @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
 473     def test_disconnect_network_nics(self, NP):
 474         net_id = net_recv['network']['id']
 475         nics = ['nic1', 'nic2', 'nic3']
 476         with patch.object(
 477                 cyclades.CycladesClient,
 478                 'list_network_nics',
 479                 return_value=nics) as LNN:
 480             self.client.disconnect_network_nics(net_id)
 481             LNN.assert_called_once_with(net_id)
 482             for i in range(len(nics)):
 483                 expected = call(net_id, 'action', json_data=dict(
 484                     remove=dict(attachment=nics[i])))
 485                 self.assertEqual(expected, NP.mock_calls[i])
 486
 487     @patch('%s.networks_get' % cyclades_pkg, return_value=FR())
 488     def test_get_network_details(self, NG):
 489         FR.json = net_recv
 490         net_id = net_recv['network']['id']
 491         r = self.client.get_network_details(net_id)
 492         NG.assert_called_once_with(network_id=net_id)
 493         self.assert_dicts_are_equal(r, net_recv['network'])
 494
 495     @patch('%s.networks_put' % cyclades_pkg, return_value=FR())
 496     def test_update_network_name(self, NP):
 497         net_id = net_recv['network']['id']
 498         new_name = '%s_new' % net_id
 499         self.client.update_network_name(net_id, new_name)
 500         NP.assert_called_once_with(
 501             network_id=net_id,
 502             json_data=dict(network=dict(name=new_name)))
 503
 504     def test_delete_network(self):
 505         net_id = net_recv['network']['id']
 506         with patch.object(
 507                 cyclades.CycladesClient, 'networks_delete',
 508                 return_value=FR()) as ND:
 509             self.client.delete_network(net_id)
 510             ND.assert_called_once_with(net_id)
 511         with patch.object(
 512                 cyclades.CycladesClient, 'networks_delete',
 513                 side_effect=ClientError('A 421 Error', 421)):
 514             try:
 515                 self.client.delete_network(421)
 516             except ClientError as err:
 517                 self.assertEqual(err.status, 421)
 518                 self.assertEqual(err.details, [
 519                     'Network may be still connected to at least one server'])
 520
 521     @patch('%s.floating_ip_pools_get' % cyclades_pkg, return_value=FR())
 522     def test_get_floating_ip_pools(self, get):
 523         r = self.client.get_floating_ip_pools()
 524         self.assert_dicts_are_equal(r, FR.json)
 525         self.assertEqual(get.mock_calls[-1], call())
 526
 527     @patch('%s.floating_ips_get' % cyclades_pkg, return_value=FR())
 528     def test_get_floating_ips(self, get):
 529         r = self.client.get_floating_ips()
 530         self.assert_dicts_are_equal(r, FR.json)
 531         self.assertEqual(get.mock_calls[-1], call())
 532
 533     @patch('%s.floating_ips_post' % cyclades_pkg, return_value=FR())
 534     def test_alloc_floating_ip(self, post):
 535         FR.json = dict(floating_ip=dict(
 536             fixed_ip='fip',
 537             id=1,
 538             instance_id='lala',
 539             ip='102.0.0.1',
 540             pool='pisine'))
 541         for args in product(
 542                 (None, 'pisine'),
 543                 (None, 'Iwannanip')):
 544             r = self.client.alloc_floating_ip(*args)
 545             pool, address = args
 546             self.assert_dicts_are_equal(r, FR.json['floating_ip'])
 547             json_data = dict()
 548             if pool:
 549                 json_data['pool'] = pool
 550             if address:
 551                 json_data['address'] = address
 552             self.assertEqual(post.mock_calls[-1], call(json_data))
 553
 554     @patch('%s.floating_ips_get' % cyclades_pkg, return_value=FR())
 555     def test_get_floating_ip(self, get):
 556         FR.json = dict(floating_ip=dict(
 557             fixed_ip='fip',
 558             id=1,
 559             instance_id='lala',
 560             ip='102.0.0.1',
 561             pool='pisine'))
 562         self.assertRaises(AssertionError, self.client.get_floating_ip, None)
 563         fip = 'fip'
 564         r = self.client.get_floating_ip(fip)
 565         self.assert_dicts_are_equal(r, FR.json['floating_ip'])
 566         self.assertEqual(get.mock_calls[-1], call(fip))
 567
 568     @patch('%s.floating_ips_delete' % cyclades_pkg, return_value=FR())
 569     def test_delete_floating_ip(self, delete):
 570         self.assertRaises(AssertionError, self.client.delete_floating_ip, None)
 571         fip = 'fip'
 572         r = self.client.delete_floating_ip(fip)
 573         self.assert_dicts_are_equal(r, FR.headers)
 574         self.assertEqual(delete.mock_calls[-1], call(fip))
 575
 576     @patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
 577     def test_attach_floating_ip(self, spost):
 578         vmid, addr = 42, 'anIpAddress'
 579         for err, args in {
 580                 ValueError: ['not a server id', addr],
 581                 TypeError: [None, addr],
 582                 AssertionError: [vmid, None],
 583                 AssertionError: [vmid, '']}.items():
 584             self.assertRaises(
 585                 err, self.client.attach_floating_ip, *args)
 586         r = self.client.attach_floating_ip(vmid, addr)
 587         self.assert_dicts_are_equal(r, FR.headers)
 588         expected = dict(addFloatingIp=dict(address=addr))
 589         self.assertEqual(spost.mock_calls[-1], call(vmid, json_data=expected))
 590
 591     @patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
 592     def test_detach_floating_ip(self, spost):
 593         vmid, addr = 42, 'anIpAddress'
 594         for err, args in {
 595                 ValueError: ['not a server id', addr],
 596                 TypeError: [None, addr],
 597                 AssertionError: [vmid, None],
 598                 AssertionError: [vmid, '']}.items():
 599             self.assertRaises(
 600                 err, self.client.detach_floating_ip, *args)
 601         r = self.client.detach_floating_ip(vmid, addr)
 602         self.assert_dicts_are_equal(r, FR.headers)
 603         expected = dict(removeFloatingIp=dict(address=addr))
 604         self.assertEqual(spost.mock_calls[-1], call(vmid, json_data=expected))
 605
 606
 607 if __name__ == '__main__':
 608     from sys import argv
 609     from kamaki.clients.test import runTestCase
 610     not_found = True
 611     if not argv[1:] or argv[1] == 'CycladesClient':
 612         not_found = False
 613         runTestCase(CycladesNetworkClient, 'Cyclades Client', argv[2:])
 614     if not argv[1:] or argv[1] == 'CycladesNetworkClient':
 615         not_found = False
 616         runTestCase(CycladesNetworkClient, 'CycladesNetwork Client', argv[2:])
 617     if not argv[1:] or argv[1] == 'CycladesRestClient':
 618         not_found = False
 619         runTestCase(CycladesRestClient, 'CycladesRest Client', argv[2:])
 620     if not_found:
 621         print('TestCase %s not found' % argv[1])