return self._wait(
server_id, current_status, get_status, delay, max_wait, wait_cb)
- def get_floating_ip_pools(self):
- """
- :returns: (dict) {floating_ip_pools:[{name: ...}, ...]}
- """
- r = self.floating_ip_pools_get()
- return r.json
-
- def get_floating_ips(self):
- """
- :returns: (dict) {floating_ips: [fixed_ip: , id: , ip: , pool: ]}
- """
- r = self.floating_ips_get()
- return r.json
-
- def alloc_floating_ip(self, pool=None, address=None):
- """
- :param pool: (str) pool of ips to allocate from
-
- :param address: (str) ip address to request
-
- :returns: (dict) {
- fixed_ip: ..., id: ..., instance_id: ..., ip: ..., pool: ...}
- """
- json_data = dict()
- if pool:
- json_data['pool'] = pool
- if address:
- json_data['address'] = address
- r = self.floating_ips_post(json_data)
- return r.json['floating_ip']
-
- def get_floating_ip(self, fip_id):
- """
- :param fip_id: (str) floating ip id
-
- :returns: (dict)
- {fixed_ip: ..., id: ..., instance_id: ..., ip: ..., pool: ...},
-
- :raises AssertionError: if fip_id is emtpy
- """
- assert fip_id, 'floating ip id is needed for get_floating_ip'
- r = self.floating_ips_get(fip_id)
- return r.json['floating_ip']
-
- def delete_floating_ip(self, fip_id=None):
- """
- :param fip_id: (str) floating ip id (if None, all ips are deleted)
-
- :returns: (dict) request headers
-
- :raises AssertionError: if fip_id is emtpy
- """
- assert fip_id, 'floating ip id is needed for delete_floating_ip'
- r = self.floating_ips_delete(fip_id)
- return r.headers
-
- def attach_floating_ip(self, server_id, address):
- """Associate the address ip to server with server_id
-
- :param server_id: (int)
-
- :param address: (str) the ip address to assign to server (vm)
-
- :returns: (dict) request headers
-
- :raises ValueError: if server_id cannot be converted to int
-
- :raises ValueError: if server_id is not of a int-convertable type
-
- :raises AssertionError: if address is emtpy
- """
- server_id = int(server_id)
- assert address, 'address is needed for attach_floating_ip'
- req = dict(addFloatingIp=dict(address=address))
- r = self.servers_action_post(server_id, json_data=req)
- return r.headers
-
- def detach_floating_ip(self, server_id, address):
- """Disassociate an address ip from the server with server_id
-
- :param server_id: (int)
-
- :param address: (str) the ip address to assign to server (vm)
-
- :returns: (dict) request headers
-
- :raises ValueError: if server_id cannot be converted to int
-
- :raises ValueError: if server_id is not of a int-convertable type
-
- :raises AssertionError: if address is emtpy
- """
- server_id = int(server_id)
- assert address, 'address is needed for detach_floating_ip'
- req = dict(removeFloatingIp=dict(address=address))
- r = self.servers_action_post(server_id, json_data=req)
- return r.headers
-
class CycladesNetworkClient(NetworkClient):
"""Cyclades Network API extentions"""
port['security_groups'] = security_groups
r = self.ports_put(port_id, json_data=dict(port=port), success=201)
return r.json['port']
+
+ def list_floatingip(self):
+ r = self.floatingips_get(success=200)
+ return r['floatingips']
+
+ def get_floatingip_details(self, floatingip_id):
+ r = self.floatingips_get(floatingip_id, success=201)
+ return r['floatingip']
+
+ def create_floatingip(
+ self, floating_network_id,
+ floating_ip_address='', port_id='', fixed_ip_address=''):
+ floatingip = dict(floating_network_id=floating_network_id)
+ if floating_ip_address != '':
+ floatingip['floating_ip_address'] = floating_ip_address
+ if port_id != '':
+ floatingip['port_id'] = port_id
+ if fixed_ip_address != '':
+ floatingip['fixed_ip_address'] = fixed_ip_address
+ r = self.floatingips_post(
+ json_data=dict(floatingip=floatingip), success=201)
+ return r.json['floatingip']
+
+ def update_floatingip(
+ self, floatingip_id,
+ floating_ip_address='', port_id='', fixed_ip_address=''):
+ """To nullify something optional, use None"""
+ floatingip = dict()
+ if floating_ip_address != '':
+ floatingip['floating_ip_address'] = floating_ip_address
+ if port_id != '':
+ floatingip['port_id'] = port_id
+ if fixed_ip_address != '':
+ floatingip['fixed_ip_address'] = fixed_ip_address
+ r = self.floatingips_put(
+ floatingip_id, json_data=dict(floatingip=floatingip), success=201)
+ return r['floatingip']
+
+ def delete_floatingip(self, floatingip_id):
+ r = self.floatingips_delete(floatingip_id, success=204)
+ return r.headers
def ports_delete(self, port_id, **kwargs):
return self.delete(path4url('ports', port_id), **kwargs)
+
+ # floatingips (L3) extentions
+
+ def floatingips_get(self, floatingip_id=None, **kwargs):
+ return self.get(path4url('floatingips', floatingip_id or ''), **kwargs)
+
+ def floatingips_post(self, json_data, **kwargs):
+ return self.post(path4url('floatingips'), json=json_data, **kwargs)
+
+ def floatingips_put(self, floatingip_id, json_data, **kwargs):
+ return self.put(
+ path4url('floatingips', floatingip_id), json=json_data, **kwargs)
+
+ def floatingips_delete(self, floatingip_id, **kwargs):
+ return self.delete(path4url('floatingips', floatingip_id), **kwargs)
'ret val')
self._assert(put, '/ports/%s' % port_id, json=json_data, **kwargs)
+ @patch('kamaki.clients.Client.get', return_value='ret val')
+ def test_floatingips_get(self, get):
+ for kwargs in (dict(), dict(k1='v1'), dict(k2='v2', k3='v3')):
+ self.assertEqual(self.client.floatingips_get(**kwargs), 'ret val')
+ self._assert(get, '/floatingips', **kwargs)
+
+ floatingip_id = 'port id'
+ self.assertEqual(
+ self.client.floatingips_get(
+ floatingip_id=floatingip_id, **kwargs),
+ 'ret val')
+ self._assert(get, '/floatingips/%s' % floatingip_id, **kwargs)
+
+ @patch('kamaki.clients.Client.post', return_value='ret val')
+ def test_floatingips_post(self, post):
+ for kwargs in (dict(), dict(k1='v1'), dict(k2='v2', k3='v3')):
+ json_data = dict(id='some id', other_param='other val')
+ self.assertEqual(
+ self.client.floatingips_post(json_data=json_data, **kwargs),
+ 'ret val')
+ self._assert(post, '/floatingips', json=json_data, **kwargs)
+
+ @patch('kamaki.clients.Client.put', return_value='ret val')
+ def test_floatingips_put(self, put):
+ floatingip_id = 'portid'
+ for kwargs in (dict(), dict(k1='v1'), dict(k2='v2', k3='v3')):
+ json_data = dict(id='some id', other_param='other val')
+ self.assertEqual(
+ self.client.floatingips_put(
+ floatingip_id, json_data=json_data, **kwargs),
+ 'ret val')
+ self._assert(
+ put, '/floatingips/%s' % floatingip_id,
+ json=json_data, **kwargs)
+
class FakeObject(object):
projects / kamaki / commitdiff