TestON/tests/CHO/CHOtest/dependencies/CHOtestFunctions.py

"""
Wrapper functions for CHOtest
Author: you@onlab.us
"""

def __init__( self ):
    self.default = ''

def installHostIntents():
    """
    Install one host intent for each combination of hosts in the topology
    """
    import itertools
    import time

    hostCombos = list( itertools.combinations( main.hostMACs, 2 ) )
    intentIdList = []
    for i in xrange( 0, len( hostCombos ), int( main.numCtrls ) ):
        pool = []
        for cli in main.CLIs:
            if i >= len( hostCombos ):
                break
            t = main.Thread( target=cli.addHostIntent,
                             threadID=main.threadID,
                             name="addHostIntent",
                             args=[hostCombos[i][0],
                                   hostCombos[i][1]])
            pool.append(t)
            t.start()
            i = i + 1
            main.threadID = main.threadID + 1
        for thread in pool:
            thread.join()
            intentIdList.append( thread.result )

    return intentIdList

def installPointIntents():
    """
    Install one point intent for each permutation of devices in the topology
    """
    import itertools
    import time

    if main.prefix == 2:
        # Spine-leaf topology is a special case
        for i in range( len( main.hostMACs ) ):
            main.MACsDict[ main.deviceDPIDs[ i+10 ] ] = main.hostMACs[ i ].split('/')[0]
        deviceCombos = list( itertools.permutations( main.deviceDPIDs[10:], 2 ) )
    else:
        deviceCombos = list( itertools.permutations( main.deviceDPIDs, 2 ) )
    intentIdList = []
    time1 = time.time()
    for i in xrange( 0, len( deviceCombos ), int( main.numCtrls ) ):
        pool = []
        for cli in main.CLIs:
            if i >= len( deviceCombos ):
                break
            t = main.Thread( target=cli.addPointIntent,
                             threadID=main.threadID,
                             name="addPointIntent",
                             args=[ deviceCombos[i][0],
                                    deviceCombos[i][1],
                                    1, 1, '',
                                    main.MACsDict.get( deviceCombos[i][0] ),
                                    main.MACsDict.get( deviceCombos[i][1] ) ] )
            pool.append(t)
            t.start()
            i = i + 1
            main.threadID = main.threadID + 1
        for thread in pool:
            thread.join()
            intentIdList.append( thread.result )
    time2 = time.time()
    main.log.info("Time taken for adding point intents: %2f seconds" %( time2 - time1 ) )

    return intentIdList

def checkIntents():
    """
    Check if all the intents are in INSTALLED state
    """
    import time

    intentResult = main.TRUE
    for i in range( main.intentCheck ):
        if i != 0:
            main.log.warn( "Verification failed. Retrying..." )
        main.log.info("Waiting for onos to install intents...")
        time.sleep( main.checkIntentsDelay )

        intentResult = main.TRUE
        for e in range(int(main.numCtrls)):
            main.log.info( "Checking intents on CLI %s" % (e+1) )
            intentResultIndividual = main.CLIs[e].checkIntentState( intentsId=main.intentIds )
            if not intentResultIndividual:
                main.log.warn( "Not all intents installed on ONOS%s" % (e+1) )
            intentResult = intentResult and intentResultIndividual
        if intentResult:
            break
    if not intentResult:
        main.log.info( "**** Intent Summary ****\n" + str(main.ONOScli1.intents( jsonFormat=False, summary=True)) )

    return intentResult

def checkPingall( protocol="IPv4" ):
    """
    Verify ping across all hosts
    """
    import time

    pingResult = main.TRUE
    for i in range( main.numPings ):
        if i != 0:
            main.log.warn( "Pingall failed. Retrying..." )
            main.log.info( "Giving ONOS some time...")
            time.sleep( main.pingSleep )

        pingResult = main.Mininet1.pingall( protocol=protocol, timeout=main.pingTimeout )
        if pingResult:
            break

    return pingResult

def checkLinkEvents( linkEvent, linkNum ):
    """
    Verify link down/up events are correctly discovered by ONOS
    linkNum: the correct link number after link down/up
    """
    import time

    linkResult = main.TRUE
    for i in range( main.linkCheck ):
        if i != 0:
            main.log.warn( "Verification failed. Retrying..." )
            main.log.info( "Giving ONOS some time..." )
            time.sleep( main.linkSleep )

        linkResult = main.TRUE
        for e in range( int( main.numCtrls ) ):
            main.log.info( "Checking link number on ONOS%s" % (e+1) )
            topology_output = main.CLIs[e].topology()
            linkResultIndividual = main.ONOSbench.checkStatus( topology_output,
                                                               main.numMNswitches,
                                                               str( linkNum ) )
            if not linkResultIndividual:
                main.log.warn( "Link %s not discovered by ONOS%s" % ( linkEvent, (e+1) ) )
            linkResult = linkResult and linkResultIndividual
        if linkResult:
            break

    return linkResult