Help us improve your experience.

Let us know what you think.

Do you have time for a two-minute survey?

Navigation
Guide That Contains This Content
[+] Expand All
[-] Collapse All

    JET Management Application

    sampleManagementApp.py performs the following operations for OpenConfig:

    1. Executes GetDataEncodings RPC

    2. Executes GetModels RPC

    3. Executes Get RPC

    4. Executes Set RPC

    sampleManagementApp.py performs the following operations for management services:

    1. Executes ExecuteOpCommand RPC to get software information

    2. Executes GetEphemeralConfig RPC

    3. Executes EditEphemeral RPC to add an interface

    4. Executes EditEphemeral RPC to delete an interface

    To run the application, update the following in the example:

    • JSD_IP = '10.221.131.129' - With IP address of the device

    • USERNAME = 'foo'

    • PASSWORD = ''

    #!/usr/bin/env python
    #
    #
    #
    #
    
    """
    Sample application for testing the Management services exposed by Junos in
    management_service.proto and openconfig_service.proto files.
    Usage:
    $python sampleManagementApp.py --user <username> --password <password> --address <Router IP Address> --port <JSD Port>
    """
    
    import argparse
    import os
    import logging
    import sys
    
    
    # Import Python GRPC module
    from grpc.beta import implementations
    
    # Import python modules generated from proto files
    import mgd_service_pb2
    import openconfig_service_pb2
    import authentication_service_pb2
    
    # IMPORTANT: Following are the dummy parameters that will be used for testing
    # Please change these parameters for proper testing
    
    # Device details and login credentials
    JSD_IP = '10.221.131.129'
    JSD_PORT = 32767 
    USERNAME = 'foo'
    PASSWORD = ''
    CLIENT_ID = 'test_client'
    _TIMEOUT_SECONDS = 20
    
    # Enable Logging to a file
    #logging.basicConfig(filename=DEFAULT_LOG_FILE_NAME, level=DEFAULT_LOG_LEVEL)
    
    # To display the messages from junos-jet-api package on the screen uncomment the below lines
    """
    myLogHandler = logging.getLogger()
    myLogHandler.setLevel(logging.INFO)
    logChoice = logging.StreamHandler(sys.stdout)
    logChoice.setLevel(logging.DEBUG)
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    logChoice.setFormatter(formatter)
    myLogHandler.addHandler(logChoice)
    """
    
    def EstablishChannel(args):
        channel = implementations.insecure_channel(args.address, args.port)
        login_stub = authentication_service_pb2.beta_create_Login_stub(channel)
        login_request = authentication_service_pb2.LoginRequest(user_name=args.user, password=args.password, client_id=args.client_id)
        login_response = login_stub.LoginCheck(login_request, _TIMEOUT_SECONDS)
        return channel
    
    def SetRPCs(channel):
        # Example to configure under interface with json configuration
        stub = openconfig_service_pb2.beta_create_OpenconfigRpcApi_stub(channel)
    
      jsonCfgValue = """ <configuration-json>
        {
        "configuration" : {
            "interfaces" : {
                "interface" : [
                {
                    "name" : "ge-1/1/2",
                    "unit" : [
                    {
                        "name" : "0",
                        "description" : "foo"
                    }
                    ]
                }
                ]
            }
       }
        }</configuration-json> """
        jsonCfgRequest = openconfig_service_pb2.SetRequest.ConfigOperationList(operation_id = "jcfg", operation = 0, path = "/", value = jsonCfgValue)
        request = openconfig_service_pb2.SetRequest(request_id = 1000, encoding = 1, config_operation = [jsonCfgRequest])
        response = stub.Set(request, _TIMEOUT_SECONDS)
        print response
    
        # Example to delete confiuration under interface with json
        jsonDelCfgValue = """ <configuration-json>
        {
        "configuration" : {
                "interfaces" : {
                    "interface" : [
                        {
                            "name" : "ge-1/1/2",
                            "@": {
                                "operation": "delete"
                            }
                        }
                    ]
                }
            }
        }
        </configuration-json> """
        jsonDelCfgRequest = openconfig_service_pb2.SetRequest.ConfigOperationList(operation_id = "jdcfg", operation = 0, path = "/", value = jsonDelCfgValue)
        request = openconfig_service_pb2.SetRequest(request_id = 1000, encoding = 1, config_operation = [jsonDelCfgRequest])
        response = stub.Set(request, _TIMEOUT_SECONDS)
        print response
    
        # Example to configure under interface with xml
    
    def ManagementTests(channel):
        # All Mangement and Openconfig RPCs will be tested here. The function will
        # make use of the channel that is passed to it to execute the RPCs
        # Create a stub for Openconfig RPCs
        metadata = [(b'user', b'foo'), (b'password', b'bar')]
        stub = openconfig_service_pb2.beta_create_OpenconfigRpcApi_stub(channel)
    
        # Execute GetDataEncodings RPC
        getEncodingsRequest = openconfig_service_pb2.GetDataEncodingsRequest(request_id = 1000)
        response = stub.GetDataEncodings(getEncodingsRequest, _TIMEOUT_SECONDS, metadata=metadata)
        print response
    
        # Execute GetModels RPC
        getModelsRequest = openconfig_service_pb2.GetModelsRequest(request_id = 1000)
        response = stub.GetModels(getModelsRequest, _TIMEOUT_SECONDS, metadata=metadata)
        print response
    
        # Execute Get RPC
        getRequest = openconfig_service_pb2.GetRequestList(operation_id = "getReqconf", operation = 1, path = "/configuration")
        request = openconfig_service_pb2.GetRequest(request_id = 1000, encoding = 0, get_request = [getRequest])
        response = stub.Get(request, _TIMEOUT_SECONDS, metadata=metadata)
        print response
    
        # Execute Set RPC
        SetRPCs(channel)
    
        # Create a stub for Management RPC
        stub = mgd_service_pb2.beta_create_ManagementRpcApi_stub(channel)
    
        # Execute ExecuteOpCommand RPC
        executeOpCommandXMLrequest = mgd_service_pb2.ExecuteOpCommandRequest(xml_command = "<get-software-information></get-software-information>",
                             out_format = 0, request_id = 1000)
    
        for response in stub.ExecuteOpCommand(executeOpCommandXMLrequest, _TIMEOUT_SECONDS, metadata=metadata):
       print response
    
        # Execute GetEphemeralConfig RPC
        ephCfgRequest = mgd_service_pb2.EphConfigRequestList (operation_id = "getEphConf", path = "/configuration")
        request = mgd_service_pb2.GetEphemeralConfigRequest (request_id = 1000, encoding = 0, eph_config_requests = [ephCfgRequest])
        response = stub.GetEphemeralConfig (request, _TIMEOUT_SECONDS, metadata=metadata)
        print response
    
        # Execute EditEphemeral RPC
        xmlCfgValue = """ <configuration>
                         <interfaces>
                         <interface>
                           <name>ge-1/0/0</name>
                           <unit>
                            <name>0</name>
                            <description>foo</description>
                           </unit>
                          </interface>
                         </interfaces>
                       </configuration>"""
        xmlCfgRequest = mgd_service_pb2.EditEphemeralConfigRequest.ConfigOperationList(operation_id = "xmlEditEph", operation = 0, path = "/", xml_config = xmlCfgValue)
        request = mgd_service_pb2.EditEphemeralConfigRequest (request_id = 1000, eph_config_operations = [xmlCfgRequest])
        response = stub.EditEphemeralConfig(request, _TIMEOUT_SECONDS)
        print response
    
        xmlCfgDelValue = """ <configuration>
                         <interfaces>
                          <interface delete="delete">
                           <name>ge-1/0/0</name>
                          </interface>
                         </interfaces>
                       </configuration>"""
        xmlCfgDelRequest = mgd_service_pb2.EditEphemeralConfigRequest.ConfigOperationList(operation_id = "xmlDelEph", operation = 0, path = "/", xml_config = xmlCfgDelValue)
        request = mgd_service_pb2.EditEphemeralConfigRequest (request_id = 1000, eph_config_operations = [xmlCfgDelRequest])
        response = stub.EditEphemeralConfig(request, _TIMEOUT_SECONDS)
        print response
    
    def Main():
        parser = argparse.ArgumentParser(prog=os.path.basename(__file__), description='Sample JET application')
        parser.add_argument("--user", nargs='?', help="Username for authentication by JSD (default:%(default)s)",
                            type=str, default=USERNAME)
        parser.add_argument("--password", nargs='?', help="Password for authentication by JSD (default:%(default)s",
                            type=str, default=PASSWORD)
        parser.add_argument("--address", nargs='?', help="Address of the JSD server. (default: %(default)s)",
                            type=str, default=JSD_IP)
        parser.add_argument("--port", nargs='?', help="Port number of the JSD request-response server. (default: %(default)s)",
                            type=int, default=JSD_PORT)
        parser.add_argument("--client-id", nargs='?', help="Client Id to uniquely identify the client (default: %(default)s)",
             type=str, default=CLIENT_ID)
        args = parser.parse_args()
    
        #Establish a connection and authenticate the channel
        channel = EstablishChannel(args)
    
        ManagementTests(channel)
    
    if __name__ == '__main__':
        Main()

    Modified: 2018-02-22