""" Description: This test is to check onos set configuration and flows with ovsdb connection. List of test cases: CASE1: Compile ONOS and push it to the test machines CASE2: Test ovsdb connection and tearDown CASE3: Test default br-int configuration and vxlan port CASE4: Test default openflow configuration CASE5: Test default flows CASE6: Configure Network Subnet Port CASE7: Test host go online and ping each other CASE8: Clear ovs configuration and host configuration zhanghaoyu7@huawei.com """ import os class FUNCovsdbtest: def __init__( self ): self.default = '' def CASE1( self, main ): """ CASE1 is to compile ONOS and push it to the test machines Startup sequence: cell onos-verify-cell NOTE: temporary - onos-remove-raft-logs onos-uninstall start mininet git pull mvn clean install onos-package onos-install -f onos-wait-for-start start cli sessions start ovsdb start vtn apps """ import os main.log.info( "ONOS Single node start " + "ovsdb test - initialization" ) main.case( "Setting up test environment" ) main.caseExplanation = "Setup the test environment including " +\ "installing ONOS, start ONOS." # load some variables from the params file PULLCODE = False if main.params[ 'GIT' ][ 'pull' ] == 'True': PULLCODE = True gitBranch = main.params[ 'GIT' ][ 'branch' ] cellName = main.params[ 'ENV' ][ 'cellName' ] ipList = os.getenv( main.params[ 'CTRL' ][ 'ip1' ] ) OVSDB1Ip = os.getenv( main.params[ 'OVSDB' ][ 'ip1' ] ) OVSDB2Ip = os.getenv( main.params[ 'OVSDB' ][ 'ip2' ] ) main.step( "Create cell file" ) cellAppString = main.params[ 'ENV' ][ 'cellApps' ] main.ONOSbench.createCellFile( main.ONOSbench.ip_address, cellName, main.OVSDB1.ip_address, cellAppString, ipList ) main.step( "Applying cell variable to environment" ) cellResult = main.ONOSbench.setCell( cellName ) verifyResult = main.ONOSbench.verifyCell() main.log.info( "Removing raft logs" ) main.ONOSbench.onosRemoveRaftLogs() main.CLIs = [] main.nodes = [] main.numCtrls= 1 for i in range( 1, main.numCtrls + 1 ): try: main.CLIs.append( getattr( main, 'ONOScli' + str( i ) ) ) main.nodes.append( getattr( main, 'ONOS' + str( i ) ) ) ipList.append( main.nodes[ -1 ].ip_address ) except AttributeError: break main.log.info( "Uninstalling ONOS" ) for node in main.nodes: main.ONOSbench.onosUninstall( node.ip_address ) # Make sure ONOS process is not running main.log.info( "Killing any ONOS processes" ) killResults = main.TRUE for node in main.nodes: killed = main.ONOSbench.onosKill( node.ip_address ) killResults = killResults and killed cleanInstallResult = main.TRUE gitPullResult = main.TRUE main.step( "Git checkout and pull" + gitBranch ) if PULLCODE: main.ONOSbench.gitCheckout( gitBranch ) gitPullResult = main.ONOSbench.gitPull() # values of 1 or 3 are good utilities.assert_lesser( expect=0, actual=gitPullResult, onpass="Git pull successful", onfail="Git pull failed" ) main.ONOSbench.getVersion( report=True ) main.step( "Using mvn clean install" ) cleanInstallResult = main.TRUE if PULLCODE and gitPullResult == main.TRUE: cleanInstallResult = main.ONOSbench.cleanInstall() else: main.log.warn( "Did not pull new code so skipping mvn" + "clean install" ) utilities.assert_equals( expect=main.TRUE, actual=cleanInstallResult, onpass="MCI successful", onfail="MCI failed" ) main.step( "Creating ONOS package" ) packageResult = main.ONOSbench.onosPackage() utilities.assert_equals( expect=main.TRUE, actual=packageResult, onpass="Successfully created ONOS package", onfail="Failed to create ONOS package" ) main.step( "Installing ONOS package" ) onosInstallResult = main.ONOSbench.onosInstall( options="-f", node=main.nodes[0].ip_address ) utilities.assert_equals( expect=main.TRUE, actual=onosInstallResult, onpass="ONOS install successful", onfail="ONOS install failed" ) main.step( "Checking if ONOS is up yet" ) print main.nodes[0].ip_address for i in range( 2 ): onos1Isup = main.ONOSbench.isup( main.nodes[0].ip_address ) if onos1Isup: break utilities.assert_equals( expect=main.TRUE, actual=onos1Isup, onpass="ONOS startup successful", onfail="ONOS startup failed" ) main.log.step( "Starting ONOS CLI sessions" ) print main.nodes[0].ip_address cliResults = main.ONOScli1.startOnosCli( main.nodes[0].ip_address ) utilities.assert_equals( expect=main.TRUE, actual=cliResults, onpass="ONOS cli startup successful", onfail="ONOS cli startup failed" ) main.step( "App Ids check" ) appCheck = main.ONOScli1.appToIDCheck() if appCheck !=main.TRUE: main.log.warn( main.CLIs[0].apps() ) main.log.warn( main.CLIs[0].appIDs() ) utilities.assert_equals( expect=main.TRUE, actual=appCheck, onpass="App Ids seem to be correct", onfail="Something is wrong with app Ids" ) if cliResults == main.FALSE: main.log.error( "Failed to start ONOS,stopping test" ) main.cleanup() main.exit() main.step( "Install onos-ovsdatabase" ) installResults = main.ONOScli1.activateApp( "org.onosproject.ovsdb" ) utilities.assert_equals( expect=main.TRUE, actual=installResults, onpass="Install onos-ovsdatabase successful", onfail="Install onos-ovsdatabase failed" ) main.step( "Install onos-app-vtnrsc" ) installResults = main.ONOScli1.activateApp( "org.onosproject.vtnrsc" ) utilities.assert_equals( expect=main.TRUE, actual=installResults, onpass="Install onos-app-vtnrsc successful", onfail="Install onos-app-vtnrsc failed" ) main.step( "Install onos-app-vtn" ) installResults = main.ONOScli1.activateApp( "org.onosproject.vtn" ) utilities.assert_equals( expect=main.TRUE, actual=installResults, onpass="Install onos-app-vtn successful", onfail="Install onos-app-vtn failed" ) main.step( "Install onos-app-vtnweb" ) installResults = main.ONOScli1.activateApp( "org.onosproject.vtnweb" ) utilities.assert_equals( expect=main.TRUE, actual=installResults, onpass="Install onos-app-vtnweb successful", onfail="Install onos-app-vtnweb failed" ) def CASE2( self, main ): """ Test ovsdb connection and teardown """ import os,sys import re import time main.case( "Test ovsdb connection and teardown" ) main.caseExplanation = "Test ovsdb connection create and delete" +\ " over ovsdb node and onos node " ctrlip = os.getenv( main.params[ 'CTRL' ][ 'ip1' ] ) ovsdbport = main.params[ 'CTRL' ][ 'ovsdbport' ] delaytime = main.params[ 'TIMER' ][ 'delaytime' ] main.step( "Set ovsdb node manager" ) assignResult = main.OVSDB1.setManager( ip=ctrlip, port=ovsdbport, delaytime=delaytime ) stepResult = assignResult utilities.assert_equals( expect=main.TRUE, actual=stepResult, onpass="Set ovsdb node manager sucess", onfail="Set ovsdb node manager failed" ) main.step( "Check ovsdb node manager is " + str( ctrlip ) ) response = main.OVSDB1.getManager() if re.search( ctrlip, response ): stepResult = main.TRUE else: stepResult = main.FALSE utilities.assert_equals( expect=main.TRUE, actual=stepResult, onpass="Check ovsdb node manager is " + str( response ) , onfail="Check ovsdb node manager failed" ) main.step( "Delete ovsdb node manager" ) deleteResult = main.OVSDB1.delManager( delaytime=delaytime ) stepResult = deleteResult utilities.assert_equals( expect=main.TRUE, actual=stepResult, onpass="ovsdb node delete manager sucess", onfail="ovsdb node delete manager failed" ) main.step( "Check ovsdb node delete manager " + str( ctrlip ) ) response = main.OVSDB1.getManager() if not re.search( ctrlip, response ): stepResult = main.TRUE else: stepResult = main.FALSE utilities.assert_equals( expect=main.TRUE, actual=stepResult, onpass="Check ovsdb node delete manager sucess", onfail="Check ovsdb node delete manager failed" )