diff --git a/py-json/cv_test_manager.py b/py-json/cv_test_manager.py index 64e7cd74..b23ab94c 100644 --- a/py-json/cv_test_manager.py +++ b/py-json/cv_test_manager.py @@ -23,48 +23,66 @@ class cv_test(LFCliBase): "name": "Wifi-Capacity-"+config_name, "text": text } - print(data) rsp = self.json_post(req_url, data) time.sleep(1) def create_test(self, test_name, instance): cmd = "cv create '{0}' '{1}'".format(test_name, instance) - self.run_cv_cmd(cmd) + self.run_cv_cmd(str(cmd)) def load_test_scenario(self, instance, scenario): cmd = "cv load '{0}' '{1}'".format(instance, scenario) - print("cmd: ",cmd) self.run_cv_cmd(cmd) def load_test_config(self, test_config, instance): cmd = "cv load '{0}' '{1}'".format(instance, test_config) - print("cmd: ", cmd) self.run_cv_cmd(cmd) def start_test(self, instance): cmd = "cv click '%s' Start" % instance - print("cmd: ",cmd) self.run_cv_cmd(cmd) def close_test(self, instance): cmd = "cv click '%s' 'Close'" % instance - print(cmd) self.run_cv_cmd(cmd) def cancel_test(self, instance): cmd = "cv click '%s' Cancel" % instance - print(cmd) self.run_cv_cmd(cmd) def run_cv_cmd(self, command): # Send chamber view commands print(command) + response_json = [] req_url = "/gui-json/cmd" data = { "cmd": command } - rsp = self.json_post(req_url, data) - print(rsp) - print(data) + debug_par = "" + rsp = self.json_post("/gui-json/cmd%s" % debug_par, data, debug_=False, response_json_list_=response_json) + return response_json + + def auto_save_report(self, instance): + cmd = "cv click %s 'Auto Save Report'" % instance + self.run_cv_cmd(cmd) + + def get_report_location(self, instance): + cmd = "cv get %s 'Report Location:'" % instance + location = self.run_cv_cmd(cmd) + return location + + def save_html(self, instance): + cmd = "cv click %s 'Save HTML'" % instance + self.run_cv_cmd(cmd) + + def close_instance(self, instance): + cmd = "cv click %s 'Close'" % instance + print(cmd) + self.run_cv_cmd(cmd) + + def cancel_instance(self, instance): + cmd = "cv click %s 'Cancel'" % instance + print(cmd) + self.run_cv_cmd(cmd) def check_ports(self): response=self.json_get("/ports/") @@ -74,7 +92,9 @@ class cv_test(LFCliBase): req_url = "/cli-json/show_text_blob" data = { "type": "Plugin-Settings", - "name": "Wifi-Capacity-"+config_name, + "name": config_name,#config name "brief": "brief" } - rsp = self.json_post(req_url, data) \ No newline at end of file + rsp = self.json_post(req_url, data) + print(rsp) + diff --git a/py-json/cvtest_reports.py b/py-json/cvtest_reports.py new file mode 100644 index 00000000..84a2bd38 --- /dev/null +++ b/py-json/cvtest_reports.py @@ -0,0 +1,13 @@ +from paramiko import SSHClient +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.load_system_host_keys() + ssh.connect(hostname=hostname,username=username,password=password) + + with SCPClient(ssh.get_transport()) as scp: + scp.get(report_location,recursive=True) + scp.close() diff --git a/py-scripts/run_test.py b/py-scripts/run_test.py index e5d519fd..4b65c64c 100644 --- a/py-scripts/run_test.py +++ b/py-scripts/run_test.py @@ -3,6 +3,7 @@ import os import argparse import time import json +from os import path if sys.version_info[0] != 3: print("This script requires Python 3") @@ -12,44 +13,42 @@ if 'py-json' not in sys.path: sys.path.append(os.path.join(os.path.abspath('..'), 'py-json')) from cv_test_manager import cv_test as cvtest +from chamberview import chamberview as cv +from cvtest_reports import lanforge_reports as lf_rpt def main(): + config_name = "WFC_scenario1" # Test Config Name (new) + instance_name = "wfc_instance" # Test Instance name + test_name = "WiFi Capacity" # Test name - #Test related settings - batch_size = "3" - loop_iter = "3" - protocol = "UDP-IPv4" - duration = " 7000" - dict = {"batch_size": "batch_size:"+" "+str(batch_size), - "loop_iter": "loop_iter:"+" "+str(loop_iter), - "protocol": "protocol:"+" "+str(protocol), - "duration": "duration:"+" "+str(duration)} + # Test related settings + batch_size = "5" + loop_iter = "1" + protocol = "TCP-IPv4" + duration = " 5000" + dict = {"batch_size": "batch_size:" + " " + str(batch_size), + "loop_iter": "loop_iter:" + " " + str(loop_iter), + "protocol": "protocol:" + " " + str(protocol), + "duration": "duration:" + " " + str(duration)} - config_name = "Test_29" #Test Config Name (new) - instance_name = "wifi_capacity_instance" #Test Instance name - test_name = "WiFi Capacity" #Test name - run_test = cvtest("192.168.200.15","8080") + + run_test = cvtest("192.168.200.21", "8080") + createCV = cv("192.168.200.21", "8080"); # Create a object + port_list = [] response = run_test.check_ports(); - print(response) port_size = json.dumps(len(response["interfaces"])) - print(port_size) - print(response) - print((int(port_size))) - 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")): - print(list_val_) port_list.append(list_val_) - print("54",port_list) for i in range(len(port_list)): add_port = "sel_port-" + str(i) + ":" + " " + port_list[i] @@ -60,19 +59,35 @@ def main(): run_test.create_test_config(config_name, value) time.sleep(0.2) - run_test.create_test(test_name,instance_name) - run_test.load_test_config("DEFAULT",instance_name) - time.sleep(1) - run_test.load_test_config(config_name, instance_name) - time.sleep(1) - run_test.load_test_config(config_name, instance_name) - time.sleep(1) - run_test.load_test_config("DEFAULT", instance_name) + + 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) + 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) + break + + time.sleep(60) + report = lf_rpt() + report.pull_reports(hostname="192.168.200.21", username="lanforge", password="lanforge", + report_location=location) - print(port_list) if __name__ == "__main__": main()