Dynamic routing CLIs OSC transition

Neutron is transitioning its CLI to OSC, should update dynamic routing
CLIs to align with OSC plugin.

Change-Id: Ic1a633ca103d9e97e28ddf23e0f528e0261ce497
Co-Authored-By: Ryan Tidwell <ryan@6tidwells.com>
Co-Authored-By: Roey Chen <roeyc@vmware.com>
Partially-Implements: blueprint bgp-spinout
Partial-Bug: #1560003
This commit is contained in:
Na 2016-07-12 02:50:06 -05:00 committed by Akihiro Motoki
parent 0907ccc4df
commit 33947da4b9
10 changed files with 1104 additions and 0 deletions

View File

@ -0,0 +1,50 @@
===================
BGP Dynamic Routing
===================
BGP dynamic routing enables announcement of project subnet prefixes
via BGP. Admins create BGP speakers and BGP peers. BGP peers can be
associated with BGP speakers, thereby enabling peering sessions with
operator infrastructure. BGP speakers can be associated with networks,
which controls which routes are announced to peers.
Network v2
.. autoprogram-cliff:: openstack.neutronclient.v2
:command: bgp speaker create
.. autoprogram-cliff:: openstack.neutronclient.v2
:command: bgp speaker delete
.. autoprogram-cliff:: openstack.neutronclient.v2
:command: bgp speaker list
.. autoprogram-cliff:: openstack.neutronclient.v2
:command: bgp speaker set
.. autoprogram-cliff:: openstack.neutronclient.v2
:command: bgp speaker show
.. autoprogram-cliff:: openstack.neutronclient.v2
:command: bgp speaker show dragents
.. autoprogram-cliff:: openstack.neutronclient.v2
:command: bgp speaker add network
.. autoprogram-cliff:: openstack.neutronclient.v2
:command: bgp speaker remove network
.. autoprogram-cliff:: openstack.neutronclient.v2
:command: bgp speaker add peer
.. autoprogram-cliff:: openstack.neutronclient.v2
:command: bgp speaker remove peer
.. autoprogram-cliff:: openstack.neutronclient.v2
:command: bgp speaker list advertised routes
.. autoprogram-cliff:: openstack.neutronclient.v2
:command: bgp peer *
.. autoprogram-cliff:: openstack.neutronclient.v2
:command: bgp dragent *

View File

@ -0,0 +1,99 @@
# 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.
#
from osc_lib.command import command
from osc_lib import utils
from neutronclient._i18n import _
from neutronclient.osc.v2.dynamic_routing import constants
def _format_alive_state(item):
return ':-)' if item else 'XXX'
_formatters = {
'alive': _format_alive_state
}
def add_common_args(parser):
parser.add_argument('dragent_id',
metavar='<agent-id>',
help=_("ID of the dynamic routing agent"))
parser.add_argument('bgp_speaker',
metavar='<bgp-speaker>',
help=_("ID or name of the BGP speaker"))
class AddBgpSpeakerToDRAgent(command.Command):
"""Add a BGP speaker to a dynamic routing agent"""
def get_parser(self, prog_name):
parser = super(AddBgpSpeakerToDRAgent, self).get_parser(prog_name)
add_common_args(parser)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
speaker_id = client.find_resource(constants.BGP_SPEAKER,
parsed_args.bgp_speaker)['id']
client.add_bgp_speaker_to_dragent(
parsed_args.dragent_id, {'bgp_speaker_id': speaker_id})
class RemoveBgpSpeakerFromDRAgent(command.Command):
"""Removes a BGP speaker from a dynamic routing agent"""
def get_parser(self, prog_name):
parser = super(RemoveBgpSpeakerFromDRAgent, self).get_parser(
prog_name)
add_common_args(parser)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
speaker_id = client.find_resource(constants.BGP_SPEAKER,
parsed_args.bgp_speaker)['id']
client.remove_bgp_speaker_from_dragent(parsed_args.dragent_id,
speaker_id)
class ListDRAgentsHostingBgpSpeaker(command.Lister):
"""List dynamic routing agents hosting a BGP speaker"""
resource = 'agent'
list_columns = ['id', 'host', 'admin_state_up', 'alive']
unknown_parts_flag = False
def get_parser(self, prog_name):
parser = super(ListDRAgentsHostingBgpSpeaker,
self).get_parser(prog_name)
parser.add_argument('bgp_speaker',
metavar='<bgp-speaker>',
help=_("ID or name of the BGP speaker"))
return parser
def take_action(self, parsed_args):
search_opts = {}
client = self.app.client_manager.neutronclient
speaker_id = client.find_resource(constants.BGP_SPEAKER,
parsed_args.bgp_speaker)['id']
search_opts['bgp_speaker'] = speaker_id
data = client.list_dragents_hosting_bgp_speaker(**search_opts)
headers = ('ID', 'Host', 'State', 'Alive')
columns = ('id', 'host', 'admin_state_up', 'alive')
return (headers,
(utils.get_dict_properties(
s, columns, formatters=_formatters,
) for s in data['agents']))

View File

@ -0,0 +1,188 @@
# 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.
#
from osc_lib.command import command
from osc_lib import utils
from neutronclient._i18n import _
from neutronclient.common import exceptions
from neutronclient.common import utils as nc_utils
from neutronclient.osc import utils as nc_osc_utils
from neutronclient.osc.v2.dynamic_routing import constants
def _get_attrs(client_manager, parsed_args):
attrs = {}
# Validate password
if 'auth_type' in parsed_args:
if parsed_args.auth_type != 'none':
if 'password' not in parsed_args or parsed_args.password is None:
raise exceptions.CommandError(_('Must provide password if '
'auth-type is specified.'))
if (
parsed_args.auth_type == 'none' and
parsed_args.password is not None
):
raise exceptions.CommandError(_('Must provide auth-type if '
'password is specified.'))
attrs['auth_type'] = parsed_args.auth_type
if parsed_args.name is not None:
attrs['name'] = parsed_args.name
if 'remote_as' in parsed_args:
attrs['remote_as'] = parsed_args.remote_as
if 'peer_ip' in parsed_args:
attrs['peer_ip'] = parsed_args.peer_ip
if 'password' in parsed_args:
attrs['password'] = parsed_args.password
if 'project' in parsed_args and parsed_args.project is not None:
identity_client = client_manager.identity
project_id = nc_osc_utils.find_project(
identity_client,
parsed_args.project,
parsed_args.project_domain,
).id
attrs['tenant_id'] = project_id
return attrs
class CreateBgpPeer(command.ShowOne):
_description = _("Create a BGP peer")
def get_parser(self, prog_name):
parser = super(CreateBgpPeer, self).get_parser(prog_name)
parser.add_argument(
'name',
metavar='<name>',
help=_("Name of the BGP peer to create"))
parser.add_argument(
'--peer-ip',
metavar='<peer-ip-address>',
required=True,
help=_("Peer IP address"))
parser.add_argument(
'--remote-as',
required=True,
metavar='<peer-remote-as>',
help=_("Peer AS number. (Integer in [%(min_val)s, %(max_val)s] "
"is allowed)") % {'min_val': constants.MIN_AS_NUM,
'max_val': constants.MAX_AS_NUM})
parser.add_argument(
'--auth-type',
metavar='<peer-auth-type>',
choices=['none', 'md5'],
type=nc_utils.convert_to_lowercase,
default='none',
help=_("Authentication algorithm. Supported algorithms: "
"none (default), md5"))
parser.add_argument(
'--password',
metavar='<auth-password>',
help=_("Authentication password"))
nc_osc_utils.add_project_owner_option_to_parser(parser)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
attrs = _get_attrs(self.app.client_manager, parsed_args)
body = {constants.BGP_PEER: attrs}
obj = client.create_bgp_peer(body)[constants.BGP_PEER]
columns, display_columns = nc_osc_utils.get_columns(obj)
data = utils.get_dict_properties(obj, columns)
return display_columns, data
class DeleteBgpPeer(command.Command):
_description = _("Delete a BGP peer")
def get_parser(self, prog_name):
parser = super(DeleteBgpPeer, self).get_parser(prog_name)
parser.add_argument(
'bgp_peer',
metavar="<bgp-peer>",
help=_("BGP peer to delete (name or ID)")
)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
id = client.find_resource(constants.BGP_PEER,
parsed_args.bgp_peer)['id']
client.delete_bgp_peer(id)
class ListBgpPeer(command.Lister):
_description = _("List BGP peers")
def take_action(self, parsed_args):
data = self.app.client_manager.neutronclient.list_bgp_peers()
headers = ('ID', 'Name', 'Peer IP', 'Remote AS')
columns = ('id', 'name', 'peer_ip', 'remote_as')
return (headers,
(utils.get_dict_properties(
s, columns,
) for s in data[constants.BGP_PEERS]))
class SetBgpPeer(command.Command):
_description = _("Update a BGP peer")
resource = constants.BGP_PEER
def get_parser(self, prog_name):
parser = super(SetBgpPeer, self).get_parser(prog_name)
parser.add_argument(
'--name',
help=_("Updated name of the BGP peer"))
parser.add_argument(
'--password',
metavar='<auth-password>',
help=_("Updated authentication password"))
parser.add_argument(
'bgp_peer',
metavar="<bgp-peer>",
help=_("BGP peer to update (name or ID)")
)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
id = client.find_resource(constants.BGP_PEER,
parsed_args.bgp_peer)['id']
attrs = _get_attrs(self.app.client_manager, parsed_args)
body = {}
body[constants.BGP_PEER] = attrs
client.update_bgp_peer(id, body)
class ShowBgpPeer(command.ShowOne):
_description = _("Show information for a BGP peer")
def get_parser(self, prog_name):
parser = super(ShowBgpPeer, self).get_parser(prog_name)
parser.add_argument(
'bgp_peer',
metavar="<bgp-peer>",
help=_("BGP peer to display (name or ID)")
)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
id = client.find_resource(constants.BGP_PEER,
parsed_args.bgp_peer)['id']
obj = client.show_bgp_peer(id)[constants.BGP_PEER]
columns, display_columns = nc_osc_utils.get_columns(obj)
data = utils.get_dict_properties(obj, columns)
return display_columns, data

View File

@ -0,0 +1,320 @@
# 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.
#
from osc_lib.command import command
from osc_lib import utils
from neutronclient._i18n import _
from neutronclient.osc import utils as nc_osc_utils
from neutronclient.osc.v2.dynamic_routing import constants
def _get_attrs(client_manager, parsed_args):
attrs = {}
if parsed_args.name is not None:
attrs['name'] = str(parsed_args.name)
if 'local_as' in parsed_args:
attrs['local_as'] = parsed_args.local_as
if 'ip_version' in parsed_args:
attrs['ip_version'] = parsed_args.ip_version
if parsed_args.advertise_tenant_networks:
attrs['advertise_tenant_networks'] = True
if parsed_args.no_advertise_tenant_networks:
attrs['advertise_tenant_networks'] = False
if parsed_args.advertise_floating_ip_host_routes:
attrs['advertise_floating_ip_host_routes'] = True
if parsed_args.no_advertise_floating_ip_host_routes:
attrs['advertise_floating_ip_host_routes'] = False
if 'project' in parsed_args and parsed_args.project is not None:
identity_client = client_manager.identity
project_id = nc_osc_utils.find_project(
identity_client,
parsed_args.project,
parsed_args.project_domain,
).id
attrs['tenant_id'] = project_id
return attrs
def add_common_arguments(parser):
parser.add_argument(
'--advertise-floating-ip-host-routes',
action='store_true',
help=_("Enable the advertisement of floating IP host routes "
"by the BGP speaker. (default)"))
parser.add_argument(
'--no-advertise-floating-ip-host-routes',
action='store_true',
help=_("Disable the advertisement of floating IP host routes "
"by the BGP speaker."))
parser.add_argument(
'--advertise-tenant-networks',
action='store_true',
help=_("Enable the advertisement of tenant network routes "
"by the BGP speaker. (default)"))
parser.add_argument(
'--no-advertise-tenant-networks',
action='store_true',
help=_("Disable the advertisement of tenant network routes "
"by the BGP speaker."))
class AddNetworkToSpeaker(command.Command):
_description = _("Add a network to a BGP speaker")
def get_parser(self, prog_name):
parser = super(AddNetworkToSpeaker, self).get_parser(prog_name)
parser.add_argument(
'bgp_speaker',
metavar='<bgp-speaker>',
help=_("BGP speaker (name or ID)"))
parser.add_argument(
'network',
metavar='<network>',
help=_("Network to add (name or ID)"))
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
speaker_id = client.find_resource(constants.BGP_SPEAKER,
parsed_args.bgp_speaker)['id']
net_id = client.find_resource('network',
parsed_args.network)['id']
client.add_network_to_bgp_speaker(speaker_id, {'network_id': net_id})
class AddPeerToSpeaker(command.Command):
_description = _("Add a peer to a BGP speaker")
def get_parser(self, prog_name):
parser = super(AddPeerToSpeaker, self).get_parser(prog_name)
parser.add_argument(
'bgp_speaker',
metavar='<bgp-speaker>',
help=_("BGP speaker (name or ID)"))
parser.add_argument(
'bgp_peer',
metavar='<bgp-peer>',
help=_("BGP Peer to add (name or ID)"))
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
speaker_id = client.find_resource(constants.BGP_SPEAKER,
parsed_args.bgp_speaker)['id']
peer_id = client.find_resource(constants.BGP_PEER,
parsed_args.bgp_peer)['id']
client.add_peer_to_bgp_speaker(speaker_id, {'bgp_peer_id': peer_id})
class CreateBgpSpeaker(command.ShowOne):
_description = _("Create a BGP speaker")
def get_parser(self, prog_name):
parser = super(CreateBgpSpeaker, self).get_parser(prog_name)
parser.add_argument(
'name',
metavar='<name>',
help=_("Name of the BGP speaker to create"))
parser.add_argument(
'--local-as',
metavar='<local-as>',
required=True,
help=_("Local AS number. (Integer in [%(min_val)s, %(max_val)s] "
"is allowed.)") % {'min_val': constants.MIN_AS_NUM,
'max_val': constants.MAX_AS_NUM})
parser.add_argument(
'--ip-version',
type=int, choices=[4, 6],
default=4,
help=_("IP version for the BGP speaker (default is 4)"))
add_common_arguments(parser)
nc_osc_utils.add_project_owner_option_to_parser(parser)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
attrs = _get_attrs(self.app.client_manager, parsed_args)
body = {}
body[constants.BGP_SPEAKER] = attrs
obj = client.create_bgp_speaker(body)[constants.BGP_SPEAKER]
columns, display_columns = nc_osc_utils.get_columns(obj)
data = utils.get_dict_properties(obj, columns)
return display_columns, data
class DeleteBgpSpeaker(command.Command):
_description = _("Delete a BGP speaker")
def get_parser(self, prog_name):
parser = super(DeleteBgpSpeaker, self).get_parser(prog_name)
parser.add_argument(
'bgp_speaker',
metavar="<bgp-speaker>",
help=_("BGP speaker to delete (name or ID)")
)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
id = client.find_resource(constants.BGP_SPEAKER,
parsed_args.bgp_speaker)['id']
client.delete_bgp_speaker(id)
class ListBgpSpeaker(command.Lister):
_description = _("List BGP speakers")
def get_parser(self, prog_name):
parser = super(ListBgpSpeaker, self).get_parser(prog_name)
parser.add_argument(
'--agent',
metavar='<agent-id>',
help=_("List BGP speakers hosted by an agent (ID only)"))
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
if parsed_args.agent is not None:
data = client.list_bgp_speaker_on_dragent(parsed_args.agent_id)
else:
data = client.list_bgp_speakers()
headers = ('ID', 'Name', 'Local AS', 'IP Version')
columns = ('id', 'name', 'local_as', 'ip_version')
return (headers, (utils.get_dict_properties(s, columns)
for s in data[constants.BGP_SPEAKERS]))
class ListRoutesAdvertisedBySpeaker(command.Lister):
_description = _("List routes advertised")
def get_parser(self, prog_name):
parser = super(ListRoutesAdvertisedBySpeaker,
self).get_parser(prog_name)
parser.add_argument(
'bgp_speaker',
metavar='<bgp-speaker>',
help=_("BGP speaker (name or ID)"))
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
speaker_id = client.find_resource(constants.BGP_SPEAKER,
parsed_args.bgp_speaker)['id']
data = client.list_route_advertised_from_bgp_speaker(speaker_id)
headers = ('ID', 'Destination', 'Nexthop')
columns = ('id', 'destination', 'next_hop')
return (headers, (utils.get_dict_properties(s, columns)
for s in data['advertised_routes']))
class RemoveNetworkFromSpeaker(command.Command):
_description = _("Remove a network from a BGP speaker")
def get_parser(self, prog_name):
parser = super(RemoveNetworkFromSpeaker, self).get_parser(prog_name)
parser.add_argument(
'bgp_speaker',
metavar='<bgp-speaker>',
help=_("BGP speaker (name or ID)"))
parser.add_argument(
'network',
metavar='<network>',
help=_("Network to remove (name or ID)"))
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
speaker_id = client.find_resource(constants.BGP_SPEAKER,
parsed_args.bgp_speaker)['id']
net_id = client.find_resource('network',
parsed_args.network)['id']
client.remove_network_from_bgp_speaker(speaker_id,
{'network_id': net_id})
class RemovePeerFromSpeaker(command.Command):
_description = _("Remove a peer from a BGP speaker")
def get_parser(self, prog_name):
parser = super(RemovePeerFromSpeaker, self).get_parser(prog_name)
parser.add_argument(
'bgp_speaker',
metavar='<bgp-speaker>',
help=_("BGP speaker (name or ID)"))
parser.add_argument(
'bgp_peer',
metavar='<bgp-peer>',
help=_("BGP Peer to remove (name or ID)"))
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
speaker_id = client.find_resource(constants.BGP_SPEAKER,
parsed_args.bgp_speaker)['id']
peer_id = client.find_resource(constants.BGP_PEER,
parsed_args.bgp_peer)['id']
client.remove_peer_from_bgp_speaker(speaker_id,
{'bgp_peer_id': peer_id})
class SetBgpSpeaker(command.Command):
_description = _("Set BGP speaker properties")
resource = constants.BGP_SPEAKER
def get_parser(self, prog_name):
parser = super(SetBgpSpeaker, self).get_parser(prog_name)
parser.add_argument(
'bgp_speaker',
metavar="<bgp-speaker>",
help=_("BGP speaker to update (name or ID)")
)
parser.add_argument(
'--name',
help=_("Name of the BGP speaker to update"))
add_common_arguments(parser)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
id = client.find_resource(constants.BGP_SPEAKER,
parsed_args.bgp_speaker)['id']
attrs = _get_attrs(self.app.client_manager, parsed_args)
body = {}
body[constants.BGP_SPEAKER] = attrs
client.update_bgp_speaker(id, body)
class ShowBgpSpeaker(command.ShowOne):
_description = _("Show a BGP speaker")
def get_parser(self, prog_name):
parser = super(ShowBgpSpeaker, self).get_parser(prog_name)
parser.add_argument(
'bgp_speaker',
metavar="<bgp-speaker>",
help=_("BGP speaker to display (name or ID)")
)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
id = client.find_resource(constants.BGP_SPEAKER,
parsed_args.bgp_speaker)['id']
obj = client.show_bgp_speaker(id)[constants.BGP_SPEAKER]
columns, display_columns = nc_osc_utils.get_columns(obj)
data = utils.get_dict_properties(obj, columns)
return display_columns, data

View File

@ -0,0 +1,18 @@
# 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.
BGP_SPEAKERS = 'bgp_speakers'
BGP_SPEAKER = 'bgp_speaker'
BGP_PEERS = 'bgp_peers'
BGP_PEER = 'bgp_peer'
MIN_AS_NUM = 1
MAX_AS_NUM = 65535

View File

@ -0,0 +1,96 @@
# 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 uuid
import mock
from neutronclient.tests.unit.osc.v2 import fakes
class TestNeutronDynamicRoutingOSCV2(fakes.TestNeutronClientOSCV2):
def setUp(self):
super(TestNeutronDynamicRoutingOSCV2, self).setUp()
self.neutronclient.find_resource = mock.Mock(
side_effect=lambda resource, name_or_id, project_id=None,
cmd_resource=None, parent_id=None, fields=None:
{'id': name_or_id})
class FakeBgpSpeaker(object):
"""Fake one or more bgp speakers."""
@staticmethod
def create_one_bgp_speaker(attrs=None):
attrs = attrs or {}
# Set default attributes.
bgp_speaker_attrs = {
'peers': [],
'local_as': 200,
'advertise_tenant_networks': True,
'networks': [],
'ip_version': 4,
'advertise_floating_ip_host_routes': True,
'id': uuid.uuid4().hex,
'name': 'bgp-speaker-' + uuid.uuid4().hex,
'tenant_id': uuid.uuid4().hex,
}
# Overwrite default attributes.
bgp_speaker_attrs.update(attrs)
return copy.deepcopy(bgp_speaker_attrs)
@staticmethod
def create_bgp_speakers(attrs=None, count=1):
"""Create multiple fake bgp speakers.
"""
bgp_speakers = []
for i in range(0, count):
bgp_speaker = FakeBgpSpeaker.create_one_bgp_speaker(attrs)
bgp_speakers.append(bgp_speaker)
return {'bgp_speakers': bgp_speakers}
class FakeBgpPeer(object):
"""Fake one or more bgp peers."""
@staticmethod
def create_one_bgp_peer(attrs=None):
attrs = attrs or {}
# Set default attributes.
bgp_peer_attrs = {
'auth_type': None,
'peer_ip': '1.1.1.1',
'remote_as': 100,
'id': uuid.uuid4().hex,
'name': 'bgp-peer-' + uuid.uuid4().hex,
'tenant_id': uuid.uuid4().hex,
}
# Overwrite default attributes.
bgp_peer_attrs.update(attrs)
return copy.deepcopy(bgp_peer_attrs)
@staticmethod
def create_bgp_peers(attrs=None, count=1):
"""Create one or multiple fake bgp peers."""
bgp_peers = []
for i in range(0, count):
bgp_peer = FakeBgpPeer.create_one_bgp_peer(attrs)
bgp_peers.append(bgp_peer)
return {'bgp_peers': bgp_peers}

View File

@ -0,0 +1,153 @@
# 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 neutronclient.osc.v2.dynamic_routing import bgp_peer
from neutronclient.tests.unit.osc.v2.dynamic_routing import fakes
class TestListBgpPeer(fakes.TestNeutronDynamicRoutingOSCV2):
_bgp_peers = fakes.FakeBgpPeer.create_bgp_peers(count=1)
columns = ('ID', 'Name', 'Peer IP', 'Remote AS')
data = []
for _bgp_peer in _bgp_peers['bgp_peers']:
data.append((
_bgp_peer['id'],
_bgp_peer['name'],
_bgp_peer['peer_ip'],
_bgp_peer['remote_as']))
def setUp(self):
super(TestListBgpPeer, self).setUp()
self.neutronclient.list_bgp_peers = mock.Mock(
return_value=self._bgp_peers
)
# Get the command object to test
self.cmd = bgp_peer.ListBgpPeer(self.app, self.namespace)
def test_bgp_peer_list(self):
parsed_args = self.check_parser(self.cmd, [], [])
columns, data = self.cmd.take_action(parsed_args)
self.neutronclient.list_bgp_peers.assert_called_once_with()
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, list(data))
class TestDeleteBgpPeer(fakes.TestNeutronDynamicRoutingOSCV2):
_bgp_peer = fakes.FakeBgpPeer.create_one_bgp_peer()
def setUp(self):
super(TestDeleteBgpPeer, self).setUp()
self.neutronclient.delete_bgp_peer = mock.Mock(return_value=None)
self.cmd = bgp_peer.DeleteBgpPeer(self.app, self.namespace)
def test_delete_bgp_peer(self):
arglist = [
self._bgp_peer['name'],
]
verifylist = [
('bgp_peer', self._bgp_peer['name']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.neutronclient.delete_bgp_peer.assert_called_once_with(
self._bgp_peer['name'])
self.assertIsNone(result)
class TestShowBgpPeer(fakes.TestNeutronDynamicRoutingOSCV2):
_one_bgp_peer = fakes.FakeBgpPeer.create_one_bgp_peer()
data = (
_one_bgp_peer['auth_type'],
_one_bgp_peer['id'],
_one_bgp_peer['name'],
_one_bgp_peer['peer_ip'],
_one_bgp_peer['remote_as'],
_one_bgp_peer['tenant_id']
)
_bgp_peer = {'bgp_peer': _one_bgp_peer}
_bgp_peer_name = _one_bgp_peer['name']
columns = (
'auth_type',
'id',
'name',
'peer_ip',
'remote_as',
'tenant_id'
)
def setUp(self):
super(TestShowBgpPeer, self).setUp()
self.neutronclient.show_bgp_peer = mock.Mock(
return_value=self._bgp_peer
)
bgp_peer.get_bgp_peer_id = mock.Mock(return_value=self._bgp_peer_name)
# Get the command object to test
self.cmd = bgp_peer.ShowBgpPeer(self.app, self.namespace)
def test_bgp_peer_list(self):
arglist = [
self._bgp_peer_name,
]
verifylist = [
('bgp_peer', self._bgp_peer_name),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
data = self.cmd.take_action(parsed_args)
self.neutronclient.show_bgp_peer.assert_called_once_with(
self._bgp_peer_name)
self.assertEqual(self.columns, data[0])
self.assertEqual(self.data, data[1])
class TestSetBgpPeer(fakes.TestNeutronDynamicRoutingOSCV2):
_one_bgp_peer = fakes.FakeBgpPeer.create_one_bgp_peer()
_bgp_peer_name = _one_bgp_peer['name']
def setUp(self):
super(TestSetBgpPeer, self).setUp()
self.neutronclient.update_bgp_peer = mock.Mock(return_value=None)
bgp_peer.get_bgp_peer_id = mock.Mock(return_value=self._bgp_peer_name)
self.cmd = bgp_peer.SetBgpPeer(self.app, self.namespace)
def test_set_bgp_peer(self):
arglist = [
self._bgp_peer_name,
'--name', 'noob',
]
verifylist = [
('bgp_peer', self._bgp_peer_name),
('name', 'noob'),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
attrs = {'bgp_peer': {
'name': 'noob',
'password': None}
}
self.neutronclient.update_bgp_peer.assert_called_once_with(
self._bgp_peer_name, attrs)
self.assertIsNone(result)

View File

@ -0,0 +1,157 @@
# 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 neutronclient.osc.v2.dynamic_routing import bgp_speaker
from neutronclient.tests.unit.osc.v2.dynamic_routing import fakes
class TestListBgpSpeaker(fakes.TestNeutronDynamicRoutingOSCV2):
_bgp_speakers = fakes.FakeBgpSpeaker.create_bgp_speakers()
columns = ('ID', 'Name', 'Local AS', 'IP Version')
data = []
for _bgp_speaker in _bgp_speakers['bgp_speakers']:
data.append((
_bgp_speaker['id'],
_bgp_speaker['name'],
_bgp_speaker['local_as'],
_bgp_speaker['ip_version']))
def setUp(self):
super(TestListBgpSpeaker, self).setUp()
self.neutronclient.list_bgp_speakers = mock.Mock(
return_value=self._bgp_speakers
)
# Get the command object to test
self.cmd = bgp_speaker.ListBgpSpeaker(self.app, self.namespace)
def test_bgp_speaker_list(self):
parsed_args = self.check_parser(self.cmd, [], [])
columns, data = self.cmd.take_action(parsed_args)
self.neutronclient.list_bgp_speakers.assert_called_once_with()
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, list(data))
class TestDeleteBgpSpeaker(fakes.TestNeutronDynamicRoutingOSCV2):
_bgp_speaker = fakes.FakeBgpSpeaker.create_one_bgp_speaker()
def setUp(self):
super(TestDeleteBgpSpeaker, self).setUp()
self.neutronclient.delete_bgp_speaker = mock.Mock(return_value=None)
self.cmd = bgp_speaker.DeleteBgpSpeaker(self.app, self.namespace)
def test_delete_bgp_speaker(self):
arglist = [
self._bgp_speaker['name'],
]
verifylist = [
('bgp_speaker', self._bgp_speaker['name']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.neutronclient.delete_bgp_speaker.assert_called_once_with(
self._bgp_speaker['name'])
self.assertIsNone(result)
class TestShowBgpSpeaker(fakes.TestNeutronDynamicRoutingOSCV2):
_one_bgp_speaker = fakes.FakeBgpSpeaker.create_one_bgp_speaker()
data = (
_one_bgp_speaker['advertise_floating_ip_host_routes'],
_one_bgp_speaker['advertise_tenant_networks'],
_one_bgp_speaker['id'],
_one_bgp_speaker['ip_version'],
_one_bgp_speaker['local_as'],
_one_bgp_speaker['name'],
_one_bgp_speaker['networks'],
_one_bgp_speaker['peers'],
_one_bgp_speaker['tenant_id']
)
_bgp_speaker = {'bgp_speaker': _one_bgp_speaker}
_bgp_speaker_name = _one_bgp_speaker['name']
columns = (
'advertise_floating_ip_host_routes',
'advertise_tenant_networks',
'id',
'ip_version',
'local_as',
'name',
'networks',
'peers',
'tenant_id'
)
def setUp(self):
super(TestShowBgpSpeaker, self).setUp()
self.neutronclient.show_bgp_speaker = mock.Mock(
return_value=self._bgp_speaker
)
# Get the command object to test
self.cmd = bgp_speaker.ShowBgpSpeaker(self.app, self.namespace)
def test_bgp_speaker_show(self):
arglist = [
self._bgp_speaker_name,
]
verifylist = [
('bgp_speaker', self._bgp_speaker_name),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
data = self.cmd.take_action(parsed_args)
self.neutronclient.show_bgp_speaker.assert_called_once_with(
self._bgp_speaker_name)
self.assertEqual(self.columns, data[0])
self.assertEqual(self.data, data[1])
class TestSetBgpSpeaker(fakes.TestNeutronDynamicRoutingOSCV2):
_one_bgp_speaker = fakes.FakeBgpSpeaker.create_one_bgp_speaker()
_bgp_speaker_name = _one_bgp_speaker['name']
def setUp(self):
super(TestSetBgpSpeaker, self).setUp()
self.neutronclient.update_bgp_speaker = mock.Mock(
return_value=None)
self.cmd = bgp_speaker.SetBgpSpeaker(self.app, self.namespace)
def test_set_bgp_speaker(self):
arglist = [
self._bgp_speaker_name,
'--name', 'noob',
]
verifylist = [
('bgp_speaker', self._bgp_speaker_name),
('name', 'noob'),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
attrs = {'bgp_speaker': {
'name': 'noob'}
}
self.neutronclient.update_bgp_speaker.assert_called_once_with(
self._bgp_speaker_name, attrs)
self.assertIsNone(result)

View File

@ -0,0 +1,4 @@
---
features:
- |
Add OSC plugin to support "Neutron Dynamic Routing"

View File

@ -65,6 +65,25 @@ openstack.neutronclient.v2 =
sfc_port_pair_group_show = neutronclient.osc.v2.sfc.sfc_port_pair_group:ShowSfcPortPairGroup
sfc_port_pair_group_unset = neutronclient.osc.v2.sfc.sfc_port_pair_group:UnsetSfcPortPairGroup
bgp_dragent_add_speaker = neutronclient.osc.v2.dynamic_routing.bgp_dragent:AddBgpSpeakerToDRAgent
bgp_dragent_remove_speaker = neutronclient.osc.v2.dynamic_routing.bgp_dragent:RemoveBgpSpeakerFromDRAgent
bgp_peer_create = neutronclient.osc.v2.dynamic_routing.bgp_peer:CreateBgpPeer
bgp_peer_delete = neutronclient.osc.v2.dynamic_routing.bgp_peer:DeleteBgpPeer
bgp_peer_list = neutronclient.osc.v2.dynamic_routing.bgp_peer:ListBgpPeer
bgp_peer_show = neutronclient.osc.v2.dynamic_routing.bgp_peer:ShowBgpPeer
bgp_peer_set = neutronclient.osc.v2.dynamic_routing.bgp_peer:SetBgpPeer
bgp_speaker_list_advertised_routes = neutronclient.osc.v2.dynamic_routing.bgp_speaker:ListRoutesAdvertisedBySpeaker
bgp_speaker_create = neutronclient.osc.v2.dynamic_routing.bgp_speaker:CreateBgpSpeaker
bgp_speaker_delete = neutronclient.osc.v2.dynamic_routing.bgp_speaker:DeleteBgpSpeaker
bgp_speaker_list = neutronclient.osc.v2.dynamic_routing.bgp_speaker:ListBgpSpeaker
bgp_speaker_add_network = neutronclient.osc.v2.dynamic_routing.bgp_speaker:AddNetworkToSpeaker
bgp_speaker_remove_network = neutronclient.osc.v2.dynamic_routing.bgp_speaker:RemoveNetworkFromSpeaker
bgp_speaker_add_peer = neutronclient.osc.v2.dynamic_routing.bgp_speaker:AddPeerToSpeaker
bgp_speaker_remove_peer = neutronclient.osc.v2.dynamic_routing.bgp_speaker:RemovePeerFromSpeaker
bgp_speaker_set = neutronclient.osc.v2.dynamic_routing.bgp_speaker:SetBgpSpeaker
bgp_speaker_show = neutronclient.osc.v2.dynamic_routing.bgp_speaker:ShowBgpSpeaker
bgp_speaker_show_dragents = neutronclient.osc.v2.dynamic_routing.bgp_dragent:ListDRAgentsHostingBgpSpeaker
firewall_group_create = neutronclient.osc.v2.fwaas.firewallgroup:CreateFirewallGroup
firewall_group_delete = neutronclient.osc.v2.fwaas.firewallgroup:DeleteFirewallGroup
firewall_group_list = neutronclient.osc.v2.fwaas.firewallgroup:ListFirewallGroup