mirror of
https://github.com/Telecominfraproject/wlan-lanforge-scripts.git
synced 2025-11-01 19:28:00 +00:00
Added cv_test_manager.py.py as test library
run_test can be used to run cv_test_manager.py Signed-off-by: SushantBawiskar <sushant.bawiskar@candelatech.com>
This commit is contained in:
80
py-json/cv_test_manager.py
Normal file
80
py-json/cv_test_manager.py
Normal file
@@ -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)
|
||||
78
py-scripts/run_test.py
Normal file
78
py-scripts/run_test.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user