cv_test_manager.py generic library for chamberview tests

cvtest_reports.py generic library to pull reports folder from lanforge
run_test.py to run chamberview test test

Signed-off-by: SushantBawiskar <sushant.bawiskar@candelatech.com>

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)
+        debug_par = ""
-        print(rsp)
+        rsp = self.json_post("/gui-json/cmd%s" % debug_par, data, debug_=False, response_json_list_=response_json)
-        print(data)
+        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)
+        rsp = self.json_post(req_url, data)
+        print(rsp)
+

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()

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
+    # Test related settings
-    batch_size = "3"
+    batch_size = "5"
-    loop_iter = "3"
+    loop_iter = "1"
-    protocol = "UDP-IPv4"
+    protocol = "TCP-IPv4"
-    duration = " 7000"
+    duration = " 5000"
-    dict = {"batch_size": "batch_size:"+" "+str(batch_size),
+    dict = {"batch_size": "batch_size:" + " " + str(batch_size),
-            "loop_iter": "loop_iter:"+" "+str(loop_iter),
+            "loop_iter": "loop_iter:" + " " + str(loop_iter),
-            "protocol": "protocol:"+" "+str(protocol),
+            "protocol": "protocol:" + " " + str(protocol),
-            "duration": "duration:"+" "+str(duration)}
+            "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)
+    run_test.create_test(test_name, instance_name)
-    time.sleep(1)
+    time.sleep(5)
-    run_test.load_test_config(config_name, instance_name)
+    createCV.sync_cv()
-    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)
+    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()