diff --git a/libs/lanforge/lf_tests.py b/libs/lanforge/lf_tests.py index fef4074ba..78d994ae9 100644 --- a/libs/lanforge/lf_tests.py +++ b/libs/lanforge/lf_tests.py @@ -249,7 +249,7 @@ class RunTest: def wifi_capacity(self, mode="BRIDGE", vlan_id=100, batch_size="1,5,10,20,40,64,128", instance_name="wct_instance", download_rate="1Gbps", influx_tags=[], - upload_rate="1Gbps", protocol="TCP-IPv4", duration="60000", raw_lines=[]): + upload_rate="1Gbps", protocol="TCP-IPv4", duration="60000", stations="", create_stations=True, raw_lines=[]): instance_name = ''.join(random.choices(string.ascii_uppercase + string.digits, k=S)) if mode == "BRIDGE": upstream_port = self.upstream_port @@ -276,7 +276,8 @@ class RunTest: upload_rate=upload_rate, download_rate=download_rate, sort="interleave", - create_stations=True, + stations=stations, + create_stations=create_stations, radio=None, security=None, paswd=None, diff --git a/libs/lanforge/lf_tools.py b/libs/lanforge/lf_tools.py index 4926dd3c4..f03e9e897 100644 --- a/libs/lanforge/lf_tools.py +++ b/libs/lanforge/lf_tools.py @@ -135,6 +135,27 @@ class ChamberView: self.CreateDut.ssid = temp self.CreateDut.add_ssids() + def get_station_list(self): + realm_obj = self.staConnect.localrealm + sta = realm_obj.station_list() + sta_list = [] + for i in sta: + for j in i: + sta_list.append(j) + return sta_list + + def admin_up_down(self, sta_list=[], option="up"): + realm_obj = self.staConnect.localrealm + if option == "up": + for i in sta_list: + realm_obj.admin_up(i) + time.sleep(0.005) + elif option == "down": + for j in sta_list: + realm_obj.admin_down(j) + time.sleep(0.005) + time.sleep(2) + def Chamber_View(self): if self.delete_old_scenario: self.CreateChamberview.clean_cv_scenario(type="Network-Connectivity", scenario_name=self.scenario_name) @@ -292,6 +313,32 @@ class ChamberView: result = df[column_name].values.tolist() return result + def read_csv_individual_station_throughput(self, dir_name, option): + try: + df = pd.read_csv("../reports/" + str(dir_name) + "/csv-data/data-Combined_bps__60_second_running_average-1.csv", sep=r'\t', engine='python') + print("csv file opened") + except FileNotFoundError: + print("csv file does not exist") + return False + + dict_data = {} + if option == "download": + csv_sta_names = df.iloc[[0]].values.tolist() + csv_throughput_values = df.iloc[[1]].values.tolist() + elif option == "upload": + csv_sta_names = df.iloc[[0]].values.tolist() + csv_throughput_values = df.iloc[[2]].values.tolist() + else: + print("Provide proper option: download or upload") + return + + sta_list = csv_sta_names[0][0][:-1].replace('"', '').split(",") + th_list = list(map(float, csv_throughput_values[0][0].split(","))) + for i in range(len(sta_list)): + dict_data[sta_list[i]] = th_list[i] + + return dict_data + def attach_report_graphs(self, report_name=None, pdf_name="WIFI Capacity Test PDF Report"): relevant_path = "../reports/" + report_name + "/" entries = os.listdir("../reports/" + report_name + '/') diff --git a/tests/e2e/advanced/multi_association_disassociation/__init__.py b/tests/e2e/advanced/multi_association_disassociation/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/e2e/advanced/multi_association_disassociation/wpa2_personal/__init__.py b/tests/e2e/advanced/multi_association_disassociation/wpa2_personal/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/e2e/advanced/multi_association_disassociation/wpa2_personal/test_bridge_association_disassociation.py b/tests/e2e/advanced/multi_association_disassociation/wpa2_personal/test_bridge_association_disassociation.py new file mode 100644 index 000000000..75b3054f9 --- /dev/null +++ b/tests/e2e/advanced/multi_association_disassociation/wpa2_personal/test_bridge_association_disassociation.py @@ -0,0 +1,289 @@ +import pytest +import allure +import os +import time +import pandas as pd +import threading + +pytestmark = [pytest.mark.advance, pytest.mark.multiassodisasso, pytest.mark.bridge] + +setup_params_general = { + "mode": "BRIDGE", + "ssid_modes": { + "wpa2_personal": [ + {"ssid_name": "ssid_wpa2_2g", "appliedRadios": ["2G"], "security_key": "something"}, + {"ssid_name": "ssid_wpa2_5g", "appliedRadios": ["5G"], "security_key": "something"} + ] + }, + "rf": {}, + "radius": False +} +@pytest.mark.parametrize( + 'setup_profiles', + [setup_params_general], + indirect=True, + scope="class" +) +@pytest.mark.usefixtures("setup_profiles") +class TestMultiAssoDisassoBridge(object): + + @allure.testcase(url="https://telecominfraproject.atlassian.net/browse/WIFI-5691", name="WIFI-5691") + @pytest.mark.wpa2_personal + @pytest.mark.twog + @pytest.mark.udp_upload_2g + def test_multi_station_udp_upload_2g(self, lf_test, lf_tools, create_lanforge_chamberview_dut): + # run wifi capacity test here + def thread_fun(station_list): + print(station_list) + time.sleep(60) + lf_tools.admin_up_down(sta_list=station_list, option="down") + print("stations down") + time.sleep(10) + lf_tools.admin_up_down(sta_list=station_list, option="up") + print("stations up") + + lf_tools.reset_scenario() + profile_data = setup_params_general["ssid_modes"]["wpa2_personal"][0] + ssid_name = profile_data["ssid_name"] + print(ssid_name) + mode = "BRIDGE" + vlan = 1 + lf_tools.add_stations(band="2G", num_stations=16, dut=lf_tools.dut_name, ssid_name=ssid_name) + lf_tools.Chamber_View() + sta_list = lf_tools.get_station_list() + print(sta_list) + lf_tools.admin_up_down(sta_list=sta_list, option="up") + sel_stations = ",".join(sta_list[0:8]) + val = [['ul_rate_sel: Per-Station Upload Rate:']] + thr1 = threading.Thread(target=thread_fun, args=(sta_list[8:16],)) + thr1.start() + wct_obj = lf_test.wifi_capacity(instance_name="udp_upload_2g", mode=mode, vlan_id=vlan, + download_rate="0Gbps", stations=sel_stations, raw_lines=val, batch_size="8", + upload_rate="4Mbps", protocol="UDP-IPv4", duration="120000", create_stations=False) + + report_name = wct_obj.report_name[0]['LAST']["response"].split(":::")[1].split("/")[-1] + + lf_tools.attach_report_graphs(report_name=report_name) + + csv_val = lf_tools.read_csv_individual_station_throughput(dir_name=report_name, option="upload") + print(type(csv_val)) + print(csv_val) + pass_value = 4 * 0.99 + print("pass value ", pass_value) + pass_fail = [] + if not csv_val: + print("csv file does not exist, station did not got ip, Test failed") + allure.attach(name="Csv Data", body="station did not got ip Test failed.") + assert False + else: + for i in csv_val.values(): + if i >= pass_value: + pass_fail.append(1) + else: + pass_fail.append(0) + if pass_fail.count(0) == 0: + print("Test passed successfully") + allure.attach(name="Csv Data", body=str(csv_val)) + assert True + else: + print(" valueTest failed due to lesser") + allure.attach(name="Csv Data", body=str(csv_val)) + assert False + print("Test Completed... Cleaning up Stations") + + @allure.testcase(url="https://telecominfraproject.atlassian.net/browse/WIFI-5692", name="WIFI-5692") + @pytest.mark.wpa2_personal + @pytest.mark.twog + @pytest.mark.udp_download_2g + def test_multi_station_udp_download_2g(self, lf_test, lf_tools, create_lanforge_chamberview_dut): + # run wifi capacity test here + def thread_fun(station_list): + print(station_list) + time.sleep(60) + lf_tools.admin_up_down(sta_list=station_list, option="down") + print("stations down") + time.sleep(10) + lf_tools.admin_up_down(sta_list=station_list, option="up") + print("stations up") + + lf_tools.reset_scenario() + profile_data = setup_params_general["ssid_modes"]["wpa2_personal"][0] + ssid_name = profile_data["ssid_name"] + print(ssid_name) + mode = "BRIDGE" + vlan = 1 + lf_tools.add_stations(band="2G", num_stations=16, dut=lf_tools.dut_name, ssid_name=ssid_name) + lf_tools.Chamber_View() + sta_list = lf_tools.get_station_list() + print(sta_list) + lf_tools.admin_up_down(sta_list=sta_list, option="up") + sel_stations = ",".join(sta_list[0:8]) + val = [['dl_rate_sel: Per-Station Download Rate:']] + thr1 = threading.Thread(target=thread_fun, args=(sta_list[8:16],)) + thr1.start() + wct_obj = lf_test.wifi_capacity(instance_name="udp_download_2g", mode=mode, vlan_id=vlan, + download_rate="4Mbps", stations=sel_stations, raw_lines=val, batch_size="8", + upload_rate="0Gbps", protocol="UDP-IPv4", duration="120000", + create_stations=False) + + report_name = wct_obj.report_name[0]['LAST']["response"].split(":::")[1].split("/")[-1] + + lf_tools.attach_report_graphs(report_name=report_name) + + csv_val = lf_tools.read_csv_individual_station_throughput(dir_name=report_name, option="download") + print(type(csv_val)) + print(csv_val) + pass_value = 4 * 0.99 + print("pass value ", pass_value) + pass_fail = [] + if not csv_val: + print("csv file does not exist, station did not got ip, Test failed") + allure.attach(name="Csv Data", body="station did not got ip Test failed.") + assert False + else: + for i in csv_val.values(): + if i >= pass_value: + pass_fail.append(1) + else: + pass_fail.append(0) + if pass_fail.count(0) == 0: + print("Test passed successfully") + allure.attach(name="Csv Data", body=str(csv_val)) + assert True + else: + print(" valueTest failed due to lesser") + allure.attach(name="Csv Data", body=str(csv_val)) + assert False + print("Test Completed... Cleaning up Stations") + + @allure.testcase(url="https://telecominfraproject.atlassian.net/browse/WIFI-5693", name="WIFI-5693") + @pytest.mark.wpa2_personal + @pytest.mark.fiveg + @pytest.mark.udp_upload_5g + def test_multi_station_udp_upload_5g(self, lf_test, lf_tools, create_lanforge_chamberview_dut): + # run wifi capacity test here + def thread_fun(station_list): + print(station_list) + time.sleep(60) + lf_tools.admin_up_down(sta_list=station_list, option="down") + print("stations down") + time.sleep(10) + lf_tools.admin_up_down(sta_list=station_list, option="up") + print("stations up") + + lf_tools.reset_scenario() + profile_data = setup_params_general["ssid_modes"]["wpa2_personal"][1] + ssid_name = profile_data["ssid_name"] + print(ssid_name) + mode = "BRIDGE" + vlan = 1 + lf_tools.add_stations(band="5G", num_stations=16, dut=lf_tools.dut_name, ssid_name=ssid_name) + lf_tools.Chamber_View() + sta_list = lf_tools.get_station_list() + print(sta_list) + lf_tools.admin_up_down(sta_list=sta_list, option="up") + sel_stations = ",".join(sta_list[0:8]) + val = [['ul_rate_sel: Per-Station Upload Rate:']] + thr1 = threading.Thread(target=thread_fun, args=(sta_list[8:16],)) + thr1.start() + wct_obj = lf_test.wifi_capacity(instance_name="udp_upload_5g", mode=mode, vlan_id=vlan, + download_rate="0Gbps", stations=sel_stations, raw_lines=val, batch_size="8", + upload_rate="8Mbps", protocol="UDP-IPv4", duration="120000", + create_stations=False) + + report_name = wct_obj.report_name[0]['LAST']["response"].split(":::")[1].split("/")[-1] + + lf_tools.attach_report_graphs(report_name=report_name) + + csv_val = lf_tools.read_csv_individual_station_throughput(dir_name=report_name, option="upload") + print(type(csv_val)) + print(csv_val) + pass_value = 8 * 0.99 + print("pass value ", pass_value) + pass_fail = [] + if not csv_val: + print("csv file does not exist, station did not got ip, Test failed") + allure.attach(name="Csv Data", body="station did not got ip Test failed.") + assert False + else: + for i in csv_val.values(): + if i >= pass_value: + pass_fail.append(1) + else: + pass_fail.append(0) + if pass_fail.count(0) == 0: + print("Test passed successfully") + allure.attach(name="Csv Data", body=str(csv_val)) + assert True + else: + print(" valueTest failed due to lesser") + allure.attach(name="Csv Data", body=str(csv_val)) + assert False + print("Test Completed... Cleaning up Stations") + + @allure.testcase(url="https://telecominfraproject.atlassian.net/browse/WIFI-5694", name="WIFI-5694") + @pytest.mark.wpa2_personal + @pytest.mark.fiveg + @pytest.mark.udp_download_5g + def test_multi_station_udp_download_5g(self, lf_test, lf_tools, create_lanforge_chamberview_dut): + # run wifi capacity test here + def thread_fun(station_list): + print(station_list) + time.sleep(60) + lf_tools.admin_up_down(sta_list=station_list, option="down") + print("stations down") + time.sleep(10) + lf_tools.admin_up_down(sta_list=station_list, option="up") + print("stations up") + + lf_tools.reset_scenario() + profile_data = setup_params_general["ssid_modes"]["wpa2_personal"][1] + ssid_name = profile_data["ssid_name"] + print(ssid_name) + mode = "BRIDGE" + vlan = 1 + lf_tools.add_stations(band="5G", num_stations=16, dut=lf_tools.dut_name, ssid_name=ssid_name) + lf_tools.Chamber_View() + sta_list = lf_tools.get_station_list() + print(sta_list) + lf_tools.admin_up_down(sta_list=sta_list, option="up") + sel_stations = ",".join(sta_list[0:8]) + val = [['dl_rate_sel: Per-Station Download Rate:']] + thr1 = threading.Thread(target=thread_fun, args=(sta_list[8:16],)) + thr1.start() + wct_obj = lf_test.wifi_capacity(instance_name="udp_download_5g", mode=mode, vlan_id=vlan, + download_rate="8Mbps", stations=sel_stations, raw_lines=val, batch_size="8", + upload_rate="0Gbps", protocol="UDP-IPv4", duration="120000", + create_stations=False) + + report_name = wct_obj.report_name[0]['LAST']["response"].split(":::")[1].split("/")[-1] + + lf_tools.attach_report_graphs(report_name=report_name) + + csv_val = lf_tools.read_csv_individual_station_throughput(dir_name=report_name, option="download") + print(type(csv_val)) + print(csv_val) + pass_value = 8 * 0.99 + print("pass value ", pass_value) + pass_fail = [] + if not csv_val: + print("csv file does not exist, station did not got ip, Test failed") + allure.attach(name="Csv Data", body="station did not got ip Test failed.") + assert False + else: + for i in csv_val.values(): + if i >= pass_value: + pass_fail.append(1) + else: + pass_fail.append(0) + if pass_fail.count(0) == 0: + print("Test passed successfully") + allure.attach(name="Csv Data", body=str(csv_val)) + assert True + else: + print(" valueTest failed due to lesser") + allure.attach(name="Csv Data", body=str(csv_val)) + assert False + print("Test Completed... Cleaning up Stations") + + diff --git a/tests/e2e/advanced/multi_association_disassociation/wpa2_personal/test_nat_association_disassociation.py b/tests/e2e/advanced/multi_association_disassociation/wpa2_personal/test_nat_association_disassociation.py new file mode 100644 index 000000000..9711e6e85 --- /dev/null +++ b/tests/e2e/advanced/multi_association_disassociation/wpa2_personal/test_nat_association_disassociation.py @@ -0,0 +1,289 @@ +import pytest +import allure +import os +import time +import pandas as pd +import threading + +#pytestmark = [pytest.mark.advance, pytest.mark.multiassodisasso, pytest.mark.nat] + +setup_params_general = { + "mode": "NAT", + "ssid_modes": { + "wpa2_personal": [ + {"ssid_name": "ssid_wpa2_2g", "appliedRadios": ["2G"], "security_key": "something"}, + {"ssid_name": "ssid_wpa2_5g", "appliedRadios": ["5G"], "security_key": "something"} + ] + }, + "rf": {}, + "radius": False +} +@pytest.mark.parametrize( + 'setup_profiles', + [setup_params_general], + indirect=True, + scope="class" +) +@pytest.mark.usefixtures("setup_profiles") +class TestMultiAssoDisassoNat(object): + + @allure.testcase(url="https://telecominfraproject.atlassian.net/browse/WIFI-5848", name="WIFI-5848") + @pytest.mark.wpa2_personal + @pytest.mark.twog + @pytest.mark.udp_upload_2g + def test_multi_station_NAT_udp_upload_2g(self, lf_test, lf_tools, create_lanforge_chamberview_dut): + # run wifi capacity test here + def thread_fun(station_list): + print(station_list) + time.sleep(60) + lf_tools.admin_up_down(sta_list=station_list, option="down") + print("stations down") + time.sleep(10) + lf_tools.admin_up_down(sta_list=station_list, option="up") + print("stations up") + + lf_tools.reset_scenario() + profile_data = setup_params_general["ssid_modes"]["wpa2_personal"][0] + ssid_name = profile_data["ssid_name"] + print(ssid_name) + mode = "NAT" + vlan = 1 + lf_tools.add_stations(band="2G", num_stations=16, dut=lf_tools.dut_name, ssid_name=ssid_name) + lf_tools.Chamber_View() + sta_list = lf_tools.get_station_list() + print(sta_list) + lf_tools.admin_up_down(sta_list=sta_list, option="up") + sel_stations = ",".join(sta_list[0:8]) + val = [['ul_rate_sel: Per-Station Upload Rate:']] + thr1 = threading.Thread(target=thread_fun, args=(sta_list[8:16],)) + thr1.start() + wct_obj = lf_test.wifi_capacity(instance_name="udp_NAT_upload_2g", mode=mode, vlan_id=vlan, + download_rate="0Gbps", stations=sel_stations, raw_lines=val, batch_size="8", + upload_rate="4Mbps", protocol="UDP-IPv4", duration="120000", create_stations=False) + + report_name = wct_obj.report_name[0]['LAST']["response"].split(":::")[1].split("/")[-1] + + lf_tools.attach_report_graphs(report_name=report_name) + + csv_val = lf_tools.read_csv_individual_station_throughput(dir_name=report_name, option="upload") + print(type(csv_val)) + print(csv_val) + pass_value = 4 * 0.99 + print("pass value ", pass_value) + pass_fail = [] + if not csv_val: + print("csv file does not exist, station did not got ip, Test failed") + allure.attach(name="Csv Data", body="station did not got ip Test failed.") + assert False + else: + for i in csv_val.values(): + if i >= pass_value: + pass_fail.append(1) + else: + pass_fail.append(0) + if pass_fail.count(0) == 0: + print("Test passed successfully") + allure.attach(name="Csv Data", body=str(csv_val)) + assert True + else: + print(" valueTest failed due to lesser") + allure.attach(name="Csv Data", body=str(csv_val)) + assert False + print("Test Completed... Cleaning up Stations") + + @allure.testcase(url="https://telecominfraproject.atlassian.net/browse/WIFI-5870", name="WIFI-5870") + @pytest.mark.wpa2_personal + @pytest.mark.twog + @pytest.mark.udp_download_2g + def test_multi_station_NAT_udp_download_2g(self, lf_test, lf_tools, create_lanforge_chamberview_dut): + # run wifi capacity test here + def thread_fun(station_list): + print(station_list) + time.sleep(60) + lf_tools.admin_up_down(sta_list=station_list, option="down") + print("stations down") + time.sleep(10) + lf_tools.admin_up_down(sta_list=station_list, option="up") + print("stations up") + + lf_tools.reset_scenario() + profile_data = setup_params_general["ssid_modes"]["wpa2_personal"][0] + ssid_name = profile_data["ssid_name"] + print(ssid_name) + mode = "NAT" + vlan = 1 + lf_tools.add_stations(band="2G", num_stations=16, dut=lf_tools.dut_name, ssid_name=ssid_name) + lf_tools.Chamber_View() + sta_list = lf_tools.get_station_list() + print(sta_list) + lf_tools.admin_up_down(sta_list=sta_list, option="up") + sel_stations = ",".join(sta_list[0:8]) + val = [['dl_rate_sel: Per-Station Download Rate:']] + thr1 = threading.Thread(target=thread_fun, args=(sta_list[8:16],)) + thr1.start() + wct_obj = lf_test.wifi_capacity(instance_name="udp_NAT_download_2g", mode=mode, vlan_id=vlan, + download_rate="4Mbps", stations=sel_stations, raw_lines=val, batch_size="8", + upload_rate="0Gbps", protocol="UDP-IPv4", duration="120000", + create_stations=False) + + report_name = wct_obj.report_name[0]['LAST']["response"].split(":::")[1].split("/")[-1] + + lf_tools.attach_report_graphs(report_name=report_name) + + csv_val = lf_tools.read_csv_individual_station_throughput(dir_name=report_name, option="download") + print(type(csv_val)) + print(csv_val) + pass_value = 4 * 0.99 + print("pass value ", pass_value) + pass_fail = [] + if not csv_val: + print("csv file does not exist, station did not got ip, Test failed") + allure.attach(name="Csv Data", body="station did not got ip Test failed.") + assert False + else: + for i in csv_val.values(): + if i >= pass_value: + pass_fail.append(1) + else: + pass_fail.append(0) + if pass_fail.count(0) == 0: + print("Test passed successfully") + allure.attach(name="Csv Data", body=str(csv_val)) + assert True + else: + print(" valueTest failed due to lesser") + allure.attach(name="Csv Data", body=str(csv_val)) + assert False + print("Test Completed... Cleaning up Stations") + + @allure.testcase(url="https://telecominfraproject.atlassian.net/browse/WIFI-5871", name="WIFI-5871") + @pytest.mark.wpa2_personal + @pytest.mark.fiveg + @pytest.mark.udp_upload_5g + def test_multi_station_NAT_udp_upload_5g(self, lf_test, lf_tools, create_lanforge_chamberview_dut): + # run wifi capacity test here + def thread_fun(station_list): + print(station_list) + time.sleep(60) + lf_tools.admin_up_down(sta_list=station_list, option="down") + print("stations down") + time.sleep(10) + lf_tools.admin_up_down(sta_list=station_list, option="up") + print("stations up") + + lf_tools.reset_scenario() + profile_data = setup_params_general["ssid_modes"]["wpa2_personal"][1] + ssid_name = profile_data["ssid_name"] + print(ssid_name) + mode = "NAT" + vlan = 1 + lf_tools.add_stations(band="5G", num_stations=16, dut=lf_tools.dut_name, ssid_name=ssid_name) + lf_tools.Chamber_View() + sta_list = lf_tools.get_station_list() + print(sta_list) + lf_tools.admin_up_down(sta_list=sta_list, option="up") + sel_stations = ",".join(sta_list[0:8]) + val = [['ul_rate_sel: Per-Station Upload Rate:']] + thr1 = threading.Thread(target=thread_fun, args=(sta_list[8:16],)) + thr1.start() + wct_obj = lf_test.wifi_capacity(instance_name="udp_NAT_upload_5g", mode=mode, vlan_id=vlan, + download_rate="0Gbps", stations=sel_stations, raw_lines=val, batch_size="8", + upload_rate="8Mbps", protocol="UDP-IPv4", duration="120000", + create_stations=False) + + report_name = wct_obj.report_name[0]['LAST']["response"].split(":::")[1].split("/")[-1] + + lf_tools.attach_report_graphs(report_name=report_name) + + csv_val = lf_tools.read_csv_individual_station_throughput(dir_name=report_name, option="upload") + print(type(csv_val)) + print(csv_val) + pass_value = 8 * 0.99 + print("pass value ", pass_value) + pass_fail = [] + if not csv_val: + print("csv file does not exist, station did not got ip, Test failed") + allure.attach(name="Csv Data", body="station did not got ip Test failed.") + assert False + else: + for i in csv_val.values(): + if i >= pass_value: + pass_fail.append(1) + else: + pass_fail.append(0) + if pass_fail.count(0) == 0: + print("Test passed successfully") + allure.attach(name="Csv Data", body=str(csv_val)) + assert True + else: + print(" valueTest failed due to lesser") + allure.attach(name="Csv Data", body=str(csv_val)) + assert False + print("Test Completed... Cleaning up Stations") + + @allure.testcase(url="https://telecominfraproject.atlassian.net/browse/WIFI-5872", name="WIFI-5872") + @pytest.mark.wpa2_personal + @pytest.mark.fiveg + @pytest.mark.udp_download_5g + def test_multi_station_NAT_udp_download_5g(self, lf_test, lf_tools, create_lanforge_chamberview_dut): + # run wifi capacity test here + def thread_fun(station_list): + print(station_list) + time.sleep(60) + lf_tools.admin_up_down(sta_list=station_list, option="down") + print("stations down") + time.sleep(10) + lf_tools.admin_up_down(sta_list=station_list, option="up") + print("stations up") + + lf_tools.reset_scenario() + profile_data = setup_params_general["ssid_modes"]["wpa2_personal"][1] + ssid_name = profile_data["ssid_name"] + print(ssid_name) + mode = "NAT" + vlan = 1 + lf_tools.add_stations(band="5G", num_stations=16, dut=lf_tools.dut_name, ssid_name=ssid_name) + lf_tools.Chamber_View() + sta_list = lf_tools.get_station_list() + print(sta_list) + lf_tools.admin_up_down(sta_list=sta_list, option="up") + sel_stations = ",".join(sta_list[0:8]) + val = [['dl_rate_sel: Per-Station Download Rate:']] + thr1 = threading.Thread(target=thread_fun, args=(sta_list[8:16],)) + thr1.start() + wct_obj = lf_test.wifi_capacity(instance_name="udp_NAT_download_5g", mode=mode, vlan_id=vlan, + download_rate="8Mbps", stations=sel_stations, raw_lines=val, batch_size="8", + upload_rate="0Gbps", protocol="UDP-IPv4", duration="120000", + create_stations=False) + + report_name = wct_obj.report_name[0]['LAST']["response"].split(":::")[1].split("/")[-1] + + lf_tools.attach_report_graphs(report_name=report_name) + + csv_val = lf_tools.read_csv_individual_station_throughput(dir_name=report_name, option="download") + print(type(csv_val)) + print(csv_val) + pass_value = 8 * 0.99 + print("pass value ", pass_value) + pass_fail = [] + if not csv_val: + print("csv file does not exist, station did not got ip, Test failed") + allure.attach(name="Csv Data", body="station did not got ip Test failed.") + assert False + else: + for i in csv_val.values(): + if i >= pass_value: + pass_fail.append(1) + else: + pass_fail.append(0) + if pass_fail.count(0) == 0: + print("Test passed successfully") + allure.attach(name="Csv Data", body=str(csv_val)) + assert True + else: + print(" valueTest failed due to lesser") + allure.attach(name="Csv Data", body=str(csv_val)) + assert False + print("Test Completed... Cleaning up Stations") + + diff --git a/tests/e2e/advanced/multi_association_disassociation/wpa2_personal/test_vlan_association_disassociation.py b/tests/e2e/advanced/multi_association_disassociation/wpa2_personal/test_vlan_association_disassociation.py new file mode 100644 index 000000000..cfa7a6d1f --- /dev/null +++ b/tests/e2e/advanced/multi_association_disassociation/wpa2_personal/test_vlan_association_disassociation.py @@ -0,0 +1,289 @@ +import pytest +import allure +import os +import time +import pandas as pd +import threading + +#pytestmark = [pytest.mark.advance, pytest.mark.multiassodisasso, pytest.mark.vlan] + +setup_params_general = { + "mode": "VLAN", + "ssid_modes": { + "wpa2_personal": [ + {"ssid_name": "ssid_wpa2_2g", "appliedRadios": ["2G"], "security_key": "something", "vlan":100}, + {"ssid_name": "ssid_wpa2_5g", "appliedRadios": ["5G"], "security_key": "something", "vlan":100} + ] + }, + "rf": {}, + "radius": False +} +@pytest.mark.parametrize( + 'setup_profiles', + [setup_params_general], + indirect=True, + scope="class" +) +@pytest.mark.usefixtures("setup_profiles") +class TestMultiAssoDisassoVlan(object): + + @allure.testcase(url="https://telecominfraproject.atlassian.net/browse/WIFI-5873", name="WIFI-5873") + @pytest.mark.wpa2_personal + @pytest.mark.twog + @pytest.mark.udp_upload_2g + def test_multi_station_VLAN_udp_upload_2g(self, lf_test, lf_tools, create_lanforge_chamberview_dut): + # run wifi capacity test here + def thread_fun(station_list): + print(station_list) + time.sleep(60) + lf_tools.admin_up_down(sta_list=station_list, option="down") + print("stations down") + time.sleep(10) + lf_tools.admin_up_down(sta_list=station_list, option="up") + print("stations up") + + lf_tools.reset_scenario() + profile_data = setup_params_general["ssid_modes"]["wpa2_personal"][0] + ssid_name = profile_data["ssid_name"] + print(ssid_name) + mode = "VLAN" + vlan = 100 + lf_tools.add_stations(band="2G", num_stations=16, dut=lf_tools.dut_name, ssid_name=ssid_name) + lf_tools.Chamber_View() + sta_list = lf_tools.get_station_list() + print(sta_list) + lf_tools.admin_up_down(sta_list=sta_list, option="up") + sel_stations = ",".join(sta_list[0:8]) + val = [['ul_rate_sel: Per-Station Upload Rate:']] + thr1 = threading.Thread(target=thread_fun, args=(sta_list[8:16],)) + thr1.start() + wct_obj = lf_test.wifi_capacity(instance_name="udp_VLAN_upload_2g", mode=mode, vlan_id=vlan, + download_rate="0Gbps", stations=sel_stations, raw_lines=val, batch_size="8", + upload_rate="4Mbps", protocol="UDP-IPv4", duration="120000", create_stations=False) + + report_name = wct_obj.report_name[0]['LAST']["response"].split(":::")[1].split("/")[-1] + + lf_tools.attach_report_graphs(report_name=report_name) + + csv_val = lf_tools.read_csv_individual_station_throughput(dir_name=report_name, option="upload") + print(type(csv_val)) + print(csv_val) + pass_value = 4 * 0.99 + print("pass value ", pass_value) + pass_fail = [] + if not csv_val: + print("csv file does not exist, station did not got ip, Test failed") + allure.attach(name="Csv Data", body="station did not got ip Test failed.") + assert False + else: + for i in csv_val.values(): + if i >= pass_value: + pass_fail.append(1) + else: + pass_fail.append(0) + if pass_fail.count(0) == 0: + print("Test passed successfully") + allure.attach(name="Csv Data", body=str(csv_val)) + assert True + else: + print(" valueTest failed due to lesser") + allure.attach(name="Csv Data", body=str(csv_val)) + assert False + print("Test Completed... Cleaning up Stations") + + @allure.testcase(url="https://telecominfraproject.atlassian.net/browse/WIFI-5874", name="WIFI-5874") + @pytest.mark.wpa2_personal + @pytest.mark.twog + @pytest.mark.udp_download_2g + def test_multi_station_VLAN_udp_download_2g(self, lf_test, lf_tools, create_lanforge_chamberview_dut): + # run wifi capacity test here + def thread_fun(station_list): + print(station_list) + time.sleep(60) + lf_tools.admin_up_down(sta_list=station_list, option="down") + print("stations down") + time.sleep(10) + lf_tools.admin_up_down(sta_list=station_list, option="up") + print("stations up") + + lf_tools.reset_scenario() + profile_data = setup_params_general["ssid_modes"]["wpa2_personal"][0] + ssid_name = profile_data["ssid_name"] + print(ssid_name) + mode = "VLAN" + vlan = 100 + lf_tools.add_stations(band="2G", num_stations=16, dut=lf_tools.dut_name, ssid_name=ssid_name) + lf_tools.Chamber_View() + sta_list = lf_tools.get_station_list() + print(sta_list) + lf_tools.admin_up_down(sta_list=sta_list, option="up") + sel_stations = ",".join(sta_list[0:8]) + val = [['dl_rate_sel: Per-Station Download Rate:']] + thr1 = threading.Thread(target=thread_fun, args=(sta_list[8:16],)) + thr1.start() + wct_obj = lf_test.wifi_capacity(instance_name="udp_VLAN_download_2g", mode=mode, vlan_id=vlan, + download_rate="4Mbps", stations=sel_stations, raw_lines=val, batch_size="8", + upload_rate="0Gbps", protocol="UDP-IPv4", duration="120000", + create_stations=False) + + report_name = wct_obj.report_name[0]['LAST']["response"].split(":::")[1].split("/")[-1] + + lf_tools.attach_report_graphs(report_name=report_name) + + csv_val = lf_tools.read_csv_individual_station_throughput(dir_name=report_name, option="download") + print(type(csv_val)) + print(csv_val) + pass_value = 4 * 0.99 + print("pass value ", pass_value) + pass_fail = [] + if not csv_val: + print("csv file does not exist, station did not got ip, Test failed") + allure.attach(name="Csv Data", body="station did not got ip Test failed.") + assert False + else: + for i in csv_val.values(): + if i >= pass_value: + pass_fail.append(1) + else: + pass_fail.append(0) + if pass_fail.count(0) == 0: + print("Test passed successfully") + allure.attach(name="Csv Data", body=str(csv_val)) + assert True + else: + print(" valueTest failed due to lesser") + allure.attach(name="Csv Data", body=str(csv_val)) + assert False + print("Test Completed... Cleaning up Stations") + + @allure.testcase(url="https://telecominfraproject.atlassian.net/browse/WIFI-5875", name="WIFI-5875") + @pytest.mark.wpa2_personal + @pytest.mark.fiveg + @pytest.mark.udp_upload_5g + def test_multi_station_VLAN_udp_upload_5g(self, lf_test, lf_tools, create_lanforge_chamberview_dut): + # run wifi capacity test here + def thread_fun(station_list): + print(station_list) + time.sleep(60) + lf_tools.admin_up_down(sta_list=station_list, option="down") + print("stations down") + time.sleep(10) + lf_tools.admin_up_down(sta_list=station_list, option="up") + print("stations up") + + lf_tools.reset_scenario() + profile_data = setup_params_general["ssid_modes"]["wpa2_personal"][1] + ssid_name = profile_data["ssid_name"] + print(ssid_name) + mode = "VLAN" + vlan = 100 + lf_tools.add_stations(band="5G", num_stations=16, dut=lf_tools.dut_name, ssid_name=ssid_name) + lf_tools.Chamber_View() + sta_list = lf_tools.get_station_list() + print(sta_list) + lf_tools.admin_up_down(sta_list=sta_list, option="up") + sel_stations = ",".join(sta_list[0:8]) + val = [['ul_rate_sel: Per-Station Upload Rate:']] + thr1 = threading.Thread(target=thread_fun, args=(sta_list[8:16],)) + thr1.start() + wct_obj = lf_test.wifi_capacity(instance_name="udp_VLAN_upload_5g", mode=mode, vlan_id=vlan, + download_rate="0Gbps", stations=sel_stations, raw_lines=val, batch_size="8", + upload_rate="8Mbps", protocol="UDP-IPv4", duration="120000", + create_stations=False) + + report_name = wct_obj.report_name[0]['LAST']["response"].split(":::")[1].split("/")[-1] + + lf_tools.attach_report_graphs(report_name=report_name) + + csv_val = lf_tools.read_csv_individual_station_throughput(dir_name=report_name, option="upload") + print(type(csv_val)) + print(csv_val) + pass_value = 8 * 0.99 + print("pass value ", pass_value) + pass_fail = [] + if not csv_val: + print("csv file does not exist, station did not got ip, Test failed") + allure.attach(name="Csv Data", body="station did not got ip Test failed.") + assert False + else: + for i in csv_val.values(): + if i >= pass_value: + pass_fail.append(1) + else: + pass_fail.append(0) + if pass_fail.count(0) == 0: + print("Test passed successfully") + allure.attach(name="Csv Data", body=str(csv_val)) + assert True + else: + print(" valueTest failed due to lesser") + allure.attach(name="Csv Data", body=str(csv_val)) + assert False + print("Test Completed... Cleaning up Stations") + + @allure.testcase(url="https://telecominfraproject.atlassian.net/browse/WIFI-5876", name="WIFI-5876") + @pytest.mark.wpa2_personal + @pytest.mark.fiveg + @pytest.mark.udp_download_5g + def test_multi_station_VLAN_udp_download_5g(self, lf_test, lf_tools, create_lanforge_chamberview_dut): + # run wifi capacity test here + def thread_fun(station_list): + print(station_list) + time.sleep(60) + lf_tools.admin_up_down(sta_list=station_list, option="down") + print("stations down") + time.sleep(10) + lf_tools.admin_up_down(sta_list=station_list, option="up") + print("stations up") + + lf_tools.reset_scenario() + profile_data = setup_params_general["ssid_modes"]["wpa2_personal"][1] + ssid_name = profile_data["ssid_name"] + print(ssid_name) + mode = "VLAN" + vlan = 100 + lf_tools.add_stations(band="5G", num_stations=16, dut=lf_tools.dut_name, ssid_name=ssid_name) + lf_tools.Chamber_View() + sta_list = lf_tools.get_station_list() + print(sta_list) + lf_tools.admin_up_down(sta_list=sta_list, option="up") + sel_stations = ",".join(sta_list[0:8]) + val = [['dl_rate_sel: Per-Station Download Rate:']] + thr1 = threading.Thread(target=thread_fun, args=(sta_list[8:16],)) + thr1.start() + wct_obj = lf_test.wifi_capacity(instance_name="udp_VLAN_download_5g", mode=mode, vlan_id=vlan, + download_rate="8Mbps", stations=sel_stations, raw_lines=val, batch_size="8", + upload_rate="0Gbps", protocol="UDP-IPv4", duration="120000", + create_stations=False) + + report_name = wct_obj.report_name[0]['LAST']["response"].split(":::")[1].split("/")[-1] + + lf_tools.attach_report_graphs(report_name=report_name) + + csv_val = lf_tools.read_csv_individual_station_throughput(dir_name=report_name, option="download") + print(type(csv_val)) + print(csv_val) + pass_value = 8 * 0.99 + print("pass value ", pass_value) + pass_fail = [] + if not csv_val: + print("csv file does not exist, station did not got ip, Test failed") + allure.attach(name="Csv Data", body="station did not got ip Test failed.") + assert False + else: + for i in csv_val.values(): + if i >= pass_value: + pass_fail.append(1) + else: + pass_fail.append(0) + if pass_fail.count(0) == 0: + print("Test passed successfully") + allure.attach(name="Csv Data", body=str(csv_val)) + assert True + else: + print(" valueTest failed due to lesser") + allure.attach(name="Csv Data", body=str(csv_val)) + assert False + print("Test Completed... Cleaning up Stations") + +