diff --git a/py-json/realm.py b/py-json/realm.py index b555d6e8..6f327744 100755 --- a/py-json/realm.py +++ b/py-json/realm.py @@ -1154,6 +1154,9 @@ class L3CXProfile(BaseProfile): else: return False + """ + coloumn names + """ def monitor(self, duration_sec=60, monitor_interval=1, @@ -1307,9 +1310,9 @@ class L3CXProfile(BaseProfile): import requests import ast try: - systeminfo = ast.literal_eval(requests.get('http://localhost:8080').text) + systeminfo = ast.literal_eval(requests.get('http://localhost:8090').text) except: - systeminfo = ast.literal_eval(requests.get('http://192.168.1.3:8080').text) + systeminfo = ast.literal_eval(requests.get('http://localhost:8090').text) df['LFGUI Release'] = systeminfo['VersionInfo']['BuildVersion'] df['Script Name'] = script_name df['Arguments'] = arguments diff --git a/py-json/test_utility.py b/py-json/test_utility.py new file mode 100644 index 00000000..68235da9 --- /dev/null +++ b/py-json/test_utility.py @@ -0,0 +1,333 @@ + +""" +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 StatusSession(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): + """ + Method to add new Message into a session + """ + self.json_post("/status-msg/" + self.session_id, { + "key": key, + "content-type": "text/plain", + "message": message + }) + + def read(self): + """ + Method to read all the messages for a particular session + """ + keys = [] + for i in self.json_get("/status-msg/"+self.session_id)['messages']: + keys.append(i['key']) + json_uri = "/status-msg/"+self.session_id + "/" + for i in keys: + json_uri = json_uri + i + "," + return self.json_get(json_uri)['messages'] + + + + + + + + +if __name__ == "__main__": + obj = StatusMsg(lfclient_host="localhost", lfclient_port=8090, session_id="01_18_21_20_04_20") + print(obj.read()) + + diff --git a/py-scripts/test_ipv4_variable_time.py b/py-scripts/test_ipv4_variable_time.py index e21a421b..34adc480 100755 --- a/py-scripts/test_ipv4_variable_time.py +++ b/py-scripts/test_ipv4_variable_time.py @@ -89,23 +89,9 @@ class IPV4VariableTime(LFCliBase): self.cx_profile.side_b_min_bps = side_b_min_rate self.cx_profile.side_b_max_bps = side_b_max_rate - - def __get_rx_values(self): - cx_list = self.json_get("endp?fields=name,rx+bytes", debug_=self.debug) - if self.debug: - print(self.cx_profile.created_cx.values()) - print("==============\n", cx_list, "\n==============") - cx_rx_map = {} - for cx_name in cx_list['endpoint']: - if cx_name != 'uri' and cx_name != 'handler': - for item, value in cx_name.items(): - for value_name, value_rx in value.items(): - if value_name == 'rx bytes' and item in self.cx_profile.created_cx.values(): - cx_rx_map[item] = value_rx - return cx_rx_map - def start(self, print_pass=False, print_fail=False): self.station_profile.admin_up() + #to-do- check here if upstream port got IP temp_stas = self.station_profile.station_names.copy() if self.local_realm.wait_for_ip(temp_stas): @@ -204,13 +190,9 @@ python3 ./test_ipv4_variable_time.py optional_args.add_argument('--ap',help='Used to force a connection to a particular AP') optional_args.add_argument('--report_file',help='where you want to store results') optional_args.add_argument('--output_format', help='choose either csv or xlsx') -<<<<<<< HEAD optional_args.add_argument('--a_min', help='--a_min bps rate minimum for side_a', default=256000) optional_args.add_argument('--b_min', help='--b_min bps rate minimum for side_b', default=256000) optional_args.add_argument('--test_duration', help='--test_duration sets the duration of the test', default="2m") -======= - optional_args.add_argument('--show', help='display results of test in terminal',default=True) ->>>>>>> e9a071eff8cfb397f9534ea865f8450e43078cee args = parser.parse_args() num_sta = 2 @@ -261,7 +243,6 @@ python3 ./test_ipv4_variable_time.py ip_var_test.exit_fail() ip_var_test.start(False, False) - try: layer3connections=','.join([[*x.keys()][0] for x in ip_var_test.local_realm.json_get('endp')['endpoint']]) except: