Merge branch 'master' into WIFI-6133

This commit is contained in:
Shivam Thakur
2021-12-13 12:12:52 +05:30
committed by GitHub
7 changed files with 917 additions and 2 deletions

View File

@@ -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,

View File

@@ -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 + '/')

View File

@@ -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")

View File

@@ -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")

View File

@@ -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")