openstack/python-openstackclient
Code Issues Proposed changes

Add floating IP Port Forwarding commands

Browse Source 
Add following commands:
  floating ip port forwarding create
  floating ip port forwarding delete
  floating ip port forwarding list
  floating ip port forwarding set
  floating ip port forwarding show

Closes-Bug: #1811352
Change-Id: I6a5642e8acce28fc830410d4fa3180597b862761
This commit is contained in:
LIU Yulong 2018-09-12 06:18:04 +08:00
parent c9cc8b0ae2
commit f044016e29
6 changed files with 1138 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

173
doc/source/cli/command-objects/floating-ip-port-forwarding.rst Normal file
View File

@ -0,0 +1,173 @@
===========================
floating ip port forwarding
===========================
Network v2
floating ip port forwarding create
----------------------------------
Create floating IP Port Forwarding
.. program:: floating ip port forwarding create
.. code:: bash
openstack floating ip port forwarding create
--internal-ip-address <internal-ip-address>
--port <port>
--internal-protocol-port <port-number>
--external-protocol-port <port-number>
--protocol <protocol>
<floating-ip>
.. describe:: --internal-ip-address <internal-ip-address>
The fixed IPv4 address of the network port associated
to the floating IP port forwarding
.. describe:: --port <port>
The name or ID of the network port associated to the
floating IP port forwarding
.. describe:: --internal-protocol-port <port-number>
The protocol port number of the network port fixed
IPv4 address associated to the floating IP port
forwarding
.. describe:: --external-protocol-port <port-number>
The protocol port number of the port forwarding's
floating IP address
.. describe:: --protocol <protocol>
The protocol used in the floating IP port forwarding,
for instance: TCP, UDP
.. describe:: <floating-ip>
Floating IP that the port forwarding belongs to (IP
address or ID)
floating ip port forwarding delete
----------------------------------
Delete floating IP Port Forwarding(s)
.. program:: floating ip port forwarding delete
.. code:: bash
openstack floating ip port forwarding delete <floating-ip>
<port-forwarding-id> [<port-forwarding-id> ...]
.. describe:: <floating-ip>
Floating IP that the port forwarding belongs to (IP
address or ID)
.. describe:: <port-forwarding-id>
The ID of the floating IP port forwarding
floating ip port forwarding list
--------------------------------
List floating IP Port Forwarding(s)
.. program:: floating ip port forwarding list
.. code:: bash
openstack floating ip port forwarding list
[--port <port>]
[--external-protocol-port <port-number>]
[--protocol protocol]
<floating-ip>
.. option:: --port <port>
The ID of the network port associated to the floating
IP port forwarding
.. option:: --protocol <protocol>
The IP protocol used in the floating IP port
forwarding
.. describe:: <floating-ip>
Floating IP that the port forwarding belongs to (IP
address or ID)
floating ip port forwarding set
-------------------------------
Set floating IP Port Forwarding properties
.. program:: floating ip port forwarding set
.. code:: bash
openstack floating ip port forwarding set
[--port <port>]
[--internal-ip-address <internal-ip-address>]
[--internal-protocol-port <port-number>]
[--external-protocol-port <port-number>]
[--protocol <protocol>]
<floating-ip>
<port-forwarding-id>
.. option:: --port <port>
The ID of the network port associated to the floating
IP port forwarding
.. option:: --internal-ip-address <internal-ip-address>
The fixed IPv4 address of the network port associated
to the floating IP port forwarding
.. option:: --internal-protocol-port <port-number>
The TCP/UDP/other protocol port number of the network
port fixed IPv4 address associated to the floating IP
port forwarding
.. option:: --external-protocol-port <port-number>
The TCP/UDP/other protocol port number of the port
forwarding's floating IP address
.. option:: --protocol <protocol>
The IP protocol used in the floating IP port
forwarding
.. describe:: <floating-ip>
Floating IP that the port forwarding belongs to (IP
address or ID)
.. describe:: <port-forwarding-id>
The ID of the floating IP port forwarding
floating ip port forwarding show
--------------------------------
Display floating IP Port Forwarding details
.. program:: floating ip port forwarding show
.. code:: bash
openstack floating ip show <floating-ip> <port-forwarding-id>
.. describe:: <floating-ip>
Floating IP that the port forwarding belongs to (IP
address or ID)
.. describe:: <port-forwarding-id>
The ID of the floating IP port forwarding

371
openstackclient/network/v2/floating_ip_port_forwarding.py Normal file
View File

@ -0,0 +1,371 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
"""Floating IP Port Forwarding action implementations"""
import logging
from osc_lib.command import command
from osc_lib import exceptions
from osc_lib import utils
from openstackclient.i18n import _
from openstackclient.network import sdk_utils
LOG = logging.getLogger(__name__)
def _get_columns(item):
column_map = {
'tenant_id': 'project_id',
}
return sdk_utils.get_osc_show_columns_for_sdk_resource(item, column_map)
class CreateFloatingIPPortForwarding(command.ShowOne):
_description = _("Create floating IP port forwarding")
def get_parser(self, prog_name):
parser = super(CreateFloatingIPPortForwarding,
self).get_parser(prog_name)
parser.add_argument(
'--internal-ip-address',
required=True,
metavar='<internal-ip-address>',
help=_("The fixed IPv4 address of the network "
"port associated to the floating IP port forwarding")
)
parser.add_argument(
'--port',
metavar='<port>',
required=True,
help=_("The name or ID of the network port associated "
"to the floating IP port forwarding")
)
parser.add_argument(
'--internal-protocol-port',
type=int,
metavar='<port-number>',
required=True,
help=_("The protocol port number "
"of the network port fixed IPv4 address "
"associated to the floating IP port forwarding")
)
parser.add_argument(
'--external-protocol-port',
type=int,
metavar='<port-number>',
required=True,
help=_("The protocol port number of "
"the port forwarding's floating IP address")
)
parser.add_argument(
'--protocol',
metavar='<protocol>',
required=True,
help=_("The protocol used in the floating IP "
"port forwarding, for instance: TCP, UDP")
)
parser.add_argument(
'floating_ip',
metavar='<floating-ip>',
help=_("Floating IP that the port forwarding belongs to "
"(IP address or ID)")
)
return parser
def take_action(self, parsed_args):
attrs = {}
client = self.app.client_manager.network
floating_ip = client.find_ip(
parsed_args.floating_ip,
ignore_missing=False,
)
if parsed_args.internal_protocol_port is not None:
if (parsed_args.internal_protocol_port <= 0 or
parsed_args.internal_protocol_port > 65535):
msg = _("The port number range is <1-65535>")
raise exceptions.CommandError(msg)
attrs['internal_port'] = parsed_args.internal_protocol_port
if parsed_args.external_protocol_port is not None:
if (parsed_args.external_protocol_port <= 0 or
parsed_args.external_protocol_port > 65535):
msg = _("The port number range is <1-65535>")
raise exceptions.CommandError(msg)
attrs['external_port'] = parsed_args.external_protocol_port
if parsed_args.port:
port = client.find_port(parsed_args.port,
ignore_missing=False)
attrs['internal_port_id'] = port.id
attrs['internal_ip_address'] = parsed_args.internal_ip_address
attrs['protocol'] = parsed_args.protocol
obj = client.create_floating_ip_port_forwarding(
floating_ip.id,
**attrs
)
display_columns, columns = _get_columns(obj)
data = utils.get_item_properties(obj, columns)
return (display_columns, data)
class DeleteFloatingIPPortForwarding(command.Command):
_description = _("Delete floating IP port forwarding")
def get_parser(self, prog_name):
parser = super(DeleteFloatingIPPortForwarding,
self).get_parser(prog_name)
parser.add_argument(
'floating_ip',
metavar='<floating-ip>',
help=_("Floating IP that the port forwarding belongs to "
"(IP address or ID)")
)
parser.add_argument(
'port_forwarding_id',
nargs="+",
metavar="<port-forwarding-id>",
help=_("The ID of the floating IP port forwarding(s) to delete")
)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.network
floating_ip = client.find_ip(
parsed_args.floating_ip,
ignore_missing=False,
)
result = 0
for port_forwarding_id in parsed_args.port_forwarding_id:
try:
client.delete_floating_ip_port_forwarding(
floating_ip.id,
port_forwarding_id,
ignore_missing=False,
)
except Exception as e:
result += 1
LOG.error(_("Failed to delete floating IP port forwarding "
"'%(port_forwarding_id)s': %(e)s"),
{'port_forwarding_id': port_forwarding_id, 'e': e})
if result > 0:
total = len(parsed_args.port_forwarding_id)
msg = (_("%(result)s of %(total)s Port forwarding failed "
"to delete.") % {'result': result, 'total': total})
raise exceptions.CommandError(msg)
class ListFloatingIPPortForwarding(command.Lister):
_description = _("List floating IP port forwarding")
def get_parser(self, prog_name):
parser = super(ListFloatingIPPortForwarding,
self).get_parser(prog_name)
parser.add_argument(
'floating_ip',
metavar='<floating-ip>',
help=_("Floating IP that the port forwarding belongs to "
"(IP address or ID)")
)
parser.add_argument(
'--port',
metavar='<port>',
help=_("Filter the list result by the ID or name of "
"the internal network port")
)
parser.add_argument(
'--external-protocol-port',
metavar='<port-number>',
dest='external_protocol_port',
help=_("Filter the list result by the "
"protocol port number of the floating IP")
)
parser.add_argument(
'--protocol',
metavar='protocol',
help=_("Filter the list result by the port protocol")
)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.network
columns = (
'id',
'internal_port_id',
'internal_ip_address',
'internal_port',
'external_port',
'protocol',
)
headers = (
'ID',
'Internal Port ID',
'Internal IP Address',
'Internal Port',
'External Port',
'Protocol',
)
query = {}
if parsed_args.port:
port = client.find_port(parsed_args.port,
ignore_missing=False)
query['internal_port_id'] = port.id
if parsed_args.external_protocol_port is not None:
query['external_port'] = parsed_args.external_protocol_port
if parsed_args.protocol is not None:
query['protocol'] = parsed_args.protocol
obj = client.find_ip(
parsed_args.floating_ip,
ignore_missing=False,
)
data = client.floating_ip_port_forwardings(obj, **query)
return (headers,
(utils.get_item_properties(
s, columns,
formatters={},
) for s in data))
class SetFloatingIPPortForwarding(command.Command):
_description = _("Set floating IP Port Forwarding Properties")
def get_parser(self, prog_name):
parser = super(SetFloatingIPPortForwarding,
self).get_parser(prog_name)
parser.add_argument(
'floating_ip',
metavar='<floating-ip>',
help=_("Floating IP that the port forwarding belongs to "
"(IP address or ID)")
)
parser.add_argument(
'port_forwarding_id',
metavar='<port-forwarding-id>',
help=_("The ID of the floating IP port forwarding")
)
parser.add_argument(
'--port',
metavar='<port>',
help=_("The ID of the network port associated to "
"the floating IP port forwarding")
)
parser.add_argument(
'--internal-ip-address',
metavar='<internal-ip-address>',
help=_("The fixed IPv4 address of the network port "
"associated to the floating IP port forwarding")
)
parser.add_argument(
'--internal-protocol-port',
metavar='<port-number>',
type=int,
help=_("The TCP/UDP/other protocol port number of the "
"network port fixed IPv4 address associated to "
"the floating IP port forwarding")
)
parser.add_argument(
'--external-protocol-port',
type=int,
metavar='<port-number>',
help=_("The TCP/UDP/other protocol port number of the "
"port forwarding's floating IP address")
)
parser.add_argument(
'--protocol',
metavar='<protocol>',
choices=['tcp', 'udp'],
help=_("The IP protocol used in the floating IP port forwarding")
)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.network
floating_ip = client.find_ip(
parsed_args.floating_ip,
ignore_missing=False,
)
attrs = {}
if parsed_args.port:
port = client.find_port(parsed_args.port,
ignore_missing=False)
attrs['internal_port_id'] = port.id
if parsed_args.internal_ip_address:
attrs['internal_ip_address'] = parsed_args.internal_ip_address
if parsed_args.internal_protocol_port is not None:
if (parsed_args.internal_protocol_port <= 0 or
parsed_args.internal_protocol_port > 65535):
msg = _("The port number range is <1-65535>")
raise exceptions.CommandError(msg)
attrs['internal_port'] = parsed_args.internal_protocol_port
if parsed_args.external_protocol_port is not None:
if (parsed_args.external_protocol_port <= 0 or
parsed_args.external_protocol_port > 65535):
msg = _("The port number range is <1-65535>")
raise exceptions.CommandError(msg)
attrs['external_port'] = parsed_args.external_protocol_port
if parsed_args.protocol:
attrs['protocol'] = parsed_args.protocol
client.update_floating_ip_port_forwarding(
floating_ip.id, parsed_args.port_forwarding_id, **attrs)
class ShowFloatingIPPortForwarding(command.ShowOne):
_description = _("Display floating IP Port Forwarding details")
def get_parser(self, prog_name):
parser = super(ShowFloatingIPPortForwarding,
self).get_parser(prog_name)
parser.add_argument(
'floating_ip',
metavar='<floating-ip>',
help=_("Floating IP that the port forwarding belongs to "
"(IP address or ID)")
)
parser.add_argument(
'port_forwarding_id',
metavar="<port-forwarding-id>",
help=_("The ID of the floating IP port forwarding")
)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.network
floating_ip = client.find_ip(
parsed_args.floating_ip,
ignore_missing=False,
)
obj = client.find_floating_ip_port_forwarding(
floating_ip,
parsed_args.port_forwarding_id,
ignore_missing=False,
)
display_columns, columns = _get_columns(obj)
data = utils.get_item_properties(obj, columns)
return (display_columns, data)

77
openstackclient/tests/unit/network/v2/fakes.py
View File

@ -1816,3 +1816,80 @@ class FakeQuota(object):
info=copy.deepcopy(quota_attrs),
loaded=True)
return quota
class FakeFloatingIPPortForwarding(object):
""""Fake one or more Port forwarding"""
@staticmethod
def create_one_port_forwarding(attrs=None):
"""Create a fake Port Forwarding.
:param Dictionary attrs:
A dictionary with all attributes
:return:
A FakeResource object with name, id, etc.
"""
attrs = attrs or {}
floatingip_id = (
attrs.get('floatingip_id') or'floating-ip-id-' + uuid.uuid4().hex
)
# Set default attributes.
port_forwarding_attrs = {
'id': uuid.uuid4().hex,
'floatingip_id': floatingip_id,
'internal_port_id': 'internal-port-id-' + uuid.uuid4().hex,
'internal_ip_address': '192.168.1.2',
'internal_port': randint(1, 65535),
'external_port': randint(1, 65535),
'protocol': 'tcp',
}
# Overwrite default attributes.
port_forwarding_attrs.update(attrs)
port_forwarding = fakes.FakeResource(
info=copy.deepcopy(port_forwarding_attrs),
loaded=True
)
return port_forwarding
@staticmethod
def create_port_forwardings(attrs=None, count=2):
"""Create multiple fake Port Forwarding.
:param Dictionary attrs:
A dictionary with all attributes
:param int count:
The number of Port Forwarding rule to fake
:return:
A list of FakeResource objects faking the Port Forwardings
"""
port_forwardings = []
for i in range(0, count):
port_forwardings.append(
FakeFloatingIPPortForwarding.create_one_port_forwarding(attrs)
)
return port_forwardings
@staticmethod
def get_port_forwardings(port_forwardings=None, count=2):
"""Get a list of faked Port Forwardings.
If port forwardings list is provided, then initialize the Mock object
with the list. Otherwise create one.
:param List port forwardings:
A list of FakeResource objects faking port forwardings
:param int count:
The number of Port Forwardings to fake
:return:
An iterable Mock object with side_effect set to a list of faked
Port Forwardings
"""
if port_forwardings is None:
port_forwardings = (
FakeFloatingIPPortForwarding.create_port_forwardings(count)
)
return mock.Mock(side_effect=port_forwardings)

502
openstackclient/tests/unit/network/v2/test_floating_ip_port_forwarding.py Normal file
View File

@ -0,0 +1,502 @@
# Copyright (c) 2018 China Telecom Corporation
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
from mock import call
from osc_lib import exceptions
from openstackclient.network.v2 import floating_ip_port_forwarding
from openstackclient.tests.unit.identity.v2_0 import fakes as identity_fakes_v2
from openstackclient.tests.unit.network.v2 import fakes as network_fakes
from openstackclient.tests.unit import utils as tests_utils
class TestFloatingIPPortForwarding(network_fakes.TestNetworkV2):
def setUp(self):
super(TestFloatingIPPortForwarding, self).setUp()
self.network = self.app.client_manager.network
self.floating_ip = (network_fakes.FakeFloatingIP.
create_one_floating_ip())
self.port = network_fakes.FakePort.create_one_port()
self.project = identity_fakes_v2.FakeProject.create_one_project()
self.network.find_port = mock.Mock(return_value=self.port)
class TestCreateFloatingIPPortForwarding(TestFloatingIPPortForwarding):
def setUp(self):
project_id = ''
super(TestCreateFloatingIPPortForwarding, self).setUp()
self.new_port_forwarding = (
network_fakes.FakeFloatingIPPortForwarding.
create_one_port_forwarding(
attrs={
'internal_port_id': self.port.id,
'floatingip_id': self.floating_ip.id,
}
)
)
self.network.create_floating_ip_port_forwarding = mock.Mock(
return_value=self.new_port_forwarding)
self.network.find_ip = mock.Mock(
return_value=self.floating_ip
)
# Get the command object to test
self.cmd = floating_ip_port_forwarding.CreateFloatingIPPortForwarding(
self.app, self.namespace)
self.columns = (
'external_port',
'floatingip_id',
'id',
'internal_ip_address',
'internal_port',
'internal_port_id',
'project_id',
'protocol'
)
self.data = (
self.new_port_forwarding.external_port,
self.new_port_forwarding.floatingip_id,
self.new_port_forwarding.id,
self.new_port_forwarding.internal_ip_address,
self.new_port_forwarding.internal_port,
self.new_port_forwarding.internal_port_id,
project_id,
self.new_port_forwarding.protocol,
)
def test_create_no_options(self):
arglist = []
verifylist = []
# Missing required args should bail here
self.assertRaises(tests_utils.ParserException, self.check_parser,
self.cmd, arglist, verifylist)
def test_create_all_options(self):
arglist = [
'--port', self.new_port_forwarding.internal_port_id,
'--internal-protocol-port',
str(self.new_port_forwarding.internal_port),
'--external-protocol-port',
str(self.new_port_forwarding.external_port),
'--protocol', self.new_port_forwarding.protocol,
self.new_port_forwarding.floatingip_id,
'--internal-ip-address',
self.new_port_forwarding.internal_ip_address,
]
verifylist = [
('port', self.new_port_forwarding.internal_port_id),
('internal_protocol_port', self.new_port_forwarding.internal_port),
('external_protocol_port', self.new_port_forwarding.external_port),
('protocol', self.new_port_forwarding.protocol),
('floating_ip', self.new_port_forwarding.floatingip_id),
('internal_ip_address', self.new_port_forwarding.
internal_ip_address),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.network.create_floating_ip_port_forwarding.\
assert_called_once_with(
self.new_port_forwarding.floatingip_id,
**{
'external_port': self.new_port_forwarding.external_port,
'internal_ip_address': self.new_port_forwarding.
internal_ip_address,
'internal_port': self.new_port_forwarding.internal_port,
'internal_port_id': self.new_port_forwarding.
internal_port_id,
'protocol': self.new_port_forwarding.protocol,
})
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
class TestDeleteFloatingIPPortForwarding(TestFloatingIPPortForwarding):
def setUp(self):
super(TestDeleteFloatingIPPortForwarding, self).setUp()
self._port_forwarding = (
network_fakes.FakeFloatingIPPortForwarding.create_port_forwardings(
count=2, attrs={
'floatingip_id': self.floating_ip.id,
}
)
)
self.network.delete_floating_ip_port_forwarding = mock.Mock(
return_value=None
)
self.network.find_ip = mock.Mock(
return_value=self.floating_ip
)
# Get the command object to test
self.cmd = floating_ip_port_forwarding.DeleteFloatingIPPortForwarding(
self.app, self.namespace)
def test_port_forwarding_delete(self):
arglist = [
self.floating_ip.id,
self._port_forwarding[0].id,
]
verifylist = [
('floating_ip', self.floating_ip.id),
('port_forwarding_id', [self._port_forwarding[0].id]),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.network.delete_floating_ip_port_forwarding.\
assert_called_once_with(
self.floating_ip.id,
self._port_forwarding[0].id,
ignore_missing=False
)
self.assertIsNone(result)
def test_multi_port_forwardings_delete(self):
arglist = []
pf_id = []
arglist.append(str(self.floating_ip))
for a in self._port_forwarding:
arglist.append(a.id)
pf_id.append(a.id)
verifylist = [
('floating_ip', str(self.floating_ip)),
('port_forwarding_id', pf_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
calls = []
for a in self._port_forwarding:
calls.append(call(a.floatingip_id, a.id, ignore_missing=False))
self.network.delete_floating_ip_port_forwarding.assert_has_calls(calls)
self.assertIsNone(result)
def test_multi_port_forwarding_delete_with_exception(self):
arglist = [
self.floating_ip.id,
self._port_forwarding[0].id,
'unexist_port_forwarding_id',
]
verifylist = [
('floating_ip', self.floating_ip.id),
('port_forwarding_id',
[self._port_forwarding[0].id, 'unexist_port_forwarding_id']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
delete_mock_result = [None, exceptions.CommandError]
self.network.delete_floating_ip_port_forwarding = (
mock.MagicMock(side_effect=delete_mock_result)
)
try:
self.cmd.take_action(parsed_args)
self.fail('CommandError should be raised.')
except exceptions.CommandError as e:
self.assertEqual(
'1 of 2 Port forwarding failed to delete.',
str(e)
)
self.network.delete_floating_ip_port_forwarding.\
assert_any_call(
self.floating_ip.id,
'unexist_port_forwarding_id',
ignore_missing=False
)
self.network.delete_floating_ip_port_forwarding.\
assert_any_call(
self.floating_ip.id,
self._port_forwarding[0].id,
ignore_missing=False
)
class TestListFloatingIPPortForwarding(TestFloatingIPPortForwarding):
columns = (
'ID',
'Internal Port ID',
'Internal IP Address',
'Internal Port',
'External Port',
'Protocol'
)
def setUp(self):
super(TestListFloatingIPPortForwarding, self).setUp()
self.port_forwardings = (
network_fakes.FakeFloatingIPPortForwarding.create_port_forwardings(
count=3, attrs={
'internal_port_id': self.port.id,
'floatingip_id': self.floating_ip.id,
}
)
)
self.data = []
for port_forwarding in self.port_forwardings:
self.data.append((
port_forwarding.id,
port_forwarding.internal_port_id,
port_forwarding.internal_ip_address,
port_forwarding.internal_port,
port_forwarding.external_port,
port_forwarding.protocol,
))
self.network.floating_ip_port_forwardings = mock.Mock(
return_value=self.port_forwardings
)
self.network.find_ip = mock.Mock(
return_value=self.floating_ip
)
# Get the command object to test
self.cmd = floating_ip_port_forwarding.ListFloatingIPPortForwarding(
self.app,
self.namespace
)
def test_port_forwarding_list(self):
arglist = [
self.floating_ip.id
]
verifylist = [
('floating_ip', self.floating_ip.id)
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.network.floating_ip_port_forwardings.assert_called_once_with(
self.floating_ip,
**{}
)
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, list(data))
def test_port_forwarding_list_all_options(self):
arglist = [
'--port', self.port_forwardings[0].internal_port_id,
'--external-protocol-port',
str(self.port_forwardings[0].external_port),
'--protocol', self.port_forwardings[0].protocol,
self.port_forwardings[0].floatingip_id,
]
verifylist = [
('port', self.port_forwardings[0].internal_port_id),
('external_protocol_port',
str(self.port_forwardings[0].external_port)),
('protocol', self.port_forwardings[0].protocol),
('floating_ip', self.port_forwardings[0].floatingip_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
query = {
'internal_port_id': self.port_forwardings[0].internal_port_id,
'external_port': str(self.port_forwardings[0].external_port),
'protocol': self.port_forwardings[0].protocol,
}
self.network.floating_ip_port_forwardings.assert_called_once_with(
self.floating_ip,
**query
)
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, list(data))
class TestSetFloatingIPPortForwarding(TestFloatingIPPortForwarding):
# The Port Forwarding to set.
def setUp(self):
super(TestSetFloatingIPPortForwarding, self).setUp()
self._port_forwarding = (
network_fakes.FakeFloatingIPPortForwarding.
create_one_port_forwarding(
attrs={
'floatingip_id': self.floating_ip.id,
}
)
)
self.network.update_floating_ip_port_forwarding = mock.Mock(
return_value=None
)
self.network.find_floating_ip_port_forwarding = mock.Mock(
return_value=self._port_forwarding)
self.network.find_ip = mock.Mock(
return_value=self.floating_ip
)
# Get the command object to test
self.cmd = floating_ip_port_forwarding.SetFloatingIPPortForwarding(
self.app,
self.namespace
)
def test_set_nothing(self):
arglist = [
self._port_forwarding.floatingip_id,
self._port_forwarding.id,
]
verifylist = [
('floating_ip', self._port_forwarding.floatingip_id),
('port_forwarding_id', self._port_forwarding.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
attrs = {}
self.network.update_floating_ip_port_forwarding.assert_called_with(
self._port_forwarding.floatingip_id,
self._port_forwarding.id,
**attrs
)
self.assertIsNone(result)
def test_set_all_thing(self):
arglist = [
'--port', self.port.id,
'--internal-ip-address', 'new_internal_ip_address',
'--internal-protocol-port', '100',
'--external-protocol-port', '200',
'--protocol', 'tcp',
self._port_forwarding.floatingip_id,
self._port_forwarding.id,
]
verifylist = [
('port', self.port.id),
('internal_ip_address', 'new_internal_ip_address'),
('internal_protocol_port', 100),
('external_protocol_port', 200),
('protocol', 'tcp'),
('floating_ip', self._port_forwarding.floatingip_id),
('port_forwarding_id', self._port_forwarding.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
attrs = {
'internal_port_id': self.port.id,
'internal_ip_address': 'new_internal_ip_address',
'internal_port': 100,
'external_port': 200,
'protocol': 'tcp',
}
self.network.update_floating_ip_port_forwarding.assert_called_with(
self._port_forwarding.floatingip_id,
self._port_forwarding.id,
**attrs
)
self.assertIsNone(result)
class TestShowFloatingIPPortForwarding(TestFloatingIPPortForwarding):
# The port forwarding to show.
columns = (
'external_port',
'floatingip_id',
'id',
'internal_ip_address',
'internal_port',
'internal_port_id',
'project_id',
'protocol',
)
def setUp(self):
project_id = ''
super(TestShowFloatingIPPortForwarding, self).setUp()
self._port_forwarding = (
network_fakes.FakeFloatingIPPortForwarding.
create_one_port_forwarding(
attrs={
'floatingip_id': self.floating_ip.id,
}
)
)
self.data = (
self._port_forwarding.external_port,
self._port_forwarding.floatingip_id,
self._port_forwarding.id,
self._port_forwarding.internal_ip_address,
self._port_forwarding.internal_port,
self._port_forwarding.internal_port_id,
project_id,
self._port_forwarding.protocol,
)
self.network.find_floating_ip_port_forwarding = mock.Mock(
return_value=self._port_forwarding
)
self.network.find_ip = mock.Mock(
return_value=self.floating_ip
)
# Get the command object to test
self.cmd = floating_ip_port_forwarding.ShowFloatingIPPortForwarding(
self.app,
self.namespace
)
def test_show_no_options(self):
arglist = []
verifylist = []
# Missing required args should bail here
self.assertRaises(tests_utils.ParserException, self.check_parser,
self.cmd, arglist, verifylist)
def test_show_default_options(self):
arglist = [
self._port_forwarding.floatingip_id,
self._port_forwarding.id,
]
verifylist = [
('floating_ip', self._port_forwarding.floatingip_id),
('port_forwarding_id', self._port_forwarding.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.network.find_floating_ip_port_forwarding.assert_called_once_with(
self.floating_ip,
self._port_forwarding.id,
ignore_missing=False
)
self.assertEqual(self.columns, columns)
self.assertEqual(list(self.data), list(data))

9
releasenotes/notes/add-fip-portforwarding-commands-6e4d8ace698ee308.yaml Normal file
View File

@ -0,0 +1,9 @@
---
features:
- |
Add floating IP Port Forwarding commands:
``floating ip port forwarding create``,
``floating ip port forwarding delete``,
``floating ip port forwarding list``,
``floating ip port forwarding set`` and
``floating ip port forwarding show``.

6
setup.cfg
View File

@ -384,6 +384,12 @@ openstack.network.v2 =
floating_ip_pool_list = openstackclient.network.v2.floating_ip_pool:ListFloatingIPPool
floating_ip_port_forwarding_create = openstackclient.network.v2.floating_ip_port_forwarding:CreateFloatingIPPortForwarding
floating_ip_port_forwarding_delete = openstackclient.network.v2.floating_ip_port_forwarding:DeleteFloatingIPPortForwarding
floating_ip_port_forwarding_list = openstackclient.network.v2.floating_ip_port_forwarding:ListFloatingIPPortForwarding
floating_ip_port_forwarding_set = openstackclient.network.v2.floating_ip_port_forwarding:SetFloatingIPPortForwarding
floating_ip_port_forwarding_show = openstackclient.network.v2.floating_ip_port_forwarding:ShowFloatingIPPortForwarding
ip_availability_list = openstackclient.network.v2.ip_availability:ListIPAvailability
ip_availability_show = openstackclient.network.v2.ip_availability:ShowIPAvailability