connection_test.py : moved to sandbox

Signed-off-by: Chuck SmileyRekiere <chuck.smileyrekiere@candelatech.com>

py-scripts/sandbox/connection_test.py

@@ -0,0 +1,278 @@
+#!/usr/bin/env python3
+"""
+Candela Technologies Inc.
+
+Info : Standard Script for Connection Testing -  Creates HTML and pdf report as a result (Used for web-console)
+
+"""
+import sys
+import os
+import importlib
+import argparse
+import datetime
+import time
+from test_utility import CreateHTML, StatusMsg
+import pdfkit
+
+if sys.version_info[0] != 3:
+    print("This script requires Python 3")
+    exit(1)
+
+ 
+sys.path.append(os.path.join(os.path.abspath(__file__ + "../../../")))
+
+LFUtils = importlib.import_module("py-json.LANforge.LFUtils")
+lfcli_base = importlib.import_module("py-json.LANforge.lfcli_base")
+LFCliBase = lfcli_base.LFCliBase
+realm = importlib.import_module("py-json.realm")
+Realm = realm.Realm
+
+webconsole_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.getcwd())))
+
+
+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", 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, _exit_on_fail=_exit_on_fail)
+        self.host = lfclient_host
+        self.port = lfclient_port
+        self.radio = radio
+        self.upstream = upstream
+        self.monitor_interval = 1
+        self.sta_prefix = sta_prefix
+        self.sta_start_id = start_id
+        self.pass_criteria = pass_criteria
+        self.num_sta = num_sta
+        self.name_prefix = name_prefix
+        self.ssid = dut_ssid
+        self.security = dut_security
+        self.password = dut_passwd
+        self.session_id = session_id
+        self.test_name = test_name
+        self.test_duration = 1
+        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))
+        self.station_data = dict.fromkeys(station_list)
+        for i in station_list:
+            self.station_data[i] = "None"
+
+        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 + '/'
+
+
+        if not os.path.exists(self.reports_path):
+            os.makedirs(self.reports_path)
+        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)
+        try:
+            self.status_msg.update('2', {"data": 'Initialized...', "data": [], "label": "Client Connectivity Time"})
+        except:
+            pass
+
+    def precleanup(self):
+        sta_list = []
+        for i in self.local_realm.station_list():
+            if (list(i.keys())[0] == '1.1.wlan0'):
+                pass
+            elif (list(i.keys())[0] == '1.1.wlan1'):
+                pass
+            else:
+                sta_list.append(list(i.keys())[0])
+        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)
+        try:
+            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)
+        self.station_profile.set_number_template("00")
+        self.station_profile.set_command_flag("add_sta", "create_admin_down", 1)
+        self.station_profile.set_command_param("set_port", "report_timer", 1500)
+        self.station_profile.set_command_flag("set_port", "rpt_timer", 1)
+        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")
+        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:
+            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']
+        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 = {}
+        cx_time = {}
+        self.timeout = 60
+        for sec in range(self.timeout):
+            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)
+
+                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] == ':'):
+                    associated_map[sta_name] = 1
+                if (sta_status['interface']['ip'] != '0.0.0.0'):
+                    self.ip_map[sta_name] = 1
+            if (len(self.station_profile.station_names) == len(self.ip_map)) and (
+                    len(self.station_profile.station_names) == len(associated_map)):
+                break
+            else:
+                time.sleep(1)
+
+
+        if (len(self.station_profile.station_names) == len(self.ip_map)) and (
+                len(self.station_profile.station_names) == len(associated_map)):
+
+            #("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)
+                #(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)
+                    # #(sta_status)
+                    continue
+                cx_time[sta_name] = sta_status['interface']['cx time (us)']
+        else:
+            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)
+                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)
+                    # #(sta_status)
+                    continue
+                cx_time[sta_name] = sta_status['interface']['cx time (us)']
+        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)
+            self.test_result_data.append(sta_status['interface'])
+
+        offset = 0
+        self.chart_data = {}
+        for data in self.test_result_data:
+            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']] = float(data["cx time (us)"]) / 1000
+                offset +=1
+                data['Result'] = "FAIL"
+            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'
+
+        if offset == 0:
+            summary_result = 'PASS ' + str(len(self.ip_map)) + "/" + str(self.num_sta) + ' Clients are Connected in less than ' + str(self.pass_criteria) + " ms"
+            self.pass_fail = "FAIL"
+        else:
+            summary_result = 'FAIL ' + str(len(self.ip_map)) + "/" + str(self.num_sta) + ' Clients are Connected, and/or Some might got connected in more than ' + str(self.pass_criteria) + " ms"
+            self.pass_fail = "FAIL"
+
+
+        self.html = open(self.reports_path + self.test_name + "_" + self.session_id + ".html", 'w')
+        self.html_data = CreateHTML(path=self.reports_path, test_name=self.test_name, time_snap=str(datetime.datetime.now()), dut_ssid=self.ssid, test_conf_data={"Number of Clients":str(self.num_sta)},
+                                   objective=objective, test_results={"summary": summary_result, "detail": {"keys": self.keys, "data": self.test_result_data}}, chart_data=self.chart_data,
+                                   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)
+
+        try:
+            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)
+        try:
+            self.status_msg.update('7', {"data": 'None', "data": [], "label": "Client Connectivity Time"})
+        except:
+            pass
+    def postcleanup(self):
+        self.station_profile.cleanup(delay=1)
+        try:
+            self.status_msg.update('8', {"data": 'None', "data": [], "label": "Client Connectivity Time"})
+        except:
+            pass
+
+def main():
+    parser = LFCliBase.create_bare_argparse(prog="connection_test.py", formatter_class=argparse.RawTextHelpFormatter,
+                                            epilog="About This Script")
+
+    # Adding More Arguments for custom use
+    parser.add_argument('--ssid', help='--ssid of DUT', default="lexusdut")
+    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=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=300)
+    args = parser.parse_args()
+
+    # Start Test
+    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)
+    obj.precleanup()
+    obj.build()
+    obj.start()
+    obj.stop()
+    obj.postcleanup()
+
+    # #(obj.chart_data)
+    try:
+        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("/", "_").split("P")[0].replace(" ","")
+    return id
+
+
+if __name__ == '__main__':
+    main()