wlan-lanforge-scripts/py-scripts/connection_test.py

"""
Candela Technologies Inc.

Info : Standard Script for Connection Testing
Date :
Author : Shivam Thakur

"""

import sys

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 Realm
import argparse
import datetime
import time
import os
from test_utility import CreateHTML
import matplotlib.pyplot as plt

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


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



    def precleanup(self):
        print("precleanup started")
        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])
        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")

    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)
        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")
        print("Test Build done")

    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']


    def start(self, print_pass=False, print_fail=False):
        print("Test is starting")
        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)
                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] == ':':
                    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 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")
            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)
                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)
                    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)
                    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)
            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
                data['Result'] = "PASS"
            else:
                self.chart_data[data['alias']] = 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"

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

    def stop(self):
        print("Stopping Test")
        self.station_profile.admin_down()
        LFUtils.wait_until_ports_admin_down(port_list=self.station_profile.station_names)

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


def main():
    # This has --mgr, --mgr_port and --debug
    parser = LFCliBase.create_bare_argparse(prog="layer3_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="local")
    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)
    args = parser.parse_args()
    # args.session_id = "local";
    print(args)

    # Start Test
    obj = ConnectionTest(lfclient_host="192.168.200.15", 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()


if __name__ == '__main__':
    main()