Commit d10cfa05 authored by Mohamad Rahhal's avatar Mohamad Rahhal
Browse files

Device component - gNMI Nokia SR Linux Driver: 

- Activated log lines in InterfaceHandler
- Activated relevant interface management test instructions
parent 8f98c670

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

+4 −60
Original line number Diff line number Diff line
@@ -62,12 +62,12 @@ class InterfaceHandler(_Handler): 
        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)))

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

            LOGGER.info('json_interface = {:s}'.format(json.dumps(json_interface)))



            interface = {}
@@ -111,7 +111,7 @@ class InterfaceHandler(_Handler): 
            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)))

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



                subinterface = {}
@@ -156,7 +156,7 @@ class InterfaceHandler(_Handler): 


                #ipv4_addresses = []

                for json_ipv4_address in json_ipv4_address_list:

                    #LOGGER.info('json_ipv4_address = {:s}'.format(json.dumps(json_ipv4_address)))

                    LOGGER.info('json_ipv4_address = {:s}'.format(json.dumps(json_ipv4_address)))



                    STATE_FIELDS = ('state', 'openconfig-if-ip:state', 'ociip:state')

                    json_ipv4_address_state = dict_get_first(json_ipv4_address, STATE_FIELDS, default={})
@@ -190,59 +190,3 @@ class InterfaceHandler(_Handler): 
            response.append(('/interface[{:s}]'.format(interface['name']), interface))



        return response
 


    def parse_counters(self, json_data : Dict) -> List[Tuple[str, Dict[str, Any]]]:

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



            interface = {}



            NAME_FIELDS = ('name', 'openconfig-interface:name', 'oci:name')

            interface_name = dict_get_first(json_interface, NAME_FIELDS)

            if interface_name is None: continue

            interface['name'] = interface_name



            STATE_FIELDS = ('state', 'openconfig-interface:state', 'oci:state')

            json_state = dict_get_first(json_interface, STATE_FIELDS, default={})



            COUNTERS_FIELDS = ('counters', 'openconfig-interface:counters', 'oci:counters')

            json_counters = dict_get_first(json_state, COUNTERS_FIELDS, default={})



            IN_PKTS_FIELDS = ('in-pkts', 'openconfig-interface:in-pkts', 'oci:in-pkts')

            interface_in_pkts = dict_get_first(json_counters, IN_PKTS_FIELDS)

            if interface_in_pkts is not None: interface['in-pkts'] = int(interface_in_pkts)



            IN_OCTETS_FIELDS = ('in-octets', 'openconfig-interface:in-octets', 'oci:in-octets')

            interface_in_octets = dict_get_first(json_counters, IN_OCTETS_FIELDS)

            if interface_in_octets is not None: interface['in-octets'] = int(interface_in_octets)



            IN_ERRORS_FIELDS = ('in-errors', 'openconfig-interface:in-errors', 'oci:in-errors')

            interface_in_errors = dict_get_first(json_counters, IN_ERRORS_FIELDS)

            if interface_in_errors is not None: interface['in-errors'] = int(interface_in_errors)



            OUT_OCTETS_FIELDS = ('out-octets', 'openconfig-interface:out-octets', 'oci:out-octets')

            interface_out_octets = dict_get_first(json_counters, OUT_OCTETS_FIELDS)

            if interface_out_octets is not None: interface['out-octets'] = int(interface_out_octets)



            OUT_PKTS_FIELDS = ('out-pkts', 'openconfig-interface:out-pkts', 'oci:out-pkts')

            interface_out_pkts = dict_get_first(json_counters, OUT_PKTS_FIELDS)

            if interface_out_pkts is not None: interface['out-pkts'] = int(interface_out_pkts)



            OUT_ERRORS_FIELDS = ('out-errors', 'openconfig-interface:out-errors', 'oci:out-errors')

            interface_out_errors = dict_get_first(json_counters, OUT_ERRORS_FIELDS)

            if interface_out_errors is not None: interface['out-errors'] = int(interface_out_errors)



            OUT_DISCARDS_FIELDS = ('out-discards', 'openconfig-interface:out-discards', 'oci:out-discards')

            interface_out_discards = dict_get_first(json_counters, OUT_DISCARDS_FIELDS)

            if interface_out_discards is not None: interface['out-discards'] = int(interface_out_discards)



            #LOGGER.info('[parse_counters] interface = {:s}'.format(str(interface)))



            if len(interface) == 0: continue

            response.append(('/interface[{:s}]'.format(interface['name']), interface))



        return response

src/device/tests/test_gnmi_nokia_srlinux.py

+49 −33
Original line number Diff line number Diff line
@@ -16,9 +16,9 @@ import logging, os, sys, time 
from typing import Dict, Tuple

os.environ['DEVICE_EMULATED_ONLY'] = 'YES'

from device.service.drivers.gnmi_nokia_srlinux.GnmiNokiaSrLinuxDriver import GnmiNokiaSrLinuxDriver # pylint: disable=wrong-import-position

#from device.service.driver_api._Driver import (

#    RESOURCE_ENDPOINTS, RESOURCE_INTERFACES, RESOURCE_NETWORK_INSTANCES, RESOURCE_ROUTING_POLICIES, RESOURCE_SERVICES

#)

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__)
@@ -54,41 +54,51 @@ def network_instance_interface(ni_name, if_name, sif_index) -> Tuple[str, Dict]: 
    str_data = {'name': ni_name, 'if_name': if_name, 'sif_index': sif_index}

    return str_path, str_data



def main():

def test_gnmi_nokia_srlinux():

    driver_settings = {

        'protocol': 'gnmi',

        'username': 'admin',

        'password': 'NokiaSrl1!',

        'use_tls' : True,

    }

    driver = GnmiNokiaSrLinuxDriver('172.100.100.102', 57400, **driver_settings)

    driver.Connect()

    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_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 = driver.GetConfig(resources_to_get)

    #LOGGER.info('results_getconfig = {:s}'.format(str(results_getconfig)))

    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 = [

    resources_to_set = [

        #network_instance('test-svc', 'L3VRF'),

    #

    #    interface('ethernet-1/1', 0, '172.16.0.1', 24, True),



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



        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 = driver.SetConfig(resources_to_set)

    #LOGGER.info('results_setconfig = {:s}'.format(str(results_setconfig)))

    ]

    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'),
@@ -100,16 +110,22 @@ def main(): 
        interface('ethernet-1/1', 0, '172.16.1.1', 24, True),

        interface('ethernet-1/2', 0, '172.0.0.2', 24, True),

    

        network_instance('20f66fb5', 'L3VRF'),

        #network_instance('20f66fb5', 'L3VRF'),

    ]

    LOGGER.info('resources_to_delete = {:s}'.format(str(resources_to_delete)))

    results_deleteconfig = driver.DeleteConfig(resources_to_delete)

    results_deleteconfig = dev1_driver.DeleteConfig(resources_to_delete)

    LOGGER.info('results_deleteconfig = {:s}'.format(str(results_deleteconfig)))



    time.sleep(1)

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



    driver.Disconnect()

    return 0

    time.sleep(1)



if __name__ == '__main__':

    sys.exit(main())
 
    dev1_driver.Disconnect()