# 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='', help=_("Name of the BGP peer to create")) parser.add_argument( '--peer-ip', metavar='', required=True, help=_("Peer IP address")) parser.add_argument( '--remote-as', required=True, metavar='', 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='', choices=['none', 'md5'], type=nc_utils.convert_to_lowercase, default='none', help=_("Authentication algorithm. Supported algorithms: " "none (default), md5")) parser.add_argument( '--password', metavar='', 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.network attrs = _get_attrs(self.app.client_manager, parsed_args) obj = client.create_bgp_peer(**attrs) display_columns, 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="", help=_("BGP peer to delete (name or ID)") ) return parser def take_action(self, parsed_args): client = self.app.client_manager.network id = client.find_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.network.bgp_peers(retrieve_all=True) 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)) 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='', help=_("Updated authentication password")) parser.add_argument( 'bgp_peer', metavar="", help=_("BGP peer to update (name or ID)") ) return parser def take_action(self, parsed_args): client = self.app.client_manager.network id = client.find_bgp_peer(parsed_args.bgp_peer)['id'] attrs = _get_attrs(self.app.client_manager, parsed_args) client.update_bgp_peer(id, **attrs) 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="", help=_("BGP peer to display (name or ID)") ) return parser def take_action(self, parsed_args): client = self.app.client_manager.network id = client.find_bgp_peer(parsed_args.bgp_peer, ignore_missing=False).id obj = client.get_bgp_peer(id) display_columns, columns = nc_osc_utils._get_columns(obj) data = utils.get_dict_properties(obj, columns) return display_columns, data