diff --git a/py-scripts/connection_test.py b/py-scripts/connection_test.py index ef52e6f9..1719bd6d 100644 --- a/py-scripts/connection_test.py +++ b/py-scripts/connection_test.py @@ -1,9 +1,7 @@ """ Candela Technologies Inc. -Info : Standard Script for Connection Testing -Date : -Author : Shivam Thakur +Info : Standard Script for Connection Testing - Creates HTML and pdf report as a result (Used for web-console) """ @@ -22,7 +20,8 @@ import datetime import time import os from test_utility import CreateHTML -from test_utility import RuntimeUpdates +# from test_utility import RuntimeUpdates +from test_utility import StatusMsg import pdfkit webconsole_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.getcwd()))) @@ -31,12 +30,11 @@ class ConnectionTest(LFCliBase): def __init__(self, lfclient_host="localhost", lfclient_port=8080, radio="wiphy1", sta_prefix="sta", start_id=0, num_sta=2, - dut_ssid="lexusdut", dut_security="open", dut_passwd="[BLANK]", upstream="eth1", _test_update=None, name_prefix="L3Test", + dut_ssid="lexusdut", dut_security="open", dut_passwd="[BLANK]", upstream="eth1", name_prefix="L3Test", session_id="Layer3Test", test_name="Client/s Connectivity Test", pass_criteria=20, _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) - print("Test is about to start") self.host = lfclient_host self.port = lfclient_port self.radio = radio @@ -53,37 +51,34 @@ class ConnectionTest(LFCliBase): self.session_id = session_id self.test_name = test_name self.test_duration = 1 - self.test_update = _test_update self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port) self.station_profile = self.local_realm.new_station_profile() self.pass_fail = "" - + self.status_msg = StatusMsg(lfclient_host=self.host, lfclient_port=self.port, session_id=self.session_id) station_list = [] for i in range(0, self.num_sta): station_list.append(self.sta_prefix + str(i).zfill(4)) - print(station_list) self.station_data = dict.fromkeys(station_list) for i in station_list: self.station_data[i] = "None" - print(self.station_data) - - self.test_update.send_update({"test_status": '1', "data": 'None', "data": [], "label": "Client Connectivity Time"}) + try: + self.status_msg.update('1', {"data": 'Initializing...', "data": [], "label": "Client Connectivity Time"}) + except: + pass self.reports_path = webconsole_dir+"/reports/" + self.test_name + "_" + self.session_id + '/' - print(self.reports_path) if not os.path.exists(self.reports_path): os.makedirs(self.reports_path) - print("Test is Initialized") self.station_list = LFUtils.portNameSeries(prefix_=self.sta_prefix, start_id_=self.sta_start_id, end_id_=self.num_sta - 1, padding_number_=10000, radio=self.radio) - print(self.station_profile.station_names) - self.test_update.send_update({"test_status": '2', "data": 'None', "data": [], "label": "Client Connectivity Time"}) - + try: + self.status_msg.update('2', {"data": 'Initialized...', "data": [], "label": "Client Connectivity Time"}) + except: + pass def precleanup(self): - print("precleanup started") sta_list = [] for i in self.local_realm.station_list(): if (list(i.keys())[0] == '1.1.wlan0'): @@ -92,19 +87,18 @@ class ConnectionTest(LFCliBase): pass else: sta_list.append(list(i.keys())[0]) - print(sta_list) for sta in sta_list: self.local_realm.rm_port(sta, check_exists=True) time.sleep(1) LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=sta_list, debug=self.debug) - print("precleanup done") - self.test_update.send_update({"test_status": '3', "data": 'None', "data": [], "label": "Client Connectivity Time"}) - + try: + self.status_msg.update('3', {"data": 'Building...', "data": [], "label": "Client Connectivity Time"}) + except: + pass def build(self): - print("Building Test Configuration") self.station_profile.use_security(self.security, self.ssid, self.password) self.station_profile.set_number_template("00") self.station_profile.set_command_flag("add_sta", "create_admin_down", 1) @@ -113,18 +107,19 @@ class ConnectionTest(LFCliBase): self.station_profile.create(radio=self.radio, sta_names_=self.station_list, debug=self.debug) self.local_realm.wait_until_ports_appear(sta_list=self.station_list) self.update(status="build complete") - print("Test Build done") - self.test_update.send_update({"test_status": '4', "data": 'None', "data": [], "label": "Client Connectivity Time"}) - + try: + 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: - print(self.json_get("port/1/1/" + i + "/?fields=ip,ap,down")) self.station_data[i.split(".")[2]] = \ self.json_get("port/1/1/" + i.split(".")[2] + "/?fields=ip,ap,down,phantom&cx%20time%20(us)")['interface'] - self.test_update.send_update({"test_status": '5', "data": 'None', "data": [], "label": "Client Connectivity Time"}) - - def start(self, print_pass=False, print_fail=False): - print("Test is starting") + try: + self.status_msg.update('5', {"data": 'None', "data": [], "label": "Client Connectivity Time"}) + except: + pass + def start(self): self.station_profile.admin_up() associated_map = {} self.ip_map = {} @@ -134,7 +129,7 @@ class ConnectionTest(LFCliBase): for sta_name in self.station_profile.station_names: sta_status = self.json_get("port/1/1/" + str(sta_name).split(".")[2] + "?fields=port,alias,ip,ap", debug_=self.debug) - print(sta_status) + if (sta_status is None or sta_status['interface'] is None) or (sta_status['interface']['ap'] is None): continue if (len(sta_status['interface']['ap']) == 17) and (sta_status['interface']['ap'][-3] == ':'): @@ -147,65 +142,50 @@ class ConnectionTest(LFCliBase): else: time.sleep(1) - if self.debug: - print("sta_list", len(self.station_profile.station_names), self.station_profile.station_names) - print("ip_map", len(self.ip_map), self.ip_map) - print("associated_map", len(associated_map), associated_map) + if (len(self.station_profile.station_names) == len(self.ip_map)) and ( len(self.station_profile.station_names) == len(associated_map)): - self._pass("PASS: All stations associated with IP", print_pass) - print("Test Passed") + + #("Test Passed") for sta_name in self.station_profile.station_names: sta_status = self.json_get("port/1/1/" + str(sta_name).split(".")[2] + "?fields=cx%20time%20(us)", debug_=self.debug) - print(sta_status) + #(sta_status) while sta_status['interface']['cx time (us)'] == 0: sta_status = self.json_get("port/1/1/" + str(sta_name).split(".")[2] + "?fields=cx%20time%20(us)", debug_=self.debug) - print(sta_status) + # #(sta_status) continue cx_time[sta_name] = sta_status['interface']['cx time (us)'] else: - self._fail("FAIL: Not all stations able to associate/get IP", print_fail) - print("sta_list", self.station_profile.station_names) - print("ip_map", self.ip_map) for sta_name in self.ip_map.keys(): sta_status = self.json_get("port/1/1/" + str(sta_name).split(".")[2] + "?fields=cx%20time%20(us)", debug_=self.debug) - print(sta_status) while sta_status['interface']['cx time (us)'] == 0: sta_status = self.json_get("port/1/1/" + str(sta_name).split(".")[2] + "?fields=cx%20time%20(us)", debug_=self.debug) - print(sta_status) + # #(sta_status) continue cx_time[sta_name] = sta_status['interface']['cx time (us)'] - print("associated_map", associated_map) - print("Test Failed") - print(self.ip_map) - print(associated_map) - print("cx time:", cx_time) self.test_result_data = [] self.keys = ["Client Name", "BSSID", "Channel", "Connection Time (ms)", "DHCP (ms)", "IPv4 Address", "MAC Address", "Mode", "Result"] for sta_name in self.station_profile.station_names: sta_status = self.json_get( "port/1/1/" + str(sta_name).split(".")[2] + "?fields=alias,ap,channel,cx%20time%20(us),ip,mac,mode,dhcp%20(ms)", debug_=self.debug) - print("ironman") - print(sta_status['interface']) self.test_result_data.append(sta_status['interface']) - print(self.test_result_data) offset = 0 self.chart_data = {} for data in self.test_result_data: - if (data["cx time (us)"]/1000 <= self.pass_criteria) and (data["cx time (us)"]/1000 > 0): - self.chart_data[data['alias']] = data["cx time (us)"]/1000 + if (int(data["cx time (us)"])/1000 <= self.pass_criteria) and (int(data["cx time (us)"])/1000 > 0): + self.chart_data[data['alias']] = float(data["cx time (us)"])/1000 data['Result'] = "PASS" else: - self.chart_data[data['alias']] = data["cx time (us)"] / 1000 + self.chart_data[data['alias']] = float(data["cx time (us)"]) / 1000 offset +=1 data['Result'] = "FAIL" - data["cx time (us)"] = str(data["cx time (us)"]/1000)+" / "+str(self.pass_criteria)+"ms" + data["cx time (us)"] = str(float(data["cx time (us)"])/1000)+" / "+str(self.pass_criteria)+"ms" objective = 'The Client Connectivity Test is designed to test the Performance of the Access Point. It will tell the Average Connection time that station takes to connect to Wifi Access Point. It will tell you Pass/Fail Criteria and detailed Report for Client Connection' @@ -223,26 +203,31 @@ class ConnectionTest(LFCliBase): chart_params={"chart_head": "Client Connection Time", "xlabel": "Clients", "ylabel": "Connection Time"}) self.html.write(self.html_data.report) self.html.close() + options = { + "enable-local-file-access": None + } + pdfkit.from_file(self.reports_path + self.test_name + "_" + self.session_id + ".html", + self.reports_path + self.test_name + "_" + self.session_id + '_report.pdf', options=options) - self.test_update.send_update({"test_status": '6', "data": 'None', "data": [], "label": "Client Connectivity Time"}) - + try: + self.status_msg.update('6', {"data": 'None', "data": [], "label": "Client Connectivity Time"}) + except: + pass def stop(self): - print("Stopping Test") self.station_profile.admin_down() LFUtils.wait_until_ports_admin_down(port_list=self.station_profile.station_names) - self.test_update.send_update({"test_status": '7', "data": 'None', "data": [], "label": "Client Connectivity Time"}) - + try: + self.status_msg.update('7', {"data": 'None', "data": [], "label": "Client Connectivity Time"}) + except: + pass def postcleanup(self): - self.station_profile.cleanup() - LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, - port_list=self.station_profile.station_names, - debug=self.debug) - print("Test Completed") - self.test_update.send_update({"test_status": '8', "data": 'None', "data": [], "label": "Client Connectivity Time"}) - + self.station_profile.cleanup(delay=1) + try: + self.status_msg.update('8', {"data": 'None', "data": [], "label": "Client Connectivity Time"}) + except: + pass def main(): - # This has --mgr, --mgr_port and --debug parser = LFCliBase.create_bare_argparse(prog="connection_test.py", formatter_class=argparse.RawTextHelpFormatter, epilog="About This Script") @@ -251,27 +236,34 @@ def main(): parser.add_argument('--passwd', help='--passwd of dut', default="[BLANK]") parser.add_argument('--radio', help='--radio to use on LANforge', default="wiphy1") parser.add_argument('--security', help='--security of dut', default="open") - parser.add_argument('--session_id', help='--session_id is for websocket', default="local") + parser.add_argument('--session_id', help='--session_id is for websocket', default=getSessionID()) parser.add_argument('--test_name', help='--test_name is for webconsole reports', default="Client Connectivity Test") parser.add_argument('--num_clients', type=int, help='--num_sta is number of stations you want to create', default=2) - parser.add_argument('--pass_criteria', type=int, help='--pass_criteria is pass criteria for connection Time', default=50) + parser.add_argument('--pass_criteria', type=int, help='--pass_criteria is pass criteria for connection Time', default=300) args = parser.parse_args() - # args.session_id = "local"; - print(args) - update = RuntimeUpdates(args.session_id, {"test_status": '0', "data": 'None', "data": [], "label": "Client Connectivity Time"}) + # Start Test - obj = ConnectionTest(lfclient_host="192.168.200.12", lfclient_port=args.mgr_port, + obj = ConnectionTest(lfclient_host=args.mgr, lfclient_port=args.mgr_port, session_id=args.session_id, test_name=args.test_name, dut_ssid=args.ssid, dut_passwd=args.passwd, dut_security=args.security, - num_sta=args.num_clients, radio=args.radio, pass_criteria=args.pass_criteria, _test_update=update) + num_sta=args.num_clients, radio=args.radio, pass_criteria=args.pass_criteria) obj.precleanup() obj.build() obj.start() obj.stop() obj.postcleanup() - print(obj.chart_data) - update.send_update({"test_status": '10', "data": obj.chart_data, "label": ["Client Names","Client Connectivity Time (ms)"], "result": obj.pass_fail}) + # #(obj.chart_data) + try: + obj.status_msg.update('10', {"data": 'done...', "data": [], "label": "Client Connectivity Time"}) + except: + pass + +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("/", "_") + return id if __name__ == '__main__': main() diff --git a/py-scripts/test_utility.py b/py-scripts/test_utility.py index 2f9ae5fd..f684528b 100644 --- a/py-scripts/test_utility.py +++ b/py-scripts/test_utility.py @@ -9,6 +9,9 @@ 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 @@ -282,21 +285,33 @@ class RuntimeUpdates(): f.close() -if __name__ == "__main__": - thread1 = ClientVisualization(lfclient_host="192.168.200.15", thread_id=1) - thread1.start() - for i in range(0, 100): - time.sleep(1) - #print(thread1.client_data) - thread1.stop() - # obj = RuntimeUpdates("1", {"test_status": 1, "data": "None"}) - # for i in range(0, 10): - # time.sleep(3) - # print(i) - # obj.send_update({"test_status": i, "data": "None"}) - # thread1 = ClientVisualization(lfclient_host="192.168.200.15", thread_id=1) - # thread1.start() - # for i in range(30): - # print(thread1.client_data) - # thread1.stop() +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() +