diff --git a/scripts/run_tests_locally-device-gnmi-nokia-srlinux.sh b/scripts/run_tests_locally-device-gnmi-nokia-srlinux.sh
index d40358e901cd4fa8caa9775a6b5ce1ec5a0c7659..beae4f72d0e867f88067237cbd92b54f68c10902 100755
--- a/scripts/run_tests_locally-device-gnmi-nokia-srlinux.sh
+++ b/scripts/run_tests_locally-device-gnmi-nokia-srlinux.sh
@@ -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
diff --git a/src/device/service/driver_api/_Driver.py b/src/device/service/driver_api/_Driver.py
index 0aa1a6c5a8697d4c75f7044981221c6dd47e3aff..9720d7d4017de7ea0b9287f05ecb4517bacaaf26 100644
--- a/src/device/service/driver_api/_Driver.py
+++ b/src/device/service/driver_api/_Driver.py
@@ -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:
diff --git a/src/device/service/drivers/gnmi_nokia_srlinux/GnmiSessionHandler.py b/src/device/service/drivers/gnmi_nokia_srlinux/GnmiSessionHandler.py
index 04dae4f5fcc6427c735b528b0ab32ba1c967709a..e0a95ff9c9d5d1973b46192696219507ba994950 100644
--- a/src/device/service/drivers/gnmi_nokia_srlinux/GnmiSessionHandler.py
+++ b/src/device/service/drivers/gnmi_nokia_srlinux/GnmiSessionHandler.py
@@ -12,7 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import copy, grpc, json, logging, queue, threading
+import copy,grpc,json, logging, queue, threading
from typing import Any, Dict, List, Optional, Tuple, Union
from common.tools.grpc.Tools import grpc_message_to_json_string
from common.type_checkers.Checkers import chk_float, chk_length, chk_string, chk_type
@@ -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)]
diff --git a/src/device/service/drivers/gnmi_nokia_srlinux/handlers/Interface.py b/src/device/service/drivers/gnmi_nokia_srlinux/handlers/Interface.py
index 977833cac2dadf564a9b802e9c5b5bbffdeeb027..26f8e5080c9bb46cb0ccb948500229769c0b1482 100644
--- a/src/device/service/drivers/gnmi_nokia_srlinux/handlers/Interface.py
+++ b/src/device/service/drivers/gnmi_nokia_srlinux/handlers/Interface.py
@@ -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 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
+ 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['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
+
+ str_path = '/interface[name={:s}]'.format(if_name)
+ json_interface = {}
+
+ if 'if_name' in resource_value:
+ #if_name=str(resource_value['if_name'])
+ json_interface['name']=if_name
+
+
+ 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
+
+ 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)
+
+
+ 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
+
+
+
+ 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)
+ })
+
+ 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
+
+
+ if 'subif_type' in resource_value:
+ subif_type = str(resource_value['subif_type'])
+ subif['type'] =subif_type
+
+
+ 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
+
+
+ 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()
+
+
+ return str_path,json.dumps(json_interface)
+
+############################PARSE##############################################################
+
+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 = []
+
+ 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
- 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
-
- 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', [])
-
- 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_interface)))
- continue
- interface['name'] = interface_name
-
- 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={})
-
- 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)
-
- 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)
-
- 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', [])
-
- for json_subinterface in json_subinterface_list:
- LOGGER.info('json_subinterface = {:s}'.format(json.dumps(json_subinterface)))
-
- subinterface = {}
-
- 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
-
- 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={})
-
- MATCH_FIELDS = ('match', 'openconfig-vlan:match', 'ocv:match')
- json_vlan = dict_get_first(json_vlan, MATCH_FIELDS, default={})
-
- SIN_TAG_FIELDS = ('single-tagged', 'openconfig-vlan:single-tagged', 'ocv:single-tagged')
- json_vlan = dict_get_first(json_vlan, SIN_TAG_FIELDS, default={})
-
- CONFIG_FIELDS = ('config', 'openconfig-vlan:config', 'ocv:config')
- json_vlan = dict_get_first(json_vlan, CONFIG_FIELDS, default={})
-
- 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
-
-
- # TODO: implement support for multiple IP addresses per subinterface
+ admin_state = json_interface.get('admin-state')
+ if admin_state is not None:
+ interface['admin-state'] = bool(admin_state)
- IPV4_FIELDS = ('ipv4', 'openconfig-if-ip:ipv4', 'ociip:ipv4')
- json_ipv4 = dict_get_first(json_subinterface, IPV4_FIELDS, default={})
-
- IPV4_ADDRESSES_FIELDS = ('addresses', 'openconfig-if-ip:addresses', 'ociip:addresses')
- json_ipv4_addresses = dict_get_first(json_ipv4, IPV4_ADDRESSES_FIELDS, default={})
+ vlan_tagging= json_interface.get('vlan-tagging')
+ if vlan_tagging is not None:
+ interface['vlan_tagging'] = bool(vlan_tagging)
- 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=[])
- #ipv4_addresses = []
- for json_ipv4_address in json_ipv4_address_list:
- LOGGER.info('json_ipv4_address = {:s}'.format(json.dumps(json_ipv4_address)))
+ 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)
- STATE_FIELDS = ('state', 'openconfig-if-ip:state', 'ociip:state')
- json_ipv4_address_state = dict_get_first(json_ipv4_address, STATE_FIELDS, default={})
+ #VLAN_FIELDS = ('vlan', 'NokiaSRL-vlan:vlan', 'ocv:vlan')
+ #json_vlan = dict_get_first(json_subinterface, VLAN_FIELDS, default={})
- #ipv4_address = {}
+ #ENCAP_FIELDS = ('encap', 'NokiaSRL-vlan:match', 'ocv:encap')
+ #json_vlan = dict_get_first(json_vlan, ENCAP_FIELDS, default={})
- #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
+ #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:
+ continue
+ subinterface['type'] = vlan_type
- 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
+#
+ #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
- 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)
+ #IPV4_FIELDS = ('ipv4', 'NokiaSRL-if-ip:ipv4', 'ociip:ipv4')
+ #json_ipv4 = dict_get_first(json_subinterface, IPV4_FIELDS, default={})
- #if len(ipv4_address) == 0: continue
- #ipv4_addresses.append(ipv4_address)
+ #ipv4_enable = json_subinterface.get('state', {}).get('enable', False)
+ #subinterface['admin-state'] = bool(ipv4_enable)
- #subinterface['ipv4_addresses'] = ipv4_addresses
+
+ #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)
- if len(subinterface) == 0: continue
- resource_key = '/interface[{:s}]/subinterface[{:s}]'.format(interface['name'], str(subinterface['index']))
- response.append((resource_key, subinterface))
+ #IPV4_ADDRESS_FIELDS = ('address', 'NokiaSRL-if-ip:address', 'ociip:address')
+ #json_ipv4_address_list = json_ipv4.get(IPV4_ADDRESS_FIELDS[0], [])
- if len(interface) == 0: continue
- response.append(('/interface[{:s}]'.format(interface['name']), interface))
+ #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
+
+ 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)
+
+ return response
- return response
diff --git a/src/device/service/drivers/gnmi_nokia_srlinux/handlers/NetworkInstance.py b/src/device/service/drivers/gnmi_nokia_srlinux/handlers/NetworkInstance.py
index aed821a06fa7fcafe96a21ad5f5fa06be2902038..b7360616f2264629f68387f75aa86ff5118ff431 100644
--- a/src/device/service/drivers/gnmi_nokia_srlinux/handlers/NetworkInstance.py
+++ b/src/device/service/drivers/gnmi_nokia_srlinux/handlers/NetworkInstance.py
@@ -19,44 +19,366 @@ from ._Handler import _Handler
LOGGER = logging.getLogger(__name__)
class NetworkInstanceHandler(_Handler):
- def get_resource_key(self) -> str: return '/network_instance'
- def get_path(self) -> str: return '/network-instances/network-instance'
+ def get_resource_key(self) -> str: return '/network-instance'
+ def get_path(self) -> str: return '/network-instance'
def compose(self, resource_key : str, resource_value : Dict, delete : bool = False) -> Tuple[str, str]:
- ni_name = str(resource_value['name']) # test-svc
+ if_name= str(resource_value['if_name'])
+ json_networkinstance = {}
if delete:
- PATH_TMPL = '/network-instances/network-instance[name={:s}]'
- str_path = PATH_TMPL.format(ni_name)
+ PATH_TMPL = '/network-instance[name={:s}]'
+ str_path = PATH_TMPL.format(if_name)
str_data = json.dumps({})
return str_path, str_data
+
+ str_path = '/network-instance[name={:s}]'.format(if_name)
- ni_type = str(resource_value['type']) # L3VRF / L2VSI / ...
-
- # not works: [FailedPrecondition] unsupported identifier 'DIRECTLY_CONNECTED'
- #protocols = [self._compose_directly_connected()]
-
- MAP_OC_NI_TYPE = {
- 'L3VRF': 'openconfig-network-instance-types:L3VRF',
- }
- ni_type = MAP_OC_NI_TYPE.get(ni_type, ni_type)
-
- str_path = '/network-instances/network-instance[name={:s}]'.format(ni_name)
- str_data = json.dumps({
- 'name': ni_name,
- 'config': {'name': ni_name, 'type': ni_type},
- #'protocols': {'protocol': protocols},
- })
- return str_path, str_data
-
- def _compose_directly_connected(self, name=None, enabled=True) -> Dict:
- identifier = 'DIRECTLY_CONNECTED'
- if name is None: name = 'DIRECTLY_CONNECTED'
- return {
- 'identifier': identifier, 'name': name,
- 'config': {'identifier': identifier, 'name': name, 'enabled': enabled},
- }
-
- def parse(self, json_data : Dict) -> List[Tuple[str, Dict[str, Any]]]:
- response = []
- return response
+ if 'if_interface1' in resource_value:
+ if_interface1= str(resource_value['if_interface1'])
+ subif_interface_default_list=json_networkinstance.setdefault('interface',list())
+ subif_interface1={'name':if_interface1}
+ subif_interface_default_list.append(subif_interface1)
+
+ if 'if_interface2' in resource_value:
+ if_interface2= str(resource_value['if_interface2'])
+ subif_interface2={'name':if_interface2}
+ subif_interface_default_list.append(subif_interface2)
+
+ if 'if_name' in resource_value:
+ json_networkinstance['name']=if_name
+
+ if 'if_admin_state_bgp' in resource_value:
+ bool_admin_state = bool(resource_value['if_admin_state_bgp'])
+ str_admin_state = 'enable' if bool_admin_state else 'disable'
+ subif_protocol=json_networkinstance.setdefault('protocols',dict())
+ subif_bgp=subif_protocol.setdefault('bgp',dict())
+ subif_afisafi=subif_bgp.setdefault('afi-safi',list())
+ subif_state = {'admin-state': str_admin_state}
+
+ if 'if_afi_safi_name_bgp' in resource_value:
+ if_afi_safi_name_bgp = str(resource_value['if_afi_safi_name_bgp'])
+ subif_state['afi-safi-name'] = if_afi_safi_name_bgp
+ subif_afisafi.append(subif_state)
+
+
+ if 'if_autonomous_system_bgp' in resource_value:
+ if_autonomous_system_bgp = int(resource_value['if_autonomous_system_bgp'])
+ subif_bgp['autonomous-system']=if_autonomous_system_bgp
+
+
+
+ if 'if_export_policy' in resource_value:
+ if_export_policy = str(resource_value['if_export_policy'])
+ subif_group1 = {'export-policy': if_export_policy}
+ subif_group=subif_bgp.setdefault('group',list())
+
+
+ if 'if_group_name' in resource_value:
+ if_group_name = str(resource_value['if_group_name'])
+ subif_group1['group-name']=if_group_name
+
+
+
+ if 'if_import_policy' in resource_value:
+ if_import_policy = str(resource_value['if_import_policy'])
+ subif_group1['import-policy']= if_import_policy
+
+ if 'if_peer_as' in resource_value:
+ if_peer_as = int(resource_value['if_peer_as'])
+ subif_group1['peer-as']= if_peer_as
+ subif_group.append(subif_group1)
+
+ if 'if_admin_state_group' in resource_value:
+ bool_admin_state = bool(resource_value['if_admin_state_group'])
+ str_admin_state = 'enable' if bool_admin_state else 'disable'
+ subif_as = {'admin-state': str_admin_state}
+ subif_dictonary={}
+ subif_group.append(subif_dictonary)
+ afi_safi_list=subif_dictonary.setdefault('afi-safi',list())
+
+
+
+ if 'if_afi_safi_name_group' in resource_value:
+ if_afi_safi_name_group = str(resource_value['if_afi_safi_name_group'])
+ subif_as['afi-safi-name']=if_afi_safi_name_group
+ afi_safi_list.append(subif_as)
+
+
+ if 'if_admin_state_group_2' in resource_value:
+ bool_admin_state = bool(resource_value['if_admin_state_group_2'])
+ str_admin_state_group2 = 'enable' if bool_admin_state else 'disable'
+ subif_dictonary2={}
+ afi_safi_list.append(subif_dictonary2)
+ subif_dictonary2['admin-state']=str_admin_state_group2
+
+ if 'if_afi_safi_name_group_2' in resource_value:
+ if_afi_safi_name_group_2 = str(resource_value['if_afi_safi_name_group_2'])
+ subif_dictonary2['afi-safi-name']=if_afi_safi_name_group_2
+
+ if 'if_export_policy_2' in resource_value:
+ if_export_policy_2 = str(resource_value['if_export_policy_2'])
+ subif_dictonary['export-policy']=if_export_policy_2
+
+ if 'if_group_name_2' in resource_value:
+ if_group_name_2 = str(resource_value['if_group_name_2'])
+ subif_dictonary['group-name']=if_group_name_2
+
+ if 'if_import_policy_2' in resource_value:
+ if_import_policy_2 = str(resource_value['if_import_policy_2'])
+ subif_dictonary['import-policy']=if_import_policy_2
+
+ if 'if_as_number' in resource_value:
+ if_as_number = int(resource_value['if_as_number'])
+ subif_dictonary['local-as'] = {}
+ subif_dictonary['local-as']['as-number']=if_as_number
+
+ if 'if_peer_as_2' in resource_value:
+ if_peer_as_2 = int(resource_value['if_peer_as_2'])
+ subif_dictonary['peer-as']=if_peer_as_2
+
+
+ if 'if_minimum_advertisement_interval' in resource_value:
+ if_minimum_advertisement_interval = int(resource_value['if_minimum_advertisement_interval'])
+ subif_dictonary['timers']={}
+ subif_dictonary['timers']['minimum-advertisement-interval']=if_minimum_advertisement_interval
+
+
+
+ if 'if_admin_state_neighbor' in resource_value:
+ bool_admin_state = bool(resource_value['if_admin_state_neighbor'])
+ str_admin_state_n= 'enable' if bool_admin_state else 'disable'
+ subif_neighbor=subif_bgp.setdefault('neighbor',list())
+ subif_state_n = {'admin-state': str_admin_state_n}
+
+
+ if 'if_peer_address_neighbor' in resource_value:
+ if_peer_address_neighbor = str(resource_value['if_peer_address_neighbor'])
+ subif_state_n['peer-address']=if_peer_address_neighbor
+ #subif_protocol=json_networkinstance.setdefault('protocols',dict())
+ #subif_bgp=subif_protocol.setdefault('bgp',dict())
+ #subif_neighbor=subif_bgp.setdefault('neighbor',list())
+
+ if 'if_peer_group_neighbor' in resource_value:
+ if_peer_group_neighbor = str(resource_value['if_peer_group_neighbor'])
+ subif_state_n['peer-group']=if_peer_group_neighbor
+ #subif_protocol=json_networkinstance.setdefault('protocols',dict())
+ #subif_bgp=subif_protocol.setdefault('bgp',dict())
+ #subif_neighbor=subif_bgp.setdefault('neighbor',list())
+
+ if 'if_local_address_neighbor' in resource_value:
+ if_local_address_neighbor = str(resource_value['if_local_address_neighbor'])
+ subif_state_n['transport']={}
+ subif_state_n['transport']['local-address']=if_local_address_neighbor
+ #subif_protocol=json_networkinstance.setdefault('protocols',dict())
+ #subif_bgp=subif_protocol.setdefault('bgp',dict())
+ #subif_neighbor=subif_bgp.setdefault('neighbor',list())
+ #subif_transport={'transport':subif_lan}
+ subif_neighbor.append(subif_state_n)
+
+ if 'if_peer_address_neighbor_2' in resource_value:
+ if_peer_address_neighbor_2 = str(resource_value['if_peer_address_neighbor_2'])
+ #subif_pan2= {'peer-address':if_peer_address_neighbor_2}
+ #subif_protocol=json_networkinstance.setdefault('protocols',dict())
+ #subif_bgp=subif_protocol.setdefault('bgp',dict())
+ #subif_neighbor=subif_bgp.setdefault('neighbor',list())
+ subif_dicneighbor={}
+ subif_dicneighbor['peer-address']=if_peer_address_neighbor_2
+
+
+ if 'if_peer_group_neighbor_2' in resource_value:
+ if_peer_group_neighbor_2 = str(resource_value['if_peer_group_neighbor_2'])
+ #subifpgn2 = {'peer-group':if_peer_group_neighbor_2}
+ # subif_protocol=json_networkinstance.setdefault('protocols',dict())
+ #subif_bgp=subif_protocol.setdefault('bgp',dict())
+ #subif_neighbor=subif_bgp.setdefault('neighbor',list())
+ subif_dicneighbor['peer-group']=if_peer_group_neighbor_2
+ subif_neighbor.append(subif_dicneighbor)
+
+
+ if 'if_router_id' in resource_value:
+ if_router_id = str(resource_value['if_router_id'])
+ #subif_protocol=json_networkinstance.setdefault('protocols',dict())
+ #subif_bgp=subif_protocol.setdefault('bgp',dict())
+ subif_bgp['router-id'] = if_router_id
+
+ ###############VRF###############3
+
+
+ if 'if_name' in resource_value and 'if_type' in resource_value:
+ if_name= str(resource_value['if_name'])
+ if_type=str(resource_value['if_type'])
+ json_networkinstance = {}
+ json_networkinstance['name']=if_name
+ json_networkinstance['type']=if_type
+
+
+ #if 'if_type' in resource_value:
+ # if_type=str(resource_value['if_type'])
+ # json_networkinstance['type']=if_type
+
+ if 'if_admin_state_vrf' in resource_value:
+ bool_admin_state= bool(resource_value['if_admin_state_vrf'])
+ str_admin_state = 'enable' if bool_admin_state else 'disable'
+ json_networkinstance ['admin-state']=str_admin_state
+
+ if 'if_interface1_name' in resource_value:
+ if_interface1_name= str(resource_value['if_interface1_name'])
+ if_interface1_list=json_networkinstance.setdefault('interface',list())
+ subif_interface_name={'name': if_interface1_name}
+ if_interface1_list.append(subif_interface_name)
+
+ if 'if_vxlaninterface_name' in resource_value:
+ if_vxlaninterface_name= str(resource_value['if_vxlaninterface_name'])
+ if_if_vxlaninterface_name_list=json_networkinstance.setdefault('vxlan-interface',list())
+ subif_vxlaninterface_name={'name': if_vxlaninterface_name}
+ if_if_vxlaninterface_name_list.append(subif_vxlaninterface_name)
+
+
+ if 'if_bgp_evpn_instance_id' in resource_value:
+ if_bgp_evpn_instance_id = int(resource_value['if_bgp_evpn_instance_id'])
+ subif_protocol=json_networkinstance.setdefault('protocols',dict())
+ subif_bgp_evpn=subif_protocol.setdefault('bgp-evpn',dict())
+ subif_bgpinstance=subif_bgp_evpn.setdefault('bgp-instance',list())
+ subif_dictonary3={}
+ subif_bgpinstance.append(subif_dictonary3)
+ subif_dictonary3['id']=if_bgp_evpn_instance_id
+
+
+ if 'if_bgp_evpn_instance_admin_state' in resource_value:
+ bool_admin_state = bool(resource_value['if_bgp_evpn_instance_admin_state'])
+ str_admin_state = 'enable' if bool_admin_state else 'disable'
+ subif_dictonary3['admin-state']=str_admin_state
+
+
+ if 'if_bgp_evpn_instance_vxlan_interface' in resource_value:
+ if_bgp_evpn_instance_vxlan_interface = str(resource_value['if_bgp_evpn_instance_vxlan_interface'])
+ subif_dictonary3['vxlan-interface']=if_bgp_evpn_instance_vxlan_interface
+
+ if 'if_bgp_evpn_instance_evi' in resource_value:
+ if_bgp_evpn_instance_evi = int(resource_value['if_bgp_evpn_instance_evi'])
+ subif_dictonary3['evi']=if_bgp_evpn_instance_evi
+
+
+ if 'if_bgp_vpn_instance_id' in resource_value:
+ if_bgp_vpn_instance_id = int(resource_value['if_bgp_vpn_instance_id'])
+ subif_bgp_vpn=subif_protocol.setdefault('bgp-vpn',dict())
+ subif_bgpinstance=subif_bgp_vpn.setdefault('bgp-instance',list())
+ subif_dictonary4={}
+ subif_dictonary4['id']=if_bgp_vpn_instance_id
+
+
+ if 'if_bgp_vpn_instance_export_rt' in resource_value and 'if_bgp_vpn_instance_import_rt' in resource_value:
+ if_bgp_vpn_instance_export_rt = str(resource_value['if_bgp_vpn_instance_export_rt'])
+ if_bgp_vpn_instance_import_rt = str(resource_value['if_bgp_vpn_instance_import_rt'])
+ subif_dictonary4['route-target']={}
+ subif_dictonary4['route-target']['export-rt']=if_bgp_vpn_instance_export_rt
+ subif_dictonary4['route-target']['import-rt']=if_bgp_vpn_instance_import_rt
+ subif_bgpinstance.append(subif_dictonary4)
+
+ #if 'if_bgp_vpn_instance_import_rt' in resource_value:
+ # if_bgp_vpn_instance_import_rt = str(resource_value['if_bgp_vpn_instance_import_rt'])
+ # subif_dictonary4['route-target']['import-rt']=if_bgp_vpn_instance_import_rt
+ # subif_bgpinstance.append(subif_dictonary4)
+
+######################################SPINE##########################
+ if 'if_interface1_spine' in resource_value:
+ if_interface1_spine= str(resource_value['if_interface1_spine'])
+ subif_interface_default_list=json_networkinstance.setdefault('interface',list())
+ subif_interface1_spine={'name':if_interface1_spine}
+ subif_interface_default_list.append(subif_interface1_spine)
+
+ if 'if_interface2_spine' in resource_value:
+ if_interface2_spine= str(resource_value['if_interface2_spine'])
+ subif_interface_default_list=json_networkinstance.setdefault('interface',list())
+ subif_interface2_spine={'name':if_interface2_spine}
+ subif_interface_default_list.append(subif_interface2_spine)
+
+ if 'if_interface3_spine' in resource_value:
+ if_interface3_spine= str(resource_value['if_interface3_spine'])
+ subif_interface_default_list=json_networkinstance.setdefault('interface',list())
+ subif_interface3_spine={'name':if_interface3_spine}
+ subif_interface_default_list.append(subif_interface3_spine)
+
+ if 'if_name' in resource_value:
+ json_networkinstance['name']=if_name
+
+ if 'if_admin_state_spine' in resource_value:
+ bool_admin_state = bool(resource_value['if_admin_state_spine'])
+ str_admin_state = 'enable' if bool_admin_state else 'disable'
+ subif_protocol=json_networkinstance.setdefault('protocols',dict())
+ subif_bgp=subif_protocol.setdefault('bgp',dict())
+ subif_afisafi=subif_bgp.setdefault('afi-safi',list())
+ subif_state = {'admin-state': str_admin_state}
+
+ if 'if_afi_safi_name_bgp_spine' in resource_value:
+ if_afi_safi_name_bgp_spine = str(resource_value['if_afi_safi_name_bgp_spine'])
+ subif_state['afi-safi-name'] = if_afi_safi_name_bgp_spine
+ subif_afisafi.append(subif_state)
+
+
+ if 'if_autonomous_system_bgp_spine' in resource_value:
+ if_autonomous_system_bgp_spine = int(resource_value['if_autonomous_system_bgp_spine'])
+ subif_bgp['autonomous-system']=if_autonomous_system_bgp_spine
+
+
+
+ if 'if_export_policy_spine' in resource_value:
+ if_export_policy_spine = str(resource_value['if_export_policy_spine'])
+ subif_group_spine = {'export-policy': if_export_policy_spine}
+ subif_group_spine_list=subif_bgp.setdefault('group',list())
+
+
+ if 'if_group_name_spine' in resource_value:
+ if_group_name_spine = str(resource_value['if_group_name_spine'])
+ subif_group_spine['group-name']=if_group_name_spine
+
+
+
+ if 'if_import_policy_spine' in resource_value:
+ if_import_policy_spine = str(resource_value['if_import_policy_spine'])
+ subif_group_spine['import-policy']= if_import_policy_spine
+ subif_group_spine_list.append(subif_group_spine)
+
+
+ if 'if_peer_address_neighbor_spine' in resource_value:
+ if_peer_address_neighbor_spine = str(resource_value['if_peer_address_neighbor_spine'])
+ subif_neighbor_spine=subif_bgp.setdefault('neighbor',list())
+ subif_state_spine={'peer-address':if_peer_address_neighbor_spine}
+
+
+ if 'if_peeras_group_neighbor_spine' in resource_value:
+ if_peeras_group_neighbor_spine = int(resource_value['if_peeras_group_neighbor_spine'])
+ subif_state_spine['peer-as']=if_peeras_group_neighbor_spine
+
+ if 'if_peer_group_neighbor_spine' in resource_value:
+ if_peer_group_neighbor_spine = str(resource_value['if_peer_group_neighbor_spine'])
+ subif_state_spine['peer-group']=if_peer_group_neighbor_spine
+ subif_neighbor_spine.append(subif_state_spine)
+#######Neighbor2
+ if 'if_peer_address2_neighbor_spine' in resource_value:
+ if_peer_address2_neighbor_spine = str(resource_value['if_peer_address2_neighbor_spine'])
+ subif_neighbor2_spine={}
+ subif_neighbor2_spine['peer-address']=if_peer_address2_neighbor_spine
+
+
+ if 'if_peeras_group_neighbor2_spine' in resource_value:
+ if_peeras_group_neighbor2_spine = int(resource_value['if_peeras_group_neighbor2_spine'])
+ subif_neighbor2_spine['peer-as']=if_peeras_group_neighbor2_spine
+
+ if 'if_peer_group_neighbor2_spine' in resource_value:
+ if_peer_group_neighbor2_spine = str(resource_value['if_peer_group_neighbor2_spine'])
+ subif_neighbor2_spine['peer-group']=if_peer_group_neighbor2_spine
+ subif_neighbor_spine.append(subif_neighbor2_spine)
+
+ if 'if_router_id_spine' in resource_value:
+ if_router_id_spine = str(resource_value['if_router_id_spine'])
+ subif_bgp['router-id'] = if_router_id_spine
+
+
+
+
+
+
+
+ return str_path, json.dumps(json_networkinstance)
diff --git a/src/device/service/drivers/gnmi_nokia_srlinux/handlers/NetworkInstanceInterface.py b/src/device/service/drivers/gnmi_nokia_srlinux/handlers/NetworkInstanceInterface.py
index 205373fca870ea7338a3c9c043c60306b535c1c0..03e3cb01aa880633640e32b5402502ed57bfea57 100644
--- a/src/device/service/drivers/gnmi_nokia_srlinux/handlers/NetworkInstanceInterface.py
+++ b/src/device/service/drivers/gnmi_nokia_srlinux/handlers/NetworkInstanceInterface.py
@@ -19,27 +19,22 @@ from ._Handler import _Handler
LOGGER = logging.getLogger(__name__)
class NetworkInstanceInterfaceHandler(_Handler):
- def get_resource_key(self) -> str: return '/network_instance/interface'
- def get_path(self) -> str: return '/network-instances/network-instance/interfaces'
+ def get_resource_key(self) -> str: return '/network-instance/interface'
+ def get_path(self) -> str: return '/network-instance/interface'
def compose(self, resource_key : str, resource_value : Dict, delete : bool = False) -> Tuple[str, str]:
- ni_name = str(resource_value['name' ]) # test-svc
- if_name = str(resource_value['if_name' ]) # ethernet-1/1
- sif_index = int(resource_value['sif_index']) # 0
- if_id = '{:s}.{:d}'.format(if_name, sif_index)
+ ni_name = str(resource_value['name' ]) # default
+ if_name = str(resource_value['if_name' ]) # ethernet-1/49.0
+
if delete:
- PATH_TMPL = '/network-instances/network-instance[name={:s}]/interfaces/interface[id={:s}]'
- str_path = PATH_TMPL.format(ni_name, if_id)
+ PATH_TMPL = '/network-instances/network-instance[name={:s}]/interface[if_name={:s}]'
+ str_path = PATH_TMPL.format(ni_name,if_name)
str_data = json.dumps({})
return str_path, str_data
- str_path = '/network-instances/network-instance[name={:s}]/interfaces/interface[id={:s}]'.format(ni_name, if_id)
- str_data = json.dumps({
- 'id': if_id,
- 'config': {'id': if_id, 'interface': if_name, 'subinterface': sif_index},
- })
- return str_path, str_data
+ str_path = 'network-instance[name={:s}]/interface'.format(ni_name, if_name)
+ str_data = json.dumps({})
def parse(self, json_data : Dict) -> List[Tuple[str, Dict[str, Any]]]:
response = []
diff --git a/src/device/service/drivers/gnmi_nokia_srlinux/handlers/NetworkInstanceVxlanInterface.py b/src/device/service/drivers/gnmi_nokia_srlinux/handlers/NetworkInstanceVxlanInterface.py
new file mode 100644
index 0000000000000000000000000000000000000000..288a0fddbf89ff73ade2e240bc27058c398329b9
--- /dev/null
+++ b/src/device/service/drivers/gnmi_nokia_srlinux/handlers/NetworkInstanceVxlanInterface.py
@@ -0,0 +1,41 @@
+# 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.
+
+import json, logging
+from typing import Any, Dict, List, Tuple
+from ._Handler import _Handler
+
+LOGGER = logging.getLogger(__name__)
+
+class NetworkInstanceVxlanInterfacehandler(_Handler):
+ def get_resource_key(self) -> str: return '/network_instance/interface'
+ def get_path(self) -> str: return '/network-instance/interface'
+
+ def compose(self, resource_key : str, resource_value : Dict, delete : bool = False) -> Tuple[str, str]:
+ ni_name = str(resource_value['name' ]) # vrf-1
+ if_name = str(resource_value['if_name' ]) # vxlan1.1
+
+
+ if delete:
+ PATH_TMPL = '/network-instances/network-instance[name={:s}]/interface[if_name={:s}]'
+ str_path = PATH_TMPL.format(ni_name,if_name)
+ str_data = json.dumps({})
+ return str_path, str_data
+
+ str_path = 'network-instance[name={:s}]/vxlan-interface'.format(ni_name, if_name)
+ str_data = json.dumps({})
+
+ def parse(self, json_data : Dict) -> List[Tuple[str, Dict[str, Any]]]:
+ response = []
+ return response
diff --git a/src/device/service/drivers/gnmi_nokia_srlinux/handlers/TunnelInterface.py b/src/device/service/drivers/gnmi_nokia_srlinux/handlers/TunnelInterface.py
new file mode 100644
index 0000000000000000000000000000000000000000..f6dcdf37b309321bcdf3d8342e2f322b092acd0a
--- /dev/null
+++ b/src/device/service/drivers/gnmi_nokia_srlinux/handlers/TunnelInterface.py
@@ -0,0 +1,59 @@
+# 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.
+
+import json, logging
+from typing import Any, Dict, List, Tuple
+from ._Handler import _Handler
+
+LOGGER = logging.getLogger(__name__)
+
+class TunnelInterfaceHandler(_Handler):
+ def get_resource_key(self) -> str:
+ return '/tunnel-interface'
+
+ def get_path(self) -> str:
+ return '/tunnel-interface'
+
+ def compose(self, resource_key: str, resource_value: Dict, delete: bool = False) -> Tuple[str, str]:
+ if_name = str(resource_value['if_name'])
+
+ if delete:
+ PATH_TMPL = '/tunnel-interface[name={:s}]'
+ str_path = PATH_TMPL.format(if_name)
+ str_data = json.dumps({})
+ return str_path, str_data
+
+
+ str_path = '/tunnel-interface[name={:s}]'.format(if_name)
+
+
+ if 'if_index' in resource_value:
+ if_index = str(resource_value['if_index'])
+ json_tunnel={}
+ subif_vxlaninterface = json_tunnel.setdefault('vxlan-interface', list())
+ subif = {'index': if_index}
+
+ if 'if_type_tunnel' in resource_value:
+ if_type_tunnel = str(resource_value['if_type_tunnel'])
+ subif['type'] = if_type_tunnel
+
+ if 'if_vni_tunnel' in resource_value:
+ if_vni_tunnel = int(resource_value['if_vni_tunnel'])
+ subif['ingress']={}
+ subif['ingress']['vni'] = if_vni_tunnel
+ subif_vxlaninterface.append(subif)
+
+
+ return str_path, json.dumps(json_tunnel)
+
diff --git a/src/device/service/drivers/gnmi_nokia_srlinux/handlers/__init__.py b/src/device/service/drivers/gnmi_nokia_srlinux/handlers/__init__.py
index 39cd7c66ad5e8c16e89192ad0f2ffb7c43ae6c50..97ca2db370085c16cfcae5298b1cf45b842d69fe 100644
--- a/src/device/service/drivers/gnmi_nokia_srlinux/handlers/__init__.py
+++ b/src/device/service/drivers/gnmi_nokia_srlinux/handlers/__init__.py
@@ -14,41 +14,60 @@
import logging
from typing import Dict, List, Optional, Tuple, Union
-from device.service.driver_api._Driver import RESOURCE_ENDPOINTS, RESOURCE_INTERFACES, RESOURCE_NETWORK_INSTANCES
+
+from grpc import RpcContext
+from device.service.driver_api._Driver import RESOURCE_ENDPOINTS, RESOURCE_INTERFACES, RESOURCE_NETWORK_INSTANCES,RESOURCE_NETWORK_INSTANCE_VXLAN,RESOURCE_ROUTING_POLICY,RESOURCE_TUNNEL_INTERFACE
from ._Handler import _Handler
from .Component import ComponentHandler
from .Interface import InterfaceHandler
from .InterfaceCounter import InterfaceCounterHandler
from .NetworkInstance import NetworkInstanceHandler
from .NetworkInstanceInterface import NetworkInstanceInterfaceHandler
-from .NetworkInstanceStaticRoute import NetworkInstanceStaticRouteHandler
+from .NetworkInstanceVxlanInterface import NetworkInstanceVxlanInterfacehandler
+from .routingpolicy import routingpolicyhandler
+from .TunnelInterface import TunnelInterfaceHandler
from .Tools import get_schema
+logging.basicConfig(level=logging.DEBUG)
LOGGER = logging.getLogger(__name__)
+LOGGER.setLevel(logging.DEBUG)
comph = ComponentHandler()
ifaceh = InterfaceHandler()
ifctrh = InterfaceCounterHandler()
nih = NetworkInstanceHandler()
niifh = NetworkInstanceInterfaceHandler()
-nisrh = NetworkInstanceStaticRouteHandler()
+nivi = NetworkInstanceVxlanInterfacehandler()
+rp = routingpolicyhandler()
+ti = TunnelInterfaceHandler()
+
ALL_RESOURCE_KEYS = [
RESOURCE_ENDPOINTS,
RESOURCE_INTERFACES,
RESOURCE_NETWORK_INSTANCES,
+ RESOURCE_NETWORK_INSTANCE_VXLAN,
+ RESOURCE_ROUTING_POLICY,
+ RESOURCE_TUNNEL_INTERFACE,
]
RESOURCE_KEY_MAPPER = {
RESOURCE_ENDPOINTS : comph.get_resource_key(),
RESOURCE_INTERFACES : ifaceh.get_resource_key(),
RESOURCE_NETWORK_INSTANCES : nih.get_resource_key(),
+ RESOURCE_NETWORK_INSTANCE_VXLAN : nivi.get_resource_key(),
+ RESOURCE_ROUTING_POLICY : rp.get_resource_key(),
+ RESOURCE_TUNNEL_INTERFACE : ti.get_resource_key(),
}
PATH_MAPPER = {
'/components' : comph.get_path(),
- '/interfaces' : ifaceh.get_path(),
- '/network-instances' : nih.get_path(),
+ '/interface' : ifaceh.get_path(),
+ '/network-instance' : nih.get_path(),
+ '/network-instance/interface' : nivi.get_path(),
+ '/routing-policy' : rp.get_path(),
+ '/tunnel-interface' : ti.get_path(),
+
}
RESOURCE_KEY_TO_HANDLER = {
@@ -57,7 +76,10 @@ RESOURCE_KEY_TO_HANDLER = {
ifctrh.get_resource_key() : ifctrh,
nih.get_resource_key() : nih,
niifh.get_resource_key() : niifh,
- nisrh.get_resource_key() : nisrh,
+ nivi.get_resource_key() : nivi,
+ rp.get_resource_key() : rp,
+ ti.get_resource_key() :ti,
+
}
PATH_TO_HANDLER = {
@@ -66,38 +88,60 @@ PATH_TO_HANDLER = {
ifctrh.get_path() : ifctrh,
nih.get_path() : nih,
niifh.get_path() : niifh,
- nisrh.get_path() : nisrh,
+ nivi.get_path() : nivi,
+ rp.get_path() :rp,
+ ti.get_path() :ti,
}
def get_handler(
- resource_key : Optional[str] = None, path : Optional[str] = None, raise_if_not_found=True
+ resource_key: Optional[str] = None, path: Optional[str] = None, raise_if_not_found=True
) -> Optional[_Handler]:
if (resource_key is None) == (path is None):
MSG = 'Exactly one of resource_key({:s}) or path({:s}) must be specified'
- raise Exception(MSG.format(str(resource_key), str(path))) # pylint: disable=broad-exception-raised
+ raise Exception(MSG.format(str(resource_key), str(path))) # pylint: disable=broad-exception-raised
if resource_key is not None:
resource_key_schema = get_schema(resource_key)
resource_key_schema = RESOURCE_KEY_MAPPER.get(resource_key_schema, resource_key_schema)
handler = RESOURCE_KEY_TO_HANDLER.get(resource_key_schema)
if handler is None and raise_if_not_found:
MSG = 'Handler not found: resource_key={:s} resource_key_schema={:s}'
- # pylint: disable=broad-exception-raised
raise Exception(MSG.format(str(resource_key), str(resource_key_schema)))
elif path is not None:
path_schema = get_schema(path)
path_schema = PATH_MAPPER.get(path_schema, path_schema)
+ LOGGER.debug("Original path: %s, Schema path: %s", path, path_schema)
handler = PATH_TO_HANDLER.get(path_schema)
+ LOGGER.debug("Mapped path schema: %s", path_schema)
if handler is None and raise_if_not_found:
- MSG = 'Handler not found: resource_key={:s} resource_key_schema={:s}'
- # pylint: disable=broad-exception-raised
- raise Exception(MSG.format(str(resource_key), str(resource_key_schema)))
+ MSG = 'Handler not found: path={:s} path_schema={:s}'
+ LOGGER.error(MSG.format(str(path), str(path_schema)))
+ raise Exception(MSG.format(str(path), str(path_schema)))
+ LOGGER.debug("Handler found for path: %s", handler.get_path())
return handler
-def get_path(resource_key : str) -> str:
- return get_handler(resource_key=resource_key).get_path()
-def parse(str_path : str, value : Union[Dict, List]):
- return get_handler(path=str_path).parse(value)
-def compose(resource_key : str, resource_value : Union[Dict, List], delete : bool = False) -> Tuple[str, str]:
- return get_handler(resource_key=resource_key).compose(resource_key, resource_value, delete=delete)
+def get_path(resource_key: str) -> str:
+ handler = get_handler(resource_key=resource_key)
+ path = handler.get_path()
+ LOGGER.debug("get_path: resource_key=%s, path=%s", resource_key, path)
+ return path
+
+def parse(str_path: str, value: Union[Dict, List]):
+ handler = get_handler(path=str_path)
+ LOGGER.debug("parse: str_path=%s, value=%s", str_path, value)
+ return handler.parse(value)
+
+def compose(resource_key: str, resource_value: Union[Dict, List], delete: bool = False) -> Tuple[str, str]:
+ handler = get_handler(resource_key=resource_key)
+ LOGGER.debug("compose: resource_key=%s, resource_value=%s, delete=%s", resource_key, resource_value, delete)
+ return handler.compose(resource_key, resource_value, delete=delete)
+
+#def get_path(resource_key : str) -> str:
+# return get_handler(resource_key=resource_key).get_path()
+#
+#def parse(str_path : str, value : Union[Dict, List]):
+# return get_handler(path=str_path).parse(value)
+#
+#def compose(resource_key : str, resource_value : Union[Dict, List], delete : bool = False) -> Tuple[str, str]:
+# return get_handler(resource_key=resource_key).compose(resource_key, resource_value, delete=delete)
diff --git a/src/device/service/drivers/gnmi_nokia_srlinux/handlers/routingpolicy.py b/src/device/service/drivers/gnmi_nokia_srlinux/handlers/routingpolicy.py
new file mode 100644
index 0000000000000000000000000000000000000000..daeeb9fb210f7846410b049045e00e250ddc48f4
--- /dev/null
+++ b/src/device/service/drivers/gnmi_nokia_srlinux/handlers/routingpolicy.py
@@ -0,0 +1,209 @@
+# 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.
+
+import json, logging
+from typing import Any, Dict, List, Tuple
+from ._Handler import _Handler
+from .Tools import dict_get_first
+
+LOGGER = logging.getLogger(__name__)
+
+class routingpolicyhandler(_Handler):
+ def get_resource_key(self) -> str:
+ return '/routing-policy'
+
+ def get_path(self) -> str:
+ return '/routing-policy'
+
+ def compose(self, resource_key: str, resource_value: Dict, delete: bool = False) -> Tuple[str, str]:
+ if delete:
+ #PATH_TMPL = '/routing-policy'
+ #str_path = PATH_TMPL.format()
+ str_path='/routing-policy'
+ str_data = json.dumps({})
+ return str_path, str_data
+
+ str_path='/routing-policy'
+
+ #if 'if_name' in resource_value:
+ # if_name = str(resource_value['if_name']),
+ # json_routing={}
+ # subif_policy=json_routing.setdefault('policy',list())
+ # subif_name = {'name':if_name}
+ # subif_policy.append(subif_name)
+
+ # if 'if_policy_result' in resource_value:
+ # if_policy_result = str(resource_value['if_policy_result']),
+ # subif_defaultaction={'default-action':if_policy_result}
+ # subif_policy.append(subif_defaultaction)
+ if 'if_name' in resource_value:
+ if_name = str(resource_value['if_name'])
+ json_routing = {
+ 'policy': {
+ 'name': if_name,
+ 'default-action': {
+ 'policy-result': ''
+ }
+ }
+ }
+
+ if 'if_policy_result' in resource_value:
+ if_policy_result = str(resource_value['if_policy_result'])
+ if 'policy' not in json_routing:
+ json_routing['policy'] = {}
+ json_routing['policy']['default-action'] = {
+ 'policy-result': if_policy_result
+ }
+
+
+
+
+ return str_path, json.dumps(json_routing)
+ #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', [])
+
+ # 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_interface)))
+ # continue
+ # interface['name'] = interface_name
+
+ # CONFIG_FIELDS = ('config', 'NokiaSRL-interface:config', 'oci:config')
+ # json_config: Dict = dict_get_first(json_interface, CONFIG_FIELDS, default={})
+
+ # STATE_FIELDS = ('state', 'NokiaSRL-interface:state', 'oci:state')
+ # json_state: Dict = dict_get_first(json_interface, STATE_FIELDS, default={})
+
+ # 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)
+
+ # 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)
+
+ # 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', [])
+
+ # for json_subinterface in json_subinterface_list:
+ # LOGGER.info('json_subinterface = {:s}'.format(json.dumps(json_subinterface)))
+
+ # subinterface = {}
+
+ # 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
+
+ # subinterface_enabled = json_subinterface.get('state', {}).get('enabled', False)
+ # subinterface['enabled'] = bool(subinterface_enabled)
+
+ # VLAN_FIELDS = ('vlan', 'NokiaSRL-vlan:vlan', 'ocv:vlan')
+ # json_vlan = dict_get_first(json_subinterface, VLAN_FIELDS, default={})
+
+ # MATCH_FIELDS = ('match', 'NokiaSRL-vlan:match', 'ocv:match')
+ # json_vlan = dict_get_first(json_vlan, MATCH_FIELDS, default={})
+
+ # SIN_TAG_FIELDS = ('single-tagged', 'NokiaSRL-vlan:single-tagged', 'ocv:single-tagged')
+ # json_vlan = dict_get_first(json_vlan, SIN_TAG_FIELDS, default={})
+
+ # CONFIG_FIELDS = ('config', 'NokiaSRL-vlan:config', 'ocv:config')
+ # json_vlan = dict_get_first(json_vlan, CONFIG_FIELDS, default={})
+
+ # 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
+
+
+ # # TODO: implement support for multiple IP addresses per subinterface
+
+ # IPV4_FIELDS = ('ipv4', 'NokiaSRL-if-ip:ipv4', 'ociip:ipv4')
+ # json_ipv4 = dict_get_first(json_subinterface, IPV4_FIELDS, default={})
+ #
+ # IPV4_ADDRESSES_FIELDS = ('addresses', 'NokiaSRL-if-ip:addresses', 'ociip:addresses')
+ # json_ipv4_addresses = dict_get_first(json_ipv4, IPV4_ADDRESSES_FIELDS, default={})
+
+ # IPV4_ADDRESS_FIELDS = ('address', 'NokiaSRL-if-ip:address', 'ociip:address')
+ # json_ipv4_address_list : List[Dict] = dict_get_first(json_ipv4_addresses, IPV4_ADDRESS_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)))
+
+ # STATE_FIELDS = ('state', 'NokiaSRL-if-ip:state', 'ociip:state')
+ # json_ipv4_address_state = dict_get_first(json_ipv4_address, STATE_FIELDS, default={})
+
+ # #ipv4_address = {}
+
+ # #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
+
+ # 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: ipv4_address['address_ip'] = ipv4_address_ip
+ # 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: 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)
+
+ # #subinterface['ipv4_addresses'] = ipv4_addresses
+
+ # 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_interface.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
diff --git a/src/device/tests/test_gnmi_nokia_srlinux.py b/src/device/tests/test_gnmi_nokia_srlinux.py
index 1222749800dafb52babea5d1feec6944acd56afd..274bd168efd5217d646ee6881022e14c8d358f3d 100644
--- a/src/device/tests/test_gnmi_nokia_srlinux.py
+++ b/src/device/tests/test_gnmi_nokia_srlinux.py
@@ -20,112 +20,251 @@ from device.service.driver_api._Driver import (
RESOURCE_ENDPOINTS, RESOURCE_INTERFACES, RESOURCE_NETWORK_INSTANCES, RESOURCE_ROUTING_POLICIES, RESOURCE_SERVICES
)
+
logging.basicConfig(level=logging.DEBUG)
LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)
-# +---+---------------------------+--------------+---------------------------------+-------+---------+--------------------+--------------+
-# | # | Name | Container ID | Image | Kind | State | IPv4 Address | IPv6 Address |
-# +---+---------------------------+--------------+---------------------------------+-------+---------+--------------------+--------------+
-# | 1 | clab-tfs-scenario-client1 | a8d48ec3265a | ghcr.io/hellt/network-multitool | linux | running | 172.100.100.201/24 | N/A |
-# | 2 | clab-tfs-scenario-client2 | fc88436d2b32 | ghcr.io/hellt/network-multitool | linux | running | 172.100.100.202/24 | N/A |
-# | 3 | clab-tfs-scenario-srl1 | b995b9bdadda | ghcr.io/nokia/srlinux | srl | running | 172.100.100.101/24 | N/A |
-# | 4 | clab-tfs-scenario-srl2 | aacfc38cc376 | ghcr.io/nokia/srlinux | srl | running | 172.100.100.102/24 | N/A |
-# +---+---------------------------+--------------+---------------------------------+-------+---------+--------------------+--------------+
-
-def interface(if_name, sif_index, ipv4_address, ipv4_prefix, enabled) -> Tuple[str, Dict]:
- str_path = '/interface[{:s}]'.format(if_name)
- str_data = {'name': if_name, 'enabled': enabled, 'sub_if_index': sif_index, 'sub_if_enabled': enabled,
- 'sub_if_ipv4_enabled': enabled, 'sub_if_ipv4_address': ipv4_address, 'sub_if_ipv4_prefix': ipv4_prefix}
+
+#+---+-----------------+----------+--------------------+--------------+---------------------------------+-------+---------+------------------+----------------------+
+#| # | Topo Path | Lab Name | Name | Container ID | Image | Kind | State | IPv4 Address | IPv6 Address |
+#+---+-----------------+----------+--------------------+--------------+---------------------------------+-------+---------+------------------+----------------------+
+#| 1 | evpn01.clab.yml | evpn01 | clab-evpn01-leaf1 | 8e8e6611e681 | ghcr.io/nokia/srlinux:23.10.1 | srl | running | 172.20.20.102/24 | 2001:172:20:20::2/64 |
+#| 2 | | | clab-evpn01-leaf2 | 26180fc54efe | ghcr.io/nokia/srlinux:23.10.1 | srl | running | 172.20.20.103/24 | 2001:172:20:20::3/64 |
+#| 3 | | | clab-evpn01-spine1 | 981ef9d27d62 | ghcr.io/nokia/srlinux:23.10.1 | srl | running | 172.20.20.101/24 | 2001:172:20:20::4/64 |
+#| 4 | | | clab-evpn01-srv1 | 9a730d7b6ac3 | ghcr.io/hellt/network-multitool | linux | running | 172.20.20.2/24 | 2001:172:20:20::5/64 |
+#| 5 | | | clab-evpn01-srv2 | b90c423c4f23 | ghcr.io/hellt/network-multitool | linux | running | 172.20.20.3/24 | 2001:172:20:20::6/64 |
+#+---+-----------------+----------+--------------------+--------------+---------------------------------+-------+---------+----
+
+def interface(if_name,if_admin_state,subif_index,subif_ipv4_admin_state,subif_ipv4_address,subif_ipv4_prefix ) -> Tuple[str, Dict]:
+ str_path = f'/interface[name={if_name}]'
+
+ str_data = {
+ 'if_name': if_name,
+ 'if_admin_state': if_admin_state,
+ 'subif_index': subif_index,
+ 'subif_ipv4_admin_state': subif_ipv4_admin_state,
+ 'subif_ipv4_address': subif_ipv4_address,
+ 'subif_ipv4_prefix': subif_ipv4_prefix,
+ }
+
+ return str_path, str_data
+def vlan_interface(if_name,if_vlan_tagging,subif_index,subif_type,subif_vlan_admin_state, subif_vlan_encap) -> Tuple[str, Dict]:
+ str_path = f'/interface[name={if_name}]'
+ str_data = { 'if_name':if_name,
+ 'if_vlan_tagging': if_vlan_tagging,
+ 'subif_index': subif_index,
+ 'subif_type': subif_type,
+ 'subif_vlan_admin_state': subif_vlan_admin_state,
+ 'subif_vlan_encap': subif_vlan_encap,
+ }
+ return str_path, str_data
+
+
+
+def routing_policy(if_name,if_policy_result) -> Tuple[str, Dict]:
+ str_path = f'/routing-policy'
+ str_data = {'if_name': if_name,'if_policy_result': if_policy_result}
+ return str_path, str_data
+
+
+def network_instance_default(if_name,if_interface1,if_interface2,if_admin_state_bgp,if_afi_safi_name_bgp,if_autonomous_system_bgp,if_export_policy,if_group_name,if_import_policy,if_peer_as,if_admin_state_group,if_afi_safi_name_group,if_admin_state_group_2,if_afi_safi_name_group_2,if_export_policy_2,if_group_name_2,if_import_policy_2,if_as_number,if_peer_as_2,if_minimum_advertisement_interval,if_admin_state_neighbor,if_peer_address_neighbor,if_peer_group_neighbor,if_local_address_neighbor,if_peer_address_neighbor_2,if_peer_group_neighbor_2,if_router_id) -> Tuple[str, Dict]:
+ str_path = f'/network-instance[name={if_name}]'
+ str_data = {
+ 'if_name':if_name,
+ 'if_interface1':if_interface1,
+ 'if_interface2':if_interface2,
+ 'if_admin_state_bgp': if_admin_state_bgp,
+ 'if_afi_safi_name_bgp':if_afi_safi_name_bgp,
+ 'if_autonomous_system_bgp':if_autonomous_system_bgp,
+ 'if_export_policy': if_export_policy,
+ 'if_group_name': if_group_name,
+ 'if_import_policy':if_import_policy,
+ 'if_peer_as': if_peer_as,
+ 'if_admin_state_group': if_admin_state_group,
+ 'if_afi_safi_name_group': if_afi_safi_name_group,
+ 'if_admin_state_group_2': if_admin_state_group_2,
+ 'if_afi_safi_name_group_2': if_afi_safi_name_group_2,
+ 'if_export_policy_2': if_export_policy_2,
+ 'if_group_name_2': if_group_name_2,
+ 'if_import_policy_2': if_import_policy_2,
+ 'if_as_number': if_as_number,
+ 'if_peer_as_2': if_peer_as_2,
+ 'if_minimum_advertisement_interval':if_minimum_advertisement_interval,
+ 'if_admin_state_neighbor': if_admin_state_neighbor,
+ 'if_peer_address_neighbor': if_peer_address_neighbor,
+ 'if_peer_group_neighbor': if_peer_group_neighbor,
+ 'if_local_address_neighbor': if_local_address_neighbor,
+ 'if_peer_address_neighbor_2': if_peer_address_neighbor_2,
+ 'if_peer_group_neighbor_2':if_peer_group_neighbor_2,
+ 'if_router_id': if_router_id,
+ }
+ return str_path, str_data
+def network_instance_vrf(if_name,if_type,if_admin_state_vrf,if_interface1_name,if_vxlaninterface_name,if_bgp_evpn_instance_id,if_bgp_evpn_instance_admin_state,if_bgp_evpn_instance_vxlan_interface,if_bgp_evpn_instance_evi,if_bgp_vpn_instance_id,if_bgp_vpn_instance_export_rt,if_bgp_vpn_instance_import_rt) -> Tuple[str, Dict]:
+ str_path = f'/network-instance[name={if_name}]'
+ str_data = {'if_name': if_name,
+ 'if_type': if_type,
+ 'if_admin_state_vrf':if_admin_state_vrf,
+ 'if_interface1_name':if_interface1_name,
+ 'if_vxlaninterface_name':if_vxlaninterface_name,
+ 'if_bgp_evpn_instance_id': if_bgp_evpn_instance_id,
+ 'if_bgp_evpn_instance_admin_state':if_bgp_evpn_instance_admin_state,
+ 'if_bgp_evpn_instance_vxlan_interface': if_bgp_evpn_instance_vxlan_interface,
+ 'if_bgp_evpn_instance_evi': if_bgp_evpn_instance_evi,
+ 'if_bgp_vpn_instance_id':if_bgp_vpn_instance_id,
+ 'if_bgp_vpn_instance_export_rt':if_bgp_vpn_instance_export_rt,
+ 'if_bgp_vpn_instance_import_rt':if_bgp_vpn_instance_import_rt }
+
+ return str_path, str_data
+def tunnel_interface(if_name, if_index,if_type_tunnel,if_vni_tunnel) -> Tuple[str, Dict]:
+ str_path = f'/tunnel-interface[name={if_name}]'
+ str_data = {'if_name':if_name,
+ 'if_index':if_index,
+ 'if_type_tunnel':if_type_tunnel,
+ 'if_vni_tunnel':if_vni_tunnel }
+
return str_path, str_data
-def network_instance(ni_name, ni_type) -> Tuple[str, Dict]:
- str_path = '/network_instance[{:s}]'.format(ni_name)
- str_data = {'name': ni_name, 'type': ni_type}
+def network_instance_interface(name,if_name)-> Tuple[str, Dict]:
+ str_path=f'/network-instance[name={name} /interface{if_name}'
+ str_data = {'name':name,'if_name':if_name}
return str_path, str_data
-def network_instance_static_route(ni_name, prefix, next_hop, next_hop_index=0) -> Tuple[str, Dict]:
- str_path = '/network_instance[{:s}]/static_route[{:s}]'.format(ni_name, prefix)
- str_data = {'name': ni_name, 'prefix': prefix, 'next_hop': next_hop, 'next_hop_index': next_hop_index}
+def network_instance_Vxlaninterface(name,if_name)-> Tuple[str, Dict]:
+ str_path=f'/network-instance[name={name} /vxlan-interface{if_name}'
+ str_data = {'name':name,'if_name':if_name}
return str_path, str_data
-def network_instance_interface(ni_name, if_name, sif_index) -> Tuple[str, Dict]:
- str_path = '/network_instance[{:s}]/interface[{:s}.{:d}]'.format(ni_name, if_name, sif_index)
- str_data = {'name': ni_name, 'if_name': if_name, 'sif_index': sif_index}
+def network_instance_default_spine(if_interface1_spine,if_interface2_spine,if_interface3_spine,if_name,if_admin_state_spine,if_afi_safi_name_bgp_spine,if_autonomous_system_bgp_spine,if_export_policy_spine,if_group_name_spine,if_import_policy_spine,if_peer_address_neighbor_spine,if_peeras_group_neighbor_spine,if_peer_group_neighbor_spine,if_peer_address2_neighbor_spine,if_peeras_group_neighbor2_spine,if_peer_group_neighbor2_spine,if_router_id_spine)-> Tuple[str, Dict]:
+ str_path = f'/network-instance[name={if_name}]'
+ str_data = {'if_interface1_spine':if_interface1_spine,
+ 'if_interface2_spine':if_interface2_spine,
+ 'if_interface3_spine':if_interface3_spine,
+ 'if_name':if_name,
+ 'if_admin_state_spine':if_admin_state_spine,
+ 'if_afi_safi_name_bgp_spine':if_afi_safi_name_bgp_spine,
+ 'if_autonomous_system_bgp_spine':if_autonomous_system_bgp_spine,
+ 'if_export_policy_spine':if_export_policy_spine,
+ 'if_group_name_spine':if_group_name_spine,
+ 'if_import_policy_spine':if_import_policy_spine,
+ 'if_peer_address_neighbor_spine':if_peer_address_neighbor_spine,
+ 'if_peeras_group_neighbor_spine':if_peeras_group_neighbor_spine,
+ 'if_peer_group_neighbor_spine':if_peer_group_neighbor_spine,
+ 'if_peer_address2_neighbor_spine':if_peer_address2_neighbor_spine,
+ 'if_peeras_group_neighbor2_spine':if_peeras_group_neighbor2_spine,
+ 'if_peer_group_neighbor2_spine':if_peer_group_neighbor2_spine,
+ 'if_router_id_spine':if_router_id_spine}
return str_path, str_data
+
def test_gnmi_nokia_srlinux():
- driver_settings = {
- 'protocol': 'gnmi',
- 'username': 'admin',
- 'password': 'NokiaSrl1!',
- 'use_tls' : True,
- }
- dev1_driver = GnmiNokiaSrLinuxDriver('172.100.100.102', 57400, **driver_settings)
- dev1_driver.Connect()
-
- #resources_to_get = []
- #resources_to_get = [RESOURCE_ENDPOINTS]
- resources_to_get = [RESOURCE_INTERFACES]
- #resources_to_get = [RESOURCE_NETWORK_INSTANCES]
- #resources_to_get = [RESOURCE_ROUTING_POLICIES]
- #resources_to_get = [RESOURCE_SERVICES]
- LOGGER.info('resources_to_get = {:s}'.format(str(resources_to_get)))
- results_getconfig = dev1_driver.GetConfig(resources_to_get)
- LOGGER.info('results_getconfig = {:s}'.format(str(results_getconfig)))
-
- resources_to_set = [
- #network_instance('test-svc', 'L3VRF'),
-
- interface('ethernet-1/1', 0, '172.16.0.1', 24, True),
- #network_instance_interface('test-svc', 'ethernet-1/1', 0),
-
- interface('ethernet-1/2', 0, '172.0.0.1', 24, True),
- #network_instance_interface('test-svc', 'ethernet-1/2', 0),
-
- #network_instance_static_route('test-svc', '172.0.0.0/24', '172.16.0.2'),
- #network_instance_static_route('test-svc', '172.2.0.0/24', '172.16.0.3'),
- ]
- LOGGER.info('resources_to_set = {:s}'.format(str(resources_to_set)))
- results_setconfig = dev1_driver.SetConfig(resources_to_set)
- LOGGER.info('results_setconfig = {:s}'.format(str(results_setconfig)))
-
- #resources_to_get = []
- #resources_to_get = [RESOURCE_ENDPOINTS]
- resources_to_get = [RESOURCE_INTERFACES]
- #resources_to_get = [RESOURCE_NETWORK_INSTANCES]
- #resources_to_get = [RESOURCE_ROUTING_POLICIES]
- #resources_to_get = [RESOURCE_SERVICES]
- LOGGER.info('resources_to_get = {:s}'.format(str(resources_to_get)))
- results_getconfig = dev1_driver.GetConfig(resources_to_get)
- LOGGER.info('results_getconfig = {:s}'.format(str(results_getconfig)))
-
- resources_to_delete = [
- #network_instance_static_route('d35fc1d9', '172.0.0.0/24', '172.16.0.2'),
- #network_instance_static_route('d35fc1d9', '172.2.0.0/24', '172.16.0.3'),
-
- #network_instance_interface('d35fc1d9', 'ethernet-1/1', 0),
- #network_instance_interface('d35fc1d9', 'ethernet-1/2', 0),
-
- interface('ethernet-1/1', 0, '172.16.1.1', 24, True),
- interface('ethernet-1/2', 0, '172.0.0.2', 24, True),
+ try:
+ driver_settings_leaf1 = {
+ 'protocol': 'gnmi',
+ 'username': 'admin',
+ 'password': 'NokiaSrl1!',
+ 'use_tls': True,
+ }
+ dev1_driver = GnmiNokiaSrLinuxDriver('172.20.20.102', 57400, **driver_settings_leaf1)
+ dev1_driver.Connect()
+ resources_to_get_leaf1 = [RESOURCE_INTERFACES]
+ LOGGER.info('resources_to_get = {:s}'.format(str(resources_to_get_leaf1)))
+ results_getconfig_leaf1 = dev1_driver.GetConfig(resources_to_get_leaf1)
+ LOGGER.info('results_getconfig = {:s}'.format(str(results_getconfig_leaf1)))
+ except Exception as e:
+
+ logging.exception("Exception occurred", exc_info=True)
+ #resources_to_set_leaf1 = [
+ #####LEAF1#####
+ #interface('ethernet-1/49',True,0,True,'192.168.11.1','30'),
+ #routing_policy('all','accept'),
+ #network_instance_default('default','ethernet-1/49.0','system0.0',True,'ipv4-unicast',101,'all','eBGP-underlay','all',201,True,'evpn',False,'ipv4-unicast','all','iBGP-overlay','all',100,100,1,True,'10.0.0.2','iBGP-overlay','10.0.0.1','192.168.11.2','eBGP-underlay','10.0.0.1'),
+ #interface('system0',True,0,True,'10.0.0.1','32'),
+ #vlan_interface('ethernet-1/1',True,0,'bridged',True,'untagged'),
+ #network_instance_vrf('vrf-1','mac-vrf',True,'ethernet-1/1.0','vxlan1.1',1,True,'vxlan1.1',111,1,'target:100:111','target:100:111'),
+ #tunnel_interface('vxlan1', 1, 'bridged', 1),
+#]
+ #results_setconfig_leaf1 = dev1_driver.SetConfig(resources_to_set_leaf1)
+ #LOGGER.info('Leaf 1 results_setconfig = {:s}'.format(str(results_setconfig_leaf1)))
+
+ time.sleep(1)
+ dev1_driver.Disconnect()
+######LEAF2##############################
+# driver_settings_leaf2 = {
+# 'protocol': 'gnmi',
+# 'username': 'admin',
+# 'password': 'NokiaSrl1!',
+# 'use_tls': True,
+# }
+# dev2_driver = GnmiNokiaSrLinuxDriver('172.20.20.103', 57400, **driver_settings_leaf2)
+# dev2_driver.Connect()
+#
+# resources_to_set_leaf2 = [
+# interface('ethernet-1/49',True,0,True,'192.168.12.1','30'),
+# routing_policy('all','accept'),
+# network_instance_default('default','ethernet-1/49.0','system0.0',True,'ipv4-unicast',102,'all','eBGP-underlay','all',201,True,'evpn',False,'ipv4-unicast','all','iBGP-overlay','all',100,100,1,True,'10.0.0.1','iBGP-overlay','10.0.0.2','192.168.12.2','eBGP-underlay','10.0.0.2'),
+# interface('system0',True,0,True,'10.0.0.2','32'),
+# vlan_interface('ethernet-1/1',True,0,'bridged',True,'untagged'),
+# network_instance_vrf('vrf-1','mac-vrf',True,'ethernet-1/1.0','vxlan1.1',1,True,'vxlan1.1',111,1,'target:100:111','target:100:111'),
+# tunnel_interface('vxlan1', 1, 'bridged', 1),
+#
+# ]
+# results_setconfig_leaf2 = dev2_driver.SetConfig(resources_to_set_leaf2)
+# LOGGER.info('Leaf 2 results_setconfig = {:s}'.format(str(results_setconfig_leaf2)))
+#
+# time.sleep(1)
+# dev2_driver.Disconnect()
+#############SPINE############################################
+# driver_settings_spine = {
+# 'protocol': 'gnmi',
+# 'username': 'admin',
+# 'password': 'NokiaSrl1!',
+# 'use_tls': True,
+# }
+# spine_driver = GnmiNokiaSrLinuxDriver('172.20.20.101', 57400, **driver_settings_spine)
+# spine_driver.Connect()
+#
+# resources_to_set_spine = [
+# interface('ethernet-1/1',True,0,True,'192.168.11.2','30'),
+# interface('ethernet-1/2',True,0,True,'192.168.12.2','30'),
+# network_instance_default_spine('ethernet-1/1.0','ethernet-1/2.0','system0.0','default',True,'ipv4-unicast',201,'all','eBGP-underlay','all','192.168.11.1',101,'eBGP-underlay','192.168.12.1',102,'eBGP-underlay','10.0.1.1'),
+# interface('system0',True,0,True,'10.0.1.1',32),
+# routing_policy('all','accept'),
+#
+# ]
+# results_setconfig_spine = spine_driver.SetConfig(resources_to_set_spine)
+# LOGGER.info('Spine results_setconfig = {:s}'.format(str(results_setconfig_spine)))
+#
+#
+# time.sleep(1)
+# spine_driver.Disconnect()
+#
+#
+# #results_setconfig = dev1_driver.SetConfig(resources_to_set)
+ #LOGGER.info('results_setconfig = {:s}'.format(str(results_setconfig)))
- #network_instance('20f66fb5', 'L3VRF'),
- ]
- LOGGER.info('resources_to_delete = {:s}'.format(str(resources_to_delete)))
- results_deleteconfig = dev1_driver.DeleteConfig(resources_to_delete)
- LOGGER.info('results_deleteconfig = {:s}'.format(str(results_deleteconfig)))
-
- #resources_to_get = []
- #resources_to_get = [RESOURCE_ENDPOINTS]
- resources_to_get = [RESOURCE_INTERFACES]
- #resources_to_get = [RESOURCE_NETWORK_INSTANCES]
- #resources_to_get = [RESOURCE_ROUTING_POLICIES]
- #resources_to_get = [RESOURCE_SERVICES]
- LOGGER.info('resources_to_get = {:s}'.format(str(resources_to_get)))
- results_getconfig = dev1_driver.GetConfig(resources_to_get)
- LOGGER.info('results_getconfig = {:s}'.format(str(results_getconfig)))
-
- time.sleep(1)
-
- dev1_driver.Disconnect()
+ #resources_to_get = [RESOURCE_INTERFACES]
+ #LOGGER.info('resources_to_get = {:s}'.format(str(resources_to_get)))
+ #results_getconfig = dev1_driver.GetConfig(resources_to_get)
+ #LOGGER.info('results_getconfig = {:s}'.format(str(results_getconfig)))
+
+ #resources_to_delete = [
+ # interface('ethernet-1/49', 0, '192.168.11.1', 30, True),
+ #routing_policy('all',True)
+ #]
+
+ #LOGGER.info('resources_to_delete = {:s}'.format(str(resources_to_delete)))
+ #results_deleteconfig = dev1_driver.DeleteConfig(resources_to_delete)
+ #LOGGER.info('results_deleteconfig = {:s}'.format(str(results_deleteconfig)))
+
+ #resources_to_get = [RESOURCE_INTERFACES]
+ #LOGGER.info('resources_to_get = {:s}'.format(str(resources_to_get)))
+ #results_getconfig = dev1_driver.GetConfig(resources_to_get)
+ #LOGGER.info('results_getconfig = {:s}'.format(str(results_getconfig)))
+
+ #time.sleep(1)
+ #dev1_driver.Disconnect()
+ return 0
+
+
+