#!/usr/bin/env python3 """ NAME: ftp_test.py PURPOSE: will create stations and endpoints to generate and verify layer-4 traffic over an ftp connection. find out download/upload time of each client according to file size. This script will monitor the bytes-rd attribute of the endpoints. SETUP: Create a file to be downloaded linux: fallocate -l example fallocate -l 2M ftp_test.txt EXAMPLE: './lf_ftp_test.py --ssid "jedway-wap2-x2048-5-3" --passwd "jedway-wpa2-x2048-5-3" --security wpa2 --bands "5G" --direction "Download" \ --file_size "2MB" --num_stations 2 INCLUDE_IN_README -Jitendrakumar Kushavah Copyright 2021 Candela Technologies Inc License: Free to distribute and modify. LANforge systems must be licensed. """ import sys import os import importlib import paramiko import argparse from datetime import datetime import time 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 ftp_html = importlib.import_module("py-scripts.ftp_html") class ftp_test(LFCliBase): def __init__(self, lfclient_host="localhost", lfclient_port=8080, radio = "wiphy0", sta_prefix="sta", start_id=0, num_sta= None, dut_ssid=None,dut_security=None, dut_passwd=None, file_size=None, band=None, upstream="eth1",_debug_on=False, _exit_on_error=False, _exit_on_fail=False, direction= None,user='lanforge'): super().__init__(lfclient_host, lfclient_port, _debug=_debug_on, _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.sta_prefix = sta_prefix self.sta_start_id = start_id self.num_sta = num_sta self.ssid = dut_ssid self.security = dut_security self.password = dut_passwd self.requests_per_ten = 1 self.band=band self.file_size=file_size self.direction=direction self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port) self.station_profile = self.local_realm.new_station_profile() self.cx_profile = self.local_realm.new_http_profile() self.port_util = realm.PortUtils(self.local_realm) self.cx_profile.requests_per_ten = self.requests_per_ten self.user = user print("Test is Initialized") def set_values(self): #This method will set values according user input if self.band == "5G": self.radio = ["wiphy0"] # need to pass in the radios if self.file_size == "2MB": #providing time duration for Pass or fail criteria self.duration = self.convert_min_in_time(1) elif self.file_size == "500MB": self.duration = self.convert_min_in_time(1) # 30 elif self.file_size == "1000MB": self.duration = self.convert_min_in_time(1) # 50 else: self.duration = self.convert_min_in_time(10) # 10 elif self.band == "2.4G": self.radio = ["wiphy0"] # need to pass in the radios if self.file_size == "2MB": self.duration = self.convert_min_in_time(1) # 2 elif self.file_size == "500MB": self.duration = self.convert_min_in_time(1) # 60 elif self.file_size == "1000MB": self.duration = self.convert_min_in_time(1) # 80 else: self.duration = self.convert_min_in_time(10) # 10 elif self.band == "Both": self.radio = ["wiphy0", "wiphy0"] # need to pass in the radios #if Both then number of stations are half for 2.4G and half for 5G self.num_sta = self.num_sta // 2 print(self.num_sta) if self.file_size == "2MB": self.duration = self.convert_min_in_time(1) # 2 elif self.file_size == "500MB": self.duration = self.convert_min_in_time(1) # 60 elif self.file_size == "1000MB": self.duration = self.convert_min_in_time(1) # 80 else: self.duration = self.convert_min_in_time(10) # 10 self.file_size_bytes=int(self.convert_file_size_in_Bytes(self.file_size)) def precleanup(self): self.count=0 #delete everything in the GUI before starting the script '''try: self.local_realm.load("BLANK") except: print("Couldn't load 'BLANK' Test configurations") ''' for rad in self.radio: if rad == "wiphy0": #select mode(All stations will connects to 5G) self.station_profile.mode = 10 self.count=self.count+1 elif rad == "wiphy0": # This probably is not the best selection mode # select mode(All stations will connects to 2.4G) self.station_profile.mode = 6 self.count = self.count + 1 #check Both band if both band then for 2.4G station id start with 20 if self.count == 2: self.sta_start_id = self.num_sta self.num_sta = 2 * (self.num_sta) #if Both band then first 20 stations will connects to 5G self.station_profile.mode = 10 self.cx_profile.cleanup() #create station list with sta_id 20 self.station_list1 = LFUtils.portNameSeries(prefix_=self.sta_prefix, start_id_=self.sta_start_id, end_id_=self.num_sta - 1, padding_number_=10000, radio=rad) #cleanup station list which started sta_id 20 self.station_profile.cleanup(self.station_list1, debug_=self.debug) LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=self.station_list, debug=self.debug) return #clean layer4 ftp traffic self.cx_profile.cleanup() 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=rad) #cleans stations self.station_profile.cleanup(self.station_list , delay=1, debug_=self.debug) LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=self.station_list, debug=self.debug) time.sleep(1) print("precleanup done") def build(self): #set ftp self.port_util.set_ftp(port_name=self.local_realm.name_to_eid(self.upstream)[2], resource=1, on=True) for rad in self.radio: #station build 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=rad, sta_names_=self.station_list, debug=self.debug) self.local_realm.wait_until_ports_appear(sta_list=self.station_list) self.station_profile.admin_up() if self.local_realm.wait_for_ip(self.station_list): self._pass("All stations got IPs") else: self._fail("Stations failed to get IPs") exit(1) #building layer4 self.cx_profile.direction ="dl" self.cx_profile.dest = "/dev/null" print('DIRECTION',self.direction) if self.direction == "Download": self.cx_profile.create(ports=self.station_profile.station_names, ftp_ip="10.40.0.1/ftp_test.txt", sleep_time=.5,debug_=self.debug,suppress_related_commands_=True, ftp=True, user="lanforge", passwd="lanforge", source="") elif self.direction == "Upload": dict_sta_and_ip = {} #data from GUI for find out ip addr of each station data = self.json_get("ports/list?fields=IP") # This loop for find out proper ip addr and station name for i in self.station_list: for j in data['interfaces']: for k in j: if i == k: dict_sta_and_ip[k] = j[i]['ip'] #list of ip addr of all stations ip = list(dict_sta_and_ip.values()) eth_list = [] client_list = [] #list of all stations for i in range(len(self.station_list)): client_list.append(self.station_list[i][4:]) #list of upstream port eth_list.append(self.upstream) #create layer for connection for upload for client_num in range(len(self.station_list)): self.cx_profile.create(ports=eth_list, ftp_ip=ip[client_num] + "/ftp_test_upload.txt", sleep_time=.5, debug_=self.debug, suppress_related_commands_=True, ftp=True, user="lanforge", passwd="lanforge", source="", upload_name=client_list[client_num]) #check Both band present then build stations with another station list if self.count == 2: self.station_list = self.station_list1 # if Both band then another 20 stations will connects to 2.4G self.station_profile.mode = 6 print("Test Build done") def start(self, print_pass=False, print_fail=False): for rad in self.radio: self.cx_profile.start_cx() print("Test Started") def stop(self): self.cx_profile.stop_cx() self.station_profile.admin_down() def postcleanup(self): self.cx_profile.cleanup() self.local_realm.load("BLANK") self.station_profile.cleanup(self.station_profile.station_names, delay=1, debug_=self.debug) LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=self.station_profile.station_names, debug=self.debug) #Create file for given file size def file_create(self): if os.path.isfile("/home/lanforge/ftp_test.txt"): os.remove("/home/lanforge/ftp_test.txt") os.system("fallocate -l " +self.file_size +" /home/lanforge/ftp_test.txt") print("File creation done", self.file_size) #convert file size MB or GB into Bytes def convert_file_size_in_Bytes(self,size): if (size.endswith("MB")) or (size.endswith("Mb")) or (size.endswith("GB")) or (size.endswith("Gb")): if (size.endswith("MB")) or (size.endswith("Mb")): return float(size[:-2]) * 10**6 elif (size.endswith("GB")) or (size.endswith("Gb")): return float(size[:-2]) * 10**9 def my_monitor(self,time1): #data in json format data = self.json_get("layer4/list?fields=bytes-rd") print("layer4/list?fields=bytes-read: {}".format(data)) #list of layer 4 connections name self.data1 = [] for i in range(self.num_sta): if self.num_sta == 1: # the station num is 1 , yet the station is 0 print("i: {} self.num_sta: {}".format(i,self.num_sta)) print("data['endpoint'][{}]: {}".format(i,data['endpoint'])) print("data list: {}".format((str(list(data['endpoint'].keys())))[2:-2])) print("data list: {}".format((str(list(data['endpoint'].keys())))[2:-2])) #self.data1.append((str(list(data['endpoint']['name'])))) self.data1.append((str((data['endpoint']['name'])))) else: # the station num is 1 , yet the station is 0 print("i: {} self.num_sta: {}".format(i,self.num_sta)) print("data['endpoint'][{}]: {}".format(i,data['endpoint'][i])) print("data list: {}".format((str(list(data['endpoint'][i].keys())))[2:-2])) self.data1.append((str(list(data['endpoint'][i].keys())))[2:-2]) data2 = self.data1 print("data1: {}".format(self.data1)) print("data2: {}".format(data2)) list_of_time = [] list1 = [] list2 = [] for i in range(self.num_sta): list_of_time.append(0) print("list_of_time: {}".format(list_of_time)) num_sta_finished = 0 while list_of_time.count(0) != 0: #run script upto given time if str(datetime.now()- time1) >= self.duration: break for i in range(self.num_sta): data = self.json_get("layer4/list?fields=bytes-rd") #print("data from bytes-rd: {}".format(data)) if self.num_sta == 1: #reading uc-avg data in json format uc_avg= self.json_get("layer4/list?fields=uc-avg") #print("layer4/list?fields=uc-avg: {}".format(uc_avg)) if data['endpoint']['bytes-rd'] <= self.file_size_bytes: data = self.json_get("layer4/list?fields=bytes-rd") if data['endpoint']['bytes-rd'] >= self.file_size_bytes: list1.append(i) print("list1: {} list2: {}".format(list1,list2)) if list1.count(i) == 1: list2.append(i) list1 = list2 print("CX_{} list1: {} list2: {}".format(data2[0],list1,list2)) #stop station after download or upload file with particular size self.json_post("/cli-json/set_cx_state", { "test_mgr": "default_tm", "cx_name": "CX_" + data2[0], "cx_state": "STOPPED" }, debug_=self.debug) list_of_time[i] = round(int(uc_avg['endpoint']['uc-avg'])/1000,1) num_sta_finished += 1 if num_sta_finished >= self.num_sta: break else: #reading uc-avg data in json format uc_avg= self.json_get("layer4/list?fields=uc-avg") if data['endpoint'][i][data2[i]]['bytes-rd'] <= self.file_size_bytes: data = self.json_get("layer4/list?fields=bytes-rd") if data['endpoint'][i][data2[i]]['bytes-rd'] >= self.file_size_bytes: list1.append(i) print("list1: {} list2: {}".format(list1,list2)) if list1.count(i) == 1: list2.append(i) list1 = list2 print("CX_{} list1: {} list2: {}".format(data2[i],list1,list2)) #stop station after download or upload file with particular size self.json_post("/cli-json/set_cx_state", { "test_mgr": "default_tm", "cx_name": "CX_" + data2[i], "cx_state": "STOPPED" }, debug_=self.debug) list_of_time[i] = round(int(uc_avg['endpoint'][i][data2[i]]['uc-avg'])/1000,1) time.sleep(0.5) # print(".", end='') #return list of download/upload time in seconds return list_of_time #Method for arrange ftp download/upload time data in dictionary def ftp_test_data(self, list_time, pass_fail, bands, file_sizes, directions, num_stations): #creating dictionary for single iteration create_dict={} create_dict["band"] = self.band create_dict["direction"] = self.direction create_dict["file_size"] = self.file_size create_dict["time"] = list_time create_dict["duration"] = self.time_test create_dict["result"] = pass_fail create_dict["bands"] = bands create_dict["file_sizes"] = file_sizes create_dict["directions"] = directions create_dict["num_stations"] = num_stations return create_dict #Method for AP reboot def ap_reboot(self, ip, user, pswd): print("starting AP reboot") # creating shh client object we use this object to connect to router ssh = paramiko.SSHClient() # automatically adds the missing host key ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh.connect(ip, port=22, username=user, password=pswd, banner_timeout=600) stdin, stdout, stderr = ssh.exec_command('reboot') output = stdout.readlines() ssh.close() # print('\n'.join(output)) time.sleep(180) print("AP rebooted") def convert_min_in_time(self,total_minutes): #saving time in minutus self.time_test = total_minutes # Get hours with floor division hours = total_minutes // 60 # Get additional minutes with modulus minutes = total_minutes % 60 # Create time as a string time_string = str("%d:%02d" % (divmod(total_minutes, 60))) + ":00" + ":000000" return time_string def pass_fail_check(self,time_list): if time_list.count(0) == 0: return "Pass" else: return "Fail" def main(): # This has --mgr, --mgr_port and --debug parser = LFCliBase.create_bare_argparse(prog="netgear-ftp", formatter_class=argparse.RawTextHelpFormatter, epilog="About This Script") # Adding More Arguments for custom use parser.add_argument('--ssid',type=str, help='--ssid', default="TestAP-Jitendra") parser.add_argument('--passwd',type=str, help='--passwd', default="BLANK") parser.add_argument('--security', type=str, help='--security', default="open") parser.add_argument('--radios',nargs="+",help='--radio to use on LANforge for 5G and 2G', default=["wiphy0"]) # Test variables parser.add_argument('--bands', nargs="+", help='--bands defaults ["5G","2.4G","Both"]', default=["5G","2.4G","Both"]) parser.add_argument('--directions', nargs="+", help='--directions defaults ["Download","Upload"]', default=["Download","Upload"]) parser.add_argument('--file_sizes', nargs="+", help='--File Size defaults ["2MB","500MB","1000MB"]', default=["2MB","500MB","1000MB"]) parser.add_argument('--num_stations', type=int, help='--num_client is number of stations', default=40) parser.add_argument('--user', default='lanforge') args = parser.parse_args() # 1st time stamp for test duration time_stamp1 = datetime.now() #use for creating ftp_test dictionary iteraration_num=0 #empty dictionary for whole test data ftp_data={} #For all combinations ftp_data of directions, file size and client counts, run the test for band in args.bands: for direction in args.directions: for file_size in args.file_sizes: # Start Test obj = ftp_test(lfclient_host=args.mgr, lfclient_port=args.mgr_port, dut_ssid=args.ssid, dut_passwd=args.passwd, dut_security=args.security, num_sta= args.num_stations, band=band, file_size=file_size, direction=direction, user=args.user ) iteraration_num=iteraration_num+1 obj.file_create() obj.set_values() obj.precleanup() #if file_size != "2MB": #obj.ap_reboot("192.168.213.190","root","Password@123xzsawq@!") obj.build() if not obj.passes(): print(obj.get_fail_message()) exit(1) #First time stamp time1 = datetime.now() obj.start(False, False) #return list of download/upload completed time stamp time_list = obj.my_monitor(time1) # check pass or fail pass_fail = obj.pass_fail_check(time_list) #dictionary of whole data ftp_data[iteraration_num] = obj.ftp_test_data(time_list,pass_fail,args.bands,args.file_sizes,args.directions,args.num_stations) obj.stop() obj.postcleanup() #2nd time stamp for test duration time_stamp2 = datetime.now() #total time for test duration test_duration = str(time_stamp2 - time_stamp1)[:-7] print("FTP Test Data", ftp_data) date = str(datetime.now()).split(",")[0].replace(" ", "-").split(".")[0] test_setup_info = { "AP Name": "vap5", "SSID": args.ssid, "Number of Stations": args.num_stations, "Test Duration": test_duration } input_setup_info = { "IP": "192.168.213.190" , "user": "root", "Contact": "support@candelatech.com" } ftp_html.generate_report(ftp_data, date, test_setup_info, input_setup_info, graph_path="/home/%s/html-reports/FTP-Test" % args.user) if __name__ == '__main__': main()