diff --git a/py-json/cv_test_manager.py b/py-json/cv_test_manager.py index c4fe6e36..13f8d1a8 100644 --- a/py-json/cv_test_manager.py +++ b/py-json/cv_test_manager.py @@ -26,13 +26,15 @@ class cv_test(LFCliBase): } print("adding- " + text + " " + "to test config") + rsp = self.json_post(req_url, data) - time.sleep(1) + # time.sleep(1) #create a test def create_test(self, test_name, instance): cmd = "cv create '{0}' '{1}'".format(test_name, instance) - self.run_cv_cmd(str(cmd)) + return self.run_cv_cmd(str(cmd)) + #load test scenario def load_test_scenario(self, instance, scenario): @@ -47,7 +49,7 @@ class cv_test(LFCliBase): #start the test def start_test(self, instance): cmd = "cv click '%s' Start" % instance - self.run_cv_cmd(cmd) + return self.run_cv_cmd(cmd) #close test def close_test(self, instance): @@ -61,7 +63,6 @@ class cv_test(LFCliBase): # Send chamber view commands def run_cv_cmd(self, command): - print(command) response_json = [] req_url = "/gui-json/cmd" data = { @@ -90,26 +91,32 @@ class cv_test(LFCliBase): #close the test instance def close_instance(self, instance): cmd = "cv click %s 'Close'" % instance - print(cmd) self.run_cv_cmd(cmd) #To cancel instance def cancel_instance(self, instance): cmd = "cv click %s 'Cancel'" % instance - print(cmd) self.run_cv_cmd(cmd) #Check total ports - def check_ports(self): + def get_ports(self): response = self.json_get("/ports/") return response - def show_changes(self, config_name): + def show_text_blob(self, config_name, blob_test_name): req_url = "/cli-json/show_text_blob" data = { "type": "Plugin-Settings", - "name": config_name, # config name + "name": str(blob_test_name + config_name), # config name "brief": "brief" } rsp = self.json_post(req_url, data) - print(rsp) + + + def rm_text_blob(self, config_name, blob_test_name): + req_url = "/cli-json/rm_text_blob" + data = { + "type": "Plugin-Settings", + "name": str(blob_test_name + config_name), # config name + } + rsp = self.json_post(req_url, data) diff --git a/py-json/cv_test_reports.py b/py-json/cv_test_reports.py index 84a2bd38..16133fbf 100644 --- a/py-json/cv_test_reports.py +++ b/py-json/cv_test_reports.py @@ -1,11 +1,12 @@ -from paramiko import SSHClient +import paramiko from scp import SCPClient class lanforge_reports: def pull_reports(self,hostname="localhost", username="lanforge", password="lanforge",report_location="/home/lanforge/html-reports/"): - ssh = SSHClient() + ssh = paramiko.SSHClient() ssh.load_system_host_keys() + ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh.connect(hostname=hostname,username=username,password=password) with SCPClient(ssh.get_transport()) as scp: diff --git a/py-scripts/create_chamberview.py b/py-scripts/create_chamberview.py index 348b1a40..77861efc 100755 --- a/py-scripts/create_chamberview.py +++ b/py-scripts/create_chamberview.py @@ -7,6 +7,8 @@ Note: To Run this script gui should be opened with pwd (Output : /home/lanforge/LANforgeGUI_5.4.3) ./lfclient.bash -cli-socket 3990 +Note: Scenario names should be different, for each run of this script. + in case of same scenario name scenario will be appended to the same name. Note: Script for creating a chamberview scenario. Run this script to set/create a chamber view scenario. diff --git a/py-scripts/lf_wifi_capacity_test.py b/py-scripts/lf_wifi_capacity_test.py index 04f423e5..2480c8be 100755 --- a/py-scripts/lf_wifi_capacity_test.py +++ b/py-scripts/lf_wifi_capacity_test.py @@ -1,9 +1,27 @@ """ +Note: To Run this script gui should be opened with + + path: cd LANforgeGUI_5.4.3 (5.4.3 can be changed with GUI version) + pwd (Output : /home/lanforge/LANforgeGUI_5.4.3) + ./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: - ./lf_wifi_capacity_test.py -lfmgr "localhost -port "8080" -test_name "WiFi Capacity" -instance_name - "test_instance" -config_name "config_file" -pull_report y -batch_size "1" -loop_iter 1 -protocol "TCP-IPv4" -duration 5000 - if -r flag is set to y. at end of the script you will receive a report dir. from lanforge to your local machine + ./lf_wifi_capacity.py --lfmgr "localhost" --port 8080 --lf_usr lanforge --lf_pswd lanforge + --instance_name "this_inst" --config_name "test_con" --upstream eth1 --batch_size 1 --loop_iter 1 + --protocol "UDP-IPv4" --duration 6000 --pull_report y --auto_add n --stations sta0000 + +Note: + --pull_report == keep it to y, if you want wifi capacity reports at end of the test. + This will pull reports from lanforge to your code directory, + from where you are running this code + keep this to n, if you are running this from lanforge + + --auto_add == if you dont want to add stations manually to wifi capacity test. + keep this as y: This will automatically add all the stations to test + if selected as n: Give station names in --stations argument + --stations == if --auto_add is n, enter stations to use for wifi capacity + """ import sys import os @@ -23,39 +41,56 @@ from cv_test_manager import cv_test as cvtest from cv_commands import chamberview as cv from cv_test_reports import lanforge_reports as lf_rpt + def main(): - parser = argparse.ArgumentParser(description=""" - ./lf_wifi_capacity_test.py -lfmgr "localhost -port "8080" -test_name "WiFi Capacity" -instance_name - "test_instance" -config_name "config_file" -pull_report y -batch_size "1" -loop_iter 1 -protocol "TCP-IPv4" -duration 5000""") - parser.add_argument("-m", "--lfmgr", type=str, default="localhost", + global batch_size, loop_iter, protocol, duration, lf_host, lf_hostport, config_name, auto_add, upstream, stations, instance_name, pull_report, lf_usr, lf_pswd + parser = argparse.ArgumentParser( + description=""" + ./lf_wifi_capacity.py --lfmgr "localhost" --port 8080 --lf_usr lanforge --lf_pswd lanforge + --instance_name "instance" --config_name "wifi_config" --upstream eth1 --batch_size 1 --loop_iter 1 + --protocol "UDP-IPv4" --duration 6000 --pull_report y --auto_add n --stations sta0000 + """) + parser.add_argument("-m", "--lfmgr", type=str, help="address of the LANforge GUI machine (localhost is default)") - parser.add_argument("-o", "--port", type=int, default="8080", + parser.add_argument("-o", "--port", type=int, help="IP Port the LANforge GUI is listening on (8080 is default)") - parser.add_argument("-t", "--test_name", type=str, default="WiFi Capacity", - help="name of test to be run ex. \"WiFi Capacity\"") - parser.add_argument("-i", "--instance_name", type=str, required=True, - help="name of test instance (by default: test_ref)") - parser.add_argument("-c", "--config_name", type=str, required=True, - help="Test config name (by default: DEFAULT)") - parser.add_argument("-r", "--pull_report", type=str, required=True, - help="pull reports from lanforge (by default: y)") - parser.add_argument("-b", "--batch_size", type=str, required=True, - help="config batch size (by default: 1)") - parser.add_argument("-l", "--loop_iter", type=str, required=True, - help="config loop iter (by default: 1)") - parser.add_argument("-p", "--protocol", type=str, required=True, - help="config protocol (by default: TCP-IPv4)") - parser.add_argument("-d", "--duration", type=str, required=True, - help="config duration (by default: 5000)") + parser.add_argument("-lf", "--lf_usr", type=str, + help="Lanforge username to pull reports") + parser.add_argument("-lf_pw", "--lf_pswd", type=str, + help="Lanforge Password to pull reports") + parser.add_argument("-i", "--instance_name", type=str, + help="create test instance") + parser.add_argument("-c", "--config_name", type=str, + help="Config file name") + parser.add_argument("-u", "--upstream", type=str, + help="Upstream port for wifi capacity test ex. eth1") + parser.add_argument("-b", "--batch_size", type=str, + help="station increment ex. 1,2,3") + parser.add_argument("-l", "--loop_iter", type=str, + help="Loop iteration ex. 1") + parser.add_argument("-p", "--protocol", type=str, + help="Protocol ex.TCP-IPv4") + parser.add_argument("-d", "--duration", type=str, + help="duration in ms. ex. 5000") + parser.add_argument("-r", "--pull_report", type=str, + help="Enter y if test reports are need to be pulled from lanforge after test") + parser.add_argument("-a", "--auto_add", type=str, + help="Enter y if all available stations are needs to be added , " + "Enter n if you want to give stations manually in stations argument") + parser.add_argument("-s", "--stations", type=str, + help="in case if you selected n in auto_add enter stations name here ex.sta0000,sta0001") + args = parser.parse_args() + if args.lfmgr is not None: lf_host = args.lfmgr if args.port is not None: lf_hostport = args.port try: - test_name = args.test_name + lf_usr = args.lf_usr + lf_pswd = args.lf_pswd instance_name = args.instance_name config_name = args.config_name batch_size = args.batch_size @@ -63,10 +98,13 @@ def main(): protocol = args.protocol duration = args.duration pull_report = args.pull_report + upstream = args.upstream + stations = args.stations + auto_add = args.auto_add except: - print("Wrong arguments entered") - exit(1) + raise Exception("Wrong argument entered") + test_name = "WiFi Capacity" # Test related settings dict = {"batch_size": "batch_size:" + " " + str(batch_size), @@ -77,58 +115,90 @@ def main(): run_test = cvtest(lf_host, lf_hostport) createCV = cv(lf_host, lf_hostport); # Create a object - port_list = [] + available_ports = [] + stripped_ports = [] - response = run_test.check_ports(); - port_size = json.dumps(len(response["interfaces"])) + run_test.rm_text_blob(config_name, "Wifi-Capacity-") # To delete old config with same name + response = run_test.get_ports(); - for i in range(int(port_size)): - list_val = json.dumps(response["interfaces"][i]) - list_val_ = json.loads(list_val).keys() - list_val_ = str(list_val_).replace("dict_keys(['", "") - list_val_ = str(list_val_).replace("'])", "") - if (list_val_.__contains__("sta") or list_val_.__contains__("eth1")): - port_list.append(list_val_) + ports = response["interfaces"] + d1 = {k: v for e in ports for (k, v) in e.items()} + all_ports = list(d1.keys()) - for i in range(len(port_list)): - add_port = "sel_port-" + str(i) + ":" + " " + port_list[i] - run_test.create_test_config(config_name,"Wifi-Capacity-",add_port) - time.sleep(0.2) + if auto_add == "yes" or auto_add == "y" or auto_add == "Y": + for port in d1.keys(): + if port.__contains__("sta") or port.__contains__(upstream): + available_ports.append(port) + + for i in range(len(available_ports)): + add_port = "sel_port-" + str(i) + ":" + " " + available_ports[i] + run_test.create_test_config(config_name, "Wifi-Capacity-", add_port) + else: + available_ports = [] + stations = stations.split(",") + for str_port in all_ports: + stripped_ports.append(str_port[4:]) # removing Resource from names + + if upstream in stripped_ports: + available_ports.append(all_ports[stripped_ports.index(upstream)]) + else: + raise Exception("Could not find upstream port") + + for sta in range(len(stations)): + if stations[sta] in stripped_ports: + available_ports.append(all_ports[stripped_ports.index(stations[sta])]) + else: + raise Exception("%s not available" % stations[sta]) + + if len(available_ports) == 0: + print("No stations are given") + exit(1) + + for count in range(len(available_ports)): + add_port = "sel_port-" + str(count) + ":" + " " + available_ports[count] + run_test.create_test_config(config_name, "Wifi-Capacity-", add_port) for key, value in dict.items(): - run_test.create_test_config(config_name,"Wifi-Capacity-",value) - time.sleep(0.2) + run_test.create_test_config(config_name, "Wifi-Capacity-", value) + + for i in range(60): + response = run_test.create_test(test_name, instance_name) + 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.create_test(test_name, instance_name) - time.sleep(5) createCV.sync_cv() time.sleep(2) run_test.load_test_config(config_name, instance_name) - time.sleep(2) run_test.auto_save_report(instance_name) - time.sleep(4) - run_test.start_test(instance_name) + + 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:"): + exit(1) while (True): check = run_test.get_report_location(instance_name) location = json.dumps(check[0]["LAST"]["response"]) - print("WiFi Capacity Test Running...") if location != "\"Report Location:::\"": location = location.replace("Report Location:::", "") - print(location) - time.sleep(1) run_test.close_instance(instance_name) - time.sleep(1) run_test.cancel_instance(instance_name) - time.sleep(4) location = location.strip("\"") report = lf_rpt() print(location) - if (pull_report == "yes" ) or (pull_report == "y") or (pull_report == "Y"): - report.pull_reports(hostname=lf_host, username="lanforge", password="lanforge", - report_location=location) + try: + if (pull_report == "yes") or (pull_report == "y") or (pull_report == "Y"): + report.pull_reports(hostname=lf_host, username=lf_usr, password=lf_pswd, + 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 + if __name__ == "__main__": main()