<s> import inspect <EOL> import logging <EOL> import netaddr <EOL> import re <EOL> import time <EOL> import xml . etree . ElementTree as ET <EOL> from oslo_config import cfg <EOL> from oslo_utils import importutils <EOL> from networking_cisco . _i18n import _LE , _LI , _LW <EOL> from networking_cisco . plugins . cisco . cfg_agent import cfg_exceptions as cfg_exc <EOL> from networking_cisco . plugins . cisco . cfg_agent . device_drivers import ( <EOL> devicedriver_api ) <EOL> from networking_cisco . plugins . cisco . cfg_agent . device_drivers . csr1kv import ( <EOL> cisco_csr1kv_snippets as snippets ) <EOL> from networking_cisco . plugins . cisco . common . htparser import HTParser <EOL> from networking_cisco . plugins . cisco . extensions import ha <EOL> ncclient = importutils . try_import ( '<STR_LIT>' ) <EOL> manager = importutils . try_import ( '<STR_LIT>' ) <EOL> LOG = logging . getLogger ( __name__ ) <EOL> T1_PORT_NAME_PREFIX = '<STR_LIT>' <EOL> T2_PORT_NAME_PREFIX = '<STR_LIT>' <EOL> class IosXeRoutingDriver ( devicedriver_api . RoutingDriverBase ) : <EOL> """<STR_LIT>""" <EOL> DEV_NAME_LEN = <NUM_LIT> <EOL> def __init__ ( self , ** device_params ) : <EOL> try : <EOL> self . _host_ip = device_params [ '<STR_LIT>' ] <EOL> self . _host_ssh_port = device_params [ '<STR_LIT>' ] <EOL> credentials = device_params . get ( '<STR_LIT>' , { } ) <EOL> self . _username = credentials . get ( '<STR_LIT>' ) <EOL> self . _password = credentials . get ( '<STR_LIT:password>' ) <EOL> self . _timeout = ( device_params . get ( '<STR_LIT>' ) or <EOL> cfg . CONF . cfg_agent . device_connection_timeout ) <EOL> self . _ncc_connection = None <EOL> self . _itfcs_enabled = False <EOL> except KeyError as e : <EOL> LOG . error ( _LE ( "<STR_LIT>" <EOL> "<STR_LIT>" ) , e ) <EOL> raise cfg_exc . InitializationException ( ) <EOL> def router_added ( self , ri ) : <EOL> self . _create_vrf ( ri ) <EOL> def router_removed ( self , ri ) : <EOL> self . _remove_vrf ( ri ) <EOL> def internal_network_added ( self , ri , port ) : <EOL> self . _create_sub_interface ( ri , port ) <EOL> if port . get ( ha . HA_INFO ) is not None and ri . get ( ha . ENABLED , False ) : <EOL> self . _add_ha ( ri , port ) <EOL> def internal_network_removed ( self , ri , port ) : <EOL> self . _remove_sub_interface ( port ) <EOL> def external_gateway_added ( self , ri , ext_gw_port ) : <EOL> self . _create_sub_interface ( ri , ext_gw_port ) <EOL> self . _add_default_route ( ri , ext_gw_port ) <EOL> def external_gateway_removed ( self , ri , ext_gw_port ) : <EOL> self . _remove_default_route ( ri , ext_gw_port ) <EOL> self . _remove_sub_interface ( ext_gw_port ) <EOL> def enable_internal_network_NAT ( self , ri , port , ext_gw_port ) : <EOL> self . _add_internal_nw_nat_rules ( ri , port , ext_gw_port ) <EOL> def disable_internal_network_NAT ( self , ri , port , ext_gw_port ) : <EOL> self . _remove_internal_nw_nat_rules ( ri , [ port ] , ext_gw_port ) <EOL> def floating_ip_added ( self , ri , ext_gw_port , floating_ip , fixed_ip ) : <EOL> self . _add_floating_ip ( ri , floating_ip , fixed_ip ) <EOL> def floating_ip_removed ( self , ri , ext_gw_port , floating_ip , fixed_ip ) : <EOL> self . _remove_floating_ip ( ri , ext_gw_port , floating_ip , fixed_ip ) <EOL> def routes_updated ( self , ri , action , route ) : <EOL> self . _update_routing_table ( ri , action , route ) <EOL> def clear_connection ( self ) : <EOL> self . _ncc_connection = None <EOL> def cleanup_invalid_cfg ( self , hd , routers ) : <EOL> return <EOL> def get_configuration ( self ) : <EOL> return self . _get_running_config ( split = False ) <EOL> def _create_sub_interface ( self , ri , port ) : <EOL> vrf_name = self . _get_vrf_name ( ri ) <EOL> ip_cidr = port [ '<STR_LIT>' ] <EOL> net_mask = netaddr . IPNetwork ( ip_cidr ) . netmask <EOL> gateway_ip = ip_cidr . split ( '<STR_LIT:/>' ) [ <NUM_LIT:0> ] <EOL> sub_interface = self . _get_interface_name_from_hosting_port ( port ) <EOL> vlan = self . _get_interface_vlan_from_hosting_port ( port ) <EOL> self . _do_create_sub_interface ( sub_interface , vlan , vrf_name , <EOL> gateway_ip , net_mask ) <EOL> def _remove_sub_interface ( self , port ) : <EOL> sub_interface = self . _get_interface_name_from_hosting_port ( port ) <EOL> self . _do_remove_sub_interface ( sub_interface ) <EOL> def _add_ha ( self , ri , port ) : <EOL> func_dict = { <EOL> ha . HA_HSRP : self . _add_ha_hsrp , <EOL> ha . HA_VRRP : self . _add_ha_vrrp , <EOL> ha . HA_GLBP : self . _add_ha_gblp <EOL> } <EOL> func_dict [ ri [ ha . DETAILS ] [ ha . TYPE ] ] ( self , ri , port ) <EOL> def _add_ha_hsrp ( self , ri , port ) : <EOL> priority = ri [ ha . DETAILS ] [ ha . PRIORITY ] <EOL> port_ha_info = port [ ha . HA_INFO ] <EOL> group = port_ha_info [ '<STR_LIT>' ] <EOL> ip = port_ha_info [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> if ip and group and priority : <EOL> vrf_name = self . _get_vrf_name ( ri ) <EOL> sub_interface = self . _get_interface_name_from_hosting_port ( port ) <EOL> self . _do_set_ha_hsrp ( sub_interface , vrf_name , priority , group , ip ) <EOL> def _add_ha_vrrp ( self , ri , port ) : <EOL> raise NotImplementedError ( ) <EOL> def _add_ha_gblp ( self , ri , port ) : <EOL> raise NotImplementedError ( ) <EOL> def _remove_ha ( self , ri , port ) : <EOL> pass <EOL> def _get_acl_name_from_vlan ( self , vlan ) : <EOL> return "<STR_LIT>" % vlan <EOL> def _add_internal_nw_nat_rules ( self , ri , port , ext_port ) : <EOL> vrf_name = self . _get_vrf_name ( ri ) <EOL> in_vlan = self . _get_interface_vlan_from_hosting_port ( port ) <EOL> acl_no = self . _get_acl_name_from_vlan ( in_vlan ) <EOL> internal_cidr = port [ '<STR_LIT>' ] <EOL> internal_net = netaddr . IPNetwork ( internal_cidr ) . network <EOL> net_mask = netaddr . IPNetwork ( internal_cidr ) . hostmask <EOL> inner_itfc = self . _get_interface_name_from_hosting_port ( port ) <EOL> outer_itfc = self . _get_interface_name_from_hosting_port ( ext_port ) <EOL> self . _nat_rules_for_internet_access ( acl_no , internal_net , <EOL> net_mask , inner_itfc , <EOL> outer_itfc , vrf_name ) <EOL> def _remove_internal_nw_nat_rules ( self , ri , ports , ext_port ) : <EOL> acls = [ ] <EOL> for port in ports : <EOL> in_itfc_name = self . _get_interface_name_from_hosting_port ( port ) <EOL> inner_vlan = self . _get_interface_vlan_from_hosting_port ( port ) <EOL> acls . append ( self . _get_acl_name_from_vlan ( inner_vlan ) ) <EOL> self . _remove_interface_nat ( in_itfc_name , '<STR_LIT>' ) <EOL> LOG . debug ( "<STR_LIT>" ) <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> self . _remove_dyn_nat_translations ( ) <EOL> vrf_name = self . _get_vrf_name ( ri ) <EOL> ext_itfc_name = self . _get_interface_name_from_hosting_port ( ext_port ) <EOL> for acl in acls : <EOL> self . _remove_dyn_nat_rule ( acl , ext_itfc_name , vrf_name ) <EOL> def _add_default_route ( self , ri , ext_gw_port ) : <EOL> ext_gw_ip = ext_gw_port [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> if ext_gw_ip : <EOL> vrf_name = self . _get_vrf_name ( ri ) <EOL> conf_str = snippets . DEFAULT_ROUTE_CFG % ( vrf_name , ext_gw_ip ) <EOL> if not self . _cfg_exists ( conf_str ) : <EOL> conf_str = snippets . SET_DEFAULT_ROUTE % ( vrf_name , ext_gw_ip ) <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' ) <EOL> def _remove_default_route ( self , ri , ext_gw_port ) : <EOL> ext_gw_ip = ext_gw_port [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> if ext_gw_ip : <EOL> vrf_name = self . _get_vrf_name ( ri ) <EOL> conf_str = snippets . DEFAULT_ROUTE_CFG % ( vrf_name , ext_gw_ip ) <EOL> if self . _cfg_exists ( conf_str ) : <EOL> conf_str = snippets . REMOVE_DEFAULT_ROUTE % ( vrf_name , <EOL> ext_gw_ip ) <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' ) <EOL> def _add_floating_ip ( self , ri , floating_ip , fixed_ip ) : <EOL> vrf_name = self . _get_vrf_name ( ri ) <EOL> self . _do_add_floating_ip ( floating_ip , fixed_ip , vrf_name ) <EOL> def _remove_floating_ip ( self , ri , ext_gw_port , floating_ip , fixed_ip ) : <EOL> vrf_name = self . _get_vrf_name ( ri ) <EOL> out_itfc_name = self . _get_interface_name_from_hosting_port ( ext_gw_port ) <EOL> self . _remove_interface_nat ( out_itfc_name , '<STR_LIT>' ) <EOL> self . _remove_dyn_nat_translations ( ) <EOL> self . _do_remove_floating_ip ( floating_ip , fixed_ip , vrf_name ) <EOL> self . _add_interface_nat ( out_itfc_name , '<STR_LIT>' ) <EOL> def _update_routing_table ( self , ri , action , route ) : <EOL> vrf_name = self . _get_vrf_name ( ri ) <EOL> destination_net = netaddr . IPNetwork ( route [ '<STR_LIT>' ] ) <EOL> dest = destination_net . network <EOL> dest_mask = destination_net . netmask <EOL> next_hop = route [ '<STR_LIT>' ] <EOL> if action is '<STR_LIT:replace>' : <EOL> self . _add_static_route ( dest , dest_mask , next_hop , vrf_name ) <EOL> elif action is '<STR_LIT>' : <EOL> self . _remove_static_route ( dest , dest_mask , next_hop , vrf_name ) <EOL> else : <EOL> LOG . error ( _LE ( '<STR_LIT>' ) , action ) <EOL> def _create_vrf ( self , ri ) : <EOL> vrf_name = self . _get_vrf_name ( ri ) <EOL> self . _do_create_vrf ( vrf_name ) <EOL> def _remove_vrf ( self , ri ) : <EOL> vrf_name = self . _get_vrf_name ( ri ) <EOL> self . _do_remove_vrf ( vrf_name ) <EOL> def _get_vrf_name ( self , ri ) : <EOL> return ri . router_name ( ) [ : self . DEV_NAME_LEN ] <EOL> def _get_connection ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> if self . _ncc_connection and self . _ncc_connection . connected : <EOL> return self . _ncc_connection <EOL> else : <EOL> self . _ncc_connection = manager . connect ( <EOL> host = self . _host_ip , port = self . _host_ssh_port , <EOL> username = self . _username , password = self . _password , <EOL> device_params = { '<STR_LIT:name>' : "<STR_LIT>" } , timeout = self . _timeout ) <EOL> if not self . _itfcs_enabled : <EOL> self . _itfcs_enabled = self . _enable_itfcs ( <EOL> self . _ncc_connection ) <EOL> return self . _ncc_connection <EOL> except Exception as e : <EOL> conn_params = { '<STR_LIT:host>' : self . _host_ip , '<STR_LIT:port>' : self . _host_ssh_port , <EOL> '<STR_LIT:user>' : self . _username , <EOL> '<STR_LIT>' : self . _timeout , '<STR_LIT>' : e . message } <EOL> raise cfg_exc . ConnectionException ( ** conn_params ) <EOL> def _get_interface_name_from_hosting_port ( self , port ) : <EOL> vlan = self . _get_interface_vlan_from_hosting_port ( port ) <EOL> int_no = self . _get_interface_no_from_hosting_port ( port ) <EOL> return '<STR_LIT>' % ( int_no , vlan ) <EOL> @ staticmethod <EOL> def _get_interface_vlan_from_hosting_port ( port ) : <EOL> return port [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> @ staticmethod <EOL> def _get_interface_no_from_hosting_port ( port ) : <EOL> """<STR_LIT>""" <EOL> _name = port [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> if_type = _name . split ( '<STR_LIT::>' ) [ <NUM_LIT:0> ] + '<STR_LIT::>' <EOL> if if_type == T1_PORT_NAME_PREFIX : <EOL> return str ( int ( _name . split ( '<STR_LIT::>' ) [ <NUM_LIT:1> ] ) * <NUM_LIT:2> ) <EOL> elif if_type == T2_PORT_NAME_PREFIX : <EOL> return str ( int ( _name . split ( '<STR_LIT::>' ) [ <NUM_LIT:1> ] ) * <NUM_LIT:2> + <NUM_LIT:1> ) <EOL> else : <EOL> params = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:value>' : _name } <EOL> raise cfg_exc . CSR1kvUnknownValueException ( ** params ) <EOL> def _get_interfaces ( self ) : <EOL> """<STR_LIT>""" <EOL> ios_cfg = self . _get_running_config ( ) <EOL> parse = HTParser ( ios_cfg ) <EOL> itfcs_raw = parse . find_lines ( "<STR_LIT>" ) <EOL> itfcs = [ raw_if . strip ( ) . split ( '<STR_LIT:U+0020>' ) [ <NUM_LIT:1> ] for raw_if in itfcs_raw ] <EOL> LOG . debug ( "<STR_LIT>" , itfcs ) <EOL> return itfcs <EOL> def _get_interface_ip ( self , interface_name ) : <EOL> """<STR_LIT>""" <EOL> ios_cfg = self . _get_running_config ( ) <EOL> parse = HTParser ( ios_cfg ) <EOL> children = parse . find_children ( "<STR_LIT>" % interface_name ) <EOL> for line in children : <EOL> if '<STR_LIT>' in line : <EOL> ip_address = line . strip ( ) . split ( '<STR_LIT:U+0020>' ) [ <NUM_LIT:2> ] <EOL> LOG . debug ( "<STR_LIT>" , ip_address ) <EOL> return ip_address <EOL> LOG . warning ( _LW ( "<STR_LIT>" ) , interface_name ) <EOL> return None <EOL> def _interface_exists ( self , interface ) : <EOL> """<STR_LIT>""" <EOL> ios_cfg = self . _get_running_config ( ) <EOL> parse = HTParser ( ios_cfg ) <EOL> itfcs_raw = parse . find_lines ( "<STR_LIT>" + interface ) <EOL> return len ( itfcs_raw ) > <NUM_LIT:0> <EOL> def _enable_itfcs ( self , conn ) : <EOL> """<STR_LIT>""" <EOL> interfaces = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> try : <EOL> for i in interfaces : <EOL> conf_str = snippets . ENABLE_INTF % i <EOL> rpc_obj = conn . edit_config ( target = '<STR_LIT>' , config = conf_str ) <EOL> if self . _check_response ( rpc_obj , '<STR_LIT>' ) : <EOL> LOG . info ( _LI ( "<STR_LIT>" ) , i ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> except Exception : <EOL> return False <EOL> return True <EOL> def _get_vrfs ( self ) : <EOL> """<STR_LIT>""" <EOL> vrfs = [ ] <EOL> ios_cfg = self . _get_running_config ( ) <EOL> parse = HTParser ( ios_cfg ) <EOL> vrfs_raw = parse . find_lines ( "<STR_LIT>" ) <EOL> for line in vrfs_raw : <EOL> vrf_name = line . strip ( ) . split ( '<STR_LIT:U+0020>' ) [ <NUM_LIT:2> ] <EOL> vrfs . append ( vrf_name ) <EOL> LOG . info ( _LI ( "<STR_LIT>" ) , vrfs ) <EOL> return vrfs <EOL> def _get_capabilities ( self ) : <EOL> """<STR_LIT>""" <EOL> conn = self . _get_connection ( ) <EOL> capabilities = [ ] <EOL> for c in conn . server_capabilities : <EOL> capabilities . append ( c ) <EOL> LOG . debug ( "<STR_LIT>" , capabilities ) <EOL> return capabilities <EOL> def _get_running_config ( self , split = True ) : <EOL> """<STR_LIT>""" <EOL> conn = self . _get_connection ( ) <EOL> config = conn . get_config ( source = "<STR_LIT>" ) <EOL> if config : <EOL> root = ET . fromstring ( config . _raw ) <EOL> running_config = root [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> if split is True : <EOL> rgx = re . compile ( "<STR_LIT>" ) <EOL> ioscfg = rgx . split ( running_config . text ) <EOL> else : <EOL> ioscfg = running_config . text <EOL> return ioscfg <EOL> def _check_acl ( self , acl_no , network , netmask ) : <EOL> """<STR_LIT>""" <EOL> exp_cfg_lines = [ '<STR_LIT>' + str ( acl_no ) , <EOL> '<STR_LIT>' + str ( network ) + '<STR_LIT:U+0020>' + str ( netmask ) ] <EOL> ios_cfg = self . _get_running_config ( ) <EOL> parse = HTParser ( ios_cfg ) <EOL> acls_raw = parse . find_children ( exp_cfg_lines [ <NUM_LIT:0> ] ) <EOL> if acls_raw : <EOL> if exp_cfg_lines [ <NUM_LIT:1> ] in acls_raw : <EOL> return True <EOL> LOG . error ( _LE ( "<STR_LIT>" ) , acl_no ) <EOL> return False <EOL> LOG . debug ( "<STR_LIT>" , acl_no ) <EOL> return False <EOL> def _cfg_exists ( self , cfg_str ) : <EOL> """<STR_LIT>""" <EOL> ios_cfg = self . _get_running_config ( ) <EOL> parse = HTParser ( ios_cfg ) <EOL> cfg_raw = parse . find_lines ( "<STR_LIT>" + cfg_str ) <EOL> LOG . debug ( "<STR_LIT>" , cfg_raw ) <EOL> return len ( cfg_raw ) > <NUM_LIT:0> <EOL> def _set_interface ( self , name , ip_address , mask ) : <EOL> conf_str = snippets . SET_INTC % ( name , ip_address , mask ) <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' ) <EOL> def _do_create_vrf ( self , vrf_name ) : <EOL> conf_str = snippets . CREATE_VRF % vrf_name <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' ) <EOL> def _do_remove_vrf ( self , vrf_name ) : <EOL> if vrf_name in self . _get_vrfs ( ) : <EOL> conf_str = snippets . REMOVE_VRF % vrf_name <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' ) <EOL> def _do_create_sub_interface ( self , sub_interface , vlan_id , vrf_name , ip , <EOL> mask ) : <EOL> if vrf_name not in self . _get_vrfs ( ) : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) , vrf_name ) <EOL> conf_str = snippets . CREATE_SUBINTERFACE % ( sub_interface , vlan_id , <EOL> vrf_name , ip , mask ) <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' ) <EOL> def _do_remove_sub_interface ( self , sub_interface ) : <EOL> if self . _interface_exists ( sub_interface ) : <EOL> conf_str = snippets . REMOVE_SUBINTERFACE % sub_interface <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' ) <EOL> def _do_set_ha_hsrp ( self , sub_interface , vrf_name , priority , group , ip ) : <EOL> if vrf_name not in self . _get_vrfs ( ) : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) , vrf_name ) <EOL> conf_str = snippets . SET_INTC_HSRP % ( sub_interface , vrf_name , group , <EOL> priority , group , ip ) <EOL> action = "<STR_LIT>" % ( group , priority ) <EOL> self . _edit_running_config ( conf_str , action ) <EOL> def _do_remove_ha_hsrp ( self , sub_interface , group ) : <EOL> conf_str = snippets . REMOVE_INTC_HSRP % ( sub_interface , group ) <EOL> action = ( "<STR_LIT>" <EOL> % ( sub_interface , group ) ) <EOL> self . _edit_running_config ( conf_str , action ) <EOL> def _get_interface_cfg ( self , interface ) : <EOL> ios_cfg = self . _get_running_config ( ) <EOL> parse = HTParser ( ios_cfg ) <EOL> return parse . find_children ( '<STR_LIT>' + interface ) <EOL> def _nat_rules_for_internet_access ( self , acl_no , network , <EOL> netmask , <EOL> inner_itfc , <EOL> outer_itfc , <EOL> vrf_name ) : <EOL> """<STR_LIT>""" <EOL> acl_present = self . _check_acl ( acl_no , network , netmask ) <EOL> if not acl_present : <EOL> conf_str = snippets . CREATE_ACL % ( acl_no , network , netmask ) <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' ) <EOL> conf_str = snippets . SET_DYN_SRC_TRL_INTFC % ( acl_no , outer_itfc , <EOL> vrf_name ) <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' ) <EOL> conf_str = snippets . SET_NAT % ( inner_itfc , '<STR_LIT>' ) <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' ) <EOL> conf_str = snippets . SET_NAT % ( outer_itfc , '<STR_LIT>' ) <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' ) <EOL> def _add_interface_nat ( self , itfc_name , itfc_type ) : <EOL> conf_str = snippets . SET_NAT % ( itfc_name , itfc_type ) <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' + itfc_type ) <EOL> def _remove_interface_nat ( self , itfc_name , itfc_type ) : <EOL> conf_str = snippets . REMOVE_NAT % ( itfc_name , itfc_type ) <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' + itfc_type ) <EOL> def _remove_dyn_nat_rule ( self , acl_no , outer_itfc_name , vrf_name ) : <EOL> conf_str = snippets . SNAT_CFG % ( acl_no , outer_itfc_name , vrf_name ) <EOL> if self . _cfg_exists ( conf_str ) : <EOL> conf_str = snippets . REMOVE_DYN_SRC_TRL_INTFC % ( <EOL> acl_no , outer_itfc_name , vrf_name ) <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' ) <EOL> conf_str = snippets . REMOVE_ACL % acl_no <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' ) <EOL> def _remove_dyn_nat_translations ( self ) : <EOL> conf_str = snippets . CLEAR_DYN_NAT_TRANS <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' ) <EOL> def _do_add_floating_ip ( self , floating_ip , fixed_ip , vrf ) : <EOL> conf_str = snippets . SET_STATIC_SRC_TRL % ( fixed_ip , floating_ip , vrf ) <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' ) <EOL> def _do_remove_floating_ip ( self , floating_ip , fixed_ip , vrf ) : <EOL> conf_str = snippets . REMOVE_STATIC_SRC_TRL % ( <EOL> fixed_ip , floating_ip , vrf ) <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' ) <EOL> def _get_floating_ip_cfg ( self ) : <EOL> ios_cfg = self . _get_running_config ( ) <EOL> parse = HTParser ( ios_cfg ) <EOL> res = parse . find_lines ( '<STR_LIT>' ) <EOL> return res <EOL> def _add_static_route ( self , dest , dest_mask , next_hop , vrf ) : <EOL> conf_str = snippets . SET_IP_ROUTE % ( vrf , dest , dest_mask , next_hop ) <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' ) <EOL> def _remove_static_route ( self , dest , dest_mask , next_hop , vrf ) : <EOL> conf_str = snippets . REMOVE_IP_ROUTE % ( vrf , dest , dest_mask , next_hop ) <EOL> self . _edit_running_config ( conf_str , '<STR_LIT>' ) <EOL> def _get_static_route_cfg ( self ) : <EOL> ios_cfg = self . _get_running_config ( ) <EOL> parse = HTParser ( ios_cfg ) <EOL> return parse . find_lines ( '<STR_LIT>' ) <EOL> def caller_name ( self , skip = <NUM_LIT:2> ) : <EOL> """<STR_LIT>""" <EOL> stack = inspect . stack ( ) <EOL> start = <NUM_LIT:0> + skip <EOL> if len ( stack ) < start + <NUM_LIT:1> : <EOL> return '<STR_LIT>' <EOL> parentframe = stack [ start ] [ <NUM_LIT:0> ] <EOL> name = [ ] <EOL> module = inspect . getmodule ( parentframe ) <EOL> if module : <EOL> name . append ( module . __name__ ) <EOL> if '<STR_LIT>' in parentframe . f_locals : <EOL> name . append ( parentframe . f_locals [ '<STR_LIT>' ] . __class__ . __name__ ) <EOL> codename = parentframe . f_code . co_name <EOL> if codename != '<STR_LIT>' : <EOL> name . append ( codename ) <EOL> del parentframe <EOL> return "<STR_LIT:.>" . join ( name ) <EOL> def _edit_running_config ( self , conf_str , snippet ) : <EOL> conn = self . _get_connection ( ) <EOL> LOG . info ( _LI ( "<STR_LIT>" <EOL> "<STR_LIT>" ) , <EOL> { '<STR_LIT>' : self . hosting_device [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : snippet , <EOL> '<STR_LIT>' : conf_str , <EOL> '<STR_LIT>' : self . caller_name ( ) } ) <EOL> try : <EOL> rpc_obj = conn . edit_config ( target = '<STR_LIT>' , config = conf_str ) <EOL> self . _check_response ( rpc_obj , snippet , conf_str = conf_str ) <EOL> except Exception as e : <EOL> if re . search ( r"<STR_LIT>" , snippet ) : <EOL> LOG . warning ( _LW ( "<STR_LIT>" ) , snippet ) <EOL> pass <EOL> elif isinstance ( e , ncclient . operations . rpc . RPCError ) : <EOL> e_tag = e . tag <EOL> e_type = e . type <EOL> params = { '<STR_LIT>' : snippet , '<STR_LIT:type>' : e_type , '<STR_LIT>' : e_tag , <EOL> '<STR_LIT>' : self . hosting_device [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : self . _host_ip , '<STR_LIT>' : conf_str } <EOL> raise cfg_exc . CSR1kvConfigException ( ** params ) <EOL> def _check_response ( self , rpc_obj , snippet_name , conf_str = None ) : <EOL> """<STR_LIT>""" <EOL> LOG . debug ( "<STR_LIT>" , <EOL> { '<STR_LIT>' : snippet_name , '<STR_LIT>' : rpc_obj . xml } ) <EOL> xml_str = rpc_obj . xml <EOL> if "<STR_LIT>" in xml_str : <EOL> LOG . info ( _LI ( "<STR_LIT>" ) , snippet_name ) <EOL> return True <EOL> e_type = rpc_obj . _root [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . text <EOL> e_tag = rpc_obj . _root [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] . text <EOL> params = { '<STR_LIT>' : snippet_name , '<STR_LIT:type>' : e_type , '<STR_LIT>' : e_tag , <EOL> '<STR_LIT>' : self . hosting_device [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : self . _host_ip , '<STR_LIT>' : conf_str } <EOL> raise cfg_exc . CSR1kvConfigException ( ** params ) </s>
