scottk/wlan-lanforge-scripts
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-lanforge-scripts.git synced 2025-11-01 03:07:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

- added stations creation to wifi capacity test

Browse Source 
- added example

Signed-off-by: SushantBawiskar <sushant.bawiskar@candelatech.com>
This commit is contained in:
SushantBawiskar
2021-04-14 18:25:56 +05:30
parent 6109a5fcc1
commit 2a24c02ca8
1 changed files with 203 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

311
py-scripts/lf_wifi_capacity_test.py
View File

@@ -8,11 +8,17 @@ Note: To Run this script gui should be opened with
./lfclient.bash -cli-socket 3990
Note: This is a test file which will run a wifi capacity test.
ex. on how to run this script:
ex. on how to run this script (if stations are available in lanforge):
./lf_wifi_capacity_test.py --mgr localhost --port 8080 --lf_user lanforge --lf_password lanforge
--instance_name this_inst --config_name test_con --upstream 1.1.eth1 --batch_size 1 --loop_iter 1
--protocol UDP-IPv4 --duration 6000 --pull_report --stations 1.1.sta0000,1.1.sta0002
ex. on how to run this script (to create new stations):
./lf_wifi_capacity_test.py --mgr localhost --port 8080 --lf_user lanforge --lf_password lanforge
--instance_name wct_instance --config_name wifi_config --upstream 1.1.eth1 --batch_size 1 --loop_iter 1
--protocol UDP-IPv4 --duration 6000 --pull_report --stations 1.1.sta0000,1.1.sta0001
--create_stations --radio "wiphy0" --ssid "" --security "open" --paswd "[BLANK]"
Note:
--pull_report == If specified, this will pull reports from lanforge to your code directory,
from where you are running this code
@@ -39,13 +45,171 @@ from cv_commands import chamberview as cv
from cv_test_reports import lanforge_reports as lf_rpt
class WiFiCapacityTest(cvtest):
def __init__(self,
lf_host="localhost",
lf_port=8080,
lf_user="lanforge",
lf_password="lanforge",
instance_name="wct_instance",
config_name="wifi_config",
upstream="eth1",
batch_size="1",
loop_iter="1",
protocol="UDP-IPv4",
duration="5000",
pull_report=False,
load_old_cfg=False,
upload_rate="10Mbps",
download_rate="1Gbps",
sort="interleave",
stations="",
create_stations=False,
radio="wiphy0",
security="open",
paswd="[BLANK]",
ssid=""
):
super().__init__(lfclient_host=lf_host, lfclient_port=lf_port)
self.lf_host = lf_host
self.lf_port = lf_port
self.lf_user = lf_user
self.lf_password =lf_password
self.createCV = cv(lf_host, lf_port);
self.station_profile = self.new_station_profile()
self.pull_report = pull_report
self.load_old_cfg = load_old_cfg
self.instance_name = instance_name
self.config_name = config_name
self.test_name = "WiFi Capacity"
self.batch_size = batch_size
self.loop_iter = loop_iter
self.protocol = protocol
self.duration = duration
self.upload_rate = upload_rate
self.download_rate = download_rate
self.upstream = upstream
self.sort = sort
self.stations = stations
self.create_stations =create_stations
self.security = security
self.ssid = ssid
self.paswd = paswd
self.radio = radio
def setup(self):
if self.create_stations and self.stations != "":
sta = self.stations.split(",")
self.station_profile.cleanup(sta)
self.station_profile.use_security(self.security, self.ssid, self.paswd)
self.station_profile.create(radio=self.radio, sta_names_=sta, debug=self.debug)
self.station_profile.admin_up()
self.wait_for_ip(station_list=sta)
print("stations created")
def run(self):
self.createCV.sync_cv()
time.sleep(2)
self.rm_text_blob(self.config_name, "Wifi-Capacity-") # To delete old config with same name
self.show_text_blob(None, None, False)
# Test related settings
cfg_options = ["batch_size: " + str(self.batch_size),
"loop_iter: " + str(self.loop_iter),
"protocol: " + str(self.protocol),
"duration: " + str(self.duration),
"ul_rate: " + self.upload_rate,
"dl_rate: " + self.download_rate,
]
port_list = [self.upstream]
if self.stations == "":
stas = self.station_map() # See realm
for eid in stas.keys():
port_list.append(eid)
else:
stas = self.stations.split(",")
for s in stas:
port_list.append(s)
idx = 0
for eid in port_list:
add_port = "sel_port-" + str(idx) + ": " + eid
self.create_test_config(self.config_name, "Wifi-Capacity-", add_port)
idx += 1
for value in cfg_options:
self.create_test_config(self.config_name, "Wifi-Capacity-", value)
# Request GUI update its text blob listing.
self.show_text_blob(self.config_name, "Wifi-Capacity-", False)
# Hack, not certain if the above show returns before the action has been completed
# or not, so we sleep here until we have better idea how to query if GUI knows about
# the text blob.
time.sleep(5)
load_old = "false"
if self.load_old_cfg:
load_old = "true"
for i in range(60):
response = self.create_test(self.test_name, self.instance_name, load_old)
d1 = {k: v for e in response for (k, v) in e.items()}
if d1["LAST"]["response"] == "OK":
break
else:
time.sleep(1)
self.load_test_config(self.config_name, self.instance_name)
self.auto_save_report(self.instance_name)
if self.sort == 'linear':
cmd = "cv click '%s' 'Linear Sort'" % self.instance_name
self.run_cv_cmd(cmd)
if self.sort == 'interleave':
cmd = "cv click '%s' 'Interleave Sort'" % self.instance_name
self.run_cv_cmd(cmd)
response = self.start_test(self.instance_name)
d1 = {k: v for e in response for (k, v) in e.items()}
if d1["LAST"]["response"].__contains__("Could not find instance:"):
print("ERROR: start_test failed: ", d1["LAST"]["response"], "\n");
# pprint(response)
exit(1)
while (True):
check = self.get_report_location(self.instance_name)
location = json.dumps(check[0]["LAST"]["response"])
if location != "\"Report Location:::\"":
location = location.replace("Report Location:::", "")
self.close_instance(self.instance_name)
self.cancel_instance(self.instance_name)
location = location.strip("\"")
report = lf_rpt()
print(location)
try:
if self.pull_report:
report.pull_reports(hostname=self.lf_host, username=self.lf_user, password=self.lf_password,
report_location=location)
except:
raise Exception("Could not find Reports")
break
self.rm_text_blob(self.config_name, "Wifi-Capacity-") # To delete old config with same name
def main():
global lf_host, lf_hostport, config_name, instance_name
parser = argparse.ArgumentParser(
description="""
./lf_wifi_capacity_test.py --lfgui localhost --port 8080 --lf_user lanforge --lf_password lanforge
./lf_wifi_capacity_test.py --mgr localhost --port 8080 --lf_user lanforge --lf_password lanforge
--instance_name wct_instance --config_name wifi_config --upstream 1.1.eth1 --batch_size 1 --loop_iter 1
--protocol UDP-IPv4 --duration 6000 --pull_report --stations 1.1.sta0000,1.1.sta0001
--protocol UDP-IPv4 --duration 6000 --pull_report --stations 1.1.sta0000,1.1.sta0001
--create_stations --radio "wiphy0" --ssid "" --security "open" --paswd "[BLANK]"
""")
parser.add_argument("-m", "--mgr", type=str, default="localhost",
help="address of the LANforge GUI machine (localhost is default)")
@@ -81,112 +245,43 @@ def main():
help="Select station sorting behaviour: none | interleave | linear Default is interleave.")
parser.add_argument("-s", "--stations", type=str, default="",
help="If specified, these stations will be used. If not specified, all available stations will be selected. Example: 1.1.sta001,1.1.wlan0,...")
parser.add_argument("-cs", "--create_stations", default=False, action='store_true',
help="create stations in lanforge (by default: False)")
parser.add_argument("-radio", "--radio", default="wiphy0",
help="create stations in lanforge at this radio (by default: wiphy0)")
parser.add_argument("-ssid", "--ssid", default="",
help="ssid name")
parser.add_argument("-security", "--security", default="open",
help="ssid Security type")
parser.add_argument("-paswd", "--paswd", default="[BLANK]",
help="ssid Password")
args = parser.parse_args()
lf_host = args.mgr
lf_hostport = args.port
instance_name = args.instance_name
config_name = args.config_name
test_name = "WiFi Capacity"
run_test = cvtest(lf_host, lf_hostport)
createCV = cv(lf_host, lf_hostport); # Create a object
createCV.sync_cv()
time.sleep(2)
run_test.rm_text_blob(config_name, "Wifi-Capacity-") # To delete old config with same name
run_test.show_text_blob(None, None, False)
# Test related settings
cfg_options = ["batch_size: " + str(args.batch_size),
"loop_iter: " + str(args.loop_iter),
"protocol: " + str(args.protocol),
"duration: " + str(args.duration),
"ul_rate: " + args.upload_rate,
"dl_rate: " + args.download_rate,
]
port_list = [args.upstream]
if args.stations == "":
stas = run_test.station_map() # See realm
for eid in stas.keys():
port_list.append(eid)
else:
stas = args.stations.split(",")
for s in stas:
port_list.append(s)
idx = 0
for eid in port_list:
add_port = "sel_port-" + str(idx) + ": " + eid
run_test.create_test_config(config_name, "Wifi-Capacity-",add_port)
idx += 1
for value in cfg_options:
run_test.create_test_config(config_name, "Wifi-Capacity-", value)
# Request GUI update its text blob listing.
run_test.show_text_blob(config_name, "Wifi-Capacity-", False)
# Hack, not certain if the above show returns before the action has been completed
# or not, so we sleep here until we have better idea how to query if GUI knows about
# the text blob.
time.sleep(5)
load_old = "false"
if args.load_old_cfg:
load_old = "true"
for i in range(60):
response = run_test.create_test(test_name, instance_name, load_old)
d1 = {k: v for e in response for (k, v) in e.items()}
if d1["LAST"]["response"] == "OK":
break
else:
time.sleep(1)
run_test.load_test_config(config_name, instance_name)
run_test.auto_save_report(instance_name)
if args.sort == 'linear':
cmd = "cv click '%s' 'Linear Sort'" % instance_name
run_test.run_cv_cmd(cmd)
if args.sort == 'interleave':
cmd = "cv click '%s' 'Interleave Sort'" % instance_name
run_test.run_cv_cmd(cmd)
response = run_test.start_test(instance_name)
d1 = {k: v for e in response for (k, v) in e.items()}
if d1["LAST"]["response"].__contains__("Could not find instance:"):
print("ERROR: start_test failed: ", d1["LAST"]["response"], "\n");
#pprint(response)
exit(1)
while (True):
check = run_test.get_report_location(instance_name)
location = json.dumps(check[0]["LAST"]["response"])
if location != "\"Report Location:::\"":
location = location.replace("Report Location:::", "")
run_test.close_instance(instance_name)
run_test.cancel_instance(instance_name)
location = location.strip("\"")
report = lf_rpt()
print(location)
try:
if args.pull_report:
report.pull_reports(hostname=lf_host, username=args.lf_user, password=args.lf_password,
report_location=location)
except:
raise Exception("Could not find Reports")
break
run_test.rm_text_blob(config_name, "Wifi-Capacity-") # To delete old config with same name
WFC_Test = WiFiCapacityTest(lf_host=args.mgr,
lf_port=args.port,
lf_user=args.lf_user,
lf_password=args.lf_password,
instance_name=args.instance_name,
config_name=args.config_name,
upstream=args.upstream,
batch_size=args.batch_size,
loop_iter=args.loop_iter,
protocol=args.protocol,
duration=args.duration,
pull_report=args.pull_report,
load_old_cfg=args.load_old_cfg,
download_rate=args.download_rate,
upload_rate=args.upload_rate,
sort=args.sort,
stations=args.stations,
create_stations=args.create_stations,
radio =args.radio,
ssid=args.ssid,
security =args.security,
paswd =args.paswd ,
)
WFC_Test.setup()
WFC_Test.run()
if __name__ == "__main__":