openstack
/
python-openstackclient
Code Issues Proposed changes

Merge "Low-level Compute v2 API: floating ip"

This commit is contained in:
Jenkins 2017-04-17 21:14:05 +00:00 committed by Gerrit Code Review
parent 18206a9224 e6ea45b283
commit b0ce95745d
5 changed files with 277 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

82
openstackclient/api/compute_v2.py
View File

@ -93,6 +93,88 @@ class APIv2(api.BaseAPI):
return ret
# Flaoting IPs
def floating_ip_create(
self,
pool=None,
):
"""Create a new floating ip
https://developer.openstack.org/api-ref/compute/#create-allocate-floating-ip-address
:param pool: Name of floating IP pool
"""
url = "/os-floating-ips"
try:
return self.create(
url,
json={'pool': pool},
)['floating_ip']
except (
ksa_exceptions.NotFound,
ksa_exceptions.BadRequest,
):
msg = _("%s not found") % pool
raise exceptions.NotFound(msg)
def floating_ip_delete(
self,
floating_ip_id=None,
):
"""Delete a floating IP
https://developer.openstack.org/api-ref/compute/#delete-deallocate-floating-ip-address
:param string security_group:
Floating IP ID
"""
url = "/os-floating-ips"
if floating_ip_id is not None:
return self.delete('/%s/%s' % (url, floating_ip_id))
return None
def floating_ip_find(
self,
floating_ip=None,
):
"""Return a security group given name or ID
https://developer.openstack.org/api-ref/compute/#list-floating-ip-addresses
:param string floating_ip:
Floating IP address
:returns: A dict of the floating IP attributes
"""
url = "/os-floating-ips"
return self.find(
url,
attr='ip',
value=floating_ip,
)
def floating_ip_list(
self,
):
"""Get floating IPs
https://developer.openstack.org/api-ref/compute/#show-floating-ip-address-details
:returns:
list of security groups names
"""
url = "/os-floating-ips"
return self.list(url)["floating_ips"]
# Security Groups
def security_group_create(

27
openstackclient/network/v2/floating_ip.py
View File

@ -190,9 +190,9 @@ class CreateFloatingIP(common.NetworkAndComputeShowOne):
return (display_columns, data)
def take_action_compute(self, client, parsed_args):
obj = client.floating_ips.create(parsed_args.network)
columns = _get_columns(obj._info)
data = utils.get_dict_properties(obj._info, columns)
obj = client.api.floating_ip_create(parsed_args.network)
columns = _get_columns(obj)
data = utils.get_dict_properties(obj, columns)
return (columns, data)
@ -245,13 +245,7 @@ class DeleteFloatingIP(common.NetworkAndComputeDelete):
client.delete_ip(obj)
def take_action_compute(self, client, parsed_args):
obj = utils.find_resource(client.floating_ips, self.r)
client.floating_ips.delete(obj.id)
def take_action(self, parsed_args):
"""Implements a naive cache for the list of floating IPs"""
super(DeleteFloatingIP, self).take_action(parsed_args)
client.api.floating_ip_delete(self.r)
class DeleteIPFloating(DeleteFloatingIP):
@ -414,10 +408,10 @@ class ListFloatingIP(common.NetworkAndComputeLister):
'Pool',
)
data = client.floating_ips.list()
data = client.api.floating_ip_list()
return (headers,
(utils.get_item_properties(
(utils.get_dict_properties(
s, columns,
formatters={},
) for s in data))
@ -510,12 +504,9 @@ class ShowFloatingIP(common.NetworkAndComputeShowOne):
return (display_columns, data)
def take_action_compute(self, client, parsed_args):
obj = utils.find_resource(
client.floating_ips,
parsed_args.floating_ip,
)
columns = _get_columns(obj._info)
data = utils.get_dict_properties(obj._info, columns)
obj = client.api.floating_ip_find(parsed_args.floating_ip)
columns = _get_columns(obj)
data = utils.get_dict_properties(obj, columns)
return (columns, data)

111
openstackclient/tests/unit/api/test_compute_v2.py
View File

@ -34,6 +34,117 @@ class TestComputeAPIv2(utils.TestCase):
self.requests_mock = self.useFixture(fixture.Fixture())
class TestFloatingIP(TestComputeAPIv2):
FAKE_FLOATING_IP_RESP = {
'id': 1,
'ip': '203.0.113.11', # TEST-NET-3
'fixed_ip': '198.51.100.11', # TEST-NET-2
'pool': 'nova',
'instance_id': None,
}
FAKE_FLOATING_IP_RESP_2 = {
'id': 2,
'ip': '203.0.113.12', # TEST-NET-3
'fixed_ip': '198.51.100.12', # TEST-NET-2
'pool': 'nova',
'instance_id': None,
}
LIST_FLOATING_IP_RESP = [
FAKE_FLOATING_IP_RESP,
FAKE_FLOATING_IP_RESP_2,
]
def test_floating_ip_create(self):
self.requests_mock.register_uri(
'POST',
FAKE_URL + '/os-floating-ips',
json={'floating_ip': self.FAKE_FLOATING_IP_RESP},
status_code=200,
)
ret = self.api.floating_ip_create('nova')
self.assertEqual(self.FAKE_FLOATING_IP_RESP, ret)
def test_floating_ip_create_not_found(self):
self.requests_mock.register_uri(
'POST',
FAKE_URL + '/os-floating-ips',
status_code=404,
)
self.assertRaises(
osc_lib_exceptions.NotFound,
self.api.floating_ip_create,
'not-nova',
)
def test_floating_ip_delete(self):
self.requests_mock.register_uri(
'DELETE',
FAKE_URL + '/os-floating-ips/1',
status_code=202,
)
ret = self.api.floating_ip_delete('1')
self.assertEqual(202, ret.status_code)
self.assertEqual("", ret.text)
def test_floating_ip_delete_none(self):
ret = self.api.floating_ip_delete()
self.assertIsNone(ret)
def test_floating_ip_find_id(self):
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-floating-ips/1',
json={'floating_ip': self.FAKE_FLOATING_IP_RESP},
status_code=200,
)
ret = self.api.floating_ip_find('1')
self.assertEqual(self.FAKE_FLOATING_IP_RESP, ret)
def test_floating_ip_find_ip(self):
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-floating-ips/' + self.FAKE_FLOATING_IP_RESP['ip'],
status_code=404,
)
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-floating-ips',
json={'floating_ips': self.LIST_FLOATING_IP_RESP},
status_code=200,
)
ret = self.api.floating_ip_find(self.FAKE_FLOATING_IP_RESP['ip'])
self.assertEqual(self.FAKE_FLOATING_IP_RESP, ret)
def test_floating_ip_find_not_found(self):
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-floating-ips/1.2.3.4',
status_code=404,
)
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-floating-ips',
json={'floating_ips': self.LIST_FLOATING_IP_RESP},
status_code=200,
)
self.assertRaises(
osc_lib_exceptions.NotFound,
self.api.floating_ip_find,
'1.2.3.4',
)
def test_floating_ip_list(self):
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-floating-ips',
json={'floating_ips': self.LIST_FLOATING_IP_RESP},
status_code=200,
)
ret = self.api.floating_ip_list()
self.assertEqual(self.LIST_FLOATING_IP_RESP, ret)
class TestSecurityGroup(TestComputeAPIv2):
FAKE_SECURITY_GROUP_RESP = {

6
openstackclient/tests/unit/compute/v2/fakes.py
View File

@ -1016,11 +1016,7 @@ class FakeFloatingIP(object):
# Overwrite default attributes.
floating_ip_attrs.update(attrs)
floating_ip = fakes.FakeResource(
info=copy.deepcopy(floating_ip_attrs),
loaded=True)
return floating_ip
return floating_ip_attrs
@staticmethod
def create_floating_ips(attrs=None, count=2):

144
openstackclient/tests/unit/network/v2/test_floating_ip_compute.py
View File

@ -32,10 +32,13 @@ class TestFloatingIPCompute(compute_fakes.TestComputev2):
self.compute = self.app.client_manager.compute
@mock.patch(
'openstackclient.api.compute_v2.APIv2.floating_ip_create'
)
class TestCreateFloatingIPCompute(TestFloatingIPCompute):
# The floating ip to be deleted.
floating_ip = compute_fakes.FakeFloatingIP.create_one_floating_ip()
_floating_ip = compute_fakes.FakeFloatingIP.create_one_floating_ip()
columns = (
'fixed_ip',
@ -46,11 +49,11 @@ class TestCreateFloatingIPCompute(TestFloatingIPCompute):
)
data = (
floating_ip.fixed_ip,
floating_ip.id,
floating_ip.instance_id,
floating_ip.ip,
floating_ip.pool,
_floating_ip['fixed_ip'],
_floating_ip['id'],
_floating_ip['instance_id'],
_floating_ip['ip'],
_floating_ip['pool'],
)
def setUp(self):
@ -58,76 +61,79 @@ class TestCreateFloatingIPCompute(TestFloatingIPCompute):
self.app.client_manager.network_endpoint_enabled = False
self.compute.floating_ips.create.return_value = self.floating_ip
# self.compute.floating_ips.create.return_value = self.floating_ip
# Get the command object to test
self.cmd = fip.CreateFloatingIP(self.app, None)
def test_create_no_options(self):
def test_floating_ip_create_no_arg(self, fip_mock):
arglist = []
verifylist = []
self.assertRaises(tests_utils.ParserException, self.check_parser,
self.cmd, arglist, verifylist)
def test_create_default_options(self):
def test_floating_ip_create_default(self, fip_mock):
fip_mock.return_value = self._floating_ip
arglist = [
self.floating_ip.pool,
self._floating_ip['pool'],
]
verifylist = [
('network', self.floating_ip.pool),
('network', self._floating_ip['pool']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.compute.floating_ips.create.assert_called_once_with(
self.floating_ip.pool)
fip_mock.assert_called_once_with(self._floating_ip['pool'])
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
@mock.patch(
'openstackclient.api.compute_v2.APIv2.floating_ip_delete'
)
class TestDeleteFloatingIPCompute(TestFloatingIPCompute):
# The floating ips to be deleted.
floating_ips = compute_fakes.FakeFloatingIP.create_floating_ips(count=2)
_floating_ips = compute_fakes.FakeFloatingIP.create_floating_ips(count=2)
def setUp(self):
super(TestDeleteFloatingIPCompute, self).setUp()
self.app.client_manager.network_endpoint_enabled = False
self.compute.floating_ips.delete.return_value = None
# Return value of utils.find_resource()
self.compute.floating_ips.get = (
compute_fakes.FakeFloatingIP.get_floating_ips(self.floating_ips))
compute_fakes.FakeFloatingIP.get_floating_ips(self._floating_ips))
# Get the command object to test
self.cmd = fip.DeleteFloatingIP(self.app, None)
def test_floating_ip_delete(self):
def test_floating_ip_delete(self, fip_mock):
fip_mock.return_value = mock.Mock(return_value=None)
arglist = [
self.floating_ips[0].id,
self._floating_ips[0]['id'],
]
verifylist = [
('floating_ip', [self.floating_ips[0].id]),
('floating_ip', [self._floating_ips[0]['id']]),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.compute.floating_ips.delete.assert_called_once_with(
self.floating_ips[0].id
fip_mock.assert_called_once_with(
self._floating_ips[0]['id']
)
self.assertIsNone(result)
def test_multi_floating_ips_delete(self):
def test_floating_ip_delete_multi(self, fip_mock):
fip_mock.return_value = mock.Mock(return_value=None)
arglist = []
verifylist = []
for f in self.floating_ips:
arglist.append(f.id)
for f in self._floating_ips:
arglist.append(f['id'])
verifylist = [
('floating_ip', arglist),
]
@ -136,47 +142,44 @@ class TestDeleteFloatingIPCompute(TestFloatingIPCompute):
result = self.cmd.take_action(parsed_args)
calls = []
for f in self.floating_ips:
calls.append(call(f.id))
self.compute.floating_ips.delete.assert_has_calls(calls)
for f in self._floating_ips:
calls.append(call(f['id']))
fip_mock.assert_has_calls(calls)
self.assertIsNone(result)
def test_multi_floating_ips_delete_with_exception(self):
def test_floating_ip_delete_multi_exception(self, fip_mock):
fip_mock.return_value = mock.Mock(return_value=None)
fip_mock.side_effect = ([
mock.Mock(return_value=None),
exceptions.CommandError,
])
arglist = [
self.floating_ips[0].id,
self._floating_ips[0]['id'],
'unexist_floating_ip',
]
verifylist = [
('floating_ip',
[self.floating_ips[0].id, 'unexist_floating_ip']),
]
verifylist = [(
'floating_ip',
[self._floating_ips[0]['id'], 'unexist_floating_ip'],
)]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
find_mock_result = [self.floating_ips[0], exceptions.CommandError]
self.compute.floating_ips.get = (
mock.Mock(side_effect=find_mock_result)
)
self.compute.floating_ips.find.side_effect = exceptions.NotFound(None)
try:
self.cmd.take_action(parsed_args)
self.fail('CommandError should be raised.')
except exceptions.CommandError as e:
self.assertEqual('1 of 2 floating_ips failed to delete.', str(e))
self.compute.floating_ips.get.assert_any_call(
self.floating_ips[0].id)
self.compute.floating_ips.get.assert_any_call(
'unexist_floating_ip')
self.compute.floating_ips.delete.assert_called_once_with(
self.floating_ips[0].id
)
fip_mock.assert_any_call(self._floating_ips[0]['id'])
fip_mock.assert_any_call('unexist_floating_ip')
@mock.patch(
'openstackclient.api.compute_v2.APIv2.floating_ip_list'
)
class TestListFloatingIPCompute(TestFloatingIPCompute):
# The floating ips to be list up
floating_ips = compute_fakes.FakeFloatingIP.create_floating_ips(count=3)
_floating_ips = compute_fakes.FakeFloatingIP.create_floating_ips(count=3)
columns = (
'ID',
@ -187,13 +190,13 @@ class TestListFloatingIPCompute(TestFloatingIPCompute):
)
data = []
for ip in floating_ips:
for ip in _floating_ips:
data.append((
ip.id,
ip.ip,
ip.fixed_ip,
ip.instance_id,
ip.pool,
ip['id'],
ip['ip'],
ip['fixed_ip'],
ip['instance_id'],
ip['pool'],
))
def setUp(self):
@ -201,27 +204,29 @@ class TestListFloatingIPCompute(TestFloatingIPCompute):
self.app.client_manager.network_endpoint_enabled = False
self.compute.floating_ips.list.return_value = self.floating_ips
# Get the command object to test
self.cmd = fip.ListFloatingIP(self.app, None)
def test_floating_ip_list(self):
def test_floating_ip_list(self, fip_mock):
fip_mock.return_value = self._floating_ips
arglist = []
verifylist = []
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.compute.floating_ips.list.assert_called_once_with()
fip_mock.assert_called_once_with()
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, list(data))
@mock.patch(
'openstackclient.api.compute_v2.APIv2.floating_ip_find'
)
class TestShowFloatingIPCompute(TestFloatingIPCompute):
# The floating ip to display.
floating_ip = compute_fakes.FakeFloatingIP.create_one_floating_ip()
_floating_ip = compute_fakes.FakeFloatingIP.create_one_floating_ip()
columns = (
'fixed_ip',
@ -232,11 +237,11 @@ class TestShowFloatingIPCompute(TestFloatingIPCompute):
)
data = (
floating_ip.fixed_ip,
floating_ip.id,
floating_ip.instance_id,
floating_ip.ip,
floating_ip.pool,
_floating_ip['fixed_ip'],
_floating_ip['id'],
_floating_ip['instance_id'],
_floating_ip['ip'],
_floating_ip['pool'],
)
def setUp(self):
@ -244,22 +249,21 @@ class TestShowFloatingIPCompute(TestFloatingIPCompute):
self.app.client_manager.network_endpoint_enabled = False
# Return value of utils.find_resource()
self.compute.floating_ips.get.return_value = self.floating_ip
# Get the command object to test
self.cmd = fip.ShowFloatingIP(self.app, None)
def test_floating_ip_show(self):
def test_floating_ip_show(self, fip_mock):
fip_mock.return_value = self._floating_ip
arglist = [
self.floating_ip.id,
self._floating_ip['id'],
]
verifylist = [
('floating_ip', self.floating_ip.id),
('floating_ip', self._floating_ip['id']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
fip_mock.assert_called_once_with(self._floating_ip['id'])
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)