diff --git a/py-scripts/lf_wifi_capacity_test.py b/py-scripts/lf_wifi_capacity_test.py index e3116f70..53d6ec44 100755 --- a/py-scripts/lf_wifi_capacity_test.py +++ b/py-scripts/lf_wifi_capacity_test.py @@ -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__":