Mohamad-Rahhal-Commit-7-March-2024 (3d5ef2aa) · Commits · TFS / controller

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

0 → 100755

+24 −0

Original line number	Diff line number	Diff line
		#!/bin/bash
		# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (https://tfs.etsi.org/)
		#
		# 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.


		PROJECTDIR=`pwd`

		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=DEBUG -o log_cli=true --verbose \
		device/tests/test_gnmi_nokia_srlinux-get-ifs.py

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

+3 −1

Original line number	Diff line number	Diff line
		@@ -124,8 +124,9 @@ class GnmiSessionHandler:
		# resource_key_tuple[2] = True

		for update in notification.update:
		#self._logger.info('update={:s}'.format(grpc_message_to_json_string(update)))
		#self._logger.info('update.path={:s}'.format(grpc_message_to_json_string(update.path)))
		str_path = path_to_string(update.path)
		#self._logger.info('str_path={:s}'.format(str(str_path)))
		#resource_key_tuple = results.get(str_path)
		#if resource_key_tuple is None:
		# # pylint: disable=broad-exception-raised
		@@ -133,6 +134,7 @@ class GnmiSessionHandler:
		# raise Exception(MSG.format(str(str_path), str(resource_keys)))
		try:
		value = decode_value(update.val)
		#self._logger.info('value={:s}'.format(str(value))) # uncomment to see decoded message from the device
		#resource_key_tuple[1] = value
		#resource_key_tuple[2] = True
		results.extend(parse(str_path, value))

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

+91 −126

Original line number	Diff line number	Diff line
		@@ -14,6 +14,8 @@

		import json, logging
		from typing import Any, Dict, List, Tuple

		from requests import get
		from ._Handler import _Handler
		from .Tools import dict_get_first

		@@ -23,12 +25,12 @@ LOGGER.setLevel(logging.DEBUG)

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

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

		def compose(self, resource_key: str, resource_value: Dict, delete: bool = False) -> Tuple[str, str]:
		if_name = str(resource_value['if_name'])
		@@ -108,126 +110,89 @@ class InterfaceHandler(_Handler):

		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', [])
		# 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('srl_nokia-interfaces:interface', [])
		response = []

		for json_interface in json_interface_list:
		#LOGGER.info('json_interface = {:s}'.format(json.dumps(json_interface)))
		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
		# LOGGER.info('DISCARDED json_interface = {:s}'.format(json.dumps(json_interface)))
		continue
		interface['name'] = interface_name

		#LOGGER.info('json_interface = {:s}'.format(json.dumps(json_interface)))
		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)


		json_subinterface_list: List[Dict] = json_data.get('subinterface', [])
		if admin_state is None:
		continue
		interface['admin_state'] = str(admin_state)
		##LOGGER.info('json_interface = {:s}'.format(json.dumps(json_interface)))
		#vlan_tagging = json_interface.get('srl_nokia-interfaces-vlans:vlan-tagging')
		#if vlan_tagging is not None:
		# continue
		#interface['vlan-tagging']=bool(vlan_tagging)
		#LOGGER.info('interface = {:s}'.format(json.dumps(interface)))

		json_subinterface_list: List[Dict] = json_interface.get('subinterface', [])
		for json_subinterface in json_subinterface_list:
		LOGGER.info('json_subinterface = {:s}'.format(json.dumps(json_subinterface)))
		subinterface = []
		subinterface = {}
		subinterface_index = json_subinterface.get('index')
		if subinterface_index is None:
		continue
		subinterface['index'] = int(subinterface_index)

		#VLAN_FIELDS = ('vlan', 'NokiaSRL-vlan:vlan', 'ocv:vlan')
		#json_vlan = dict_get_first(json_subinterface, VLAN_FIELDS, default={})

		#ENCAP_FIELDS = ('encap', 'NokiaSRL-vlan:match', 'ocv:encap')
		#json_vlan = dict_get_first(json_vlan, ENCAP_FIELDS, default={})

		#WITHOUT_TAG_FIELDS = ('untagged', 'NokiaSRL-vlan:untagged', 'ocv:single-tagged')
		#json_vlan = dict_get_first(json_vlan, WITHOUT_TAG_FIELDS, default={})

		vlan_type = json_subinterface.get('type')
		if vlan_type is None:
		subinterface_adminstate = json_subinterface.get('admin-state')
		if subinterface_adminstate is None:
		continue
		subinterface['admin-state'] = str(subinterface_adminstate)
		vlan_type = json_subinterface.get('type')
		#if vlan_type is None:
		# continue
		subinterface['type'] = vlan_type
		ipv4_admin_state = json_subinterface.get('ipv4', {}).get('admin-state')
		subinterface['ipv4'] = {'admin-state': bool(ipv4_admin_state)}
		ipv4_address_list = json_subinterface.get('ipv4', {}).get('address', [])
		addresses = []
		if ipv4_address_list:
		for json_address in ipv4_address_list:
		LOGGER.info('json_address = {:s}'.format(json.dumps(json_address)))
		address_ip_prefix = json_address.get('ip-prefix')

		#
		#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

		#IPV4_FIELDS = ('ipv4', 'NokiaSRL-if-ip:ipv4', 'ociip:ipv4')
		#json_ipv4 = dict_get_first(json_subinterface, IPV4_FIELDS, default={})

		#ipv4_enable = json_subinterface.get('state', {}).get('enable', False)
		#subinterface['admin-state'] = bool(ipv4_enable)


		#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)

		#IPV4_ADDRESS_FIELDS = ('address', 'NokiaSRL-if-ip:address', 'ociip:address')
		#json_ipv4_address_list = json_ipv4.get(IPV4_ADDRESS_FIELDS[0], [])
		if address_ip_prefix is not None:
		addresses.append({'ip-prefix': str(address_ip_prefix)})
		subinterface['ipv4'] = {'address': addresses}
		subinterface['vlan'] = {}
		subinterface['vlan']['encap'] = {}
		json_vlan_encap = json_interface.get('vlan', {}).get('encap', {})
		if json_vlan_encap:
		if 'vlan' not in subinterface:
		subinterface['vlan'] = {}

		#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 'encap' not in subinterface['vlan']:
		subinterface['vlan']['encap'] = {}

		subinterface['vlan']['encap']['untagged'] = json_vlan_encap


		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

		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)

		# 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)))
		#
		return response

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

+164 −5

Original line number	Diff line number	Diff line
		@@ -376,9 +376,168 @@ class NetworkInstanceHandler(_Handler):
		subif_bgp['router-id'] = if_router_id_spine







		return str_path, json.dumps(json_networkinstance)
		def parse(self, json_data: Dict) -> List[Tuple[str, Dict[str, Any]]]:
		response = []
		json_network_instance_list = json_data.get('srl_nokia-network-instance:network-instance', [])

		for json_network_instance in json_network_instance_list:
		#LOGGER.info('json_networkinstance = {:s}'.format(json.dumps(json_network_instance)))
		network_instance = {}
		json_interface_list = json_network_instance.get('interface', [])
		json_vxlan_list=json_network_instance.get('vxlan-interface',[])
		vxlaninterface_names = []
		for json_vxlan in json_vxlan_list:
		#LOGGER.info('json_interface = {:s}'.format(json.dumps(json_interface)))
		# LOGGER.info('json_interface.keys = {:s}'.format(str(json_interface.keys())))
		#interface = {}
		vxlaninterface_name = json_vxlan.get('name')

		if interface_name is not None:
		vxlaninterface_names.append(str(vxlaninterface_name))
		interface_names = []
		for json_interface in json_interface_list:
		#LOGGER.info('json_interface = {:s}'.format(json.dumps(json_interface)))
		# LOGGER.info('json_interface.keys = {:s}'.format(str(json_interface.keys())))
		#interface = {}
		interface_name = json_interface.get('name')

		if interface_name is not None:
		interface_names.append(str(interface_name))

		network_instance_name = json_network_instance.get('name')
		network_instance['name'] = str(network_instance_name)

		network_instance_type = json_network_instance.get('type')
		network_instance['type'] = str(network_instance_type)

		network_instance_adminstate = json_network_instance.get('name')
		network_instance['name'] = str(network_instance_adminstate)

		#LOGGER.info('json_network_instance.keys = {:s}'.format(str(json_network_instance.keys())))
		json_protocols = json_network_instance.get('protocols', {})
		json_bgp_evpn=json_protocols.get('bgp-evpn',{})
		json_bgp_evpn_instance=json_bgp_evpn.get('srl_nokia-bgp-evpn:bgp-instance',[])
		json_bgp_vpn_instance=json_protocols.get('srl_nokia-bgp-vpn:bgp-vpn',{})
		json_bgp_vpn_2=json_bgp_vpn_instance.get('bgp-instance',[])
		#LOGGER.info('json_protocols.keys = {:s}'.format(str(json_protocols.keys())))
		bgp_protocols = json_protocols.get('srl_nokia-bgp:bgp', {})
		#LOGGER.info('bgp_protocols.keys = {:s}'.format(str(bgp_protocols.keys())))
		afi_safi_list: List[Dict] = bgp_protocols.get('afi-safi', [])
		#afi_safi_list = bgp_protocols.get('afi-safi', [])
		lists = []
		for afi_safi_entry in afi_safi_list:
		#LOGGER.info('afi_safi_entry = {:s}'.format(json.dumps(afi_safi_entry)))
		#afi_safi = {}
		admin_state_afi = afi_safi_entry.get('admin-state')
		afi_safi_name = afi_safi_entry.get('afi-safi-name')
		if admin_state_afi is not None and afi_safi_name is not None:
		#afi_safi['admin-state'] = bool(admin_state_afi)
		#afi_safi['afi-safi-name'] = bool(afi_safi_name)
		lists.append({'admin-state': str(admin_state_afi),'afi-safi-name': str(afi_safi_name)})
		if afi_safi_entry:
		network_instance['protocols']={}
		network_instance['bgp']={}
		#network_instance['protocols']['bgp']={'afi-safi': lists}
		network_instance['protocols'] = {'bgp': {'afi-safi': afi_safi_entry}}

		autonomous_system = bgp_protocols.get('autonomous-system')
		#LOGGER.info('bgp_protocols = {:s}'.format(json.dumps(bgp_protocols)))

		bgp_protocols['autonomous-system'] = str(autonomous_system)

		group_list = bgp_protocols.get('group', [])
		group_list1=[]
		for group_1 in group_list:
		#LOGGER.info('group = {:s}'.format(json.dumps(group_1)))
		group = {}
		export_policy = group_1.get('export-policy')
		group_name = group_1.get('group-name')
		import_policy = group_1.get('import-policy')
		peer_as_number = group_1.get('peer-as')
		localas = group_1.get('local-as').get('as-number')
		timers=group_1.get('timers').get('minimum-advertisement-interval')

		group['export-policy'] = str(export_policy)
		group['group-name'] = str(group_name)
		group['import-policy'] = str(import_policy)
		group['peer-as'] = int(peer_as_number)
		group['local-as']={}
		group['local-as']['as-number']=localas
		group['timers']={}
		group['timers']['minimum-advertisement-interval']=timers

		group_list1.append(group)

		afi_safi_list_2 = group_1.get('afi-safi', [])
		afi_safi_list_group_2=[]
		for afi_safi_group_entry in afi_safi_list_2:
		group2 = {}
		admin_state_group = afi_safi_group_entry.get('admin-state')
		afi_safi_name_group = afi_safi_group_entry.get('afi-safi-name')
		group2['admin-state'] = str(admin_state_group)
		group2['afi-safi-name'] = str(afi_safi_name_group)
		afi_safi_list_group_2.append(group2)


		neighbor_group_list = bgp_protocols.get('neighbor', [])
		neighbor_list_inside=[]
		#LOGGER.info('json_neighbor = {:s}'.format(json.dumps(json_interface)))
		for neighbor_group in neighbor_group_list:
		neighbor1 = {}
		admin_state_neighbor = neighbor_group.get('admin-state')
		peeraddress_neighbor = neighbor_group.get('peer-address')
		peergroup_neighbor = neighbor_group.get('peer-group')
		local_address_neighbor = neighbor_group.get('transport', {}).get('local-address')
		neighbor1['admin-state'] = str(admin_state_neighbor)
		neighbor1['peer-address'] = str(peeraddress_neighbor)
		neighbor1['peer-group'] = str(peergroup_neighbor)
		neighbor1['transport'] = {'local-address': str(local_address_neighbor)}

		neighbor2 = {}
		peeraddress_neighbor_2 = neighbor_group.get('peer-address')
		peergroup_neighbor_2 = neighbor_group.get('peer-group')
		neighbor2['peer-address'] = str(peeraddress_neighbor_2)
		neighbor2['peer-group'] = str(peergroup_neighbor_2)

		neighbor_list_inside.append((neighbor1))
		neighbor_list_inside.append(neighbor2)

		router_id = bgp_protocols.get('router-id')

		bgp_protocols['router-id'] = str(router_id)

		bgp_instance1_list=[]
		for json_bgp_instance in json_bgp_evpn_instance:
		instance1 = {}
		id_evpn_bgp_instance = json_bgp_instance.get('id')
		admin_state_evpn_bgp_instance = json_bgp_instance.get('admin-state')
		vxlan_interface_evpn_bgp_instance = json_bgp_instance.get('vxlan-interface')
		evi_evpn_bgp_instance = json_bgp_instance.get('evi', {})
		instance1['id'] = int(id_evpn_bgp_instance)
		instance1['admin-state'] = str(admin_state_evpn_bgp_instance)
		instance1['vxlan-interface'] = str(vxlan_interface_evpn_bgp_instance)
		instance1['evi'] =int(evi_evpn_bgp_instance)
		bgp_instance1_list.append(instance1)
		bgp_instance2_list=[]
		for json_bgp_instance2 in json_bgp_vpn_2:
		#LOGGER.info('json_bgp_instance2 = {:s}'.format(json.dumps(json_bgp_instance2)))
		instance2 = {}
		id_vpn_bgp_instance = json_bgp_instance2.get('id')
		export_target_evpn_bgp_instance = json_bgp_instance2.get('route-target',{}).get('export-rt')
		import_target_evpn_bgp_instance = json_bgp_instance2.get('route-target',{}).get('import-rt')
		instance2['id'] = int(id_vpn_bgp_instance)
		instance2['route-target'] = {}
		instance2['route-target']['export-rt'] =str(export_target_evpn_bgp_instance)
		instance2['route-target']['import-rt'] =str(import_target_evpn_bgp_instance)
		bgp_instance2_list.append(instance2)



		if len(network_instance) == 0:
		continue

		resource_key = '/network-instance[{:s}]'.format(network_instance_name)
		response.append((resource_key,vxlaninterface_names,network_instance_type,network_instance_adminstate, interface_names,network_instance_name,lists,autonomous_system,group_list1,afi_safi_list_group_2,neighbor_list_inside,router_id,bgp_instance1_list,bgp_instance2_list))

		return response

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

+29 −0

Original line number	Diff line number	Diff line
		@@ -57,3 +57,32 @@ class TunnelInterfaceHandler(_Handler):

		return str_path, json.dumps(json_tunnel)

		def parse(self, json_data: Dict) -> List[Tuple[str, Dict[str, Any]]]:
		response = []
		json_tunnel_list = json_data.get('srl_nokia-tunnel-interfaces:tunnel-interface', [])
		for json_tunnel in json_tunnel_list:
		tunnel_interface1 = {}
		name = json_tunnel.get('name', {})
		tunnel_interface1['name']=str(name)
		json_vxlan_list = json_tunnel.get('vxlan-interface', [])
		lists = []
		for json_vxlan1 in json_vxlan_list:
		tunnel_interface2={}
		index = json_vxlan1.get('index', {})
		type = json_vxlan1.get('type', {})
		ingress = json_vxlan1.get('ingress', {}).get('vni')

		tunnel_interface2['index'] = str(index)
		tunnel_interface2['type'] = str(type)
		tunnel_interface2['ingress'] = {}
		tunnel_interface2['ingress']['vni'] = int(ingress)

		lists.append(tunnel_interface2)

		if len(lists) == 0:
		continue

		resource_key = '/tunnel-interface'
		response.append((resource_key, lists))

		return response