| """ |
| |
| Copyright 2016 Open Networking Foundation (ONF) |
| |
| Please refer questions to either the onos test mailing list at <onos-test@onosproject.org>, |
| the System Testing Plans and Results wiki page at <https://wiki.onosproject.org/x/voMg>, |
| or the System Testing Guide page at <https://wiki.onosproject.org/x/WYQg> |
| |
| TestON is free software: you can redistribute it and/or modify |
| it under the terms of the GNU General Public License as published by |
| the Free Software Foundation, either version 2 of the License, or |
| (at your option) any later version. |
| |
| TestON is distributed in the hope that it will be useful, |
| but WITHOUT ANY WARRANTY; without even the implied warranty of |
| MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
| GNU General Public License for more details. |
| |
| You should have received a copy of the GNU General Public License |
| along with TestON. If not, see <http://www.gnu.org/licenses/>. |
| |
| **** Scripted by Antony Silvester - antony.silvester@huawei.com ****** |
| |
| |
| This Test check the bgp_ls functionality |
| |
| List of test cases: |
| CASE1: Compile ONOS and push it to the test machines |
| CASE2: Discovery the topology using BGPLS |
| CASE3: Addition of new Node to existing topology |
| CASE4: Verification of Links thats is discovered" |
| CASE5: Deletion of Links |
| Case6: Uninstalling the app |
| |
| |
| """ |
| class FUNCbgpls: |
| |
| def __init__( self ): |
| self.default = '' |
| |
| def CASE1( self, main ): |
| """ |
| CASE1 is to compile ONOS and push it to the test machines |
| |
| Startup sequence: |
| cell <name> |
| onos-verify-cell |
| NOTE: temporary - onos-remove-raft-logs |
| onos-uninstall |
| git pull |
| onos-package |
| onos-install -f |
| onos-wait-for-start |
| start cli sessions |
| start BGPLS apps |
| |
| """ |
| import os |
| |
| main.log.info( "ONOS Single node start " + |
| "Scapy Tool - initialization" ) |
| try: |
| from tests.dependencies.ONOSSetup import ONOSSetup |
| main.testSetUp = ONOSSetup() |
| except ImportError: |
| main.log.error( "ONOSSetup not found. exiting the test" ) |
| main.exit() |
| main.testSetUp.envSetupDescription() |
| try: |
| main.cellName = main.params['ENV']['cellName'] |
| main.ipList = os.getenv(main.params['CTRL']['ip1']) |
| main.scapy_ip = os.getenv(main.params['SCAPY']['HOSTNAMES']) |
| |
| main.apps = main.params['ENV']['cellApps'] |
| stepResult = main.testSetUp.envSetup() |
| except Exception as e: |
| main.testSetUp.envSetupException( e ) |
| |
| cliResults = main.testSetUp.ONOSSetUp( main.scapy_ip, main.Cluster, |
| cellName=main.cellName, removeLog=True ) |
| main.step( "App Ids check" ) |
| appCheck = main.Cluster.active( 0 ).CLI.appToIDCheck() |
| |
| if appCheck != main.TRUE: |
| main.log.warn( main.Cluster.active( 0 ).CLI.apps() ) |
| main.log.warn( main.Cluster.active( 0 ).CLI.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() |
| |
| def CASE2( self, main ): |
| """ |
| Discovery the topology using BGPLS |
| """ |
| import os |
| import sys |
| import re |
| import time |
| |
| main.case( "Testcase 2 : Discovery the Network Topology using BGPLS" ) |
| main.Cluster.active( 0 ).CLI.log( "\"testcase2 start\"" ) |
| |
| try: |
| from tests.FUNC.FUNCbgpls.dependencies.Nbdata import BgpLs |
| except ImportError: |
| main.log.exception( "Something wrong with import file or code error." ) |
| main.log.info( "Import Error, please check!" ) |
| main.cleanup() |
| main.exit() |
| |
| bgplsConfig = BgpLs() |
| Ne_id = bgplsConfig.Constants() |
| app = bgplsConfig.apps() |
| httpport = main.params[ 'HTTP' ][ 'port' ] |
| path = main.params[ 'HTTP' ][ 'path' ] |
| bgplsConfig.ipValue( main.ipList, main.scapy_ip ) |
| |
| bgplsConfig.Comments() |
| main.log.info( "Sending BGPLS information" ) |
| bgplsConfig.Comments() |
| |
| main.Scapy1.handle.sendline( "sudo python OnosSystemTest/TestON/tests/FUNC/FUNCbgpls/\ |
| dependencies/Scapyfiles/Topo_discovery.py" ) |
| bgplsConfig.Comments() |
| main.log.info( "Enable BGPlS plugin in ONOS" ) |
| bgplsConfig.Comments() |
| |
| main.testSetUp.startOnosClis( main.Cluster ) |
| |
| installResults = main.Cluster.active( 0 ).CLI.activateApp( app[ 0 ] ) |
| main.step( "Install onos-app-bgp" ) |
| utilities.assert_equals( expect=main.TRUE, actual=installResults, |
| onpass="Install onos-app-bgp successful", |
| onfail="Install onos-app-bgp failed" ) |
| |
| bgpls_post = bgplsConfig.DictoJson() |
| |
| bgplsConfig.Comments() |
| main.log.info( "BGPLS RestConf input" ) |
| bgplsConfig.Comments() |
| |
| print ( bgpls_post ) |
| Poststatus, result = main.Cluster.active( 0 ).REST.send( '/network/configuration/', method="POST", data=bgpls_post ) |
| main.step( "Configure BGP through RESTCONF" ) |
| |
| utilities.assert_equals( expect='200', |
| actual=Poststatus, |
| onpass="Post Port Success", |
| onfail="Post Port Failed " + str( Poststatus ) + "," + str( result ) ) |
| |
| bgplsConfig.Comments() |
| main.step( "Check Network devices are Updated in ONOS " ) |
| bgplsConfig.Comments() |
| time.sleep( 15 ) |
| response = main.Cluster.active( 0 ).CLI.devices() |
| responseCheck = main.FALSE |
| if response: |
| responseCheck = main.TRUE |
| utilities.assert_equals( expect=main.TRUE, |
| actual=responseCheck, |
| onpass="Network Devices update in ONOS successful", |
| onfail="Network Devices update in ONOS failed" ) |
| |
| main.step( "Check the nodes are discovered" ) |
| if response.find( Ne_id[ 1 ][ 0 ] ) and response.find( Ne_id[ 1 ][ 1 ] ) and response.find( Ne_id[ 1 ][ 2 ] ) != -1: |
| stepResult = main.TRUE |
| else: |
| stepResult = main.FALSE |
| utilities.assert_equals( expect=main.TRUE, |
| actual=stepResult, |
| onpass="Node " + str( Ne_id[ 1 ][ 0 ] ) + ( Ne_id[ 1 ][ 1 ] ) + ( Ne_id[ 1 ][ 2 ] ) + " sucess", |
| onfail="Node " + str( Ne_id[ 1 ][ 0 ] ) + ( Ne_id[ 1 ][ 1 ] ) + ( Ne_id[ 1 ][ 2 ] ) + " failed" ) |
| main.Cluster.active( 0 ).CLI.log( "\"testcase2 end\"" ) |
| |
| main.step( "Check for Errors or Exception in testcase2" ) |
| startStr = "testcase2 start" |
| endStr = "testcase2 end" |
| errorLog = main.ONOSbench.logReport( main.Cluster.active( 0 ).ipAddress, |
| [ "ERROR", "EXCEPT" ], "s", |
| startStr, endStr ) |
| utilities.assert_equals( expect=0, actual=errorLog, |
| onpass="No Exception or Error occured in testcase2", |
| onfail="Exception or Error occured in testcase2" ) |
| bgplsConfig.Comments() |
| main.log.info( "Kill Scapy process" ) |
| bgplsConfig.Comments() |
| |
| main.Scapy1.handle.sendline( "\x03" ) |
| time.sleep( 90 ) # This Sleep time gives time for the socket to close. |
| |
| def CASE3( self, main ): |
| """ |
| Addition of new Node to existing topology |
| """ |
| import os |
| import sys |
| import re |
| import time |
| |
| main.case( "Testcase 3: Addition of New Node to existing topology" ) |
| main.Cluster.active( 0 ).CLI.log( "\"testcase3 start\"" ) |
| try: |
| from tests.FUNC.FUNCbgpls.dependencies.Nbdata import BgpLs |
| except ImportError: |
| main.log.exception( "Something wrong with import file or code error." ) |
| main.log.info( "Import Error, please check!" ) |
| main.cleanup() |
| main.exit() |
| |
| bgplsConfig = BgpLs() |
| Ne_id = bgplsConfig.Constants() |
| app = bgplsConfig.apps() |
| |
| httpport = main.params[ 'HTTP' ][ 'port' ] |
| path = main.params[ 'HTTP' ][ 'path' ] |
| |
| bgplsConfig.ipValue( main.ipList, main.scapy_ip ) |
| |
| bgplsConfig.Comments() |
| main.log.info( "Sending BGPLS Packet " ) |
| bgplsConfig.Comments() |
| |
| main.Scapy1.handle.sendline( "sudo python OnosSystemTest/TestON/tests/FUNC/FUNCbgpls/\ |
| dependencies/Scapyfiles/Update_Node.py" ) |
| bgplsConfig.Comments() |
| main.log.info( "Enable BGPlS plugin in ONOS" ) |
| bgplsConfig.Comments() |
| |
| main.step( "UnInstall onos-app-bgp" ) |
| installResults = main.Cluster.active( 0 ).CLI.deactivateApp( app[ 0 ] ) |
| utilities.assert_equals( expect=main.TRUE, actual=installResults, |
| onpass="Uninstall onos-app-bgp successful", |
| onfail="Uninstall onos-app-bgp failed" ) |
| |
| installResults = main.Cluster.active( 0 ).CLI.activateApp( app[ 0 ] ) |
| main.step( "Install onos-app-bgp" ) |
| utilities.assert_equals( expect=main.TRUE, actual=installResults, |
| onpass="Install onos-app-bgp successful", |
| onfail="Install onos-app-bgp failed" ) |
| |
| bgpls_post = bgplsConfig.DictoJson() |
| |
| bgplsConfig.Comments() |
| main.log.info( "BGPLS RestConf input" ) |
| bgplsConfig.Comments() |
| |
| bgplsConfig.Comments() |
| main.step( "Check Network devices are Updated in ONOS" ) |
| bgplsConfig.Comments() |
| time.sleep( 120 ) |
| response = main.Cluster.active( 0 ).CLI.devices() |
| |
| responseCheck = main.FALSE |
| if response: |
| responseCheck = main.TRUE |
| utilities.assert_equals( expect=main.TRUE, |
| actual=responseCheck, |
| onpass="Network Devices update in ONOS successful", |
| onfail="Network Devices update in ONOS failed" ) |
| main.step( "Check Newly added Node is getting updated" ) |
| |
| if response.find( Ne_id[ 1 ][ 3 ] ) != -1: |
| stepResult = main.TRUE |
| else: |
| stepResult = main.FALSE |
| utilities.assert_equals( expect=main.TRUE, |
| actual=stepResult, |
| onpass="Node " + str( Ne_id[ 1 ][ 3 ] ) + " update sucess", |
| onfail="Node " + str( Ne_id[ 1 ][ 3 ] ) + " update failed" ) |
| main.Cluster.active( 0 ).CLI.log( "\"testcase3 end\"" ) |
| |
| main.step( "Check for Errors or Exception in testcase3" ) |
| startStr = "testcase3 start" |
| endStr = "testcase3 end" |
| errorLog = main.ONOSbench.logReport( main.Cluster.active( 0 ).ipAddress, |
| [ "ERROR", "EXCEPT" ], "s", |
| startStr, endStr ) |
| utilities.assert_equals( expect=0, actual=errorLog, |
| onpass="No Exception or Error occured in testcase3", |
| onfail="Exception or Error occured in testcase3" ) |
| bgplsConfig.Comments() |
| main.log.info( "Kill Scapy process" ) |
| bgplsConfig.Comments() |
| main.Scapy1.handle.sendline( "\x03" ) |
| time.sleep( 90 ) # This Sleep time gives time for the socket to close. |
| |
| def CASE4( self, main ): |
| """ |
| Verification of Links in existing topology |
| """ |
| import json |
| import time |
| import os |
| main.case( "Testcase 4: Verification of Links thats is discovered" ) |
| main.Cluster.active( 0 ).CLI.log( "\"testcase4 start\"" ) |
| try: |
| from tests.FUNC.FUNCbgpls.dependencies.Nbdata import BgpLs |
| except ImportError: |
| main.log.exception( "Something wrong with import file or code error." ) |
| main.log.info( "Import Error, please check!" ) |
| main.cleanup() |
| main.exit() |
| |
| bgplsConfig = BgpLs() |
| app = bgplsConfig.apps() |
| bgplsConfig.ipValue( main.ipList, main.scapy_ip ) |
| |
| bgplsConfig.Comments() |
| main.log.info( "Sending BGPLS Link information Packet " ) |
| bgplsConfig.Comments() |
| |
| main.Scapy1.handle.sendline( "sudo python OnosSystemTest/TestON/tests/FUNC/FUNCbgpls/dependencies/Scapyfiles/Link_Update_Node.py" ) |
| bgplsConfig.Comments() |
| main.log.info( "Enable BGPlS plugin in ONOS" ) |
| bgplsConfig.Comments() |
| |
| main.step( "UnInstall onos-app-bgp" ) |
| installResults = main.Cluster.active( 0 ).CLI.deactivateApp( app[ 0 ] ) |
| utilities.assert_equals( expect=main.TRUE, actual=installResults, |
| onpass="Uninstall onos-app-bgp successful", |
| onfail="Uninstall onos-app-bgp failed" ) |
| |
| installResults = main.Cluster.active( 0 ).CLI.activateApp( app[ 0 ] ) |
| main.step( "Install onos-app-bgp" ) |
| utilities.assert_equals( expect=main.TRUE, actual=installResults, |
| onpass="Install onos-app-bgp successful", |
| onfail="Install onos-app-bgp failed" ) |
| bgplsConfig.Comments() |
| main.step( "Checking the Link Discovery Status" ) |
| bgplsConfig.Comments() |
| time.sleep( 120 ) # Time taken to discovery the links |
| response = main.Cluster.active( 0 ).CLI.links() |
| linksResp = json.loads( response ) |
| check_link = bgplsConfig.checkLinks( linksResp ) |
| reply_Check_Link = main.FALSE |
| if check_link: |
| reply_Check_Link = main.TRUE |
| utilities.assert_equals( expect=main.TRUE, actual=reply_Check_Link, |
| onpass="Link Discovery Success.", |
| onfail="Link Discovery Failed." ) |
| main.Cluster.active( 0 ).CLI.log( "\"testcase4 end\"" ) |
| |
| main.step( "Check for Errors or Exception in testcase4 " ) |
| startStr = "testcase4 start" |
| endStr = "testcase4 end" |
| errorLog = main.ONOSbench.logReport( main.Cluster.active( 0 ).ipAddress, |
| [ "ERROR", "EXCEPT" ], "s", |
| startStr, endStr ) |
| utilities.assert_equals( expect=0, actual=errorLog, |
| onpass="No Exception or Error occured in testcase4", |
| onfail="Exception or Error occured in testcase4" ) |
| bgplsConfig.Comments() |
| main.log.info( "Kill Scapy process" ) |
| bgplsConfig.Comments() |
| main.Scapy1.handle.sendline( "\x03" ) |
| time.sleep( 90 ) |
| |
| def CASE5( self, main ): |
| """ |
| Deletion of links |
| """ |
| import json |
| import time |
| import os |
| main.case( "Testcase 5: Deletion of Link in existing topology" ) |
| |
| main.Cluster.active( 0 ).CLI.log( "\"testcase5 start\"" ) |
| try: |
| from tests.FUNC.FUNCbgpls.dependencies.Nbdata import BgpLs |
| except ImportError: |
| main.log.exception( "Something wrong with import file or code error." ) |
| main.log.info( "Import Error, please check!" ) |
| main.cleanup() |
| main.exit() |
| |
| bgplsConfig = BgpLs() |
| app = bgplsConfig.apps() |
| bgplsConfig.ipValue( main.ipList, main.scapy_ip ) |
| |
| bgplsConfig.Comments() |
| main.log.info( "Sending BGPLS Delete Link Packet " ) |
| bgplsConfig.Comments() |
| |
| main.Scapy1.handle.sendline( "sudo python OnosSystemTest/TestON/tests/FUNC/FUNCbgpls/dependencies/Scapyfiles/Deletion_Node.py" ) |
| bgplsConfig.Comments() |
| main.log.info( "Enable BGPlS plugin in ONOS " ) |
| bgplsConfig.Comments() |
| |
| main.step( "UnInstall onos-app-bgp" ) |
| installResults = main.Cluster.active( 0 ).CLI.deactivateApp( app[ 0 ] ) |
| utilities.assert_equals( expect=main.TRUE, actual=installResults, |
| onpass="Uninstall onos-app-bgp successful", |
| onfail="Uninstall onos-app-bgp failed" ) |
| |
| installResults = main.Cluster.active( 0 ).CLI.activateApp( app[ 0 ] ) |
| main.step( "Install onos-app-bgp" ) |
| utilities.assert_equals( expect=main.TRUE, actual=installResults, |
| onpass="Install onos-app-bgp successful", |
| onfail="Install onos-app-bgp failed" ) |
| bgplsConfig.Comments() |
| main.step( "Checking whether the links is deleted" ) |
| bgplsConfig.Comments() |
| time.sleep( 120 ) # Time taken to discovery the links |
| response = main.Cluster.active( 0 ).CLI.links() |
| linksResp = json.loads( response ) |
| check_link = bgplsConfig.checkLinks( linksResp ) |
| reply_Check_Link = main.FALSE |
| if not check_link: |
| reply_Check_Link = main.TRUE |
| utilities.assert_equals( expect=main.TRUE, actual=reply_Check_Link, |
| onpass="Link is Deleted Successfully.", |
| onfail="Link is Deletion Failed." ) |
| main.Cluster.active( 0 ).CLI.log( "\"testcase5 end\"" ) |
| |
| main.step( "Check for Errors or Exception in testcase5" ) |
| startStr = "testcase5 start" |
| endStr = "testcase5 end" |
| errorLog = main.ONOSbench.logReport( main.Cluster.active( 0 ).ipAddress, |
| [ "ERROR", "EXCEPT" ], "s", |
| startStr, endStr ) |
| utilities.assert_equals( expect=0, actual=errorLog, |
| onpass="No Exception or Error occured in testcase5", |
| onfail="Exception or Error occured in testcase5" ) |
| bgplsConfig.Comments() |
| main.log.info( "Kill Scapy process" ) |
| bgplsConfig.Comments() |
| main.Scapy1.handle.sendline( "\x03" ) |
| time.sleep( 90 ) |
| |
| def CASE6( self, main ): |
| """ |
| Uninstalling the app |
| """ |
| import os |
| import sys |
| import re |
| import time |
| |
| main.case( "TestCase 6: UnInstalling of app" ) |
| main.Cluster.active( 0 ).CLI.log( "\"testcase6 start\"" ) |
| try: |
| from tests.FUNC.FUNCbgpls.dependencies.Nbdata import BgpLs |
| except ImportError: |
| main.log.exception( "Something wrong with import file or code error." ) |
| main.log.info( "Import Error, please check!" ) |
| main.cleanup() |
| main.exit() |
| |
| bgplsConfig = BgpLs() |
| app = bgplsConfig.apps() |
| |
| bgplsConfig = BgpLs() |
| bgplsConfig.ipValue( main.ipList, main.scapy_ip ) |
| |
| main.testSetUp.createApplyCell( main.Cluster, True, main.cellName, main.scapy_ip, True, main.ipList ) |
| bgplsConfig.Comments() |
| main.testSetUp.startOnosClis( main.Cluster ) |
| bgplsConfig.Comments() |
| |
| main.log.info( "Uninstall onos-app-bgp" ) |
| bgplsConfig.Comments() |
| main.step( "UnInstall onos-app-bgp" ) |
| installResults = main.Cluster.active( 0 ).CLI.deactivateApp( app[ 0 ] ) |
| utilities.assert_equals( expect=main.TRUE, actual=installResults, |
| onpass="Uninstall onos-app-bgp successful", |
| onfail="Uninstall onos-app-bgp failed" ) |
| |
| main.Cluster.active( 0 ).CLI.log( "\"testcase6 end\"" ) |
| main.step( "Check for Errors or Exception in testcase6" ) |
| startStr = "testcase6 start" |
| endStr = "testcase6 end" |
| errorLog = main.ONOSbench.logReport( main.Cluster.active( 0 ).ipAddress, |
| [ "ERROR", "EXCEPT" ], "s", |
| startStr, endStr ) |
| utilities.assert_equals( expect=0, actual=errorLog, |
| onpass="No Exception or Error occured in testcase6", |
| onfail="Exception or Error occured in testcase6" ) |
| |
| main.step( "Check for Errors or Exception End of the Script" ) |
| errorLog = main.ONOSbench.logReport( main.Cluster.active( 0 ).ipAddress, |
| [ "ERROR", "EXCEPT" ] ) |
| utilities.assert_equals( expect=0, actual=errorLog, |
| onpass="No Exception or Error occured", |
| onfail="Exception or Error occured" ) |