TestON/tests/USECASE/dependencies/sdnipBaseFunction.py

class SdnBase:
    def __init__(self):
        self.default = ''
    def initSetup( self ):
        import json
        import time
        import os
        from operator import eq
        main.case( "Setting up ONOS environment" )
        try:
            from tests.dependencies.ONOSSetup import ONOSSetup
            main.testSetUp = ONOSSetup()
        except Exception:
            main.log.error( "ONOSSetup not found. exiting the test" )
            main.exit()

        main.testSetUp.envSetup()
        main.apps = main.params[ 'ENV' ][ 'appString' ]
        cellName = main.params[ 'ENV' ][ 'cellName' ]




        main.step( "Copying config files" )
        src = os.path.dirname( main.testFile ) + "/network-cfg.json"
        dst = main.ONOSbench.home + "/tools/package/config/network-cfg.json"
        status = main.ONOSbench.scp( main.ONOSbench, src, dst, direction="to" )
        utilities.assert_equals( expect=main.TRUE,
                                 actual=status,
                                 onpass="Copy config file succeeded",
                                 onfail="Copy config file failed" )
        main.testSetUp.ONOSSetUp( main.Mininet, cellName=cellName )

        main.step( "Checking if ONOS CLI is ready for issuing commands" )
        for i in range( 10 ):
            ready = True
            for cli in main.CLIs:
                output = cli.summary()
                if not output:
                    ready = False
            if ready:
                break
            time.sleep( 30 )
        utilities.assert_equals( expect=True, actual=ready,
                                 onpass="ONOS summary command succeded",
                                 onfail="ONOS summary command failed" )

        if not ready:
            main.log.error( "ONOS startup failed!" )
            main.cleanup()
            main.exit()

    def pToPIntentTest( self, intentExpectedNum ):
        '''
        point-to-point intents test for each BGP peer and BGP speaker pair
        '''
        import time
        main.case( "Check point-to-point intents" )
        main.log.info( "There are %s BGP peers in total "
                       % main.params[ 'config' ][ 'peerNum' ] )
        main.step( "Check P2P intents number from ONOS CLI" )

        getIntentsResult = main.ONOScli1.intents( jsonFormat=True )
        bgpIntentsActualNum = \
            main.QuaggaCliSpeaker1.extractActualBgpIntentNum( getIntentsResult )
        bgpIntentsExpectedNum = int( main.params[ 'config' ][ 'peerNum' ] ) * intentExpectedNum
        if bgpIntentsActualNum != bgpIntentsExpectedNum:
            time.sleep( int( main.params['timers']['RouteDelivery'] ) )
            getIntentsResult = main.ONOScli1.intents( jsonFormat=True )
            bgpIntentsActualNum = \
                main.QuaggaCliSpeaker1.extractActualBgpIntentNum( getIntentsResult )
        main.log.info( "bgpIntentsExpected num is:" )
        main.log.info( bgpIntentsExpectedNum )
        main.log.info( "bgpIntentsActual num is:" )
        main.log.info( bgpIntentsActualNum )
        utilities.assert_equals( expect=bgpIntentsExpectedNum,
                                 actual=bgpIntentsActualNum,
                                 onpass="PointToPointIntent Intent Num is correct!",
                                 onfail="PointToPointIntent Intent Num is wrong!" )

    def routeAndIntentCheck( self, allRoutesExpected, routeIntentsExpectedNum ):
        '''
        routes and intents check to all BGP peers
        '''
        import time
        getRoutesResult = main.ONOScli1.routes( jsonFormat=True )
        allRoutesActual = \
            main.QuaggaCliSpeaker1.extractActualRoutesMaster( getRoutesResult )
        allRoutesStrExpected = str( sorted( allRoutesExpected ) )
        allRoutesStrActual = str( allRoutesActual ).replace( 'u', "" )
        if allRoutesStrActual != allRoutesStrExpected:
            time.sleep( int( main.params['timers']['RouteDelivery'] ) )
            getRoutesResult = main.ONOScli1.routes( jsonFormat=True )
            allRoutesActual = \
                main.QuaggaCliSpeaker1.extractActualRoutesMaster( getRoutesResult )
            allRoutesStrActual = str( allRoutesActual ).replace( 'u', "" )

        main.step( "Check routes installed" )
        main.log.info( "Routes expected:" )
        main.log.info( allRoutesStrExpected )
        main.log.info( "Routes get from ONOS CLI:" )
        main.log.info( allRoutesStrActual )
        utilities.assertEquals( \
            expect=allRoutesStrExpected, actual=allRoutesStrActual,
            onpass="Routes are correct!",
            onfail="Routes are wrong!" )

        main.step( "Check M2S intents installed" )
        getIntentsResult = main.ONOScli1.intents( jsonFormat=True )
        routeIntentsActualNum = \
            main.QuaggaCliSpeaker1.extractActualRouteIntentNum( getIntentsResult )
        if routeIntentsActualNum != routeIntentsExpectedNum:
            time.sleep( int( main.params['timers']['RouteDelivery'] ) )
            getIntentsResult = main.ONOScli1.intents( jsonFormat=True )
            routeIntentsActualNum = \
                main.QuaggaCliSpeaker1.extractActualRouteIntentNum( getIntentsResult )

        main.log.info( "MultiPointToSinglePoint Intent Num expected is:" )
        main.log.info( routeIntentsExpectedNum )
        main.log.info( "MultiPointToSinglePoint Intent NUM Actual is:" )
        main.log.info( routeIntentsActualNum )
        utilities.assertEquals( \
            expect=routeIntentsExpectedNum,
            actual=routeIntentsActualNum,
            onpass="MultiPointToSinglePoint Intent Num is correct!",
            onfail="MultiPointToSinglePoint Intent Num is wrong!" )

        main.step( "Check whether all flow status are ADDED" )
        flowCheck = utilities.retry( main.ONOScli1.checkFlowsState,
                                     main.FALSE,
                                     kwargs={'isPENDING':False},
                                     attempts=10 )
        utilities.assertEquals( \
            expect=main.TRUE,
            actual=flowCheck,
            onpass="Flow status is correct!",
            onfail="Flow status is wrong!" )

    def linkUpDownCheck( self, link1Peer, link2Peer, link3Peer,
                         link1RouteNum, link1IntentNum,
                         link2RouteNum, link2IntentNum,
                         link3RouteNum, link3IntentNum,
                         speakers, hosts, upOrDown ):
        '''
        Cut/Recover links to peers one by one, check routes/intents
        upOrDown - "up" or "down"
        '''
        import time
        main.case( "Bring " + upOrDown + " links and check routes/intents" )
        main.step( "Bring " + upOrDown + " the link between sw32 and " + link1Peer )
        linkResult1 = main.Mininet.link( END1="sw32", END2=link1Peer,
                                         OPTION=upOrDown )
        utilities.assert_equals( expect=main.TRUE,
                                 actual=linkResult1,
                                 onpass="Bring down link succeeded!",
                                 onfail="Bring down link failed!" )

        if linkResult1 == main.TRUE:
            time.sleep( int( main.params[ 'timers' ][ 'RouteDelivery' ] ) )
            main.Functions.checkRouteNum( main, link1RouteNum )
            main.Functions.checkM2SintentNum( main, link1IntentNum )
        else:
            main.log.error( "Bring " + upOrDown + " link failed!" )
            main.cleanup()
            main.exit()

        main.step( "Bring " + upOrDown + " the link between sw8 and " + link2Peer )
        linkResult2 = main.Mininet.link( END1="sw8", END2=link2Peer,
                                         OPTION=upOrDown )
        utilities.assert_equals( expect=main.TRUE,
                                 actual=linkResult2,
                                 onpass="Bring " + upOrDown + " link succeeded!",
                                 onfail="Bring " + upOrDown + " link failed!" )
        if linkResult2 == main.TRUE:
            time.sleep( int( main.params[ 'timers' ][ 'RouteDelivery' ] ) )
            main.Functions.checkRouteNum( main, link2RouteNum )
            main.Functions.checkM2SintentNum( main, link2IntentNum )
        else:
            main.log.error( "Bring " + upOrDown + " link failed!" )
            main.cleanup()
            main.exit()

        main.step( "Bring " + upOrDown + " the link between sw28 and "+ link3Peer )
        linkResult3 = main.Mininet.link( END1="sw28", END2=link3Peer,
                                         OPTION=upOrDown )
        utilities.assert_equals( expect=main.TRUE,
                                 actual=linkResult3,
                                 onpass="Bring " + upOrDown + " link succeeded!",
                                 onfail="Bring " + upOrDown + " link failed!" )
        if linkResult3 == main.TRUE:
            time.sleep( int( main.params[ 'timers' ][ 'RouteDelivery' ] ) )
            main.Functions.checkRouteNum( main, link3RouteNum )
            main.Functions.checkM2SintentNum( main, link3IntentNum )
        else:
            main.log.error( "Bring " + upOrDown + " link failed!" )
            main.cleanup()
            main.exit()

        main.step( "Check whether all flow status are ADDED" )
        flowCheck = utilities.retry( main.ONOScli1.checkFlowsState,
                                     main.FALSE,
                                     kwargs={'isPENDING':False},
                                     attempts=10 )
        utilities.assert_equals( expect=main.TRUE,
                                 actual=flowCheck,
                                 onpass="Flow status is correct!",
                                 onfail="Flow status is wrong!" )

        # Ping test
        main.Functions.pingSpeakerToPeer( main, speakers=[ speakers ],
                       peers=[ link1Peer, link2Peer, link3Peer ],
                       expectAllSuccess=False )
        main.Functions.pingHostToHost( main,
                        hosts=hosts,
                        expectAllSuccess=False )