From e4869e715a9fe47677162ed9dc7e16f929477cfa Mon Sep 17 00:00:00 2001 From: shivam Date: Wed, 20 Jan 2021 03:38:34 +0530 Subject: [PATCH] test_utility update --- py-json/LANforge/lfcli_base.py | 1 + py-scripts/connection_test.py | 11 +- py-scripts/test_utility.py | 317 --------------------------------- 3 files changed, 10 insertions(+), 319 deletions(-) delete mode 100644 py-scripts/test_utility.py diff --git a/py-json/LANforge/lfcli_base.py b/py-json/LANforge/lfcli_base.py index fac37d88..208deb23 100644 --- a/py-json/LANforge/lfcli_base.py +++ b/py-json/LANforge/lfcli_base.py @@ -293,6 +293,7 @@ class LFCliBase: debug_=debug_, die_on_error_=self.exit_on_error) json_response = lf_r.json_delete(debug=debug_, die_on_error_=self.halt_on_error) + print(json_response) #debug_printer.pprint(json_response) if (json_response is None) and debug_: print("LFCliBase.json_delete: no entity/response, probabily status 404") diff --git a/py-scripts/connection_test.py b/py-scripts/connection_test.py index 1719bd6d..525b4fa6 100644 --- a/py-scripts/connection_test.py +++ b/py-scripts/connection_test.py @@ -24,6 +24,7 @@ from test_utility import CreateHTML from test_utility import StatusMsg import pdfkit + webconsole_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.getcwd()))) class ConnectionTest(LFCliBase): @@ -97,6 +98,7 @@ class ConnectionTest(LFCliBase): self.status_msg.update('3', {"data": 'Building...', "data": [], "label": "Client Connectivity Time"}) except: pass + def build(self): self.station_profile.use_security(self.security, self.ssid, self.password) @@ -111,6 +113,7 @@ class ConnectionTest(LFCliBase): self.status_msg.update('4', {"data": 'Starting...', "data": [], "label": "Client Connectivity Time"}) except: pass + def update(self, status="None"): for i in self.station_list: self.station_data[i.split(".")[2]] = \ @@ -119,6 +122,7 @@ class ConnectionTest(LFCliBase): self.status_msg.update('5', {"data": 'None', "data": [], "label": "Client Connectivity Time"}) except: pass + def start(self): self.station_profile.admin_up() associated_map = {} @@ -213,6 +217,7 @@ class ConnectionTest(LFCliBase): self.status_msg.update('6', {"data": 'None', "data": [], "label": "Client Connectivity Time"}) except: pass + def stop(self): self.station_profile.admin_down() LFUtils.wait_until_ports_admin_down(port_list=self.station_profile.station_names) @@ -258,12 +263,14 @@ def main(): obj.status_msg.update('10', {"data": 'done...', "data": [], "label": "Client Connectivity Time"}) except: pass - + for i in obj.status_msg.read()['messages']: + print(i) def getSessionID(): x = datetime.datetime.now() id = x.strftime("%x").replace("/","_")+"_"+x.strftime("%x") + "_" + x.strftime("%X").split(":")[0] + "_" + x.strftime("%X").split(":")[1] + "_" + x.strftime("%X").split(":")[2]+str(x).split(".")[1] - id = str(id).replace("/", "_") + id = str(id).replace("/", "_").split("P")[0].replace(" ","") return id + if __name__ == '__main__': main() diff --git a/py-scripts/test_utility.py b/py-scripts/test_utility.py deleted file mode 100644 index f684528b..00000000 --- a/py-scripts/test_utility.py +++ /dev/null @@ -1,317 +0,0 @@ - -""" -Candela Technologies Inc. - -Info : Standard Script for Webconsole Test Utility -Date : - - -""" - -import sys -from pprint import pprint -from uuid import uuid1 - -if 'py-json' not in sys.path: - sys.path.append('../py-json') -from LANforge import LFUtils -from LANforge import lfcli_base -from LANforge.lfcli_base import LFCliBase -from LANforge.LFUtils import * -import realm -from realm import PortUtils -import argparse -import datetime -import time -import matplotlib.pyplot as plt -import threading -import re -import json -import os -webconsole_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.getcwd()))) -updates_path = webconsole_dir + "/web_json/updates.js" - - -class ClientVisualization(LFCliBase, threading.Thread): - def __init__(self, lfclient_host="localhost", lfclient_port=8080, num_clients= 64, max_data= 120, thread_id=None, _debug_on=False, _exit_on_error=False, _exit_on_fail=False): - super().__init__(lfclient_host, lfclient_port, _debug=_debug_on, _halt_on_error=_exit_on_error, - _exit_on_fail=_exit_on_fail) - threading.Thread.__init__(self) - self.num_clients = num_clients - self.max_data = max_data - self._stop_event = threading.Event() - self.client_data = {"down":[], "phantom":[], "ip":[], "scanning":[]} - - def stop(self): - self._stop_event.set() - - def stopped(self): - return self._stop_event.is_set() - - def run(self): - self.start_thread() - - def start_thread(self): - while True: - self.scanning = 0 - self.ip = 0 - self.down = 0 - self.phantom = 0 - for i in self.json_get("/port/list?fields=port,alias,parent%20dev,down,phantom,ip,port%20type")['interfaces']: - - for j in i: - print(i[j]['port type']) - if i[j]['port type'] == "WIFI-STA" and i[j]['parent dev'] == "wiphy1" and i[j]['alias'] != 'wlan1': - #print(j) - if i[j]['down'] == False and i[j]['phantom'] == False and i[j]['ip'] == '0.0.0.0': - self.scanning += 1 - elif i[j]['down'] == False and i[j]['phantom'] == True: - self.phantom += 1 - elif i[j]['down'] == True and i[j]['phantom'] == True: - self.phantom += 1 - self.client_data['phantom'].append(self.phantom) - elif i[j]['down'] == True and i[j]['phantom'] == False: - self.down += 1 - elif i[j]['ip'] != "0.0.0.0": - self.ip += 1 - else: - continue - self.client_data['scanning'].append(self.scanning) - self.client_data['phantom'].append(self.phantom) - self.client_data['down'].append(self.down) - self.client_data['ip'].append(self.ip) - - - for i in self.client_data: - if len(self.client_data[i]) >= self.max_data: - self.client_data[i].pop(0) - time.sleep(1) - if self.stopped(): - break - - -class CreateHTML(): - def __init__(self, path="", test_name="", time_snap="", dut_ssid="", test_conf_data={}, objective="", test_results={}, chart_data={}, chart_params={}): - self.head = """ - - - """+test_name+""" - - -
- -
-
-

"""+test_name+"""

-

"""+time_snap+"""

-
-
-
- """ - self.test_conf = """ - - - - - - - - -
- Test Setup Information -
- Device Under Test - - - - - - - """ - - for i in test_conf_data: - self.test_conf = self.test_conf + """ - - - - """ - - self.test_conf = self.test_conf + """
- SSID - """+dut_ssid+""" -
"""+str(i)+""" - """+test_conf_data[i]+""" -
-
- """ - - self.objective = """ -

Objective

- """+objective+""" -

-
- """ - - if str(test_results['summary']).__contains__("PASS"): - self.summary_results =""" -
- - - - - - - -
- Summary Results -
- """ + test_results['summary'] + """ -
-
- """ - else: - self.summary_results = """ -
- - - - - - - -
- Summary Results -
- """ + test_results['summary'] + """ -
-
- """ - chart_d =[] - chart_label =[] - for i in chart_data: - chart_label.append(i) - chart_d.append(chart_data[i]) - - - - - self.detail_result = """ - -
Detailed Results
- - """ - for index in test_results['detail']['keys']: - self.detail_result = self.detail_result+"" - self.detail_result = self.detail_result +"" - - for data in test_results['detail']['data']: - self.detail_result = self.detail_result + "" - print("shivam") - print(data) - for i in data: - print(data[i]) - if str(data[i]).__contains__("PASS"): - self.detail_result = self.detail_result + "" - elif str(data[i]).__contains__("FAIL"): - self.detail_result = self.detail_result + "" - else: - self.detail_result = self.detail_result + "" - self.detail_result = self.detail_result +"" - - self.chart_data = chart_data - chart_values = [] - for i in self.chart_data: - chart_values.append(self.chart_data[i]) - plt.bar(list(self.chart_data.keys()), chart_values, tick_label=list(self.chart_data.keys())) - - plt.xlabel(chart_params['xlabel']) - # naming the y-axis - plt.ylabel(chart_params['ylabel']) - # plot title - plt.title(chart_params['chart_head']) - plt.xticks(rotation=90, fontsize=8) - plt.tight_layout() - # function to show the plot - plt.savefig(fname=path + "plot.png") - plt.close() - - self.chart = """""" - - - self.end = """
"+index+"
" + str(data[i]) + "" + str(data[i]) + "" + str(data[i]) + "
- - - - """ - self.report = self.head + self.test_conf + self.objective + self.summary_results + self.chart +self.detail_result + self.end - - - -class RuntimeUpdates(): - def __init__(self, session_id, init_data): - self.session_id = session_id - self.init_data = init_data - f = open(updates_path, 'r+') - data = f.read() - f.close() - obj = data[data.find('{'): data.rfind('}') + 1] - obj = re.sub('[\']', '"', obj) - data = json.loads(obj) - print(data) - data["web_updates"].append({"ID": self.session_id, "data": self.init_data}) - print(data) - f = open(updates_path, 'r+') - f.seek(0) - f.truncate() - f.write("var updates = " + str(data) + ";") - f.close() - - def send_update(self, update_data): - f = open(updates_path, 'r+') - data = f.read() - f.close() - obj = data[data.find('{'): data.rfind('}') + 1] - obj = re.sub('[\']', '"', obj) - data = json.loads(obj) - - for update in data["web_updates"]: - if update["ID"] == self.session_id: - update["data"] = update_data - print(data) - f = open(updates_path, 'r+') - f.seek(0) - f.truncate() - f.write("var updates = " + str(data) + ";") - f.close() - - -class StatusMsg(LFCliBase): - def __init__(self, lfclient_host="localhost", lfclient_port=8080, - _deep_clean=False, - session_id="0", - _debug_on=False, - _exit_on_error=False, - _exit_on_fail=False): - super().__init__(lfclient_host, lfclient_port, _debug=_debug_on, _halt_on_error=_exit_on_error, _exit_on_fail=_exit_on_fail) - self.deep_clean = _deep_clean - self.session_id = session_id - self.json_put("/status-msg/" + self.session_id, {}) - - def update(self, key, message): - self.json_post("/status-msg/" + self.session_id, { - "key": key, - "content-type": "text/plain", - "message": message - }) - pass - - - - - - -if __name__ == "__main__": - obj = StatusMsg(lfclient_host="192.168.200.15", lfclient_port=8080, session_id="Connection Test") - obj.update() - -