1. Sanity check MD-CLI show command#

#!/usr/bin/env python3

# Import the required modules
from pysros.management import connect
from pysros.pprint import Table   

# Our function from earlier that helps us print pretty show commands
def print_table(rows, cols, title):
    """
    This function uses the Table class from the pysros.pprint module to print a table
    in the style of the Nokia SROS CLI.
    It takes a list of lists as the rows.
    It takes a list of tuples which are the column names and their width.
    It takes a title string which is printed out at the top of the table.
    """
    # Initialize the Table object with the heading and columns.
    table = Table(title, cols)
    table.print(rows) # Print the table

# This is where we will place the functions we create to collect and process each set of data

def main():
    # Start the session
    c = connect()

    # Call all your functions here and collate them into data for the tables

    # Disconnect the session
    c.disconnect()

    # Print the tables here with the data you have collected

# Call the main function
if __name__ == '__main__':
    main()

#!/usr/bin/env python3

from pysros.management import connect
from pysros.pprint import Table


def print_table(rows, cols, title):
    """
    This function uses the Table class from the pysros.pprint module to print a table
    in the style of the Nokia SROS CLI.
    It takes a list of lists as the rows.
    It takes a list of tuples which are the column names and their width.
    It takes a title string which is printed out at the top of the table.
    """
    # Initialize the Table object with the heading and columns.
    table = Table(title, cols)
    table.print(rows) # Print the table

def card(c):
    """
    This function first uses the get_list_keys method to get a list of all the cards in the system.
    It then iterates through the list of cards and checks the operational state of each card.
    If the operational state is not 'in-service', it adds a failure string message within a list
    to the result list. (list of lists). The function returns the result list, which will be empty
    if all provisioned cards are operationally in-service.
    """
    top_path  = "/nokia-state:state/card"
    cards = c.running.get_list_keys(top_path)
    specific_path = '/nokia-state:state/card[slot-number="{}"]/hardware-data'
    result = []
    for card in cards:
        # Check if the card is not operationally in-service
        if c.running.get(specific_path.format(card))['oper-state'].data not in ["in-service", "empty"]:
            # If it isn't then we add to the result list to signify a failure
            result.append(["Card {} is not operationally in-service".format(card)])
    return result

def isis(c):
    """
    This function first uses the get_list_keys method to get a list of all the ISIS interfaces.
    It then iterates through the list of interfaces and checks the operational state of each
    adjacency. If the operational state is not 'up', it adds a message to the result list.
    The function returns the result list, which will be empty if all adjacencies are
    operationally up.
    """
    top_path  = '/nokia-state:state/router[router-name="Base"]/isis[isis-instance="0"]/interface'
    interfaces = c.running.get_list_keys(top_path)
    specific_path = (
        '/nokia-state:state/router[router-name="Base"]/isis[isis-instance="0"]'
        '/interface[interface-name="{}"]/adjacency[adjacency-index="1"]'
    )
    result = []
    for interface in interfaces:
        if interface != "system": # Skip the system interface
            # There could be a LookupError if there is no adjacency index at all
            try:
                # Check if the adjacency is not operationally up
                if c.running.get(specific_path.format(interface))['oper-state'].data != "up":
                    # If it isn't then we add to the result list to signify a failure
                    result.append(["The ISIS adjacency on interface {} is not up".format(interface)])
            except LookupError:
                result.append(["The ISIS adjacency on interface {} is not up".format(interface)])
    return result

def bgp(c):
    """
    This function first uses the get_list_keys method to get a list of all the BGP peers.
    It then iterates through the list of peers and checks the operational state of each peering.
    If the operational state is not 'Established', it adds a message to the result list.
    The function returns the result list, which will be empty if all BGP peerings are Established.
    """
    top_path  = '/nokia-state:state/router[router-name="Base"]/bgp/neighbor'
    peerings = c.running.get_list_keys(top_path)
    specific_path = (
        '/nokia-state:state/router[router-name="Base"]/bgp/neighbor'
        '[ip-address="{}"]/statistics'
    )
    result = []
    for peer in peerings:
        # Check if the peering is not Established
        if c.running.get(specific_path.format(peer))['session-state'].data != "Established":
            # If it isn't then we add to the result list to signify a failure
            result.append(["The BGP peer {} is not Established".format(peer)])
    return result

def cpu(c):
    """
    This function collects the cpu-usage of the businest CPU core over a 5 minute period.
    The collected data is a string, so it is converted to a float and then to an int to show a whole number
    """
    path  = '/nokia-state:state/system/cpu[sample-period="300"]/summary/busiest-core-utilization'
    result = c.running.get(path)['cpu-usage'].data
    return int(float(result))

def memory(c):
    """
    This function collects the memory in use and available from the system
    From this it calculates the memory utilization and formats it the same way as the CPU utilization
    """
    path = "/nokia-state:state/system/memory-pools/summary"
    result = c.running.get(path)
    total_size = result['current-total-size'].data
    available = result['available-memory'].data
    total_mem = total_size + available
    utilization = (total_size / total_mem) * 100
    pretty_utilization = "{}%".format(int(utilization))
    return pretty_utilization

def main():
    # Start the session
    c = connect()

    # Create two lists to hold the results table data and any failure table data
    result_table_data = []
    failure_table_data = []

    # Card data
    card_result = card(c)
    if card_result: # If the returned list is not empty, it means there was a failure
        result_table_data.append(["CARD STATUS", "FAIL"])
        failure_table_data.extend(card_result)
    else: # If the returned list is empty, it means there was no failure
        result_table_data.append(["CARD STATUS", "PASS"])

    # ISIS data
    isis_result = isis(c)
    if isis_result: # If the returned list is not empty, it means there was a failure
        result_table_data.append(["ISIS STATUS", "FAIL"])
        failure_table_data.extend(isis_result)
    else: # If the returned list is empty, it means there was no failure
        result_table_data.append(["ISIS STATUS", "PASS"])

    # BGP data
    bgp_result = bgp(c)
    if bgp_result: # If the returned list is not empty, it means there was a failure
        result_table_data.append(["BGP STATUS", "FAIL"])
        failure_table_data.extend(bgp_result)
    else: # If the returned list is empty, it means there was no failure
        result_table_data.append(["BGP STATUS", "PASS"])

    # CPU data
    result_table_data.append(["BUSIEST CPU CORE OVER 5MIN", "{}%".format(cpu(c))])

    # Memory data
    result_table_data.append(["MEMORY UTILIZATION", memory(c)])

    # Disconnect the session
    c.disconnect()

    # Print the results table
    print_table(result_table_data, [(60, "Test"), (19, "Result")], "Self Test Results")

    # Print the failure table only if there are any failures
    if failure_table_data:
        print("\n")
        print_table(failure_table_data, [(79, "Failures")], "Details Of Failed Tests")

# Call the main function
if __name__ == '__main__':
    main()

#!/usr/bin/env python3

# Import the required modules
from pysros.management import connect
from pysros.pprint import Table   

# Our function from earlier that helps us print pretty show commands
def print_table(rows, cols, title):
    """
    This function uses the Table class from the pysros.pprint module to print a table
    in the style of the Nokia SROS CLI.
    It takes a list of lists as the rows.
    It takes a list of tuples which are the column names and their width.
    It takes a title string which is printed out at the top of the table.
    """
    # Initialize the Table object with the heading and columns.
    table = Table(title, cols)
    table.print(rows) # Print the table

def card(c):
    """
    This function first uses the get_list_keys method to get a list of all the cards in the system.
    It then iterates through the list of cards and checks the operational state of each card.
    If the operational state is not 'in-service', it adds a failure string message within a list
    to the result list. (list of lists). The function returns the result list, which will be empty
    if all provisioned cards are operationally in-service.
    """
    top_path  = "/nokia-state:state/card"
    cards = c.running.get_list_keys(top_path)
    specific_path = '/nokia-state:state/card[slot-number="{}"]/hardware-data'
    result = []
    for card in cards:
        # Check if the card is not operationally in-service
        if c.running.get(specific_path.format(card))['oper-state'].data not in ["in-service", "empty"]:
            # If it isn't then we add to the result list to signify a failure
            result.append(["Card {} is not operationally in-service".format(card)])
    return result

def mda(c):
    """
    This function needs to be a bit different as we have to run get_list_keys twice, once
    to get a list of cards, and again against each card to get a list of MDAs in that card.
    MDAs are also different in that this state data also shows for MDAs that are not
    provisioned so we also filter off anything that has an oper-state of 'empty'
    """
    top_path  = "/nokia-state:state/card"
    cards = c.running.get_list_keys(top_path)
    mda_path = '/nokia-state:state/card[slot-number="{}"]/mda'
    specific_path = '/nokia-state:state/card[slot-number="{}"]/mda[mda-slot="{}"]/hardware-data'
    result = []

    for card in cards:
        mdas = c.running.get_list_keys(mda_path.format(card))
        # Check if the card is not operationally in-service
        for mda in mdas:
            if c.running.get(
                specific_path.format(card, mda)
            )['oper-state'].data not in ["in-service", "empty"]:
                # If it isn't then we add to the result list to signify a failure
                result.append(["MDA {}/{} is not operationally in-service".format(card, mda)])
    return result

def isis(c):
    """
    This function first uses the get_list_keys method to get a list of all the ISIS interfaces.
    It then iterates through the list of interfaces and checks the operational state of each
    adjacency. If the operational state is not 'up', it adds a message to the result list.
    The function returns the result list, which will be empty if all adjacencies are
    operationally up.
    """
    top_path  = '/nokia-state:state/router[router-name="Base"]/isis[isis-instance="0"]/interface'
    interfaces = c.running.get_list_keys(top_path)
    specific_path = (
        '/nokia-state:state/router[router-name="Base"]/isis[isis-instance="0"]'
        '/interface[interface-name="{}"]/adjacency[adjacency-index="1"]'
    )
    result = []
    for interface in interfaces:
        if interface != "system": # Skip the system interface
            # There could be a LookupError if there is no adjacency index at all
            try:
                # Check if the adjacency is not operationally up
                if c.running.get(specific_path.format(interface))['oper-state'].data != "up":
                    # If it isn't then we add to the result list to signify a failure
                    result.append(["The ISIS adjacency on interface {} is not up".format(interface)])
            except LookupError:
                result.append(["The ISIS adjacency on interface {} is not up".format(interface)])
    return result

def mpls(c):
    """
    This function checks that the SR-LFA coverage is 100% for the node-sid topology.
    """
    path = (
        '/nokia-state:state/router[router-name="Base"]/isis[isis-instance="0"]'
        '/sr-lfa-coverage[level-number="2"][multi-topology-id="ipv4"]'
        '[sid-type="node-sid"][protocol-version="ipv4"]'
    )
    data = c.running.get(path)['lfa-covered-percent'].data
    result = []
    if data != 100:
        # If it isn't then we add to the result list to signify a failure
        result.append(["The ISIS SR-LFA coverage is not 100%"])
    return result

def bgp(c):
    """
    This function first uses the get_list_keys method to get a list of all the BGP peers.
    It then iterates through the list of peers and checks the operational state of each peering.
    If the operational state is not 'Established', it adds a message to the result list.
    The function returns the result list, which will be empty if all BGP peerings are Established.
    """
    top_path  = '/nokia-state:state/router[router-name="Base"]/bgp/neighbor'
    peerings = c.running.get_list_keys(top_path)
    specific_path = (
        '/nokia-state:state/router[router-name="Base"]/bgp/neighbor'
        '[ip-address="{}"]/statistics'
    )
    result = []
    for peer in peerings:
        # Check if the peering is not Established
        if c.running.get(specific_path.format(peer))['session-state'].data != "Established":
            # If it isn't then we add to the result list to signify a failure
            result.append(["The BGP peer {} is not Established".format(peer)])
    return result

def svc(c):
    """
    Different service types are in different paths, so for this function we need to check both
    separately and then combine the results into one list. For this example we are just checking
    VPLS and IES services. We also need to check that the IES service is not the internal one
    which is used for the system
    """
    top_path  = "/nokia-state:state/service/{}"
    vpls_services = c.running.get_list_keys(top_path.format("vpls"))
    ies_services = c.running.get_list_keys(top_path.format("ies"))
    specific_path = '/nokia-state:state/service/{}[service-name="{}"]'
    result = []
    for vpls in vpls_services:
        # Check if the service is not up
        if c.running.get(specific_path.format('vpls', vpls))['oper-state'].data != "up":
            # If it isn't then we add to the result list to signify a failure
            result.append(["The VPLS service {} is not up".format(vpls)])
    for ies in ies_services:
        if ies != "_tmnx_InternalIesService": # Skip the internal IES service
            # Check if the service is not up
            if c.running.get(specific_path.format('ies', ies))['oper-state'].data != "up":
                # If it isn't then we add to the result list to signify a failure
                result.append(["The IES service {} is not up".format(ies)])
    return result

def ping(c):
    """
    This function takes a connection object and pings destinations of your choice.
    It uses the c.action method to ping and we extract just the packet loss from the result.
    If any packet loss is seen then we flag a failure by populating the result list and return. 
    """

    ping_path = '/nokia-oper-global:global-operations/ping'
    ping_data = {'destination': '{}',
                    'output-format': 'summary',
                    'interval': "0.01",
                    'timeout': 1}

    ping_dest  = [
                     "10.46.15.22", # PE2
                     "10.46.15.23", # PE3
                     "10.46.15.24"  # PE4
                 ]

    result = []

    for dest in ping_dest:
        ping_data['destination'] = dest
        the_ping = c.action(ping_path, ping_data)
        pktloss = the_ping['results']['summary']['statistics']['packets']['loss'].data
        if pktloss != '0.0':
            # If packet loss isn't 0.0 then we add to the result list to signify a failure
            result.append(["Ping to {} saw packet loss".format(dest)])
    return result

def cpu(c):
    """
    This function collects the cpu-usage of the businest CPU core over a 5 minute period.
    The collected data is a string, so it is converted to a float and then to an int to show a whole number
    """
    path  = '/nokia-state:state/system/cpu[sample-period="300"]/summary/busiest-core-utilization'
    result = c.running.get(path)['cpu-usage'].data
    return int(float(result))

def memory(c):
    """
    This function collects the memory in use and available from the system
    From this it calculates the memory utilization and formats it the same way as the CPU utilization
    """
    path = "/nokia-state:state/system/memory-pools/summary"
    result = c.running.get(path)
    total_size = result['current-total-size'].data
    available = result['available-memory'].data
    total_mem = total_size + available
    utilization = (total_size / total_mem) * 100
    pretty_utilization = "{}%".format(int(utilization))
    return pretty_utilization

def cf3(c):
    """
    This function collects the utilization of the CF3 flash device on the active CPM.
    You need to find out first which CPM is active in order to do this.
    """
    active_cpm = c.running.get("/nokia-state:state/system/active-cpm-slot").data
    flash_path  = '/nokia-state:state/cpm[cpm-slot="{}"]/flash[flash-id="3"]'.format(active_cpm)
    result = c.running.get(flash_path)['percent-used'].data
    return "{}%".format(result)

def fan(c):
    """
    This function first uses the get_list_keys method to get a list of all the fan trays
    It then iterates through the list of trays and gets their speed
    For each tray, the speed is added to the result list
    """
    top_path  = '/nokia-state:state/chassis[chassis-class="router"][chassis-number="1"]/fan'
    fan_trays = c.running.get_list_keys(top_path)
    specific_path = (
        '/nokia-state:state/chassis[chassis-class="router"][chassis-number="1"]/fan[fan-slot="{}"]'
    )
    result = []
    for tray in fan_trays:
        speed = c.running.get(specific_path.format(tray))['speed'].data
        result.append(["FAN TRAY {} SPEED".format(tray), "{}%".format(speed)])
    return result

def main():
    # Start the session
    c = connect()

    # Create two lists to hold the results table data and any failure table data
    result_table_data = []
    failure_table_data = []

    # Card data
    card_result = card(c)
    if card_result: # If the returned list is not empty, it means there was a failure
        result_table_data.append(["CARD STATUS", "FAIL"])
        failure_table_data.extend(card_result)
    else: # If the returned list is empty, it means there was no failure
        result_table_data.append(["CARD STATUS", "PASS"])

    # MDA data
    mda_result = mda(c)
    if mda_result: # If the returned list is not empty, it means there was a failure
        result_table_data.append(["MDA STATUS", "FAIL"])
        failure_table_data.extend(mda_result)
    else: # If the returned list is empty, it means there was no failure
        result_table_data.append(["MDA STATUS", "PASS"])

    # ISIS data
    isis_result = isis(c)
    if isis_result: # If the returned list is not empty, it means there was a failure
        result_table_data.append(["ISIS STATUS", "FAIL"])
        failure_table_data.extend(isis_result)
    else: # If the returned list is empty, it means there was no failure
        result_table_data.append(["ISIS STATUS", "PASS"])

    # BGP data
    bgp_result = bgp(c)
    if bgp_result: # If the returned list is not empty, it means there was a failure
        result_table_data.append(["BGP STATUS", "FAIL"])
        failure_table_data.extend(bgp_result)
    else: # If the returned list is empty, it means there was no failure
        result_table_data.append(["BGP STATUS", "PASS"])

    # MPLS data
    mpls_result = mpls(c)
    if mpls_result: # If the returned list is not empty, it means there was a failure
        result_table_data.append(["MPLS STATUS", "FAIL"])
        failure_table_data.extend(mpls_result)
    else: # If the returned list is empty, it means there was no failure
        result_table_data.append(["MPLS STATUS", "PASS"])

    # svc data
    svc_result = svc(c)
    if svc_result: # If the returned list is not empty, it means there was a failure
        result_table_data.append(["SERVICE STATUS", "FAIL"])
        failure_table_data.extend(svc_result)
    else: # If the returned list is empty, it means there was no failure
        result_table_data.append(["SERVICE STATUS", "PASS"])

    # ping tests
    ping_result = ping(c)
    if ping_result: # If the returned list is not empty, it means there was a failure
        result_table_data.append(["PING TESTS", "FAIL"])
        failure_table_data.extend(ping_result)
    else: # If the returned list is empty, it means there was no failure
        result_table_data.append(["PING TESTS", "PASS"])

    # CPU data
    result_table_data.append(["BUSIEST CPU CORE OVER 5MIN", "{}%".format(cpu(c))])

    # Memory data
    result_table_data.append(["MEMORY UTILIZATION", memory(c)])

    # CF3 flash data
    result_table_data.append(["CF3 UTILIZATION", cf3(c)])

    # Fan data
    # We 'extend' the result table list this time because of multiple fan trays
    result_table_data.extend(fan(c))

    # Disconnect the session
    c.disconnect()

    # Print the results table
    print_table(result_table_data, [(60, "Test"), (19, "Result")], "Self Test Results")

    # Print the failure table only if there are any failures
    if failure_table_data:
        print("\n")
        print_table(failure_table_data, [(79, "Failures")], "Details Of Failed Tests")

# Call the main function
if __name__ == '__main__':
    main()