Changes:
Features:
-
+- Implement floating ip methods for compute and cyclades clients [#3862]
+ ComputeRestClient: floating_ip_pools_get, floating_ips_get/post/delete
+ CycladesRestClient: floating_ip_pools_get, floating_ips_get/post/delete
+ ComputeClient: get_floating_ip_pools, get_floating_ips,
+ alloc/get_delete_floating_ip
+ CycladesClient: get_floating_ip_pools, get_floating_ips,
+ alloc/get_delete_floating_ip, dis/assoc_floating_ip_to_server
command = path4url('metadata', key)
r = self.images_delete(image_id, command)
return r.headers
+
+ def get_floating_ip_pools(self, tenant_id):
+ """
+ :param tenant_id: (str)
+
+ :returns: (dict) {floating_ip_pools:[{name: ...}, ...]}
+ """
+ r = self.floating_ip_pools_get(tenant_id)
+ return r.json
+
+ def get_floating_ips(self, tenant_id):
+ """
+ :param tenant_id: (str)
+
+ :returns: (dict) {floating_ips:[
+ {fixed_ip: ..., id: ..., instance_id: ..., ip: ..., pool: ...},
+ ... ]}
+ """
+ r = self.floating_ips_get(tenant_id)
+ return r.json
+
+ def alloc_floating_ip(self, tenant_id, pool=None):
+ """
+ :param tenant_id: (str)
+
+ :param pool: (str) pool of ips to allocate from
+
+ :returns: (dict) {
+ fixed_ip: ..., id: ..., instance_id: ..., ip: ..., pool: ...
+ }
+ """
+ json_data = dict(pool=pool) if pool else dict()
+ r = self.floating_ips_post(tenant_id, json_data)
+ return r.json['floating_ip']
+
+ def get_floating_ip(self, tenant_id, fip_id=None):
+ """
+ :param tenant_id: (str)
+
+ :param fip_id: (str) floating ip id (if None, all ips are returned)
+
+ :returns: (list) [
+ {fixed_ip: ..., id: ..., instance_id: ..., ip: ..., pool: ...},
+ ... ]
+ """
+ r = self.floating_ips_get(tenant_id, fip_id)
+ return r.json['floating_ips']
+
+ def delete_floating_ip(self, tenant_id, fip_id=None):
+ """
+ :param tenant_id: (str)
+
+ :param fip_id: (str) floating ip id (if None, all ips are deleted)
+
+ :returns: (dict) request headers
+ """
+ r = self.floating_ips_delete(tenant_id, fip_id)
+ return r.headers
path = path4url(tenant_id, 'os-floating-ip-pools')
return self.get(path, success=success, **kwargs)
- def floating_ips_get(self, tenant_id, success=200, **kwargs):
- path = path4url(tenant_id, 'os-floating-ips')
+ def floating_ips_get(self, tenant_id, ip='', success=200, **kwargs):
+ path = path4url(tenant_id, 'os-floating-ips', ip or '')
return self.get(path, success=success, **kwargs)
- def floating_ips_post(self, tenant_id, json_data, success=201, **kwargs):
- path = path4url(tenant_id, 'os-floating-ips')
+ def floating_ips_post(
+ self, tenant_id, json_data, ip='', success=201, **kwargs):
+ path = path4url(tenant_id, 'os-floating-ips', ip or '')
if json_data is not None:
json_data = json.dumps(json_data)
self.set_header('Content-Type', 'application/json')
self.set_header('Content-Length', len(json_data))
return self.post(path, data=json_data, success=success, **kwargs)
- def floating_ip_get(self, tenant_id, success=200, **kwargs):
- path = path4url(tenant_id, 'os-floating-ip')
- return self.get(path, success=success, **kwargs)
-
- def floating_ip_delete(self, tenant_id, success=204, **kwargs):
- path = path4url(tenant_id, 'os-floating-ip')
+ def floating_ips_delete(self, tenant_id, ip='', success=204, **kwargs):
+ path = path4url(tenant_id, 'os-floating-ips', ip or '')
return self.delete(path, success=success, **kwargs)
def test_floating_ips_get(self, get):
for args in product(
('tenant1', 'tenant2'),
+ ('', '192.193.194.195'),
(200, 204),
({}, {'k': 'v'})):
- tenant_id, success, kwargs = args
- r = self.client.floating_ips_get(tenant_id, success, **kwargs)
+ tenant_id, ip, success, kwargs = args
+ r = self.client.floating_ips_get(*args[:3], **kwargs)
self.assertTrue(isinstance(r, FR))
+ expected = '' if not ip else '/%s' % ip
self.assertEqual(get.mock_calls[-1], call(
- '/%s/os-floating-ips' % tenant_id,
+ '/%s/os-floating-ips%s' % (tenant_id, expected),
success=success, **kwargs))
@patch('%s.set_header' % rest_pkg)
for args in product(
('tenant1', 'tenant2'),
(None, [dict(json="data"), dict(data="json")]),
+ ('', '192.193.194.195'),
(202, 204),
({}, {'k': 'v'})):
- (tenant_id, json_data, success, kwargs) = args
- self.client.floating_ips_post(*args[:3], **kwargs)
+ (tenant_id, json_data, ip, success, kwargs) = args
+ self.client.floating_ips_post(*args[:4], **kwargs)
if json_data:
json_data = dumps(json_data)
self.assertEqual(SH.mock_calls[-2:], [
call('Content-Type', 'application/json'),
call('Content-Length', len(json_data))])
+ expected = '' if not ip else '/%s' % ip
self.assertEqual(post.mock_calls[-1], call(
- '/%s/os-floating-ips' % tenant_id,
+ '/%s/os-floating-ips%s' % (tenant_id, expected),
data=json_data, success=success,
**kwargs))
- @patch('%s.get' % rest_pkg, return_value=FR())
- def test_floating_ip_get(self, get):
- for args in product(
- ('tenant1', 'tenant2'),
- (200, 204),
- ({}, {'k': 'v'})):
- tenant_id, success, kwargs = args
- r = self.client.floating_ip_get(tenant_id, success, **kwargs)
- self.assertTrue(isinstance(r, FR))
- self.assertEqual(get.mock_calls[-1], call(
- '/%s/os-floating-ip' % tenant_id,
- success=success, **kwargs))
-
@patch('%s.delete' % rest_pkg, return_value=FR())
- def test_floating_ip_delete(self, delete):
+ def test_floating_ips_delete(self, delete):
for args in product(
('tenant1', 'tenant2'),
+ ('', '192.193.194.195'),
(204,),
({}, {'k': 'v'})):
- tenant_id, success, kwargs = args
- r = self.client.floating_ip_delete(tenant_id, success, **kwargs)
+ tenant_id, ip, success, kwargs = args
+ r = self.client.floating_ips_delete(*args[:3], **kwargs)
self.assertTrue(isinstance(r, FR))
+ expected = '' if not ip else '/%s' % ip
self.assertEqual(delete.mock_calls[-1], call(
- '/%s/os-floating-ip' % tenant_id,
+ '/%s/os-floating-ips%s' % (tenant_id, expected),
success=success, **kwargs))
self.client.delete_image_metadata(img_ref, key)
ID.assert_called_once_with(img_ref, '/metadata/%s' % key)
+ @patch('%s.floating_ip_pools_get' % compute_pkg, return_value=FR())
+ def test_get_floating_ip_pools(self, get):
+ tid = 't3n@nt_1d'
+ r = self.client.get_floating_ip_pools(tid)
+ self.assert_dicts_are_equal(r, FR.json)
+ self.assertEqual(get.mock_calls[-1], call(tid))
+
+ @patch('%s.floating_ips_get' % compute_pkg, return_value=FR())
+ def test_get_floating_ips(self, get):
+ tid = 't3n@nt_1d'
+ r = self.client.get_floating_ips(tid)
+ self.assert_dicts_are_equal(r, FR.json)
+ self.assertEqual(get.mock_calls[-1], call(tid))
+
+ @patch('%s.floating_ips_post' % compute_pkg, return_value=FR())
+ def test_alloc_floating_ip(self, post):
+ FR.json = dict(floating_ip=dict(
+ fixed_ip='fip',
+ id=1,
+ instance_id='lala',
+ ip='102.0.0.1',
+ pool='pisine'))
+ for args in product(
+ ('t1', 't2'),
+ (None, 'pisine')):
+ r = self.client.alloc_floating_ip(*args)
+ tenant_id, pool = args
+ self.assert_dicts_are_equal(r, FR.json['floating_ip'])
+ expected = dict(pool=pool) if pool else dict()
+ self.assertEqual(post.mock_calls[-1], call(tenant_id, expected))
+
+ @patch('%s.floating_ips_get' % compute_pkg, return_value=FR())
+ def test_get_floating_ip(self, get):
+ FR.json = dict(floating_ips=[dict(
+ fixed_ip='fip',
+ id=1,
+ instance_id='lala',
+ ip='102.0.0.1',
+ pool='pisine'), ])
+ for args in product(
+ ('t1', 't2'),
+ (None, 'fip')):
+ r = self.client.get_floating_ip(*args)
+ tenant_id, fip = args
+ self.assertEqual(r, FR.json['floating_ips'])
+ self.assertEqual(get.mock_calls[-1], call(tenant_id, fip))
+
+ @patch('%s.floating_ips_delete' % compute_pkg, return_value=FR())
+ def test_delete_floating_ip(self, delete):
+ for args in product(
+ ('t1', 't2'),
+ (None, 'fip')):
+ r = self.client.delete_floating_ip(*args)
+ tenant_id, fip = args
+ self.assertEqual(r, FR.headers)
+ self.assertEqual(delete.mock_calls[-1], call(tenant_id, fip))
+
if __name__ == '__main__':
from sys import argv
pass
return r['status']
return False
+
+ 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: ..., instance_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 assoc_floating_ip_to_server(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 assoc_floating_ip_to_server'
+ r = self.servers_post(
+ server_id, 'action',
+ json_data=dict(addFloatingIp=dict(address=address)))
+ return r.headers
+
+ def disassoc_floating_ip_to_server(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 disassoc_floating_ip_to_server'
+ r = self.servers_post(
+ server_id, 'action',
+ json_data=dict(removeFloatingIp=dict(address=address)))
+ return r.headers
path = path4url('os-floating-ip-pools')
return self.get(path, success=success, **kwargs)
- def floating_ips_get(self, success=200, **kwargs):
- path = path4url('os-floating-ips')
+ def floating_ips_get(self, fip_id='', success=200, **kwargs):
+ path = path4url('os-floating-ips', fip_id)
return self.get(path, success=success, **kwargs)
- def floating_ips_post(self, json_data, success=201, **kwargs):
- path = path4url('os-floating-ips')
+ def floating_ips_post(self, json_data, fip_id='', success=201, **kwargs):
+ path = path4url('os-floating-ips', fip_id)
if json_data is not None:
json_data = json.dumps(json_data)
self.set_header('Content-Type', 'application/json')
self.set_header('Content-Length', len(json_data))
return self.post(path, data=json_data, success=success, **kwargs)
- def floating_ip_get(self, floating_ip_id, success=200, **kwargs):
- path = path4url('os-floating-ip', floating_ip_id)
- return self.get(path, success=success, **kwargs)
-
- def floating_ip_delete(self, floating_ip_id, success=200, **kwargs):
- path = path4url('os-floating-ip', floating_ip_id)
+ def floating_ips_delete(self, fip_id, success=200, **kwargs):
+ path = path4url('os-floating-ips', fip_id)
return self.delete(path, success=success, **kwargs)
@patch('%s.set_header' % rest_pkg)
@patch('%s.post' % rest_pkg, return_value=FR())
def test_networks_post(self, post, SH):
- from json import dumps
for args in product(
('', 'net_id'),
('', 'cmd'),
@patch('%s.set_header' % rest_pkg)
@patch('%s.put' % rest_pkg, return_value=FR())
def test_networks_put(self, put, SH):
- from json import dumps
for args in product(
('', 'net_id'),
('', 'cmd'),
@patch('%s.get' % rest_pkg, return_value=FR())
def test_floating_ips_get(self, get):
for args in product(
+ ('fip', ''),
(200, 204),
({}, {'k': 'v'})):
- success, kwargs = args
- r = self.client.floating_ips_get(success, **kwargs)
+ fip, success, kwargs = args
+ r = self.client.floating_ips_get(fip, success, **kwargs)
self.assertTrue(isinstance(r, FR))
+ expected = '' if not fip else '/%s' % fip
self.assertEqual(get.mock_calls[-1], call(
- '/os-floating-ips', success=success, **kwargs))
+ '/os-floating-ips%s' % expected, success=success, **kwargs))
@patch('%s.set_header' % rest_pkg)
@patch('%s.post' % rest_pkg, return_value=FR())
def test_floating_ips_post(self, post, SH):
for args in product(
(None, [dict(json="data"), dict(data="json")]),
+ ('fip', ''),
(202, 204),
({}, {'k': 'v'})):
- (json_data, success, kwargs) = args
- self.client.floating_ips_post(*args[:2], **kwargs)
+ json_data, fip, success, kwargs = args
+ self.client.floating_ips_post(*args[:3], **kwargs)
if json_data:
json_data = dumps(json_data)
self.assertEqual(SH.mock_calls[-2:], [
call('Content-Type', 'application/json'),
call('Content-Length', len(json_data))])
+ expected = '' if not fip else '/%s' % fip
self.assertEqual(post.mock_calls[-1], call(
- '/os-floating-ips',
+ '/os-floating-ips%s' % expected,
data=json_data, success=success,
**kwargs))
- @patch('%s.get' % rest_pkg, return_value=FR())
- def test_floating_ip_get(self, get):
- for args in product(
- ('fip1', 'fip2'),
- (200, 204),
- ({}, {'k': 'v'})):
- fip, success, kwargs = args
- r = self.client.floating_ip_get(fip, success, **kwargs)
- self.assertTrue(isinstance(r, FR))
- self.assertEqual(get.mock_calls[-1], call(
- '/os-floating-ip/%s' % fip, success=success, **kwargs))
-
@patch('%s.delete' % rest_pkg, return_value=FR())
- def test_floating_ip_delete(self, delete):
+ def test_floating_ips_delete(self, delete):
for args in product(
('fip1', 'fip2'),
(200, 204),
({}, {'k': 'v'})):
fip, success, kwargs = args
- r = self.client.floating_ip_delete(fip, success, **kwargs)
+ r = self.client.floating_ips_delete(fip, success, **kwargs)
self.assertTrue(isinstance(r, FR))
self.assertEqual(delete.mock_calls[-1], call(
- '/os-floating-ip/%s' % fip, success=success, **kwargs))
+ '/os-floating-ips/%s' % fip, success=success, **kwargs))
class CycladesClient(TestCase):
self.assertEqual(err.details, [
'Network may be still connected to at least one server'])
+ @patch('%s.floating_ip_pools_get' % cyclades_pkg, return_value=FR())
+ def test_get_floating_ip_pools(self, get):
+ r = self.client.get_floating_ip_pools()
+ self.assert_dicts_are_equal(r, FR.json)
+ self.assertEqual(get.mock_calls[-1], call())
+
+ @patch('%s.floating_ips_get' % cyclades_pkg, return_value=FR())
+ def test_get_floating_ips(self, get):
+ r = self.client.get_floating_ips()
+ self.assert_dicts_are_equal(r, FR.json)
+ self.assertEqual(get.mock_calls[-1], call())
+
+ @patch('%s.floating_ips_post' % cyclades_pkg, return_value=FR())
+ def test_alloc_floating_ip(self, post):
+ FR.json = dict(floating_ip=dict(
+ fixed_ip='fip',
+ id=1,
+ instance_id='lala',
+ ip='102.0.0.1',
+ pool='pisine'))
+ for args in product(
+ (None, 'pisine'),
+ (None, 'Iwannanip')):
+ r = self.client.alloc_floating_ip(*args)
+ pool, address = args
+ self.assert_dicts_are_equal(r, FR.json['floating_ip'])
+ json_data = dict()
+ if pool:
+ json_data['pool'] = pool
+ if address:
+ json_data['address'] = address
+ self.assertEqual(post.mock_calls[-1], call(json_data))
+
+ @patch('%s.floating_ips_get' % cyclades_pkg, return_value=FR())
+ def test_get_floating_ip(self, get):
+ FR.json = dict(floating_ip=dict(
+ fixed_ip='fip',
+ id=1,
+ instance_id='lala',
+ ip='102.0.0.1',
+ pool='pisine'))
+ self.assertRaises(AssertionError, self.client.get_floating_ip, None)
+ fip = 'fip'
+ r = self.client.get_floating_ip(fip)
+ self.assert_dicts_are_equal(r, FR.json['floating_ip'])
+ self.assertEqual(get.mock_calls[-1], call(fip))
+
+ @patch('%s.floating_ips_delete' % cyclades_pkg, return_value=FR())
+ def test_delete_floating_ip(self, delete):
+ self.assertRaises(AssertionError, self.client.delete_floating_ip, None)
+ fip = 'fip'
+ r = self.client.delete_floating_ip(fip)
+ self.assert_dicts_are_equal(r, FR.headers)
+ self.assertEqual(delete.mock_calls[-1], call(fip))
+
+ @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
+ def test_assoc_floating_ip_to_server(self, spost):
+ vmid, addr = 42, 'anIpAddress'
+ for err, args in {
+ ValueError: ['not a server id', addr],
+ TypeError: [None, addr],
+ AssertionError: [vmid, None],
+ AssertionError: [vmid, '']}.items():
+ self.assertRaises(
+ err, self.client.assoc_floating_ip_to_server, *args)
+ r = self.client.assoc_floating_ip_to_server(vmid, addr)
+ self.assert_dicts_are_equal(r, FR.headers)
+ expected = dict(addFloatingIp=dict(address=addr))
+ self.assertEqual(
+ spost.mock_calls[-1], call(vmid, 'action', json_data=expected))
+
+ @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
+ def test_disassoc_floating_ip_to_server(self, spost):
+ vmid, addr = 42, 'anIpAddress'
+ for err, args in {
+ ValueError: ['not a server id', addr],
+ TypeError: [None, addr],
+ AssertionError: [vmid, None],
+ AssertionError: [vmid, '']}.items():
+ self.assertRaises(
+ err, self.client.disassoc_floating_ip_to_server, *args)
+ r = self.client.disassoc_floating_ip_to_server(vmid, addr)
+ self.assert_dicts_are_equal(r, FR.headers)
+ expected = dict(removeFloatingIp=dict(address=addr))
+ self.assertEqual(
+ spost.mock_calls[-1], call(vmid, 'action', json_data=expected))
+
if __name__ == '__main__':
from sys import argv
projects / kamaki / commitdiff