Nightly_sanity calls tests from lf_tests (#9)

This commit is contained in:
SushantBawiskar
2021-02-02 22:05:06 +05:30
committed by GitHub
parent cf88c76aad
commit c63b34c6cf
4 changed files with 162 additions and 174 deletions

View File

@@ -3,6 +3,7 @@
from JfrogHelper import *
from UnitTestBase import *
from cloudsdk import CreateAPProfiles
from lf_tests import *
parser = argparse.ArgumentParser(description="Nightly Combined Tests", add_help=False)
parser.add_argument("--default_ap_profile", type=str,
@@ -61,89 +62,6 @@ if command_line_args.testbed == None:
client: TestRail_Client = TestRail_Client(command_line_args)
###Class for Tests
class RunTest:
def Single_Client_Connectivity(self, port, radio, ssid_name, ssid_psk, security, station, test_case, rid):
'''SINGLE CLIENT CONNECTIVITY using test_connect2.py'''
staConnect = StaConnect2(lanforge_ip, lanforge_port, debug_=False)
staConnect.sta_mode = 0
staConnect.upstream_resource = 1
staConnect.upstream_port = port
staConnect.radio = radio
staConnect.resource = 1
staConnect.dut_ssid = ssid_name
staConnect.dut_passwd = ssid_psk
staConnect.dut_security = security
staConnect.station_names = station
staConnect.sta_prefix = lanforge_prefix
staConnect.runtime_secs = 10
staConnect.bringup_time_sec = 60
staConnect.cleanup_on_exit = True
# staConnect.cleanup()
staConnect.setup()
staConnect.start()
print("napping %f sec" % staConnect.runtime_secs)
time.sleep(staConnect.runtime_secs)
staConnect.stop()
staConnect.cleanup()
run_results = staConnect.get_result_list()
for result in run_results:
print("test result: " + result)
# result = 'pass'
print("Single Client Connectivity :", staConnect.passes)
if staConnect.passes() == True:
print("Single client connection to", ssid_name, "successful. Test Passed")
client.update_testrail(case_id=test_case, run_id=rid, status_id=1, msg='Client connectivity passed')
logger.info("Client connectivity to " + ssid_name + " Passed")
return ("passed")
else:
client.update_testrail(case_id=test_case, run_id=rid, status_id=5, msg='Client connectivity failed')
print("Single client connection to", ssid_name, "unsuccessful. Test Failed")
logger.warning("Client connectivity to " + ssid_name + " FAILED")
return ("failed")
def Single_Client_EAP(port, sta_list, ssid_name, radio, security, eap_type, identity, ttls_password, test_case,
rid):
eap_connect = EAPConnect(lanforge_ip, lanforge_port, _debug_on=False)
eap_connect.upstream_resource = 1
eap_connect.upstream_port = port
eap_connect.security = security
eap_connect.sta_list = sta_list
eap_connect.station_names = sta_list
eap_connect.sta_prefix = lanforge_prefix
eap_connect.ssid = ssid_name
eap_connect.radio = radio
eap_connect.eap = eap_type
eap_connect.identity = identity
eap_connect.ttls_passwd = ttls_password
eap_connect.runtime_secs = 10
eap_connect.setup()
eap_connect.start()
print("napping %f sec" % eap_connect.runtime_secs)
time.sleep(eap_connect.runtime_secs)
eap_connect.stop()
eap_connect.cleanup()
run_results = eap_connect.get_result_list()
for result in run_results:
print("test result: " + result)
# result = 'pass'
print("Single Client Connectivity :", eap_connect.passes)
if eap_connect.passes() == True:
print("Single client connection to", ssid_name, "successful. Test Passed")
client.update_testrail(case_id=test_case, run_id=rid, status_id=1, msg='Client connectivity passed')
logger.info("Client connectivity to " + ssid_name + " Passed")
return ("passed")
else:
client.update_testrail(case_id=test_case, run_id=rid, status_id=5, msg='Client connectivity failed')
print("Single client connection to", ssid_name, "unsuccessful. Test Failed")
logger.warning("Client connectivity to " + ssid_name + " FAILED")
return ("failed")
def testrail_retest(self, test_case, rid, ssid_name):
client.update_testrail(case_id=test_case, run_id=rid, status_id=4,
msg='Error in Client Connectivity Test. Needs to be Re-run')
print("Error in test for single client connection to", ssid_name)
logger.warning("ERROR testing Client connectivity to " + ssid_name)
####Use variables other than defaults for running tests on custom FW etc
@@ -177,7 +95,8 @@ if command_line_args.skip_upgrade == True:
######Testrail Project and Run ID Information ##############################
Test: RunTest = RunTest()
Test = RunTest(lanforge_ip = lanforge_ip, lanforge_port = lanforge_port, lanforge_prefix=lanforge_prefix);
projId = client.get_project_id(project_name=projectId)
print("TIP WLAN Project ID is:", projId)
@@ -515,11 +434,11 @@ for key in equipment_ids:
security = "wpa2"
eap_type = "TTLS"
try:
test_result = RunTest.Single_Client_EAP(port, sta_list, ssid_name, radio, security, eap_type, identity,
ttls_password, test_case, rid)
test_result = Test.Single_Client_EAP(port, sta_list, ssid_name, radio, security, eap_type,
identity, ttls_password, test_case, rid, client, logger)
except:
test_result = "error"
Test.testrail_retest(test_case, rid, ssid_name)
Test.testrail_retest(test_case, rid, ssid_name, client, logger)
pass
report_data['tests'][key][int(test_case)] = test_result
print(report_data['tests'][key])
@@ -535,12 +454,11 @@ for key in equipment_ids:
ssid_psk = psk_2g_wpa2
security = "wpa2"
try:
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security, station,
test_case,
rid)
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security,
station, test_case, rid, client, logger)
except:
test_result = "error"
Test.testrail_retest(test_case, rid, ssid_name)
Test.testrail_retest(test_case, rid, ssid_name, client, logger)
pass
report_data['tests'][key][int(test_case)] = test_result
print(report_data['tests'][key])
@@ -555,12 +473,11 @@ for key in equipment_ids:
ssid_psk = psk_2g_wpa
security = "wpa"
try:
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security, station,
test_case,
rid)
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security,
station, test_case, rid, client, logger)
except:
test_result = "error"
Test.testrail_retest(test_case, rid, ssid_name)
Test.testrail_retest(test_case, rid, ssid_name, client, logger)
pass
report_data['tests'][key][int(test_case)] = test_result
print(report_data['tests'][key])
@@ -576,11 +493,11 @@ for key in equipment_ids:
security = "wpa2"
eap_type = "TTLS"
try:
test_result = RunTest.Single_Client_EAP(port, sta_list, ssid_name, radio, security, eap_type, identity,
ttls_password, test_case, rid)
test_result = Test.Single_Client_EAP(port, sta_list, ssid_name, radio, security, eap_type,
identity, ttls_password, test_case, rid, client, logger)
except:
test_result = "error"
Test.testrail_retest(test_case, rid, ssid_name)
Test.testrail_retest(test_case, rid, ssid_name, client, logger)
pass
report_data['tests'][key][int(test_case)] = test_result
print(report_data['tests'][key])
@@ -595,12 +512,11 @@ for key in equipment_ids:
ssid_psk = psk_5g_wpa2
security = "wpa2"
try:
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security, station,
test_case,
rid)
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security,
station, test_case, rid, client, logger)
except:
test_result = "error"
Test.testrail_retest(test_case, rid, ssid_name)
Test.testrail_retest(test_case, rid, ssid_name, client, logger)
pass
report_data['tests'][key][int(test_case)] = test_result
print(report_data['tests'][key])
@@ -615,12 +531,11 @@ for key in equipment_ids:
ssid_psk = psk_5g_wpa
security = "wpa"
try:
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security, station,
test_case,
rid)
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security,
station, test_case, rid, client, logger)
except:
test_result = "error"
Test.testrail_retest(test_case, rid, ssid_name)
Test.testrail_retest(test_case, rid, ssid_name, client, logger)
pass
report_data['tests'][key][int(test_case)] = test_result
print(report_data['tests'][key])
@@ -844,11 +759,11 @@ for key in equipment_ids:
security = "wpa2"
eap_type = "TTLS"
try:
test_result = RunTest.Single_Client_EAP(port, sta_list, ssid_name, radio, security, eap_type, identity,
ttls_password, test_case, rid)
test_result = Test.Single_Client_EAP(port, sta_list, ssid_name, radio, security, eap_type,
identity, ttls_password, test_case, rid, client, logger)
except:
test_result = "error"
Test.testrail_retest(test_case, rid, ssid_name)
Test.testrail_retest(test_case, rid, ssid_name, client, logger)
pass
report_data['tests'][key][int(test_case)] = test_result
print(report_data['tests'][key])
@@ -863,12 +778,11 @@ for key in equipment_ids:
ssid_psk = psk_2g_wpa2_nat
security = "wpa2"
try:
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security, station,
test_case,
rid)
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security,
station, test_case, rid, client, logger)
except:
test_result = "error"
Test.testrail_retest(test_case, rid, ssid_name)
Test.testrail_retest(test_case, rid, ssid_name, client, logger)
pass
report_data['tests'][key][int(test_case)] = test_result
print(report_data['tests'][key])
@@ -887,7 +801,7 @@ for key in equipment_ids:
test_case, rid)
except:
test_result = "error"
Test.testrail_retest(test_case, rid, ssid_name)
Test.testrail_retest(test_case, rid, ssid_name, client, logger)
pass
report_data['tests'][key][int(test_case)] = test_result
print(report_data['tests'][key])
@@ -903,11 +817,11 @@ for key in equipment_ids:
security = "wpa2"
eap_type = "TTLS"
try:
test_result = RunTest.Single_Client_EAP(port, sta_list, ssid_name, radio, security, eap_type, identity,
ttls_password, test_case, rid)
test_result = Test.Single_Client_EAP(port, sta_list, ssid_name, radio, security, eap_type,
identity, ttls_password, test_case, rid, client, logger)
except:
test_result = "error"
Test.testrail_retest(test_case, rid, ssid_name)
Test.testrail_retest(test_case, rid, ssid_name, client, logger)
pass
report_data['tests'][key][int(test_case)] = test_result
print(report_data['tests'][key])
@@ -922,12 +836,11 @@ for key in equipment_ids:
ssid_psk = psk_5g_wpa2_nat
security = "wpa2"
try:
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security, station,
test_case,
rid)
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security,
station, test_case, rid, client, logger)
except:
test_result = "error"
Test.testrail_retest(test_case, rid, ssid_name)
Test.testrail_retest(test_case, rid, ssid_name, client, logger)
pass
report_data['tests'][key][int(test_case)] = test_result
print(report_data['tests'][key])
@@ -942,12 +855,11 @@ for key in equipment_ids:
ssid_psk = psk_5g_wpa_nat
security = "wpa"
try:
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security, station,
test_case,
rid)
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security,
station, test_case, rid, client, logger)
except:
test_result = "error"
Test.testrail_retest(test_case, rid, ssid_name)
Test.testrail_retest(test_case, rid, ssid_name, client, logger)
pass
report_data['tests'][key][int(test_case)] = test_result
print(report_data['tests'][key])
@@ -1173,11 +1085,11 @@ for key in equipment_ids:
security = "wpa2"
eap_type = "TTLS"
try:
test_result = RunTest.Single_Client_EAP(port, sta_list, ssid_name, radio, security, eap_type, identity,
ttls_password, test_case, rid)
test_result = Test.Single_Client_EAP(port, sta_list, ssid_name, radio, security, eap_type,
identity, ttls_password, test_case, rid, client, logger)
except:
test_result = "error"
Test.testrail_retest(test_case, rid, ssid_name)
Test.testrail_retest(test_case, rid, ssid_name, client, logger)
pass
report_data['tests'][key][int(test_case)] = test_result
print(report_data['tests'][key])
@@ -1192,12 +1104,11 @@ for key in equipment_ids:
ssid_psk = psk_2g_wpa2_vlan
security = "wpa2"
try:
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security, station,
test_case,
rid)
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security,
station, test_case, rid, client, logger)
except:
test_result = "error"
Test.testrail_retest(test_case, rid, ssid_name)
Test.testrail_retest(test_case, rid, ssid_name, client, logger)
pass
report_data['tests'][key][int(test_case)] = test_result
print(report_data['tests'][key])
@@ -1216,7 +1127,7 @@ for key in equipment_ids:
test_case, rid)
except:
test_result = "error"
Test.testrail_retest(test_case, rid, ssid_name)
Test.testrail_retest(test_case, rid, ssid_name, client, logger)
pass
report_data['tests'][key][int(test_case)] = test_result
print(report_data['tests'][key])
@@ -1232,11 +1143,11 @@ for key in equipment_ids:
security = "wpa2"
eap_type = "TTLS"
try:
test_result = RunTest.Single_Client_EAP(port, sta_list, ssid_name, radio, security, eap_type, identity,
ttls_password, test_case, rid)
test_result = Test.Single_Client_EAP(port, sta_list, ssid_name, radio, security, eap_type,
identity, ttls_password, test_case, rid, client, logger)
except:
test_result = "error"
Test.testrail_retest(test_case, rid, ssid_name)
Test.testrail_retest(test_case, rid, ssid_name, client, logger)
pass
report_data['tests'][key][int(test_case)] = test_result
print(report_data['tests'][key])
@@ -1251,12 +1162,11 @@ for key in equipment_ids:
ssid_psk = psk_5g_wpa2_vlan
security = "wpa2"
try:
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security, station,
test_case,
rid)
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security,
station, test_case, rid, client, logger)
except:
test_result = "error"
Test.testrail_retest(test_case, rid, ssid_name)
Test.testrail_retest(test_case, rid, ssid_name, client, logger)
pass
report_data['tests'][key][int(test_case)] = test_result
print(report_data['tests'][key])
@@ -1271,12 +1181,11 @@ for key in equipment_ids:
ssid_psk = psk_5g_wpa_vlan
security = "wpa"
try:
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security, station,
test_case,
rid)
test_result = Test.Single_Client_Connectivity(port, radio, ssid_name, ssid_psk, security,
station, test_case, rid, client, logger)
except:
test_result = "error"
Test.testrail_retest(test_case, rid, ssid_name)
Test.testrail_retest(test_case, rid, ssid_name, client, logger)
pass
report_data['tests'][key][int(test_case)] = test_result
print(report_data['tests'][key])
@@ -1334,4 +1243,4 @@ with open(report_path + today + '/report_data.json', 'w') as report_json_file:
json.dump(report_data, report_json_file)
print(".....End of Sanity Test.....")
logger.info("End of Sanity Test run")
logger.info("End of Sanity Test run")

View File

@@ -314,20 +314,20 @@ class EAPConnect(LFCliBase):
def stop(self):
# stop cx traffic
print("Stopping CX Traffic")
self.l3_udp_profile.stop_cx()
self.l3_tcp_profile.stop_cx()
self.create_traffic.l3_udp_profile.stop_cx()
self.create_traffic.l3_tcp_profile.stop_cx()
# Refresh stats
print("\nRefresh CX stats")
self.l3_udp_profile.refresh_cx()
self.l3_tcp_profile.refresh_cx()
self.create_traffic.l3_udp_profile.refresh_cx()
self.create_traffic.l3_tcp_profile.refresh_cx()
print("Sleeping for 5 seconds")
time.sleep(5)
# get data for endpoints JSON
self.collect_endp_stats(self.l3_udp_profile.created_cx)
self.collect_endp_stats(self.l3_tcp_profile.created_cx)
self.collect_endp_stats(self.create_traffic.l3_udp_profile.created_cx)
self.collect_endp_stats(self.create_traffic.l3_tcp_profile.created_cx)
# print("\n")
def cleanup(self):
@@ -336,12 +336,12 @@ class EAPConnect(LFCliBase):
for sta_name in self.station_names:
LFUtils.removePort(self.resource, sta_name, self.lfclient_url)
curr_endp_names = []
removeCX(self.lfclient_url, self.l3_udp_profile.get_cx_names())
removeCX(self.lfclient_url, self.l3_tcp_profile.get_cx_names())
for (cx_name, endp_names) in self.l3_udp_profile.created_cx.items():
removeCX(self.lfclient_url, self.create_traffic.l3_udp_profile.get_cx_names())
removeCX(self.lfclient_url, self.create_traffic.l3_tcp_profile.get_cx_names())
for (cx_name, endp_names) in self.create_traffic.l3_udp_profile.created_cx.items():
curr_endp_names.append(endp_names[0])
curr_endp_names.append(endp_names[1])
for (cx_name, endp_names) in self.l3_tcp_profile.created_cx.items():
for (cx_name, endp_names) in self.create_traffic.l3_tcp_profile.created_cx.items():
curr_endp_names.append(endp_names[0])
curr_endp_names.append(endp_names[1])
removeEndps(self.lfclient_url, curr_endp_names, debug= self.debug)

View File

@@ -5,26 +5,8 @@
#########################################################################################################
# create TCP and UDP traffic, run it a short amount of time,
# self.l3_udp_profile.start_cx(): To Start UDP
import sys
if sys.version_info[0] != 3:
print("This script requires Python 3")
exit(1)
if 'py-json' not in sys.path:
sys.path.append('../../py-json')
import argparse
import LANforge
from LANforge import LFUtils
# from LANforge import LFCliBase
from LANforge import lfcli_base
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
import realm
from realm import Realm
import pprint
class createTraffic:

97
unit_tests/lf_tests.py Normal file
View File

@@ -0,0 +1,97 @@
#########################################################################################################
# Used by Nightly_Sanity
# This has different types of tests like Single client connectivity, Single_Client_EAP, testrail_retest
#########################################################################################################
from UnitTestBase import *
class RunTest:
def __init__(self, lanforge_ip, lanforge_port, lanforge_prefix):
self.lanforge_ip = lanforge_ip
self.lanforge_port = lanforge_port
self.lanforge_prefix = lanforge_prefix
def Single_Client_Connectivity(self, port, radio, ssid_name, ssid_psk, security,
station, test_case, rid, client, logger):
'''SINGLE CLIENT CONNECTIVITY using test_connect2.py'''
self.staConnect = StaConnect2(self.lanforge_ip, self.lanforge_port, debug_=False)
self.staConnect.sta_mode = 0
self.staConnect.upstream_resource = 1
self.staConnect.upstream_port = port
self.staConnect.radio = radio
self.staConnect.resource = 1
self.staConnect.dut_ssid = ssid_name
self.staConnect.dut_passwd = ssid_psk
self.staConnect.dut_security = security
self.staConnect.station_names = station
self.staConnect.sta_prefix = self.lanforge_prefix
self.staConnect.runtime_secs = 10
self.staConnect.bringup_time_sec = 60
self.staConnect.cleanup_on_exit = True
# staConnect.cleanup()
self.staConnect.setup()
self.staConnect.start()
print("napping %f sec" % self.staConnect.runtime_secs)
time.sleep(self.staConnect.runtime_secs)
self.staConnect.stop()
self.staConnect.cleanup()
run_results = self.staConnect.get_result_list()
for result in run_results:
print("test result: " + result)
# result = 'pass'
print("Single Client Connectivity :", self.staConnect.passes)
if self.staConnect.passes() == True:
print("Single client connection to", ssid_name, "successful. Test Passed")
client.update_testrail(case_id=test_case, run_id=rid, status_id=1, msg='Client connectivity passed')
logger.info("Client connectivity to " + ssid_name + " Passed")
return ("passed")
else:
client.update_testrail(case_id=test_case, run_id=rid, status_id=5, msg='Client connectivity failed')
print("Single client connection to", ssid_name, "unsuccessful. Test Failed")
logger.warning("Client connectivity to " + ssid_name + " FAILED")
return ("failed")
def Single_Client_EAP(self, port, sta_list, ssid_name, radio, security, eap_type,
identity, ttls_password, test_case, rid, client, logger):
eap_connect = EAPConnect(self.lanforge_ip, self.lanforge_port, _debug_on=False)
eap_connect.upstream_resource = 1
eap_connect.upstream_port = port
eap_connect.security = security
eap_connect.sta_list = sta_list
eap_connect.station_names = sta_list
eap_connect.sta_prefix = self.lanforge_prefix
eap_connect.ssid = ssid_name
eap_connect.radio = radio
eap_connect.eap = eap_type
eap_connect.identity = identity
eap_connect.ttls_passwd = ttls_password
eap_connect.runtime_secs = 10
eap_connect.setup()
eap_connect.start()
print("napping %f sec" % eap_connect.runtime_secs)
time.sleep(eap_connect.runtime_secs)
eap_connect.stop()
eap_connect.cleanup()
run_results = eap_connect.get_result_list()
for result in run_results:
print("test result: " + result)
# result = 'pass'
print("Single Client Connectivity :", eap_connect.passes)
if eap_connect.passes() == True:
print("Single client connection to", ssid_name, "successful. Test Passed")
client.update_testrail(case_id=test_case, run_id=rid, status_id=1, msg='Client connectivity passed')
logger.info("Client connectivity to " + ssid_name + " Passed")
return ("passed")
else:
client.update_testrail(case_id=test_case, run_id=rid, status_id=5, msg='Client connectivity failed')
print("Single client connection to", ssid_name, "unsuccessful. Test Failed")
logger.warning("Client connectivity to " + ssid_name + " FAILED")
return ("failed")
def testrail_retest(self, test_case, rid, ssid_name, client, logger):
client.update_testrail(case_id=test_case, run_id=rid, status_id=4,
msg='Error in Client Connectivity Test. Needs to be Re-run')
print("Error in test for single client connection to", ssid_name)
logger.warning("ERROR testing Client connectivity to " + ssid_name)