FIRST COMMIT (410e995d) · Commits · TFS / controller

scripts/run_tests_locally-device-gnmi-nokia-srlinux.sh

+3 −1

Original line number	Diff line number	Diff line
		@@ -20,5 +20,7 @@ cd $PROJECTDIR/src
		RCFILE=$PROJECTDIR/coverage/.coveragerc

		# Run unitary tests and analyze coverage of code at same time
		coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose \
		coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO -o log_cli=true --verbose \
		device/tests/test_gnmi_nokia_srlinux.py
		#\ device/service/drivers/gnmi_nokia_srlinux/handlers/__init__.py
		#\ device/service/driver_api/_Driver.py

src/device/service/driver_api/_Driver.py

+3 −0

Original line number	Diff line number	Diff line
		@@ -25,6 +25,9 @@ RESOURCE_ROUTING_POLICIES = '__routing_policies__'
		RESOURCE_SERVICES = '__services__'
		RESOURCE_ACL = '__acl__'
		RESOURCE_INVENTORY = '__inventory__'
		RESOURCE_NETWORK_INSTANCE_VXLAN = '__network_instance_vxlan__'
		RESOURCE_ROUTING_POLICY='__routing_policy__'
		RESOURCE_TUNNEL_INTERFACE='__tunnel_interface__'


		class _Driver:

src/device/service/drivers/gnmi_nokia_srlinux/GnmiSessionHandler.py

+17 −8

Original line number	Diff line number	Diff line
		@@ -26,6 +26,11 @@ from .tools.Subscriptions import Subscriptions
		from .tools.Value import decode_value #, value_exists
		from .MonitoringThread import MonitoringThread

		logging.basicConfig(level=logging.DEBUG)
		LOGGER = logging.getLogger(__name__)
		LOGGER.setLevel(logging.DEBUG)


		class GnmiSessionHandler:
		def __init__(self, address : str, port : int, settings : Dict, logger : logging.Logger) -> None:
		self._address = address
		@@ -159,21 +164,25 @@ class GnmiSessionHandler:
		set_request = SetRequest()
		#for resource_key in resource_keys:
		for resource_key, resource_value in resources:
		self._logger.info('---1')
		self._logger.info(str(resource_key))
		self._logger.info(str(resource_value))
		self._logger.info('Resource from test script:')
		self._logger.info(' resource_key = ' + str(resource_key))
		self._logger.info(' resource_value = ' + str(resource_value))

		#resource_tuple = resource_tuples.get(resource_key)
		#if resource_tuple is None: continue
		#_, value, exists, operation_done = resource_tuple
		if isinstance(resource_value, str): resource_value = json.loads(resource_value)
		str_path, str_data = compose(resource_key, resource_value, delete=False)
		self._logger.info('---3')
		self._logger.info(str(str_path))
		self._logger.info(str(str_data))

		self._logger.info('Request being sent to device:')
		self._logger.info(' path = ' + str(str_path))
		self._logger.info(' data = ' + str(str_data))

		set_request_list = set_request.update #if exists else set_request.replace
		set_request_entry = set_request_list.add()
		set_request_entry.path.CopyFrom(path_from_string(str_path))
		set_request_entry.val.json_val = str_data.encode('UTF-8')
		set_request_entry.val.json_ietf_val = str_data.encode('UTF-8')


		self._logger.info('set_request={:s}'.format(grpc_message_to_json_string(set_request)))
		metadata = [('username', self._username), ('password', self._password)]

src/device/service/drivers/gnmi_nokia_srlinux/handlers/Interface.py

+191 −150

Original line number	Diff line number	Diff line
		@@ -17,176 +17,217 @@ from typing import Any, Dict, List, Tuple
		from ._Handler import _Handler
		from .Tools import dict_get_first

		logging.basicConfig(level=logging.DEBUG)
		LOGGER = logging.getLogger(__name__)
		LOGGER.setLevel(logging.DEBUG)

		class InterfaceHandler(_Handler):
		def get_resource_key(self) -> str: return '/interface'
		def get_path(self) -> str: return '/interfaces/interface'
		def get_resource_key(self) -> str:
		LOGGER.debug('Getting resource key for InterfaceHandler')
		return '/interface'

		def get_path(self) -> str:
		LOGGER.debug('Getting path for InterfaceHandler')
		return '/interface'

		def compose(self, resource_key: str, resource_value: Dict, delete: bool = False) -> Tuple[str, str]:
		if_name = str (resource_value['name' ]) # ethernet-1/1
		sif_index = int (resource_value.get('sub_if_index' , 0 )) # 0
		if_name = str(resource_value['if_name'])
		sif_index = int(resource_value.get('subif_index'))

		if delete:
		PATH_TMPL = '/interfaces/interface[name={:s}]/subinterfaces/subinterface[index={:d}]'
		PATH_TMPL = '/interface[name={:s}]/subinterface[index={:d}]'
		str_path = PATH_TMPL.format(if_name, sif_index)
		str_data = json.dumps({})
		return str_path, str_data

		if_enabled = bool(resource_value.get('enabled' , True)) # True/False
		sif_enabled = bool(resource_value.get('sub_if_enabled' , True)) # True/False
		sif_ipv4_enabled = bool(resource_value.get('sub_if_ipv4_enabled', True)) # True/False
		sif_ipv4_address = str (resource_value['sub_if_ipv4_address' ]) # 172.16.0.1
		sif_ipv4_prefix = int (resource_value['sub_if_ipv4_prefix' ]) # 24

		str_path = '/interfaces/interface[name={:s}]'.format(if_name)
		str_data = json.dumps({
		'name': if_name,
		'config': {'name': if_name, 'enabled': if_enabled},
		'subinterfaces': {
		'subinterface': {
		'index': sif_index,
		'config': {'index': sif_index, 'enabled': sif_enabled},
		'ipv4': {
		'config': {'enabled': sif_ipv4_enabled},
		'addresses': {
		'address': {
		'ip': sif_ipv4_address,
		'config': {'ip': sif_ipv4_address, 'prefix_length': sif_ipv4_prefix},
		}
		}
		}
		}
		}
		})
		return str_path, str_data
		str_path = '/interface[name={:s}]'.format(if_name)
		json_interface = {}

		def parse(self, json_data : Dict) -> List[Tuple[str, Dict[str, Any]]]:
		LOGGER.info('json_data = {:s}'.format(json.dumps(json_data)))
		json_interface_list : List[Dict] = json_data.get('interface', [])
		if 'if_name' in resource_value:
		#if_name=str(resource_value['if_name'])
		json_interface['name']=if_name

		response = []
		for json_interface in json_interface_list:
		LOGGER.info('json_interface = {:s}'.format(json.dumps(json_interface)))

		interface = {}
		if 'if_admin_state' in resource_value:
		bool_admin_state = bool(resource_value['if_admin_state'])
		str_admin_state = 'enable' if bool_admin_state else 'disable'
		json_interface['admin-state'] = str_admin_state

		interface_name = json_interface.get('name')
		if interface_name is None:
		LOGGER.info('DISCARDED json_interface = {:s}'.format(json.dumps(json_interface)))
		continue
		interface['name'] = interface_name
		if 'subif_index' in resource_value:
		subif_index = int(resource_value['subif_index'])
		else:
		subif_index=0
		subif_list= json_interface.setdefault('subinterface', list())
		subif={'index':subif_index}
		subif_list.append(subif)

		CONFIG_FIELDS = ('config', 'openconfig-interface:config', 'oci:config')
		json_config : Dict = dict_get_first(json_interface, CONFIG_FIELDS, default={})

		STATE_FIELDS = ('state', 'openconfig-interface:state', 'oci:state')
		json_state : Dict = dict_get_first(json_interface, STATE_FIELDS, default={})
		if 'subif_ipv4_admin_state' in resource_value:
		bool_ipv4_admin_state = bool(resource_value['subif_ipv4_admin_state'])
		str_ipv4_admin_state = 'enable' if bool_ipv4_admin_state else 'disable'
		ipv4 = subif.setdefault('ipv4', dict())
		ipv4['admin-state'] = str_ipv4_admin_state

		interface_type = json_config.get('type')
		if interface_type is None: interface_type = json_state.get('type')
		if interface_type is None:
		LOGGER.info('DISCARDED json_interface = {:s}'.format(json.dumps(json_interface)))
		continue
		interface_type = interface_type.replace('ianaift:', '')
		interface_type = interface_type.replace('iana-if-type:', '')
		interface['type'] = interface_type

		interface_mtu = json_config.get('mtu')
		if interface_mtu is None: interface_mtu = json_state.get('mtu')
		if interface_mtu is not None: interface['mtu'] = int(interface_mtu)

		interface_enabled = json_config.get('enabled')
		if interface_enabled is None: interface_enabled = json_state.get('enabled')
		interface['enabled'] = False if interface_enabled is None else bool(interface_enabled)
		if 'subif_ipv4_address' in resource_value and 'subif_ipv4_prefix' in resource_value:
		subif_ipv4_address = str(resource_value['subif_ipv4_address'])
		subif_ipv4_prefix = int(resource_value['subif_ipv4_prefix'])
		ipv4 = subif.setdefault('ipv4', dict())
		address_list= ipv4.setdefault('address', list())
		address_list.append({
		'ip-prefix': '{:s}/{:d}'.format(subif_ipv4_address, subif_ipv4_prefix)
		})

		interface_management = json_config.get('management')
		if interface_management is None: interface_management = json_state.get('management')
		interface['management'] = False if interface_management is None else bool(interface_management)
		if 'if_vlan_tagging' in resource_value:
		bool_if_vlan_tagging = bool(resource_value['if_vlan_tagging'])
		json_interface['vlan-tagging']=bool_if_vlan_tagging

		interface_descr = json_interface.get('config', {}).get('description')
		if interface_descr is not None: interface['description'] = interface_descr

		json_subinterfaces = json_interface.get('subinterfaces', {})
		json_subinterface_list : List[Dict] = json_subinterfaces.get('subinterface', [])
		if 'subif_type' in resource_value:
		subif_type = str(resource_value['subif_type'])
		subif['type'] =subif_type

		for json_subinterface in json_subinterface_list:
		LOGGER.info('json_subinterface = {:s}'.format(json.dumps(json_subinterface)))

		subinterface = {}
		if 'subif_vlan_admin_state' in resource_value:
		bool_vlan_admin_state = bool(resource_value['subif_vlan_admin_state'])
		str_vlan_admin_state = 'enable' if bool_vlan_admin_state else 'disable'
		subif['admin-state'] = str_vlan_admin_state

		subinterface_index = json_subinterface.get('state', {}).get('index')
		if subinterface_index is None: continue
		subinterface['index'] = int(subinterface_index)

		subinterface_name = json_subinterface.get('state', {}).get('name')
		if subinterface_name is None: continue
		subinterface['name'] = subinterface_name
		if 'subif_vlan_encap' in resource_value:
		subif_vlan_encap=str(resource_value['subif_vlan_encap'])
		vlan=subif.setdefault('vlan',dict())
		encap=vlan.setdefault('encap',dict())
		encap[subif_vlan_encap]=dict()

		subinterface_enabled = json_subinterface.get('state', {}).get('enabled', False)
		subinterface['enabled'] = bool(subinterface_enabled)

		VLAN_FIELDS = ('vlan', 'openconfig-vlan:vlan', 'ocv:vlan')
		json_vlan = dict_get_first(json_subinterface, VLAN_FIELDS, default={})
		return str_path,json.dumps(json_interface)

		MATCH_FIELDS = ('match', 'openconfig-vlan:match', 'ocv:match')
		json_vlan = dict_get_first(json_vlan, MATCH_FIELDS, default={})
		############################PARSE##############################################################

		SIN_TAG_FIELDS = ('single-tagged', 'openconfig-vlan:single-tagged', 'ocv:single-tagged')
		json_vlan = dict_get_first(json_vlan, SIN_TAG_FIELDS, default={})
		def parse(self, json_data: Dict) -> List[Tuple[str, Dict[str, Any]]]:
		# LOGGER.info('json_data = {:s}'.format(json.dumps(json_data)))
		LOGGER.info('Parsing json_data for InterfaceHandler')
		LOGGER.debug('json_data = {:s}'.format(json.dumps(json_data)))
		json_interface_list : List[Dict] = json_data.get('interface', [])
		response = []

		CONFIG_FIELDS = ('config', 'openconfig-vlan:config', 'ocv:config')
		json_vlan = dict_get_first(json_vlan, CONFIG_FIELDS, default={})
		for json_interface in json_interface_list:
		interface = {}
		interface_name = json_interface.get('name')
		if interface_name is None:
		LOGGER.info('DISCARDED json_interface = {:s}'.format(json.dumps(json_data)))
		return response
		interface['name'] = interface_name

		VLAN_ID_FIELDS = ('vlan-id', 'openconfig-vlan:vlan-id', 'ocv:vlan-id')
		subinterface_vlan_id = dict_get_first(json_vlan, VLAN_ID_FIELDS)
		if subinterface_vlan_id is not None: subinterface['vlan_id'] = subinterface_vlan_id
		admin_state = json_interface.get('admin-state')
		if admin_state is not None:
		interface['admin-state'] = bool(admin_state)

		vlan_tagging= json_interface.get('vlan-tagging')
		if vlan_tagging is not None:
		interface['vlan_tagging'] = bool(vlan_tagging)

		# TODO: implement support for multiple IP addresses per subinterface

		IPV4_FIELDS = ('ipv4', 'openconfig-if-ip:ipv4', 'ociip:ipv4')
		json_ipv4 = dict_get_first(json_subinterface, IPV4_FIELDS, default={})
		json_subinterface_list: List[Dict] = json_data.get('subinterface', [])
		for json_subinterface in json_subinterface_list:
		LOGGER.info('json_subinterface = {:s}'.format(json.dumps(json_subinterface)))
		subinterface = []
		subinterface_index = json_subinterface.get('index')
		if subinterface_index is None:
		continue
		subinterface['index'] = int(subinterface_index)

		IPV4_ADDRESSES_FIELDS = ('addresses', 'openconfig-if-ip:addresses', 'ociip:addresses')
		json_ipv4_addresses = dict_get_first(json_ipv4, IPV4_ADDRESSES_FIELDS, default={})
		#VLAN_FIELDS = ('vlan', 'NokiaSRL-vlan:vlan', 'ocv:vlan')
		#json_vlan = dict_get_first(json_subinterface, VLAN_FIELDS, default={})

		IPV4_ADDRESS_FIELDS = ('address', 'openconfig-if-ip:address', 'ociip:address')
		json_ipv4_address_list : List[Dict] = dict_get_first(json_ipv4_addresses, IPV4_ADDRESS_FIELDS, default=[])
		#ENCAP_FIELDS = ('encap', 'NokiaSRL-vlan:match', 'ocv:encap')
		#json_vlan = dict_get_first(json_vlan, ENCAP_FIELDS, default={})

		#ipv4_addresses = []
		for json_ipv4_address in json_ipv4_address_list:
		LOGGER.info('json_ipv4_address = {:s}'.format(json.dumps(json_ipv4_address)))
		#WITHOUT_TAG_FIELDS = ('untagged', 'NokiaSRL-vlan:untagged', 'ocv:single-tagged')
		#json_vlan = dict_get_first(json_vlan, WITHOUT_TAG_FIELDS, default={})

		STATE_FIELDS = ('state', 'openconfig-if-ip:state', 'ociip:state')
		json_ipv4_address_state = dict_get_first(json_ipv4_address, STATE_FIELDS, default={})
		vlan_type = json_subinterface.get('type')
		if vlan_type is None:
		continue
		subinterface['type'] = vlan_type

		#ipv4_address = {}
		#
		#VLAN_ID_FIELDS = ('vlan-id', 'NokiaSRL-vlan:vlan-id', 'ocv:vlan-id')
		#subinterface_vlan_id = dict_get_first(json_vlan, VLAN_ID_FIELDS)
		#if subinterface_vlan_id is not None:
		# subinterface['vlan_id'] = subinterface_vlan_id

		#ORIGIN_FIELDS = ('origin', 'openconfig-if-ip:origin', 'ociip:origin')
		#ipv4_address_origin = dict_get_first(json_ipv4_address_state, ORIGIN_FIELDS, default={})
		#if ipv4_address_origin is not None: ipv4_address['origin'] = ipv4_address_origin
		#IPV4_FIELDS = ('ipv4', 'NokiaSRL-if-ip:ipv4', 'ociip:ipv4')
		#json_ipv4 = dict_get_first(json_subinterface, IPV4_FIELDS, default={})

		IP_FIELDS = ('ip', 'openconfig-if-ip:ip', 'ociip:ip')
		ipv4_address_ip = dict_get_first(json_ipv4_address_state, IP_FIELDS)
		#if ipv4_address_ip is not None: ipv4_address['address_ip'] = ipv4_address_ip
		if ipv4_address_ip is not None: subinterface['address_ip'] = ipv4_address_ip
		#ipv4_enable = json_subinterface.get('state', {}).get('enable', False)
		#subinterface['admin-state'] = bool(ipv4_enable)

		PREFIX_FIELDS = ('prefix-length', 'openconfig-if-ip:prefix-length', 'ociip:prefix-length')
		ipv4_address_prefix = dict_get_first(json_ipv4_address_state, PREFIX_FIELDS)
		#if ipv4_address_prefix is not None: ipv4_address['address_prefix'] = int(ipv4_address_prefix)
		if ipv4_address_prefix is not None: subinterface['address_prefix'] = int(ipv4_address_prefix)

		#if len(ipv4_address) == 0: continue
		#ipv4_addresses.append(ipv4_address)
		#IPV4_ADMIN_STATE_FIELDS = ('admin-state', 'NokiaSRL-if-ip:admin-state', 'ociip:admin-state')
		#ipv4_admin_state = dict_get_first(json_ipv4, IPV4_ADMIN_STATE_FIELDS)

		#subinterface['ipv4_addresses'] = ipv4_addresses
		#IPV4_ADDRESS_FIELDS = ('address', 'NokiaSRL-if-ip:address', 'ociip:address')
		#json_ipv4_address_list = json_ipv4.get(IPV4_ADDRESS_FIELDS[0], [])

		#for json_ipv4_address in json_ipv4_address_list:
		# LOGGER.info('json_ipv4_address = {:s}'.format(json.dumps(json_ipv4_address)))
		#
		# STATE_FIELDS = ('state', 'NokiaSRL-if-ip:state', 'ociip:state')
		# json_ipv4_address_state = dict_get_first(json_ipv4_address, STATE_FIELDS, default={})
		#
		# IP_FIELDS = ('ip', 'NokiaSRL-if-ip:ip', 'ociip:ip')
		# ipv4_address_ip = dict_get_first(json_ipv4_address_state, IP_FIELDS)
		# if ipv4_address_ip is not None:
		# subinterface['address_ip'] = ipv4_address_ip
		#
		# PREFIX_FIELDS = ('prefix-length', 'NokiaSRL-if-ip:prefix-length', 'ociip:prefix-length')
		# ipv4_address_prefix = dict_get_first(json_ipv4_address_state, PREFIX_FIELDS)
		# if ipv4_address_prefix is not None:
		# subinterface['address_prefix'] = int(ipv4_address_prefix)

		#if ipv4_admin_state is not None:
		# subinterface['admin-state'] = ipv4_admin_state
		#json_ipv4_dic:[Dict] = json_subinterface.get('ipv4', {})
		ipv4={}
		ipv4_admin_state = json_subinterface.get('admin-state')
		if ipv4_admin_state is not None:
		ipv4['admin-state'] = bool(ipv4_admin_state)

		json_address_list: List[Dict] = json_subinterface.get('address', [])
		for json_address in json_address_list:
		LOGGER.info('json_address = {:s}'.format(json.dumps(json_address)))
		address = []
		address_ip_prefix = json_address.get('state', {}).get('ip-prefix')
		if address_ip_prefix is None:
		continue
		address['ip-prefix'] = int(address_ip_prefix)



		if len(subinterface) == 0:
		continue

		if len(subinterface) == 0: continue
		resource_key = '/interface[{:s}]/subinterface[{:s}]'.format(interface['name'], str(subinterface['index']))
		response.append((resource_key, subinterface))

		if len(interface) == 0: continue
		if len(interface) == 0:
		continue

		response.append(('/interface[{:s}]'.format(interface['name']), interface))
		json_protocols = json_data.get('protocols', {})
		json_bgp_list: List[Dict] = json_protocols.get('bgp', [])

		for json_bgp in json_bgp_list:
		json_afi_safi_list: List[Dict] = json_bgp.get('afi-safi', [])

		for json_afi_safi in json_afi_safi_list:
		interface['afisafiname'] = json_afi_safi.get('afi-safi-name', '')
		response.append((self.compose('/interfaces/interface', interface)))

		LOGGER.debug('Parse result: %s', response)

		return response

src/device/service/drivers/gnmi_nokia_srlinux/handlers/NetworkInstance.py

+356 −34

File changed.

Preview size limit exceeded, changes collapsed.