openstack
/
python-openstackclient
Code Issues Proposed changes

Add support for managing external gateways 

This change implements the logic to call the new API for managing
external gateways.

Relevant Neutron core change:
https://review.opendev.org/c/openstack/neutron/+/873593

Co-Authored-by: Frode Nordahl <frode.nordahl@canonical.com>
Related-Bug: #2002687
Change-Id: Ib45f30f552934a0a5c035c3b7fadfc0d522219ba
This commit is contained in:
Dmitrii Shcherbakov 2023-07-07 19:08:33 +03:00 committed by Frode Nordahl
parent 58ad3cefa7
commit 16c695045c
No known key found for this signature in database
GPG Key ID: 6A5D59A3BA48373F
3 changed files with 884 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

311
openstackclient/network/v2/router.py
View File

@ -13,6 +13,7 @@
"""Router action implementations"""
import collections
import copy
import json
import logging
@ -85,23 +86,67 @@ def _get_columns(item):
)
def is_multiple_gateways_supported(n_client):
return n_client.find_extension("external-gateway-multihoming") is not None
def _passed_multiple_gateways(extension_supported, external_gateways):
passed_multiple_gws = len(external_gateways) > 1
if passed_multiple_gws and not extension_supported:
msg = _(
'Supplying --external-gateway option multiple times is not '
'supported due to the lack of external-gateway-multihoming '
'extension at the Neutron side.'
)
raise exceptions.CommandError(msg)
return passed_multiple_gws
def _get_external_gateway_attrs(client_manager, parsed_args):
attrs = {}
if parsed_args.external_gateway:
gateway_info = {}
if parsed_args.external_gateways:
external_gateways: collections.defaultdict[
str, list[dict]
] = collections.defaultdict(list)
n_client = client_manager.network
network = n_client.find_network(
parsed_args.external_gateway, ignore_missing=False
)
gateway_info['network_id'] = network.id
if parsed_args.disable_snat:
gateway_info['enable_snat'] = False
if parsed_args.enable_snat:
gateway_info['enable_snat'] = True
first_network_id = None
for gw_net_name_or_id in parsed_args.external_gateways:
gateway_info = {}
gw_net = n_client.find_network(
gw_net_name_or_id, ignore_missing=False
)
if first_network_id is None:
first_network_id = gw_net.id
gateway_info['network_id'] = gw_net.id
if 'disable_snat' in parsed_args and parsed_args.disable_snat:
gateway_info['enable_snat'] = False
if 'enable_snat' in parsed_args and parsed_args.enable_snat:
gateway_info['enable_snat'] = True
# This option was added before multiple gateways were supported, so
# it does not have a per-gateway port granularity so just pass it
# along in gw info in case it is specified.
if 'qos_policy' in parsed_args and parsed_args.qos_policy:
qos_id = n_client.find_qos_policy(
parsed_args.qos_policy, ignore_missing=False
).id
gateway_info['qos_policy_id'] = qos_id
if 'no_qos_policy' in parsed_args and parsed_args.no_qos_policy:
gateway_info['qos_policy_id'] = None
external_gateways[gw_net.id].append(gateway_info)
multiple_gws_supported = is_multiple_gateways_supported(n_client)
# Parse the external fixed IP specs and match them to specific gateway
# ports if needed.
if parsed_args.fixed_ips:
ips = []
for ip_spec in parsed_args.fixed_ips:
# If there is only one gateway, this value will represent the
# network ID for it, otherwise it will be overridden.
ip_net_id = first_network_id
if ip_spec.get('subnet', False):
subnet_name_id = ip_spec.pop('subnet')
if subnet_name_id:
@ -109,12 +154,45 @@ def _get_external_gateway_attrs(client_manager, parsed_args):
subnet_name_id, ignore_missing=False
)
ip_spec['subnet_id'] = subnet.id
ip_net_id = subnet.network_id
if ip_spec.get('ip-address', False):
ip_spec['ip_address'] = ip_spec.pop('ip-address')
ips.append(ip_spec)
gateway_info['external_fixed_ips'] = ips
attrs['external_gateway_info'] = gateway_info
# Finally, add an ip_spec to the specific gateway identified
# by a network from the spec.
if (
'subnet_id' in ip_spec
and ip_net_id not in external_gateways
):
msg = _(
'Subnet %s does not belong to any of the networks '
'provided for --external-gateway.'
) % (ip_spec['subnet_id'])
raise exceptions.CommandError(msg)
for gw_info in external_gateways[ip_net_id]:
if 'external_fixed_ips' not in gw_info:
gw_info['external_fixed_ips'] = [ip_spec]
break
else:
# The end user has requested more fixed IPs than there are
# gateways, add multiple fixed IPs to single gateway to
# retain current behavior.
for gw_info in external_gateways[ip_net_id]:
gw_info['external_fixed_ips'].append(ip_spec)
break
# Use the newer API whenever it is supported regardless of whether one
# or multiple gateways are passed as arguments.
if multiple_gws_supported:
gateway_list = []
# Now merge the per-network-id lists of external gateway info
# dicts into one list.
for gw_info_list in external_gateways.values():
gateway_list.extend(gw_info_list)
attrs['external_gateways'] = gateway_list
else:
attrs['external_gateway_info'] = external_gateways[
first_network_id
][0]
return attrs
@ -372,7 +450,13 @@ class CreateRouter(command.ShowOne, common.NeutronCommandWithExtraArgs):
parser.add_argument(
'--external-gateway',
metavar="<network>",
help=_("External Network used as router's gateway (name or ID)"),
action='append',
help=_(
"External Network used as router's gateway (name or ID). "
"(repeat option to set multiple gateways per router "
"if the L3 service plugin in use supports it)."
),
dest='external_gateways',
)
parser.add_argument(
'--fixed-ip',
@ -384,7 +468,7 @@ class CreateRouter(command.ShowOne, common.NeutronCommandWithExtraArgs):
"Desired IP and/or subnet (name or ID) "
"on external gateway: "
"subnet=<subnet>,ip-address=<ip-address> "
"(repeat option to set multiple fixed IP addresses)"
"(repeat option to set multiple fixed IP addresses)."
),
)
snat_group = parser.add_mutually_exclusive_group()
@ -433,7 +517,7 @@ class CreateRouter(command.ShowOne, common.NeutronCommandWithExtraArgs):
self._parse_extra_properties(parsed_args.extra_properties)
)
if parsed_args.enable_ndp_proxy and not parsed_args.external_gateway:
if parsed_args.enable_ndp_proxy and not parsed_args.external_gateways:
msg = _(
"You must specify '--external-gateway' in order "
"to enable router's NDP proxy"
@ -443,15 +527,24 @@ class CreateRouter(command.ShowOne, common.NeutronCommandWithExtraArgs):
if parsed_args.enable_ndp_proxy is not None:
attrs['enable_ndp_proxy'] = parsed_args.enable_ndp_proxy
external_gateways = attrs.pop('external_gateways', None)
obj = client.create_router(**attrs)
# tags cannot be set when created, so tags need to be set later.
_tag.update_tags_for_set(client, obj, parsed_args)
# If the multiple external gateways API is intended to be used,
# do a separate API call to set the desired external gateways as the
# router creation API supports adding only one.
if external_gateways:
client.update_external_gateways(
obj, body={'router': {'external_gateways': external_gateways}}
)
if (
parsed_args.disable_snat
or parsed_args.enable_snat
or parsed_args.fixed_ips
) and not parsed_args.external_gateway:
) and not parsed_args.external_gateways:
msg = _(
"You must specify '--external-gateway' in order "
"to specify SNAT or fixed-ip values"
@ -791,7 +884,13 @@ class SetRouter(common.NeutronCommandWithExtraArgs):
parser.add_argument(
'--external-gateway',
metavar="<network>",
help=_("External Network used as router's gateway (name or ID)"),
action='append',
help=_(
"External Network used as router's gateway (name or ID). "
"(repeat option to set multiple gateways per router "
"if the L3 service plugin in use supports it)."
),
dest='external_gateways',
)
parser.add_argument(
'--fixed-ip',
@ -803,7 +902,7 @@ class SetRouter(common.NeutronCommandWithExtraArgs):
"Desired IP and/or subnet (name or ID) "
"on external gateway: "
"subnet=<subnet>,ip-address=<ip-address> "
"(repeat option to set multiple fixed IP addresses)"
"(repeat option to set multiple fixed IP addresses)."
),
)
snat_group = parser.add_mutually_exclusive_group()
@ -873,7 +972,7 @@ class SetRouter(common.NeutronCommandWithExtraArgs):
parsed_args.disable_snat
or parsed_args.enable_snat
or parsed_args.fixed_ips
) and not parsed_args.external_gateway:
) and not parsed_args.external_gateways:
msg = _(
"You must specify '--external-gateway' in order "
"to update the SNAT or fixed-ip values"
@ -882,7 +981,7 @@ class SetRouter(common.NeutronCommandWithExtraArgs):
if (
parsed_args.qos_policy or parsed_args.no_qos_policy
) and not parsed_args.external_gateway:
) and not parsed_args.external_gateways:
try:
original_net_id = obj.external_gateway_info['network_id']
except (KeyError, TypeError):
@ -893,17 +992,21 @@ class SetRouter(common.NeutronCommandWithExtraArgs):
)
raise exceptions.CommandError(msg)
else:
if not attrs.get('external_gateway_info'):
if not attrs.get('external_gateway_info') and not attrs.get(
'external_gateways'
):
attrs['external_gateway_info'] = {}
attrs['external_gateway_info']['network_id'] = original_net_id
if parsed_args.qos_policy:
check_qos_id = client.find_qos_policy(
parsed_args.qos_policy, ignore_missing=False
).id
attrs['external_gateway_info']['qos_policy_id'] = check_qos_id
if not attrs.get('external_gateways'):
attrs['external_gateway_info']['qos_policy_id'] = check_qos_id
if 'no_qos_policy' in parsed_args and parsed_args.no_qos_policy:
attrs['external_gateway_info']['qos_policy_id'] = None
if not attrs.get('external_gateways'):
attrs['external_gateway_info']['qos_policy_id'] = None
attrs.update(
self._parse_extra_properties(parsed_args.extra_properties)
@ -913,7 +1016,16 @@ class SetRouter(common.NeutronCommandWithExtraArgs):
attrs['enable_ndp_proxy'] = parsed_args.enable_ndp_proxy
if attrs:
external_gateways = attrs.pop('external_gateways', None)
client.update_router(obj, **attrs)
# If the multiple external gateways API is intended to be used,
# do a separate API call to set external gateways.
if external_gateways:
client.update_external_gateways(
obj,
body={'router': {'external_gateways': external_gateways}},
)
# tags is a subresource and it needs to be updated separately.
_tag.update_tags_for_set(client, obj, parsed_args)
@ -973,11 +1085,15 @@ class UnsetRouter(common.NeutronUnsetCommandWithExtraArgs):
"(repeat option to unset multiple routes)"
),
)
# NOTE(dmitriis): This was not extended to support selective removal
# of external gateways due to a cpython bug in argparse:
# https://github.com/python/cpython/issues/53584
parser.add_argument(
'--external-gateway',
action='store_true',
default=False,
help=_("Remove external gateway information from the router"),
dest='external_gateways',
)
parser.add_argument(
'--qos-policy',
@ -1024,7 +1140,7 @@ class UnsetRouter(common.NeutronUnsetCommandWithExtraArgs):
'qos_policy_id': None,
}
if parsed_args.external_gateway:
if parsed_args.external_gateways:
attrs['external_gateway_info'] = {}
attrs.update(
@ -1032,6 +1148,149 @@ class UnsetRouter(common.NeutronUnsetCommandWithExtraArgs):
)
if attrs:
# If removing multiple gateways per router are supported,
# use the relevant API to remove them all.
if is_multiple_gateways_supported(client):
client.remove_external_gateways(
obj,
body={'router': {'external_gateways': {}}},
)
client.update_router(obj, **attrs)
# tags is a subresource and it needs to be updated separately.
_tag.update_tags_for_unset(client, obj, parsed_args)
class AddGatewayToRouter(command.ShowOne):
_description = _("Add router gateway")
def get_parser(self, prog_name):
parser = super().get_parser(prog_name)
parser.add_argument(
'router',
metavar="<router>",
help=_("Router to modify (name or ID)."),
)
parser.add_argument(
metavar="<network>",
help=_(
"External Network to a attach a router gateway to (name or "
"ID)."
),
dest='external_gateways',
# The argument is stored in a list in order to reuse the
# common attribute parsing code.
nargs=1,
)
parser.add_argument(
'--fixed-ip',
metavar='subnet=<subnet>,ip-address=<ip-address>',
action=parseractions.MultiKeyValueAction,
optional_keys=['subnet', 'ip-address'],
dest='fixed_ips',
help=_(
"Desired IP and/or subnet (name or ID) "
"on external gateway: "
"subnet=<subnet>,ip-address=<ip-address> "
"(repeat option to set multiple fixed IP addresses)."
),
)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.network
if not is_multiple_gateways_supported(client):
msg = _(
'The external-gateway-multihoming extension is not enabled at '
'the Neutron side.'
)
raise exceptions.CommandError(msg)
router_obj = client.find_router(
parsed_args.router, ignore_missing=False
)
# Get the common attributes.
attrs = _get_external_gateway_attrs(
self.app.client_manager, parsed_args
)
if attrs:
external_gateways = attrs.pop('external_gateways')
router_obj = client.add_external_gateways(
router_obj,
body={'router': {'external_gateways': external_gateways}},
)
display_columns, columns = _get_columns(router_obj)
data = utils.get_item_properties(
router_obj, columns, formatters=_formatters
)
return (display_columns, data)
class RemoveGatewayFromRouter(command.ShowOne):
_description = _("Remove router gateway")
def get_parser(self, prog_name):
parser = super().get_parser(prog_name)
parser.add_argument(
'router',
metavar="<router>",
help=_("Router to modify (name or ID)."),
)
parser.add_argument(
metavar="<network>",
help=_(
"External Network to remove a router gateway from (name or "
"ID)."
),
dest='external_gateways',
# The argument is stored in a list in order to reuse the
# common attribute parsing code.
nargs=1,
)
parser.add_argument(
'--fixed-ip',
metavar='subnet=<subnet>,ip-address=<ip-address>',
action=parseractions.MultiKeyValueAction,
optional_keys=['subnet', 'ip-address'],
dest='fixed_ips',
help=_(
"IP and/or subnet (name or ID) on the external gateway "
"which is used to identify a particular gateway if multiple "
"are attached to the same network: subnet=<subnet>,"
"ip-address=<ip-address>."
),
)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.network
if not is_multiple_gateways_supported(client):
msg = _(
'The external-gateway-multihoming extension is not enabled at '
'the Neutron side.'
)
raise exceptions.CommandError(msg)
router_obj = client.find_router(
parsed_args.router, ignore_missing=False
)
# Get the common attributes.
attrs = _get_external_gateway_attrs(
self.app.client_manager, parsed_args
)
if attrs:
external_gateways = attrs.pop('external_gateways')
router_obj = client.remove_external_gateways(
router_obj,
body={'router': {'external_gateways': external_gateways}},
)
display_columns, columns = _get_columns(router_obj)
data = utils.get_item_properties(
router_obj, columns, formatters=_formatters
)
return (display_columns, data)

616
openstackclient/tests/unit/network/v2/test_router.py
View File

@ -75,7 +75,7 @@ class TestAddPortToRouter(TestRouter):
self._router,
**{
'port_id': self._router.port,
}
},
)
self.assertIsNone(result)
@ -130,6 +130,7 @@ class TestAddSubnetToRouter(TestRouter):
class TestCreateRouter(TestRouter):
# The new router created.
new_router = network_fakes.FakeRouter.create_one_router()
_extensions = {'fake': network_fakes.create_one_extension()}
columns = (
'admin_state_up',
@ -169,7 +170,9 @@ class TestCreateRouter(TestRouter):
return_value=self.new_router
)
self.network_client.set_tags = mock.Mock(return_value=None)
self.network_client.find_extension = mock.Mock(
side_effect=lambda name: self._extensions.get(name)
)
# Get the command object to test
self.cmd = router.CreateRouter(self.app, self.namespace)
@ -228,7 +231,7 @@ class TestCreateRouter(TestRouter):
('enable', True),
('distributed', False),
('ha', False),
('external_gateway', _network.name),
('external_gateways', [_network.name]),
('enable_snat', True),
('fixed_ips', [{'ip-address': '2001:db8::1'}]),
]
@ -1100,10 +1103,13 @@ class TestSetRouter(TestRouter):
# The router to set.
_default_route = {'destination': '10.20.20.0/24', 'nexthop': '10.20.30.1'}
_network = network_fakes.create_one_network()
_subnet = network_fakes.FakeSubnet.create_one_subnet()
_subnet = network_fakes.FakeSubnet.create_one_subnet(
attrs={'network_id': _network.id}
)
_router = network_fakes.FakeRouter.create_one_router(
attrs={'routes': [_default_route], 'tags': ['green', 'red']}
)
_extensions = {'fake': network_fakes.create_one_extension()}
def setUp(self):
super(TestSetRouter, self).setUp()
@ -1114,7 +1120,9 @@ class TestSetRouter(TestRouter):
return_value=self._network
)
self.network_client.find_subnet = mock.Mock(return_value=self._subnet)
self.network_client.find_extension = mock.Mock(
side_effect=lambda name: self._extensions.get(name)
)
# Get the command object to test
self.cmd = router.SetRouter(self.app, self.namespace)
@ -1312,7 +1320,7 @@ class TestSetRouter(TestRouter):
self._router.id,
]
verifylist = [
('external_gateway', self._network.id),
('external_gateways', [self._network.id]),
('router', self._router.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
@ -1320,7 +1328,7 @@ class TestSetRouter(TestRouter):
result = self.cmd.take_action(parsed_args)
self.network_client.update_router.assert_called_with(
self._router,
**{'external_gateway_info': {'network_id': self._network.id}}
**{'external_gateway_info': {'network_id': self._network.id}},
)
self.assertIsNone(result)
@ -1335,7 +1343,7 @@ class TestSetRouter(TestRouter):
]
verifylist = [
('router', self._router.id),
('external_gateway', self._network.id),
('external_gateways', [self._network.id]),
('fixed_ips', [{'subnet': "'abc'"}]),
('enable_snat', True),
]
@ -1354,7 +1362,7 @@ class TestSetRouter(TestRouter):
],
'enable_snat': True,
}
}
},
)
self.assertIsNone(result)
@ -1369,7 +1377,7 @@ class TestSetRouter(TestRouter):
]
verifylist = [
('router', self._router.id),
('external_gateway', self._network.id),
('external_gateways', [self._network.id]),
('fixed_ips', [{'ip-address': "10.0.1.1"}]),
('enable_snat', True),
]
@ -1388,7 +1396,7 @@ class TestSetRouter(TestRouter):
],
'enable_snat': True,
}
}
},
)
self.assertIsNone(result)
@ -1403,7 +1411,7 @@ class TestSetRouter(TestRouter):
]
verifylist = [
('router', self._router.id),
('external_gateway', self._network.id),
('external_gateways', [self._network.id]),
('fixed_ips', [{'subnet': "'abc'", 'ip-address': "10.0.1.1"}]),
('enable_snat', True),
]
@ -1423,7 +1431,7 @@ class TestSetRouter(TestRouter):
],
'enable_snat': True,
}
}
},
)
self.assertIsNone(result)
@ -1468,7 +1476,7 @@ class TestSetRouter(TestRouter):
]
verifylist = [
('router', self._router.id),
('external_gateway', self._network.id),
('external_gateways', [self._network.id]),
('qos_policy', qos_policy.id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
@ -1481,7 +1489,7 @@ class TestSetRouter(TestRouter):
'network_id': self._network.id,
'qos_policy_id': qos_policy.id,
}
}
},
)
self.assertIsNone(result)
@ -1494,7 +1502,7 @@ class TestSetRouter(TestRouter):
]
verifylist = [
('router', self._router.id),
('external_gateway', self._network.id),
('external_gateways', [self._network.id]),
('no_qos_policy', True),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
@ -1507,7 +1515,7 @@ class TestSetRouter(TestRouter):
'network_id': self._network.id,
'qos_policy_id': None,
}
}
},
)
self.assertIsNone(result)
@ -1526,7 +1534,7 @@ class TestSetRouter(TestRouter):
]
verifylist = [
('router', self._router.id),
('external_gateway', self._network.id),
('external_gateways', [self._network.id]),
('qos_policy', qos_policy.id),
('no_qos_policy', True),
]
@ -1747,6 +1755,13 @@ class TestUnsetRouter(TestRouter):
)
self.network_client.update_router = mock.Mock(return_value=None)
self.network_client.set_tags = mock.Mock(return_value=None)
self._extensions = {'fake': network_fakes.create_one_extension()}
self.network_client.find_extension = mock.Mock(
side_effect=lambda name: self._extensions.get(name)
)
self.network_client.remove_external_gateways = mock.Mock(
return_value=None
)
# Get the command object to test
self.cmd = router.UnsetRouter(self.app, self.namespace)
@ -1799,7 +1814,7 @@ class TestUnsetRouter(TestRouter):
'--external-gateway',
self._testrouter.name,
]
verifylist = [('external_gateway', True)]
verifylist = [('external_gateways', True)]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
attrs = {'external_gateway_info': {}}
@ -1808,6 +1823,33 @@ class TestUnsetRouter(TestRouter):
)
self.assertIsNone(result)
def test_unset_router_external_gateway_multiple_supported(self):
# Add the relevant extension in order to test the alternate behavior.
self._extensions = {
'external-gateway-multihoming': network_fakes.create_one_extension(
attrs={'name': 'external-gateway-multihoming'}
)
}
arglist = [
'--external-gateway',
self._testrouter.name,
]
verifylist = [('external_gateways', True)]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
# The removal of all gateways should be requested using the multiple
# gateways API.
self.network_client.remove_external_gateways.assert_called_once_with(
self._testrouter, body={'router': {'external_gateways': {}}}
)
# The compatibility API will also be called in order to potentially
# unset other parameters along with external_gateway_info which
# should already be empty at that point anyway.
self.network_client.update_router.assert_called_once_with(
self._testrouter, **{'external_gateway_info': {}}
)
self.assertIsNone(result)
def _test_unset_tags(self, with_tags=True):
if with_tags:
arglist = ['--tag', 'red', '--tag', 'blue']
@ -1895,3 +1937,539 @@ class TestUnsetRouter(TestRouter):
self.assertRaises(
exceptions.CommandError, self.cmd.take_action, parsed_args
)
class TestGatewayOps(TestRouter):
def setUp(self):
super().setUp()
self._networks = []
self._network = network_fakes.create_one_network()
self._networks.append(self._network)
self._router = network_fakes.FakeRouter.create_one_router(
{
'external_gateway_info': {
'network_id': self._network.id,
},
}
)
self._subnet = network_fakes.FakeSubnet.create_one_subnet(
attrs={'network_id': self._network.id}
)
self._extensions = {
'external-gateway-multihoming': network_fakes.create_one_extension(
attrs={'name': 'external-gateway-multihoming'}
)
}
self.network_client.find_extension = mock.Mock(
side_effect=lambda name: self._extensions.get(name)
)
self.network_client.find_router = mock.Mock(return_value=self._router)
def _find_network(name_or_id, ignore_missing):
for network in self._networks:
if name_or_id in (network.id, network.name):
return network
if ignore_missing:
return None
raise Exception('Test resource not found')
self.network_client.find_network = mock.Mock(side_effect=_find_network)
self.network_client.find_subnet = mock.Mock(return_value=self._subnet)
self.network_client.add_external_gateways = mock.Mock(
return_value=None
)
self.network_client.remove_external_gateways = mock.Mock(
return_value=None
)
class TestCreateMultipleGateways(TestGatewayOps):
_columns = (
'admin_state_up',
'availability_zone_hints',
'availability_zones',
'description',
'distributed',
'external_gateway_info',
'ha',
'id',
'name',
'project_id',
'routes',
'status',
'tags',
)
def setUp(self):
super().setUp()
self._second_network = network_fakes.create_one_network()
self._networks.append(self._second_network)
self.network_client.create_router = mock.Mock(
return_value=self._router
)
self.network_client.update_router = mock.Mock(return_value=None)
self.network_client.update_external_gateways = mock.Mock(
return_value=None
)
self._data = (
router.AdminStateColumn(self._router.admin_state_up),
format_columns.ListColumn(self._router.availability_zone_hints),
format_columns.ListColumn(self._router.availability_zones),
self._router.description,
self._router.distributed,
router.RouterInfoColumn(self._router.external_gateway_info),
self._router.ha,
self._router.id,
self._router.name,
self._router.project_id,
router.RoutesColumn(self._router.routes),
self._router.status,
format_columns.ListColumn(self._router.tags),
)
self.cmd = router.CreateRouter(self.app, self.namespace)
def test_create_one_gateway(self):
arglist = [
"--external-gateway",
self._network.id,
self._router.name,
]
verifylist = [
('name', self._router.name),
('external_gateways', [self._network.id]),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.network_client.update_external_gateways.assert_called_with(
self._router,
body={
'router': {
'external_gateways': [
{
'network_id': self._network.id,
}
]
}
},
)
self.assertEqual(self._columns, columns)
self.assertCountEqual(self._data, data)
def test_create_multiple_gateways(self):
arglist = [
self._router.name,
"--external-gateway",
self._network.id,
"--external-gateway",
self._network.id,
"--external-gateway",
self._second_network.id,
'--fixed-ip',
'subnet={},ip-address=10.0.1.1'.format(self._subnet.id),
'--fixed-ip',
'subnet={},ip-address=10.0.1.2'.format(self._subnet.id),
]
verifylist = [
('name', self._router.name),
(
'external_gateways',
[self._network.id, self._network.id, self._second_network.id],
),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
# The router will not have a gateway after the create call, but it
# will be added after the update call.
self.network_client.create_router.assert_called_once_with(
**{
'admin_state_up': True,
'name': self._router.name,
}
)
self.network_client.update_external_gateways.assert_called_with(
self._router,
body={
'router': {
'external_gateways': [
{
'network_id': self._network.id,
'external_fixed_ips': [
{
'subnet_id': self._subnet.id,
'ip_address': '10.0.1.1',
}
],
},
{
'network_id': self._network.id,
'external_fixed_ips': [
{
'subnet_id': self._subnet.id,
'ip_address': '10.0.1.2',
}
],
},
{
'network_id': self._second_network.id,
},
]
}
},
)
self.assertEqual(self._columns, columns)
self.assertCountEqual(self._data, data)
class TestUpdateMultipleGateways(TestGatewayOps):
def setUp(self):
super().setUp()
self._second_network = network_fakes.create_one_network()
self._networks.append(self._second_network)
self.network_client.update_router = mock.Mock(return_value=None)
self.network_client.update_external_gateways = mock.Mock(
return_value=None
)
self.cmd = router.SetRouter(self.app, self.namespace)
def test_update_one_gateway(self):
arglist = [
"--external-gateway",
self._network.id,
"--no-qos-policy",
self._router.name,
]
verifylist = [
('router', self._router.name),
('external_gateways', [self._network.id]),
('no_qos_policy', True),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.network_client.update_external_gateways.assert_called_with(
self._router,
body={
'router': {
'external_gateways': [
{'network_id': self._network.id, 'qos_policy_id': None}
]
}
},
)
self.assertIsNone(result)
def test_update_multiple_gateways(self):
arglist = [
self._router.name,
"--external-gateway",
self._network.id,
"--external-gateway",
self._network.id,
"--external-gateway",
self._second_network.id,
'--fixed-ip',
'subnet={},ip-address=10.0.1.1'.format(self._subnet.id),
'--fixed-ip',
'subnet={},ip-address=10.0.1.2'.format(self._subnet.id),
"--no-qos-policy",
]
verifylist = [
('router', self._router.name),
(
'external_gateways',
[self._network.id, self._network.id, self._second_network.id],
),
('no_qos_policy', True),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.network_client.update_external_gateways.assert_called_with(
self._router,
body={
'router': {
'external_gateways': [
{
'network_id': self._network.id,
'external_fixed_ips': [
{
'subnet_id': self._subnet.id,
'ip_address': '10.0.1.1',
}
],
'qos_policy_id': None,
},
{
'network_id': self._network.id,
'external_fixed_ips': [
{
'subnet_id': self._subnet.id,
'ip_address': '10.0.1.2',
}
],
'qos_policy_id': None,
},
{
'network_id': self._second_network.id,
'qos_policy_id': None,
},
]
}
},
)
self.assertIsNone(result)
class TestAddGatewayRouter(TestGatewayOps):
def setUp(self):
super().setUp()
# Get the command object to test
self.cmd = router.AddGatewayToRouter(self.app, self.namespace)
self.network_client.add_external_gateways.return_value = self._router
def test_add_gateway_network_only(self):
arglist = [
self._router.name,
self._network.id,
]
verifylist = [
('router', self._router.name),
('external_gateways', [self._network.id]),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.network_client.add_external_gateways.assert_called_with(
self._router,
body={
'router': {
'external_gateways': [{'network_id': self._network.id}]
}
},
)
self.assertEqual(result[1][result[0].index('id')], self._router.id)
def test_add_gateway_network_fixed_ip(self):
arglist = [
self._router.name,
self._network.id,
'--fixed-ip',
'subnet={},ip-address=10.0.1.1'.format(self._subnet.id),
]
verifylist = [
('router', self._router.name),
('external_gateways', [self._network.id]),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.network_client.add_external_gateways.assert_called_with(
self._router,
body={
'router': {
'external_gateways': [
{
'network_id': self._network.id,
'external_fixed_ips': [
{
'subnet_id': self._subnet.id,
'ip_address': '10.0.1.1',
}
],
}
]
}
},
)
self.assertEqual(result[1][result[0].index('id')], self._router.id)
def test_add_gateway_network_multiple_fixed_ips(self):
arglist = [
self._router.name,
self._network.id,
'--fixed-ip',
'subnet={},ip-address=10.0.1.1'.format(self._subnet.id),
'--fixed-ip',
'subnet={},ip-address=10.0.1.2'.format(self._subnet.id),
]
verifylist = [
('router', self._router.name),
('external_gateways', [self._network.id]),
(
'fixed_ips',
[
{'ip-address': '10.0.1.1', 'subnet': self._subnet.id},
{'ip-address': '10.0.1.2', 'subnet': self._subnet.id},
],
),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.network_client.add_external_gateways.assert_called_with(
self._router,
body={
'router': {
'external_gateways': [
{
'network_id': self._network.id,
'external_fixed_ips': [
{
'subnet_id': self._subnet.id,
'ip_address': '10.0.1.1',
},
{
'subnet_id': self._subnet.id,
'ip_address': '10.0.1.2',
},
],
}
]
}
},
)
self.assertEqual(result[1][result[0].index('id')], self._router.id)
def test_add_gateway_network_only_no_extension(self):
self._extensions = {}
arglist = [
self._router.name,
self._network.id,
]
verifylist = [
('router', self._router.name),
('external_gateways', [self._network.id]),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.assertRaises(
exceptions.CommandError, self.cmd.take_action, parsed_args
)
class TestRemoveGatewayRouter(TestGatewayOps):
def setUp(self):
super().setUp()
# Get the command object to test
self.cmd = router.RemoveGatewayFromRouter(self.app, self.namespace)
self.network_client.remove_external_gateways.return_value = (
self._router
)
def test_remove_gateway_network_only(self):
arglist = [
self._router.name,
self._network.id,
]
verifylist = [
('router', self._router.name),
('external_gateways', [self._network.id]),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.network_client.remove_external_gateways.assert_called_with(
self._router,
body={
'router': {
'external_gateways': [{'network_id': self._network.id}]
}
},
)
self.assertEqual(result[1][result[0].index('id')], self._router.id)
def test_remove_gateway_network_fixed_ip(self):
arglist = [
self._router.name,
self._network.id,
'--fixed-ip',
'subnet={},ip-address=10.0.1.1'.format(self._subnet.id),
]
verifylist = [
('router', self._router.name),
('external_gateways', [self._network.id]),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.network_client.remove_external_gateways.assert_called_with(
self._router,
body={
'router': {
'external_gateways': [
{
'network_id': self._network.id,
'external_fixed_ips': [
{
'subnet_id': self._subnet.id,
'ip_address': '10.0.1.1',
}
],
}
]
}
},
)
self.assertEqual(result[1][result[0].index('id')], self._router.id)
def test_remove_gateway_network_multiple_fixed_ips(self):
arglist = [
self._router.name,
self._network.id,
'--fixed-ip',
'subnet={},ip-address=10.0.1.1'.format(self._subnet.id),
'--fixed-ip',
'subnet={},ip-address=10.0.1.2'.format(self._subnet.id),
]
verifylist = [
('router', self._router.name),
('external_gateways', [self._network.id]),
(
'fixed_ips',
[
{'ip-address': '10.0.1.1', 'subnet': self._subnet.id},
{'ip-address': '10.0.1.2', 'subnet': self._subnet.id},
],
),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.network_client.remove_external_gateways.assert_called_with(
self._router,
body={
'router': {
'external_gateways': [
{
'network_id': self._network.id,
'external_fixed_ips': [
{
'subnet_id': self._subnet.id,
'ip_address': '10.0.1.1',
},
{
'subnet_id': self._subnet.id,
'ip_address': '10.0.1.2',
},
],
}
]
}
},
)
self.assertEqual(result[1][result[0].index('id')], self._router.id)
def test_remove_gateway_network_only_no_extension(self):
self._extensions = {}
arglist = [
self._router.name,
self._network.id,
]
verifylist = [
('router', self._router.name),
('external_gateways', [self._network.id]),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.assertRaises(
exceptions.CommandError, self.cmd.take_action, parsed_args
)

2
setup.cfg
View File

@ -560,12 +560,14 @@ openstack.network.v2 =
port_show = openstackclient.network.v2.port:ShowPort
port_unset = openstackclient.network.v2.port:UnsetPort
router_add_gateway = openstackclient.network.v2.router:AddGatewayToRouter
router_add_port = openstackclient.network.v2.router:AddPortToRouter
router_add_route = openstackclient.network.v2.router:AddExtraRoutesToRouter
router_add_subnet = openstackclient.network.v2.router:AddSubnetToRouter
router_create = openstackclient.network.v2.router:CreateRouter
router_delete = openstackclient.network.v2.router:DeleteRouter
router_list = openstackclient.network.v2.router:ListRouter
router_remove_gateway = openstackclient.network.v2.router:RemoveGatewayFromRouter
router_remove_port = openstackclient.network.v2.router:RemovePortFromRouter
router_remove_route = openstackclient.network.v2.router:RemoveExtraRoutesFromRouter
router_remove_subnet = openstackclient.network.v2.router:RemoveSubnetFromRouter