<s> from neutron . common import constants as const <EOL> from neutron . common import exceptions as n_exc <EOL> from neutron . db import servicetype_db as st_db <EOL> from neutron . i18n import _LI <EOL> from neutron import manager <EOL> from neutron . plugins . common import constants as plugin_constants <EOL> from neutron . services import provider_configuration as pconf <EOL> from neutron . services import service_base <EOL> from oslo_log import log <EOL> from networking_bgpvpn . neutron . extensions . bgpvpn import BGPVPNPluginBase <EOL> from networking_bgpvpn . neutron . services . common import constants <EOL> LOG = log . getLogger ( __name__ ) <EOL> class BGPVPNPlugin ( BGPVPNPluginBase ) : <EOL> supported_extension_aliases = [ "<STR_LIT>" ] <EOL> path_prefix = "<STR_LIT>" <EOL> def __init__ ( self ) : <EOL> super ( BGPVPNPlugin , self ) . __init__ ( ) <EOL> service_type_manager = st_db . ServiceTypeManager . get_instance ( ) <EOL> service_type_manager . add_provider_configuration ( <EOL> constants . BGPVPN , <EOL> pconf . ProviderConfiguration ( '<STR_LIT>' ) ) <EOL> drivers , default_provider = service_base . load_drivers ( constants . BGPVPN , <EOL> self ) <EOL> LOG . info ( _LI ( "<STR_LIT>" ) , <EOL> default_provider ) <EOL> self . driver = drivers [ default_provider ] <EOL> if len ( drivers ) > <NUM_LIT:1> : <EOL> LOG . warning ( _LI ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> def _validate_network ( self , context , net_assoc ) : <EOL> if not net_assoc or '<STR_LIT>' not in net_assoc : <EOL> msg = '<STR_LIT>' <EOL> raise n_exc . BadRequest ( resource = '<STR_LIT>' , msg = msg ) <EOL> plugin = manager . NeutronManager . get_plugin ( ) <EOL> network = plugin . get_network ( context , net_assoc [ '<STR_LIT>' ] ) <EOL> self . _validate_network_has_router_assoc ( context , network , plugin ) <EOL> return network <EOL> def _validate_network_has_router_assoc ( self , context , network , plugin ) : <EOL> filter = { '<STR_LIT>' : [ network [ '<STR_LIT:id>' ] ] , <EOL> '<STR_LIT>' : [ const . DEVICE_OWNER_ROUTER_INTF ] } <EOL> router_port = plugin . get_ports ( context , filters = filter ) <EOL> if router_port : <EOL> router_id = router_port [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> filter = { '<STR_LIT>' : [ network [ '<STR_LIT>' ] ] } <EOL> bgpvpns = self . driver . get_bgpvpns ( context , filters = filter ) <EOL> bgpvpns = [ str ( bgpvpn [ '<STR_LIT:id>' ] ) for bgpvpn in bgpvpns <EOL> if router_id in bgpvpn [ '<STR_LIT>' ] ] <EOL> if bgpvpns : <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % { '<STR_LIT>' : network [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : bgpvpns } <EOL> ) <EOL> raise n_exc . BadRequest ( resource = '<STR_LIT>' , msg = msg ) <EOL> def _validate_router ( self , context , router_assoc ) : <EOL> if not router_assoc or '<STR_LIT>' not in router_assoc : <EOL> msg = '<STR_LIT>' <EOL> raise n_exc . BadRequest ( resource = '<STR_LIT>' , msg = msg ) <EOL> l3_plugin = manager . NeutronManager . get_service_plugins ( ) . get ( <EOL> plugin_constants . L3_ROUTER_NAT ) <EOL> router = l3_plugin . get_router ( context , router_assoc [ '<STR_LIT>' ] ) <EOL> plugin = manager . NeutronManager . get_plugin ( ) <EOL> self . _validate_router_has_net_assocs ( context , router , plugin ) <EOL> return router <EOL> def _validate_router_has_net_assocs ( self , context , router , plugin ) : <EOL> filter = { '<STR_LIT>' : [ router [ '<STR_LIT:id>' ] ] , <EOL> '<STR_LIT>' : [ const . DEVICE_OWNER_ROUTER_INTF ] } <EOL> router_ports = plugin . get_ports ( context , filters = filter ) <EOL> if router_ports : <EOL> filter = { '<STR_LIT>' : [ router [ '<STR_LIT>' ] ] } <EOL> bgpvpns = self . driver . get_bgpvpns ( context , filters = filter ) <EOL> for port in router_ports : <EOL> bgpvpns = [ str ( bgpvpn [ '<STR_LIT:id>' ] ) for bgpvpn in bgpvpns <EOL> if port [ '<STR_LIT>' ] in bgpvpn [ '<STR_LIT>' ] ] <EOL> if bgpvpns : <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % { '<STR_LIT>' : router [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : port [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : bgpvpns } ) <EOL> raise n_exc . BadRequest ( resource = '<STR_LIT>' , msg = msg ) <EOL> def get_plugin_type ( self ) : <EOL> return constants . BGPVPN <EOL> def get_plugin_description ( self ) : <EOL> return "<STR_LIT>" <EOL> def create_bgpvpn ( self , context , bgpvpn ) : <EOL> bgpvpn = bgpvpn [ '<STR_LIT>' ] <EOL> return self . driver . create_bgpvpn ( context , bgpvpn ) <EOL> def get_bgpvpns ( self , context , filters = None , fields = None ) : <EOL> return self . driver . get_bgpvpns ( context , filters , fields ) <EOL> def get_bgpvpn ( self , context , id , fields = None ) : <EOL> return self . driver . get_bgpvpn ( context , id , fields ) <EOL> def update_bgpvpn ( self , context , id , bgpvpn ) : <EOL> bgpvpn = bgpvpn [ '<STR_LIT>' ] <EOL> return self . driver . update_bgpvpn ( context , id , bgpvpn ) <EOL> def delete_bgpvpn ( self , context , id ) : <EOL> self . driver . delete_bgpvpn ( context , id ) <EOL> def create_bgpvpn_network_association ( self , context , bgpvpn_id , <EOL> network_association ) : <EOL> net_assoc = network_association [ '<STR_LIT>' ] <EOL> net = self . _validate_network ( context , net_assoc ) <EOL> bgpvpn = self . get_bgpvpn ( context , bgpvpn_id ) <EOL> if net [ '<STR_LIT>' ] != bgpvpn [ '<STR_LIT>' ] : <EOL> msg = '<STR_LIT>' <EOL> raise n_exc . NotAuthorized ( resource = '<STR_LIT>' , msg = msg ) <EOL> if net_assoc [ '<STR_LIT>' ] != bgpvpn [ '<STR_LIT>' ] : <EOL> msg = '<STR_LIT>' <EOL> raise n_exc . NotAuthorized ( resource = '<STR_LIT>' , msg = msg ) <EOL> return self . driver . create_net_assoc ( context , bgpvpn_id , net_assoc ) <EOL> def get_bgpvpn_network_association ( self , context , assoc_id , bgpvpn_id , <EOL> fields = None ) : <EOL> return self . driver . get_net_assoc ( context , assoc_id , bgpvpn_id , fields ) <EOL> def get_bgpvpn_network_associations ( self , context , bgpvpn_id , <EOL> filters = None , fields = None ) : <EOL> return self . driver . get_net_assocs ( context , bgpvpn_id , filters , fields ) <EOL> def update_bgpvpn_network_association ( self , context , assoc_id , bgpvpn_id , <EOL> network_association ) : <EOL> pass <EOL> def delete_bgpvpn_network_association ( self , context , assoc_id , bgpvpn_id ) : <EOL> self . driver . delete_net_assoc ( context , assoc_id , bgpvpn_id ) <EOL> def create_bgpvpn_router_association ( self , context , bgpvpn_id , <EOL> router_association ) : <EOL> router_assoc = router_association [ '<STR_LIT>' ] <EOL> router = self . _validate_router ( context , router_assoc ) <EOL> bgpvpn = self . get_bgpvpn ( context , bgpvpn_id ) <EOL> if not bgpvpn [ '<STR_LIT:type>' ] == constants . BGPVPN_L3 : <EOL> msg = ( "<STR_LIT>" <EOL> % constants . BGPVPN_L3 ) <EOL> raise n_exc . BadRequest ( resource = '<STR_LIT>' , msg = msg ) <EOL> if not router [ '<STR_LIT>' ] == bgpvpn [ '<STR_LIT>' ] : <EOL> msg = "<STR_LIT>" <EOL> raise n_exc . NotAuthorized ( resource = '<STR_LIT>' , msg = msg ) <EOL> if not ( router_assoc [ '<STR_LIT>' ] == bgpvpn [ '<STR_LIT>' ] ) : <EOL> msg = "<STR_LIT>" "<STR_LIT>" <EOL> raise n_exc . NotAuthorized ( resource = '<STR_LIT>' , msg = msg ) <EOL> return self . driver . create_router_assoc ( context , bgpvpn_id , <EOL> router_assoc ) <EOL> def get_bgpvpn_router_association ( self , context , assoc_id , bgpvpn_id , <EOL> fields = None ) : <EOL> return self . driver . get_router_assoc ( context , assoc_id , bgpvpn_id , <EOL> fields ) <EOL> def get_bgpvpn_router_associations ( self , context , bgpvpn_id , filters = None , <EOL> fields = None ) : <EOL> return self . driver . get_router_assocs ( context , bgpvpn_id , filters , <EOL> fields ) <EOL> def delete_bgpvpn_router_association ( self , context , assoc_id , bgpvpn_id ) : <EOL> self . driver . delete_router_assoc ( context , assoc_id , bgpvpn_id ) </s>
