diff --git a/py-json/cv_test_manager.py b/py-json/cv_test_manager.py new file mode 100644 index 00000000..64e7cd74 --- /dev/null +++ b/py-json/cv_test_manager.py @@ -0,0 +1,80 @@ +from LANforge import LFRequest +from LANforge import LFUtils +from LANforge.lfcli_base import LFCliBase +import datetime +import time +import json + +from chamberview import chamberview as cv + + +class cv_test(LFCliBase): + def __init__(self, + lfclient_host="localhost", + lfclient_port=8080, + ): + super().__init__(_lfjson_host=lfclient_host, + _lfjson_port=lfclient_port) + + def create_test_config(self,config_name,text): + req_url = "/cli-json/add_text_blob" + data = { + "type": "Plugin-Settings", + "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) + + 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) + req_url = "/gui-json/cmd" + data = { + "cmd": command + } + rsp = self.json_post(req_url, data) + print(rsp) + print(data) + + def check_ports(self): + response=self.json_get("/ports/") + return response + + def show_changes(self,config_name): + req_url = "/cli-json/show_text_blob" + data = { + "type": "Plugin-Settings", + "name": "Wifi-Capacity-"+config_name, + "brief": "brief" + } + rsp = self.json_post(req_url, data) \ No newline at end of file diff --git a/py-scripts/run_test.py b/py-scripts/run_test.py new file mode 100644 index 00000000..e5d519fd --- /dev/null +++ b/py-scripts/run_test.py @@ -0,0 +1,78 @@ +import sys +import os +import argparse +import time +import json + +if sys.version_info[0] != 3: + print("This script requires Python 3") + exit(1) + +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 + +def main(): + + #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)} + + 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") + + 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] + run_test.create_test_config(config_name, add_port) + time.sleep(0.2) + + for key, value in dict.items(): + 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) + time.sleep(2) + run_test.load_test_config(config_name, instance_name) + run_test.start_test(instance_name) + + + print(port_list) +if __name__ == "__main__": + main()