From e6ea45b2833fdd57a8011154aec5c1f6b00f44ca Mon Sep 17 00:00:00 2001
From: Dean Troyer <dtroyer@gmail.com>
Date: Sat, 8 Apr 2017 11:17:30 -0500
Subject: [PATCH] Low-level Compute v2 API: floating ip
api.compute.APIv2 floating ip functions.
novaclient 8.0 is now released without support for the previously
deprecated nova-net functions, so include a new low-level REST
implementation of the removed APIs.
Change-Id: Ic461b8d15e072e0534dcd73fff6857581d83c89b
---
openstackclient/api/compute_v2.py | 82 ++++++++++
openstackclient/network/v2/floating_ip.py | 27 ++--
.../tests/unit/api/test_compute_v2.py | 111 ++++++++++++++
.../tests/unit/compute/v2/fakes.py | 6 +-
.../network/v2/test_floating_ip_compute.py | 144 +++++++++---------
5 files changed, 277 insertions(+), 93 deletions(-)
diff --git a/openstackclient/api/compute_v2.py b/openstackclient/api/compute_v2.py
index 3bf3a0d81f..20062a2f4d 100644
--- a/openstackclient/api/compute_v2.py
+++ b/openstackclient/api/compute_v2.py
@@ -64,6 +64,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(
diff --git a/openstackclient/network/v2/floating_ip.py b/openstackclient/network/v2/floating_ip.py
index eaf27420ff..05b688a63d 100644
--- a/openstackclient/network/v2/floating_ip.py
+++ b/openstackclient/network/v2/floating_ip.py
@@ -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)
diff --git a/openstackclient/tests/unit/api/test_compute_v2.py b/openstackclient/tests/unit/api/test_compute_v2.py
index f443e810c1..f48986244a 100644
--- a/openstackclient/tests/unit/api/test_compute_v2.py
+++ b/openstackclient/tests/unit/api/test_compute_v2.py
@@ -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 = {
diff --git a/openstackclient/tests/unit/compute/v2/fakes.py b/openstackclient/tests/unit/compute/v2/fakes.py
index 05cb507625..90c2e1f9da 100644
--- a/openstackclient/tests/unit/compute/v2/fakes.py
+++ b/openstackclient/tests/unit/compute/v2/fakes.py
@@ -1019,11 +1019,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):
diff --git a/openstackclient/tests/unit/network/v2/test_floating_ip_compute.py b/openstackclient/tests/unit/network/v2/test_floating_ip_compute.py
index 23cd82d213..0d58c158b2 100644
--- a/openstackclient/tests/unit/network/v2/test_floating_ip_compute.py
+++ b/openstackclient/tests/unit/network/v2/test_floating_ip_compute.py
@@ -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)