blob: 8586a7a5a4a08603d6424d1cabc2d80f33ccf312 [file] [log] [blame]
#Testing the basic functionality of ONOS Next
#For sanity and driver functionality excercises only.
import time
import sys
import os
import re
import time
class ONOSNextTest:
def __init__(self):
self.default = ''
def CASE1(self, main):
'''
Startup sequence:
git pull
mvn clean install
onos-package
cell <name>
onos-verify-cell
onos-install -f
onos-wait-for-start
'''
cell_name = main.params['ENV']['cellName']
ONOS1_ip = main.params['CTRL']['ip1']
ONOS1_port = main.params['CTRL']['port1']
main.case("Setting up test environment")
main.step("Git checkout and pull master")
#main.ONOSbench.git_checkout("master")
#git_pull_result = main.ONOSbench.git_pull()
main.step("Using mvn clean & install")
#clean_install_result = main.ONOSbench.clean_install()
clean_install_result = main.TRUE
main.step("Creating ONOS package")
package_result = main.ONOSbench.onos_package()
main.step("Creating cell file")
cell_file_result = main.ONOSbench.create_cell_file(
"10.128.20.10", "temp_cell_2", "10.128.10.90",
"10.128.10.11", "10.128.10.12", "10.128.10.13")
main.step("Applying cell variable to environment")
cell_result = main.ONOSbench.set_cell(cell_name)
verify_result = main.ONOSbench.verify_cell()
main.step("Installing ONOS package")
onos_install_result = main.ONOSbench.onos_install()
onos1_isup = main.ONOSbench.isup()
main.step("Starting ONOS service")
start_result = main.ONOSbench.onos_start(ONOS1_ip)
case1_result = (clean_install_result and package_result and\
cell_result and verify_result and onos_install_result and\
onos1_isup and start_result )
utilities.assert_equals(expect=main.TRUE, actual=case1_result,
onpass="Test startup successful",
onfail="Test startup NOT successful")
def CASE11(self, main):
'''
Cleanup sequence:
onos-service <node_ip> stop
onos-uninstall
TODO: Define rest of cleanup
'''
ONOS1_ip = main.params['CTRL']['ip1']
main.case("Cleaning up test environment")
main.step("Testing ONOS kill function")
kill_result = main.ONOSbench.onos_kill(ONOS1_ip)
main.step("Stopping ONOS service")
stop_result = main.ONOSbench.onos_stop(ONOS1_ip)
main.step("Uninstalling ONOS service")
uninstall_result = main.ONOSbench.onos_uninstall()
def CASE3(self, main):
'''
Test 'onos' command and its functionality in driver
'''
ONOS1_ip = main.params['CTRL']['ip1']
main.case("Testing 'onos' command")
main.step("Sending command 'onos -w <onos-ip> system:name'")
cmdstr1 = "system:name"
cmd_result1 = main.ONOSbench.onos_cli(ONOS1_ip, cmdstr1)
main.log.info("onos command returned: "+cmd_result1)
main.step("Sending command 'onos -w <onos-ip> onos:topology'")
cmdstr2 = "onos:topology"
cmd_result2 = main.ONOSbench.onos_cli(ONOS1_ip, cmdstr2)
main.log.info("onos command returned: "+cmd_result2)
def CASE4(self, main):
main.step("Assigning switches to controllers")
for i in range(1,29):
main.Mininet1.assign_sw_controller(sw=str(i),
ip1=ONOS1_ip, port1=ONOS1_port)
switch_mastership = main.TRUE
for i in range (1,29):
response = main.Mininet1.get_sw_controller("s"+str(i))
print("Response is " + str(response))
if re.search("tcp:"+ONOS1_ip,response):
switch_mastership = switch_mastership and main.TRUE
else:
switch_mastership = main.FALSE
#REACTIVE FWD test
main.step("Pingall")
ping_result = main.FALSE
while ping_result == main.FALSE:
time1 = time.time()
ping_result = main.Mininet1.pingall()
time2 = time.time()
print "Time for pingall: %2f seconds" % (time2 - time1)
case4_result = switch_mastership and ping_result
utilities.assert_equals(expect=main.TRUE, actual=case4_result,
onpass="Test successful",
onfail="Test NOT successful")