diff --git a/unit_tests/Nightly_Sanity.py b/unit_tests/Nightly_Sanity.py index 46c233af9..098a6074a 100755 --- a/unit_tests/Nightly_Sanity.py +++ b/unit_tests/Nightly_Sanity.py @@ -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") \ No newline at end of file diff --git a/unit_tests/eap_connect.py b/unit_tests/eap_connect.py index 7b46307d4..d5055150b 100755 --- a/unit_tests/eap_connect.py +++ b/unit_tests/eap_connect.py @@ -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) diff --git a/unit_tests/lf_lib.py b/unit_tests/lf_lib.py index 2e89605ae..33b7b1db0 100644 --- a/unit_tests/lf_lib.py +++ b/unit_tests/lf_lib.py @@ -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: diff --git a/unit_tests/lf_tests.py b/unit_tests/lf_tests.py new file mode 100644 index 000000000..212140173 --- /dev/null +++ b/unit_tests/lf_tests.py @@ -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) \ No newline at end of file