Support of the boolean advertise_extra_routes

This Boolean attribute is added to the Router Association resource
(Neutron API, 'bgpvpn' and 'bgpvpn-routes-control' API extensions) in
order to support routes control. This allow to propagate routes of
subnets not directly connected to the router
The corresponding code in networking-bgpvpn has already merged

See https://blueprints.launchpad.net/bgpvpn/+spec/routes-control

Change-Id: Icdd7f6592a9d657b6414645406f06b74b6f3bb11
Implements: blueprint routes-control
This commit is contained in:
Samuel Barré 2018-02-20 15:12:31 +01:00 committed by Akihiro Motoki
parent 2cf52672b7
commit b60283cd16
5 changed files with 391 additions and 1 deletions

View File

@ -24,8 +24,12 @@ from neutronclient.osc.v2.networking_bgpvpn.resource_association import\
DeleteBgpvpnResAssoc
from neutronclient.osc.v2.networking_bgpvpn.resource_association import\
ListBgpvpnResAssoc
from neutronclient.osc.v2.networking_bgpvpn.resource_association import\
SetBgpvpnResAssoc
from neutronclient.osc.v2.networking_bgpvpn.resource_association import\
ShowBgpvpnResAssoc
from neutronclient.osc.v2.networking_bgpvpn.resource_association import\
UnsetBgpvpnResAssoc
class BgpvpnRouterAssoc(object):
@ -38,15 +42,62 @@ class BgpvpnRouterAssoc(object):
('tenant_id', 'Project', column_util.LIST_LONG_ONLY),
('%s_id' % _assoc_res_name, '%s ID' % _assoc_res_name.capitalize(),
column_util.LIST_BOTH),
('advertise_extra_routes', 'Advertise extra routes',
column_util.LIST_LONG_ONLY),
)
_formatters = {}
def _get_common_parser(self, parser):
"""Adds to parser arguments common to create, set and unset commands.
:params ArgumentParser parser: argparse object contains all command's
arguments
"""
ADVERTISE_ROUTES = _("Routes will be advertised to the "
"BGP VPN%s") % (
_(' (default)') if self._action == 'create'
else "")
NOT_ADVERTISE_ROUTES = _("Routes from the router will not be "
"advertised to the BGP VPN")
group_advertise_extra_routes = parser.add_mutually_exclusive_group()
group_advertise_extra_routes.add_argument(
'--advertise_extra_routes',
action='store_true',
help=NOT_ADVERTISE_ROUTES if self._action == 'unset'
else ADVERTISE_ROUTES,
)
group_advertise_extra_routes.add_argument(
'--no-advertise_extra_routes',
action='store_true',
help=ADVERTISE_ROUTES if self._action == 'unset'
else NOT_ADVERTISE_ROUTES,
)
def _args2body(self, _, args):
attrs = {}
if args.advertise_extra_routes:
attrs['advertise_extra_routes'] = self._action != 'unset'
elif args.no_advertise_extra_routes:
attrs['advertise_extra_routes'] = self._action == 'unset'
return {self._resource: attrs}
class CreateBgpvpnRouterAssoc(BgpvpnRouterAssoc, CreateBgpvpnResAssoc):
_description = _("Create a BGP VPN router association")
pass
class SetBgpvpnRouterAssoc(BgpvpnRouterAssoc, SetBgpvpnResAssoc):
_description = _("Set BGP VPN router association properties")
class UnsetBgpvpnRouterAssoc(BgpvpnRouterAssoc, UnsetBgpvpnResAssoc):
_description = _("Unset BGP VPN router association properties")
class DeleteBgpvpnRouterAssoc(BgpvpnRouterAssoc, DeleteBgpvpnResAssoc):
_description = _("Delete a BGP VPN router association(s) for a given BGP "
"VPN")

View File

@ -33,6 +33,12 @@ from neutronclient.osc.v2.networking_bgpvpn.resource_association import\
ShowBgpvpnResAssoc
from neutronclient.osc.v2.networking_bgpvpn.resource_association import\
UnsetBgpvpnResAssoc
from neutronclient.osc.v2.networking_bgpvpn.router_association import\
CreateBgpvpnRouterAssoc
from neutronclient.osc.v2.networking_bgpvpn.router_association import\
SetBgpvpnRouterAssoc
from neutronclient.osc.v2.networking_bgpvpn.router_association import\
ShowBgpvpnRouterAssoc
from neutronclient.tests.unit.osc.v2 import fakes as test_fakes
@ -139,6 +145,35 @@ class ShowBgpvpnFakeResAssoc(BgpvpnFakeAssoc, ShowBgpvpnResAssoc):
pass
class BgpvpnFakeRouterAssoc(object):
_assoc_res_name = 'fake_resource'
_resource = '%s_association' % _assoc_res_name
_resource_plural = '%ss' % _resource
_attr_map = (
('id', 'ID', column_util.LIST_BOTH),
('tenant_id', 'Project', column_util.LIST_LONG_ONLY),
('%s_id' % _assoc_res_name, '%s ID' % _assoc_res_name.capitalize(),
column_util.LIST_BOTH),
('advertise_extra_routes', 'Advertise extra routes',
column_util.LIST_LONG_ONLY),
)
_formatters = {}
class CreateBgpvpnFakeRouterAssoc(BgpvpnFakeRouterAssoc,
CreateBgpvpnRouterAssoc):
pass
class SetBgpvpnFakeRouterAssoc(BgpvpnFakeRouterAssoc, SetBgpvpnRouterAssoc):
pass
class ShowBgpvpnFakeRouterAssoc(BgpvpnFakeRouterAssoc, ShowBgpvpnRouterAssoc):
pass
class FakeResource(object):
"""Fake resource with minimal attributes."""
@ -177,14 +212,19 @@ class FakeResAssoc(object):
"""Fake resource association with minimal attributes."""
@staticmethod
def create_one_resource_association(resource):
def create_one_resource_association(resource, attrs=None):
"""Create a fake resource association."""
attrs = attrs or {}
res_assoc_attrs = {
'id': 'fake_association_id',
'tenant_id': resource['tenant_id'],
'fake_resource_id': resource['id'],
}
# Overwrite default attributes.
res_assoc_attrs.update(attrs)
return copy.deepcopy(res_assoc_attrs)
@staticmethod

View File

@ -0,0 +1,291 @@
# Copyright (c) 2018 Orange SA.
# 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 copy
import operator
import mock
from osc_lib.tests.utils import ParserException
from osc_lib import utils as osc_utils
from osc_lib.utils import columns as column_util
from neutronclient.tests.unit.osc.v2.networking_bgpvpn import fakes
columns_short = tuple(col for col, _, listing_mode
in fakes.BgpvpnFakeRouterAssoc._attr_map
if listing_mode in (column_util.LIST_BOTH,
column_util.LIST_SHORT_ONLY))
columns_long = tuple(col for col, _, listing_mode
in fakes.BgpvpnFakeRouterAssoc._attr_map
if listing_mode in (column_util.LIST_BOTH,
column_util.LIST_LONG_ONLY))
headers_short = tuple(head for _, head, listing_mode
in fakes.BgpvpnFakeRouterAssoc._attr_map
if listing_mode in (column_util.LIST_BOTH,
column_util.LIST_SHORT_ONLY))
headers_long = tuple(head for _, head, listing_mode
in fakes.BgpvpnFakeRouterAssoc._attr_map
if listing_mode in (column_util.LIST_BOTH,
column_util.LIST_LONG_ONLY))
sorted_attr_map = sorted(fakes.BgpvpnFakeRouterAssoc._attr_map,
key=operator.itemgetter(1))
sorted_columns = tuple(col for col, _, _ in sorted_attr_map)
sorted_headers = tuple(head for _, head, _ in sorted_attr_map)
def _get_data(attrs, columns=sorted_columns):
return osc_utils.get_dict_properties(
attrs, columns, formatters=fakes.BgpvpnFakeAssoc._formatters)
class TestCreateRouterAssoc(fakes.TestNeutronClientBgpvpn):
def setUp(self):
super(TestCreateRouterAssoc, self).setUp()
self.cmd = fakes.CreateBgpvpnFakeRouterAssoc(self.app, self.namespace)
self.fake_bgpvpn = fakes.FakeBgpvpn.create_one_bgpvpn()
self.fake_router = fakes.FakeResource.create_one_resource()
def _build_args(self, param=None):
arglist_base = [
self.fake_bgpvpn['id'],
self.fake_router['id'],
'--project', self.fake_bgpvpn['tenant_id']
]
if param is not None:
if isinstance(param, list):
arglist_base.extend(param)
else:
arglist_base.append(param)
return arglist_base
def _build_verify_list(self, param=None):
verifylist = [
('bgpvpn', self.fake_bgpvpn['id']),
('resource', self.fake_router['id']),
('project', self.fake_bgpvpn['tenant_id'])
]
if param is not None:
verifylist.append(param)
return verifylist
def _exec_create_router_association(
self, fake_res_assoc, arglist, verifylist):
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
cols, data = self.cmd.take_action(parsed_args)
fake_res_assoc_call = copy.deepcopy(fake_res_assoc)
fake_res_assoc_call.pop('id')
self.neutronclient.create_bgpvpn_fake_resource_assoc.\
assert_called_once_with(
self.fake_bgpvpn['id'],
{fakes.BgpvpnFakeRouterAssoc._resource: fake_res_assoc_call})
return cols, data
def test_create_router_association(self):
fake_res_assoc = fakes.FakeResAssoc.create_one_resource_association(
self.fake_router)
self.neutronclient.create_bgpvpn_fake_resource_assoc = mock.Mock(
return_value={
fakes.BgpvpnFakeRouterAssoc._resource: fake_res_assoc,
'advertise_extra_routes': True})
arglist = self._build_args()
# advertise_extra_routes will be False since none
# of the mutually exclusive args present
verifylist = self._build_verify_list(('advertise_extra_routes', False))
self._exec_create_router_association(
fake_res_assoc, arglist, verifylist)
def test_create_router_association_advertise(self):
fake_res_assoc = fakes.FakeResAssoc.create_one_resource_association(
self.fake_router,
{'advertise_extra_routes': True})
self.neutronclient.create_bgpvpn_fake_resource_assoc = mock.Mock(
return_value={
fakes.BgpvpnFakeRouterAssoc._resource: fake_res_assoc})
arglist = self._build_args('--advertise_extra_routes')
verifylist = self._build_verify_list(('advertise_extra_routes', True))
cols, data = self._exec_create_router_association(
fake_res_assoc, arglist, verifylist)
self.assertEqual(sorted_headers, cols)
self.assertEqual(_get_data(fake_res_assoc), data)
def test_create_router_association_no_advertise(self):
fake_res_assoc = fakes.FakeResAssoc.create_one_resource_association(
self.fake_router,
{'advertise_extra_routes': False})
self.neutronclient.create_bgpvpn_fake_resource_assoc = mock.Mock(
return_value={
fakes.BgpvpnFakeRouterAssoc._resource: fake_res_assoc})
arglist = self._build_args('--no-advertise_extra_routes')
verifylist = self._build_verify_list(('advertise_extra_routes', False))
cols, data = self._exec_create_router_association(
fake_res_assoc, arglist, verifylist)
self.assertEqual(sorted_headers, cols)
self.assertEqual(_get_data(fake_res_assoc), data)
def test_create_router_association_advertise_fault(self):
arglist = self._build_args(
['--advertise_extra_routes', '--no-advertise_extra_routes'])
try:
self._exec_create_router_association(None, arglist, None)
except ParserException as e:
self.assertEqual(format(e), 'Argument parse failed')
def test_router_association_unknown_arg(self):
arglist = self._build_args('--unknown arg')
try:
self._exec_create_router_association(None, arglist, None)
except ParserException as e:
self.assertEqual(format(e), 'Argument parse failed')
class TestSetRouterAssoc(fakes.TestNeutronClientBgpvpn):
def setUp(self):
super(TestSetRouterAssoc, self).setUp()
self.cmd = fakes.SetBgpvpnFakeRouterAssoc(self.app, self.namespace)
self.fake_bgpvpn = fakes.FakeBgpvpn.create_one_bgpvpn()
self.fake_router = fakes.FakeResource.create_one_resource()
def _build_args(self, fake_res_assoc, param=None):
arglist_base = [
fake_res_assoc['id'],
self.fake_bgpvpn['id']
]
if param is not None:
if isinstance(param, list):
arglist_base.extend(param)
else:
arglist_base.append(param)
return arglist_base
def _build_verify_list(self, fake_res_assoc, param=None):
verifylist = [
('resource_association_id', fake_res_assoc['id']),
('bgpvpn', self.fake_bgpvpn['id'])
]
if param is not None:
verifylist.append(param)
return verifylist
def test_set_router_association_no_advertise(self):
fake_res_assoc = fakes.FakeResAssoc.create_one_resource_association(
self.fake_router,
{'advertise_extra_routes': True})
self.neutronclient.update_bgpvpn_fake_resource_assoc = mock.Mock()
arglist = self._build_args(
fake_res_assoc,
'--no-advertise_extra_routes')
verifylist = [
('resource_association_id', fake_res_assoc['id']),
('bgpvpn', self.fake_bgpvpn['id']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
fake_res_assoc_call = copy.deepcopy(fake_res_assoc)
fake_res_assoc_call.pop('id')
self.neutronclient.update_bgpvpn_fake_resource_assoc.\
assert_called_once_with(
self.fake_bgpvpn['id'],
fake_res_assoc['id'],
{
fakes.BgpvpnFakeRouterAssoc._resource: {
'advertise_extra_routes': False
}
})
self.assertIsNone(result)
def test_set_router_association_advertise(self):
fake_res_assoc = fakes.FakeResAssoc.create_one_resource_association(
self.fake_router,
{'advertise_extra_routes': False})
self.neutronclient.update_bgpvpn_fake_resource_assoc = mock.Mock()
arglist = self._build_args(
fake_res_assoc,
'--advertise_extra_routes')
verifylist = [
('resource_association_id', fake_res_assoc['id']),
('bgpvpn', self.fake_bgpvpn['id']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
fake_res_assoc_call = copy.deepcopy(fake_res_assoc)
fake_res_assoc_call.pop('id')
self.neutronclient.update_bgpvpn_fake_resource_assoc.\
assert_called_once_with(
self.fake_bgpvpn['id'],
fake_res_assoc['id'],
{
fakes.BgpvpnFakeRouterAssoc._resource: {
'advertise_extra_routes': True
}
})
self.assertIsNone(result)
class TestShowRouterAssoc(fakes.TestNeutronClientBgpvpn):
def setUp(self):
super(TestShowRouterAssoc, self).setUp()
self.cmd = fakes.ShowBgpvpnFakeRouterAssoc(self.app, self.namespace)
def test_show_router_association(self):
fake_bgpvpn = fakes.FakeBgpvpn.create_one_bgpvpn()
fake_res = fakes.FakeResource.create_one_resource()
fake_res_assoc = fakes.FakeResAssoc.create_one_resource_association(
fake_res,
{'advertise_extra_routes': True})
self.neutronclient.show_bgpvpn_fake_resource_assoc = mock.Mock(
return_value={fakes.BgpvpnFakeAssoc._resource: fake_res_assoc})
arglist = [
fake_res_assoc['id'],
fake_bgpvpn['id'],
]
verifylist = [
('resource_association_id', fake_res_assoc['id']),
('bgpvpn', fake_bgpvpn['id']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
headers, data = self.cmd.take_action(parsed_args)
self.neutronclient.show_bgpvpn_fake_resource_assoc.\
assert_called_once_with(fake_bgpvpn['id'], fake_res_assoc['id'])
self.assertEqual(sorted_headers, headers)
self.assertEqual(data, _get_data(fake_res_assoc))

View File

@ -0,0 +1,6 @@
---
features:
- |
Add optional flag to control the advertisement in BGPVPNs
of the routes defined on a Router resource
(``bgpvpn-routes-control`` API extension).

View File

@ -123,7 +123,9 @@ openstack.neutronclient.v2 =
bgpvpn_router_association_create = neutronclient.osc.v2.networking_bgpvpn.router_association:CreateBgpvpnRouterAssoc
bgpvpn_router_association_delete = neutronclient.osc.v2.networking_bgpvpn.router_association:DeleteBgpvpnRouterAssoc
bgpvpn_router_association_list = neutronclient.osc.v2.networking_bgpvpn.router_association:ListBgpvpnRouterAssoc
bgpvpn_router_association_set = neutronclient.osc.v2.networking_bgpvpn.router_association:SetBgpvpnRouterAssoc
bgpvpn_router_association_show = neutronclient.osc.v2.networking_bgpvpn.router_association:ShowBgpvpnRouterAssoc
bgpvpn_router_association_unset = neutronclient.osc.v2.networking_bgpvpn.router_association:UnsetBgpvpnRouterAssoc
bgpvpn_port_association_create = neutronclient.osc.v2.networking_bgpvpn.port_association:CreateBgpvpnPortAssoc
bgpvpn_port_association_set = neutronclient.osc.v2.networking_bgpvpn.port_association:SetBgpvpnPortAssoc
bgpvpn_port_association_unset = neutronclient.osc.v2.networking_bgpvpn.port_association:UnsetBgpvpnPortAssoc