scottk/wlan-lanforge-scripts
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-lanforge-scripts.git synced 2025-11-01 03:07:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Realm monitor

Browse Source
This commit is contained in:
Matthew Stidham
2020-12-21 16:12:24 -08:00
parent 669cb65777 afc36e0d6d
commit 14bc7dc771
13 changed files with 1162 additions and 432 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
py-scripts/example_security_connection.py
View File

@@ -1,5 +1,6 @@
#!/usr/bin/env python3
import sys
import os
if sys.version_info[0] != 3:
@@ -76,49 +77,37 @@ def main():
prog='example_security_connection.py',
formatter_class=argparse.RawTextHelpFormatter,
epilog='''\
Example flags and command line input to run the script.
This python script creates an inputted number of stations using user-inputted security. This verifies that the most basic form of security works with the LANforge device.
''',
description='''\
example_security_connection.py
--------------------
This python script creates an inputted number of stations using user-inputted security. This verifies that the most basic form of security works with the LANforge device.
--------------------
---------------------------------------------------------------------------
Generic command example:
python3 ./example_security_connection.py
Example of command line to run:
./example_security_connection.py
--mgr localhost
--mgr_port 8080
--num_stations 6
--mode 1
{"auto" : "0",
"a" : "1",
"b" : "2",
"g" : "3",
"abg" : "4",
"abgn" : "5",
"bgn" : "6",
"bg" : "7",
"abgnAC" : "8",
"anAC" : "9",
"an" : "10",
"bgnAC" : "11",
"abgnAX" : "12",
"bgnAX" : "13",
"anAX" : "14"}
--radio wiphy2
--security {open|wep|wpa|wpa2|wpa3}
--ssid netgear-wpa3
--ap "00:0e:8e:78:e1:76"
--passwd admin123-wpa3
--debug
''')
optional = parser.add_argument_group('optional arguments')
required = parser.add_argument_group('required arguments')
required.add_argument('--security', help='WiFi Security protocol: < open | wep | wpa | wpa2 | wpa3 >', required=True)
optional.add_argument('--mode',help='Used to force mode of stations')
optional.add_argument('--ap',help='Used to force a connection to a particular AP')
required=None
for agroup in parser._action_groups:
if agroup.title == "required arguments":
required = agroup
#if required is not None:
optional = None
for agroup in parser._action_groups:
if agroup.title == "optional arguments":
optional = agroup
if optional is not None:
optional.add_argument('--mode',help=LFCliBase.Help_Mode)
optional.add_argument('--ap',help='Add BSSID of access point to connect to')
args = parser.parse_args()
num_sta = 2
@@ -131,8 +120,11 @@ def main():
end_id_=num_sta-1,
padding_number_=10000,
radio=args.radio)
ip_test = IPv4Test(host=args.mgr, port=args.mgr_port, ssid=args.ssid, password=args.passwd, radio=args.radio, mode= args.mode,
security=args.security, sta_list=station_list, ap=args.ap)
ip_test = IPv4Test(host=args.mgr, port=args.mgr_port,
ssid=args.ssid, password=args.passwd,
radio=args.radio, mode= args.mode,
security=args.security, sta_list=station_list,
ap=args.ap)
ip_test.cleanup(station_list)
ip_test.timeout = 60
ip_test.build()

158
py-scripts/lf_cisco_snp.py
View File

@@ -622,11 +622,12 @@ class cisco_():
class L3VariableTime(LFCliBase):
def __init__(self, host, port, endp_type, args, tos, side_b, radio_name_list, number_of_stations_per_radio_list,
ssid_list, ssid_password_list, ssid_security_list, wifimode_list,station_lists, name_prefix, debug_on, outfile,
ssid_list, ssid_password_list, ssid_security_list, wifimode_list,station_lists, name_prefix, debug_on, outfile, results,
test_keys,test_config,
reset_port_enable_list,
reset_port_time_min_list,
reset_port_time_max_list,
csv_started=False,
side_a_min_bps=560000, side_a_max_bps=0,
side_a_min_pdu=1518,side_a_max_pdu=0,
side_b_min_bps=560000, side_b_max_bps=0,
@@ -662,7 +663,8 @@ class L3VariableTime(LFCliBase):
self.station_profiles = []
self.args = args
self.outfile = outfile
self.csv_started = False
self.results = results
self.csv_started = csv_started
self.epoch_time = int(time.time())
self.debug = debug_on
self.test_keys = test_keys
@@ -676,6 +678,10 @@ class L3VariableTime(LFCliBase):
self.csv_file = open(self.outfile, "a+")
self.csv_writer = csv.writer(self.csv_file, delimiter=",")
if self.results is not None:
self.csv_results = open(self.results, "a+")
self.csv_results_writer = csv.writer(self.csv_results, delimiter=",")
for (radio_, ssid_, ssid_password_, ssid_security_, wifimode_,\
reset_port_enable_, reset_port_time_min_, reset_port_time_max_) \
in zip(radio_name_list, ssid_list, ssid_password_list, ssid_security_list, wifimode_list,\
@@ -766,6 +772,7 @@ class L3VariableTime(LFCliBase):
csv_rx_drop_percent_data.append(rx_drop_percent[item])
self.csv_add_row(csv_rx_drop_percent_data,self.csv_writer,self.csv_file)
self.csv_add_row(csv_rx_drop_percent_data,self.csv_results_writer,self.csv_results)
def __compare_vals(self, old_list, new_list):
passes = 0
@@ -773,12 +780,14 @@ class L3VariableTime(LFCliBase):
csv_performance_values = []
csv_rx_headers = []
csv_rx_row_data = []
csv_result_row_data = []
csv_rx_delta_row_data = []
csv_rx_delta_dict = {}
test_id = ""
for key in self.test_keys:
csv_rx_row_data.append(self.test_config_dict[key])
csv_rx_delta_row_data.append(self.test_config_dict[key])
#for key in self.test_keys:
# csv_rx_row_data.append(self.test_config_dict[key])
# csv_rx_delta_row_data.append(self.test_config_dict[key])
for key in [key for key in old_list if "mtx" in key]: del old_list[key]
@@ -787,8 +796,6 @@ class L3VariableTime(LFCliBase):
filtered_values = [v for _, v in new_list.items() if v !=0]
average_rx= sum(filtered_values) / len(filtered_values) if len(filtered_values) != 0 else 0
# only evaluate upstream or downstream
new_evaluate_list = new_list.copy()
print("new_evaluate_list before",new_evaluate_list)
@@ -798,6 +805,7 @@ class L3VariableTime(LFCliBase):
elif "downstream" in self.test_config_dict.values():
for key in [key for key in new_evaluate_list if "-B" in key]: del new_evaluate_list[key]
print("downstream in dictionary values")
#follow code left in for now, provides the best 5 worst 5
'''print("new_evaluate_list after",new_evaluate_list)
csv_performance_values=sorted(new_evaluate_list.items(), key=lambda x: (x[1],x[0]), reverse=False)
csv_performance_values=self.csv_validate_list(csv_performance_values,5)
@@ -819,6 +827,7 @@ class L3VariableTime(LFCliBase):
if len(old_evaluate_list) == len(new_evaluate_list):
for item, value in old_evaluate_list.items():
expected_passes +=1
print("ITEM: {} VALUE: {}".format(item, value))
if new_evaluate_list[item] > old_evaluate_list[item]:
passes += 1
#if self.debug: logg.info(item, new_evaluate_list[item], old_evaluate_list[item], " Difference: ", new_evaluate_list[item] - old_evaluate_list[item])
@@ -835,26 +844,54 @@ class L3VariableTime(LFCliBase):
csv_header += csv_rx_headers
logg.info(csv_header)
self.csv_add_column_headers(csv_header)
csv_results = self.csv_generate_column_results_headers()
#csv_results += csv_rx_headers
self.csv_add_column_headers_results(csv_results)
print("###################################")
print(csv_results)
print("###################################")
self.csv_started = True
# need to generate list first to determine worst and best
filtered_values = [v for _, v in csv_rx_delta_dict.items() if v !=0]
#average_rx_delta= sum(filtered_values) / len(filtered_values) if len(filtered_values) != 0 else 0
for key in self.test_keys:
csv_rx_row_data.append(self.test_config_dict[key])
csv_result_row_data.append(self.test_config_dict[key])
csv_rx_delta_row_data.append(self.test_config_dict[key])
max_tp_mbps = sum(filtered_values)
csv_rx_row_data.append(max_tp_mbps)
csv_result_row_data.append(max_tp_mbps)
#To do needs to be read or passed in based on test type
expected_tp_mbps = max_tp_mbps
csv_rx_row_data.append(expected_tp_mbps)
csv_result_row_data.append(expected_tp_mbps)
#Generate TestID
for key in self.test_keys:
test_id = test_id + "_" + self.test_config_dict[key]
print("test_id: {}".format(test_id))
csv_rx_row_data.append(test_id)
csv_result_row_data.append(test_id)
# Todo pass or fail
if max_tp_mbps == expected_tp_mbps:
csv_rx_row_data.append("pass")
csv_result_row_data.append("pass")
else:
csv_rx_row_data.append("fail")
csv_result_row_data.append("fail")
csv_rx_row_data.extend([self.epoch_time, self.time_stamp(),'rx_delta'])
csv_result_row_data.extend([self.epoch_time, self.time_stamp()])
print("csv_rx_row_data {}".format(csv_rx_row_data))
#TODO: may want to pass in the information that needs to be in the csv file into the class
'''
csv_rx_row_data.extend([self.epoch_time, self.time_stamp(),'rx'])
csv_rx_delta_row_data.extend([self.epoch_time, self.time_stamp(),'rx_delta'])
@@ -880,20 +917,23 @@ class L3VariableTime(LFCliBase):
csv_rx_headers.append(item)
# note need to have all upstream and downstream in the csv table thus new_list and old_list
#csv_rx_row_data.append(new_list[item])
# provide delta
csv_rx_row_data.append(new_list[item] - old_list[item])
self.csv_add_row(csv_rx_row_data,self.csv_writer,self.csv_file)
#self.csv_add_row(csv_rx_row_data,self.csv_results_writer,self.csv_results)
#self.csv_add_row(csv_rx_delta_row_data,self.csv_writer,self.csv_file)
if passes == expected_passes:
return True
return True, max_tp_mbps, csv_result_row_data
else:
return False
return False, max_tp_mbps, csv_result_row_data
else:
print("Old-list length: %i new: %i does not match in compare-vals."%(len(old_list), len(new_list)))
print("old-list:",old_list)
print("new-list:",new_list)
return False
return False, None, None # check to see if this is valid
def verify_controller(self):
if self.args == None:
@@ -1002,6 +1042,11 @@ class L3VariableTime(LFCliBase):
self._pass("PASS: Stations build finished")
def start(self, print_pass=False, print_fail=False):
best_max_tp_mbps = 0
best_csv_rx_row_data = " "
max_tp_mbps = 0
csv_rx_row_data = " "
Result = False
logg.info("Bringing up stations")
self.local_realm.admin_up(self.side_b)
for station_profile in self.station_profiles:
@@ -1052,7 +1097,23 @@ class L3VariableTime(LFCliBase):
new_rx_values, rx_drop_percent = self.__get_rx_values()
expected_passes += 1
if self.__compare_vals(old_rx_values, new_rx_values):
'''
#self.csv_add_row(csv_rx_row_data,self.csv_results_writer,self.csv_results)
if passes == expected_passes:
return True, max_tp_mbps, csv_rx_row_data
else:
return False, max_tp_mbps, csv_rx_row_data
'''
# __compare_vals - does the calculations
Result, max_tp_mbps, csv_rx_row_data = self.__compare_vals(old_rx_values, new_rx_values)
if max_tp_mbps > best_max_tp_mbps:
best_max_tp_mbps = max_tp_mbps
best_csv_rx_row_data = csv_rx_row_data
# need to check the expected max_tp_mbps
if Result:
passes += 1
else:
self._fail("FAIL: Not all stations increased traffic", print_fail)
@@ -1062,7 +1123,7 @@ class L3VariableTime(LFCliBase):
#self.__record_rx_dropped_percent(rx_drop_percent)
cur_time = datetime.datetime.now()
self.csv_add_row(best_csv_rx_row_data,self.csv_results_writer,self.csv_results)
if passes == expected_passes:
self._pass("PASS: All tests passed", print_pass)
@@ -1083,7 +1144,7 @@ class L3VariableTime(LFCliBase):
def csv_generate_column_headers(self):
csv_rx_headers = self.test_keys.copy()
csv_rx_headers.extend
csv_rx_headers.extend(['Max TP Mbps','Expected TP','Pass Fail','Time epoch','Time','Monitor'])
csv_rx_headers.extend(['max_tp_mbps','expected_tp','test_id','pass_fail','epoch_time','time','monitor'])
'''for i in range(1,6):
csv_rx_headers.append("least_rx_data {}".format(i))
for i in range(1,6):
@@ -1091,18 +1152,35 @@ class L3VariableTime(LFCliBase):
csv_rx_headers.append("average_rx_data")'''
return csv_rx_headers
def csv_generate_column_results_headers(self):
csv_rx_headers = self.test_keys.copy()
csv_rx_headers.extend
csv_rx_headers.extend(['max_tp_mbps','expected_tp','test_id','pass_fail','epoch_time','time'])
'''for i in range(1,6):
csv_rx_headers.append("least_rx_data {}".format(i))
for i in range(1,6):
csv_rx_headers.append("most_rx_data_{}".format(i))
csv_rx_headers.append("average_rx_data")'''
return csv_rx_headers
def csv_add_column_headers(self,headers):
if self.csv_file is not None:
self.csv_writer.writerow(headers)
self.csv_file.flush()
def csv_add_column_headers_results(self,headers):
if self.csv_results is not None:
self.csv_results_writer.writerow(headers)
self.csv_results.flush()
def csv_validate_list(self, csv_list, length):
if len(csv_list) < length:
csv_list = csv_list + [('no data','no data')] * (length - len(csv_list))
return csv_list
def csv_add_row(self,row,writer,csv_file):
if self.csv_file is not None:
def csv_add_row(self,row,writer,csv_file): # can make two calls eventually
if csv_file is not None:
writer.writerow(row)
csv_file.flush()
@@ -1466,6 +1544,7 @@ Eventual Realm at Cisco
if args.csv_outfile != None:
current_time = time.strftime("%m_%d_%Y_%H_%M_%S", time.localtime())
csv_outfile = "{}_{}.csv".format(args.csv_outfile,current_time)
csv_results = "results_{}_{}.csv".format(args.csv_outfile,current_time)
print("csv output file : {}".format(csv_outfile))
if args.log:
@@ -1707,13 +1786,13 @@ Eventual Realm at Cisco
cisco_chan_widths = "20".split()
cisco_ap_modes = "local".split()
cisco_data_encryptions = "disable".split()
cisco_packet_types = "lf_udp lf_tcp".split()
#cisco_packet_types = "lf_udp".split()
#cisco_packet_types = "lf_udp lf_tcp".split()
cisco_packet_types = "lf_udp".split()
#cisco_directions = "upstream downstream".split()
cisco_directions = "upstream downstream".split()
cisco_packet_sizes = "88 512 1370 1518".split()
#cisco_packet_sizes = "1518".split()
cisco_client_densities = "1".split()
#cisco_packet_sizes = "88 512 1370 1518".split()
cisco_packet_sizes = "1518".split()
cisco_client_densities = "10".split()
cisco_data_encryptions = "disable".split()
cisco_side_a_min_bps = 500000000
@@ -1760,11 +1839,12 @@ Eventual Realm at Cisco
logg.info(cisco_client_densities)
logg.info(cisco_data_encryptions)
ap_set = None
band_set = None
chan_width_set = None
ap_mode_set = None
tx_power_set = None
__ap_set = None
__band_set = None
__chan_width_set = None
__ap_mode_set = None
__tx_power_set = None
__csv_started = False
for cisco_ap in cisco_aps:
for cisco_band in cisco_bands: # frequency
@@ -1788,26 +1868,27 @@ Eventual Realm at Cisco
cisco_band,cisco_wifimode,cisco_chan_width,cisco_data_encryption,cisco_ap_mode,cisco_client_density,
cisco_packet_type,cisco_direction,cisco_packet_size)
test_keys = ['AP','Band','wifi_mode','BW','encryption','ap_mode','clients','packet_type','direction','packet_size']
logg.info("# Cisco run settings: {}".format(test_config))
if(args.no_controller):
logg.info("################################################")
logg.info("# NO CONTROLLER SET , TEST MODE")
logg.info("################################################")
else:
if( cisco_ap != ap_set or
cisco_band != band_set or
cisco_chan_width != chan_width_set or
cisco_ap_mode != ap_mode_set or
cisco_tx_power != tx_power_set
if( cisco_ap != __ap_set or
cisco_band != __band_set or
cisco_chan_width != __chan_width_set or
cisco_ap_mode != __ap_mode_set or
cisco_tx_power != __tx_power_set
):
logg.info("###############################################")
logg.info("# NEW CONTROLLER CONFIG")
logg.info("###############################################")
ap_set = cisco_ap
band_set = cisco_band
chan_width_set = cisco_chan_width
ap_mode_set = cisco_ap_mode
tx_power_set = cisco_tx_power
__ap_set = cisco_ap
__band_set = cisco_band
__chan_width_set = cisco_chan_width
__ap_mode_set = cisco_ap_mode
__tx_power_set = cisco_tx_power
#############################################
# configure cisco controller
#############################################
@@ -1969,8 +2050,11 @@ Eventual Realm at Cisco
side_b_min_pdu =cisco_packet_size,
debug_on=debug_on,
outfile=csv_outfile,
results=csv_results,
test_keys=test_keys,
test_config=test_config)
test_config=test_config,
csv_started=__csv_started )
__csv_started = True
ip_var_test.pre_cleanup()
ip_var_test.build()
if not ip_var_test.passes():

126
py-scripts/test_all_scripts.sh
View File

@@ -1,90 +1,88 @@
#!/bin/bash
#This bash script aims to automate the test process of all Candela Technologies's test_* scripts in the lanforge-scripts directory. The script can be run 2 ways and may include (via user input) the "start_num" and "stop_num" variables to select which tests should be run.
# OPTION ONE: ./test_all_scripts.sh : this command runs all the scripts in the array "testCommands"
# OPTION TWO: ./test_all_scripts.sh 4 5 : this command runs py-script commands (in testCommands array) that include the py-script options beginning with 4 and 5 (inclusive) in case function ret_case_num.
# OPTION TWO: ./test_all_scripts.sh 4 5 : this command runs py-script commands (in testCommands array) that include the py-script options beginning with 4 and 5 (inclusive) in case function ret_case_num.
#Variables
NUM_STA=4
SSID_USED="jedway-wpa2-x2048-5-3"
PASSWD_USED="jedway-wpa2-x2048-5-3"
RADIO_USED="wiphy1"
SECURITY="wpa2"
CURR_TEST_NAME="BLANK"
CURR_TEST_NUM=0
STOP_NUM=9
START_NUM=0
CURR_TEST_NUM=0
CURR_TEST_NAME="BLANK"
STOP_NUM=9
#Test array
testCommands=("./example_wpa_connection.py --num_stations $NUM_STA --ssid jedway-r8000-36 --passwd jedway-r8000-36 --radio $RADIO_USED --security wpa"
"./example_wpa2_connection.py --num_stations $NUM_STA --ssid $SSID_USED --passwd $SSID_USED --radio $RADIO_USED --security wpa2"
"./example_wep_connection.py --num_stations $NUM_STA --ssid jedway-wep-48 --passwd jedway-wep-48 --radio $RADIO_USED --security wep"
"./example_wpa3_connection.py --num_stations $NUM_STA --ssid jedway-wpa3-1 --passwd jedway-wpa3-1 --radio $RADIO_USED --security wpa3"
"./test_ipv4_connection.py --radio wiphy2 --num_stations $NUM_STA --ssid $SSID_USED --passwd $PASSWD_USED --security $SECURITY --debug --upstream_port eth1"
"./test_generic.py --mgr localhost --mgr_port 4122 --radio $RADIO_USED --ssid SSID_USED --passwd $PASSWD_USED --num_stations $NUM_STA --type lfping --dest 10.40.0.1 --security $SECURITY"
"./test_generic.py --mgr localhost --mgr_port 4122 --radio $RADIO_USED --ssid $SSID_USED --passwd $PASSWD_USED --num_stations $NUM_STA --type speedtest --speedtest_min_up 20 --speedtest_min_dl 20 --speedtest_max_ping 150 --security $SECURITY"
"./test_ipv4_l4_urls_per_ten.py --upstream_port eth1 --radio $RADIO_USED --num_stations $NUM_STA --security $SECURITY --ssid $SSID_USED --passwd $PASSWD_USED --num_tests 1 --requests_per_ten 600 --target_per_ten 600"
"./test_ipv4_l4_wifi.py --upstream_port eth1 --radio wiphy0 --num_stations $NUM_STA --security $SECURITY --ssid jedway-wpa2-x2048-4-4 --passwd jedway-wpa2-x2048-4-4 --test_duration 3m"
"./test_ipv4_l4.py --radio wiphy3 --num_stations 4 --security wpa2 --ssid jedway-wpa2-x2048-4-1 --passwd jedway-wpa2-x2048-4-1 --url \"dl http://10.40.0.1 /dev/null\" --test_duration 2m --debug"
testCommands=("./example_wpa_connection.py --num_stations $NUM_STA --ssid jedway-r8000-36 --passwd jedway-r8000-36 --radio $RADIO_USED --security wpa"
"./example_wpa2_connection.py --num_stations $NUM_STA --ssid $SSID_USED --passwd $SSID_USED --radio $RADIO_USED --security wpa2"
"./example_wep_connection.py --num_stations $NUM_STA --ssid jedway-wep-48 --passwd jedway-wep-48 --radio $RADIO_USED --security wep"
"./example_wpa3_connection.py --num_stations $NUM_STA --ssid jedway-wpa3-1 --passwd jedway-wpa3-1 --radio $RADIO_USED --security wpa3"
"./test_ipv4_connection.py --radio wiphy2 --num_stations $NUM_STA --ssid $SSID_USED --passwd $PASSWD_USED --security $SECURITY --debug --upstream_port eth1"
"./test_generic.py --mgr localhost --mgr_port 4122 --radio $RADIO_USED --ssid SSID_USED --passwd $PASSWD_USED --num_stations $NUM_STA --type lfping --dest 10.40.0.1 --security $SECURITY"
"./test_generic.py --mgr localhost --mgr_port 4122 --radio $RADIO_USED --ssid $SSID_USED --passwd $PASSWD_USED --num_stations $NUM_STA --type speedtest --speedtest_min_up 20 --speedtest_min_dl 20 --speedtest_max_ping 150 --security $SECURITY"
"./test_ipv4_l4_urls_per_ten.py --upstream_port eth1 --radio $RADIO_USED --num_stations $NUM_STA --security $SECURITY --ssid $SSID_USED --passwd $PASSWD_USED --num_tests 1 --requests_per_ten 600 --target_per_ten 600"
"./test_ipv4_l4_wifi.py --upstream_port eth1 --radio wiphy0 --num_stations $NUM_STA --security $SECURITY --ssid jedway-wpa2-x2048-4-4 --passwd jedway-wpa2-x2048-4-4 --test_duration 3m"
"./test_ipv4_l4.py --radio wiphy3 --num_stations 4 --security wpa2 --ssid jedway-wpa2-x2048-4-1 --passwd jedway-wpa2-x2048-4-1 --url \"dl http://10.40.0.1 /dev/null\" --test_duration 2m --debug"
)
function ret_case_num(){
case $1 in
"example_wpa_connection")
echo 1 ;;
"example_wpa2_connection")
echo 2 ;;
"example_wpa3_connection")
echo 4 ;;
"example_wep_connection")
echo 3 ;;
"test_ipv4_connection")
echo 5 ;;
"test_generic")
echo 6 ;;
"test_ipv4_l4_urls_per_ten")
echo 7 ;;
"test_ipv4_l4_wifi")
echo 8 ;;
"test_ipv4_l4")
echo 9 ;;
esac
}
declare -A name_to_num
name_to_num=(
["example_wpa_connection"]=1
["example_wpa2_connection"]=2
["example_wpa3_connection"]=4
["example_wep_connection"]=3
["test_ipv4_connection"]=5
["test_generic"]=6
["test_ipv4_l4_urls_per_ten"]=7
["test_ipv4_l4_wifi"]=8
["test_ipv4_l4"]=9
)
function blank_db() {
echo "Loading blank scenario..." >> ~/test_all_output_file.txt
./scenario.py --load BLANK >> ~/test_all_output_file.txt
#check_blank.py
echo "Loading blank scenario..." >>~/test_all_output_file.txt
./scenario.py --load BLANK >>~/test_all_output_file.txt
#check_blank.py
}
function echo_print(){
echo "Beginning $CURR_TEST_NAME test..." >> ~/test_all_output_file.txt
function echo_print() {
echo "Beginning $CURR_TEST_NAME test..." >>~/test_all_output_file.txt
}
function run_test(){
for i in "${testCommands[@]}"; do
CURR_TEST_NAME=${i%%.py*}
CURR_TEST_NAME=${CURR_TEST_NAME#./*}
CURR_TEST_NUM=$(ret_case_num $CURR_TEST_NAME)
if [[ $CURR_TEST_NUM -gt $STOP_NUM ]] || [[ $STOP_NUM -eq $CURR_NUM && $STOP_NUM -ne 0 ]]; then
exit 1
fi
if [[ $CURR_TEST_NUM -gt $START_NUM ]] || [[ $CURR_TEST_NUM -eq $START_NUM ]]; then
echo_print
eval $i >> ~/test_all_output_file.txt
if [ $? -ne 0 ]; then
function run_test() {
for i in "${testCommands[@]}"; do
CURR_TEST_NAME=${i%%.py*}
CURR_TEST_NAME=${CURR_TEST_NAME#./*}
CURR_TEST_NUM="${name_to_num[$CURR_TEST_NAME]}"
echo "$CURR_TEST_NAME $CURR_TEST_NUM"
if (( $CURR_TEST_NUM > $STOP_NUM )) || (( $STOP_NUM == $CURR_TEST_NUM )) && (( $STOP_NUM != 0 )); then
exit 1
fi
if (( $CURR_TEST_NUM > $START_NUM )) || (( $CURR_TEST_NUM == $START_NUM )); then
echo_print
echo "$i"
[[ x$DEBUG != x ]] && sleep 2
eval $i >>~/test_all_output_file.txt
if [ $? -ne 0 ]; then
echo $CURR_TEST_NAME failure
else
echo $CURR_TEST_NAME success
[[ x$DEBUG != x ]] && exit 1
else
echo $CURR_TEST_NAME success
fi
if [[ "${CURR_TEST_NAME}" = @(example_wpa_connection|example_wpa2_connection|example_wpa3_connection|example_wep_connection) ]]; then
blank_db
fi
fi
if [[ "${CURR_TEST_NAME}" = @(example_wpa_connection|example_wpa2_connection|example_wpa3_connection|example_wep_connection) ]]; then
blank_db
fi
fi
done
}
function check_args(){
function check_args() {
if [ ! -z $1 ]; then
START_NUM=$1
fi
if [ ! -z $2 ]; then
STOP_NUM=$2
fi
fi
}
true > ~/test_all_output_file.txt
check_args $1 $2
true >~/test_all_output_file.txt
check_args $1 $2
run_test
#test generic and fileio are for macvlans
#test generic and fileio are for macvlans

115
py-scripts/test_fileio.py
View File

@@ -50,6 +50,7 @@ class FileIOTest(LFCliBase):
read_only_test_group=None,
port_list=[],
ip_list=None,
connections_per_port=1,
mode="both",
update_group_args={"name": None, "action": None, "cxs": None},
_debug_on=False,
@@ -66,6 +67,7 @@ class FileIOTest(LFCliBase):
self.number_template = number_template
self.test_duration = test_duration
self.port_list = []
self.connections_per_port = connections_per_port
self.use_macvlans = use_macvlans
self.mode = mode.lower()
self.ip_list = ip_list
@@ -131,6 +133,7 @@ class FileIOTest(LFCliBase):
self.wo_profile.max_write_rate_bps = LFUtils.parse_size(max_write_rate_bps)
self.wo_profile.directory = directory
self.wo_profile.server_mount = server_mount
self.wo_profile.num_connections_per_port = connections_per_port
self.ro_profile = self.wo_profile.create_ro_profile()
@@ -280,25 +283,26 @@ class FileIOTest(LFCliBase):
self.created_ports += self.station_profile.station_names
if len(self.ip_list) > 0:
if self.gateway is not None and self.netmask is not None:
print("++++++++++++++++\n", self.ip_list, "++++++++++++++++\n")
for num_port in range(len(self.port_list)):
shelf = self.local_realm.name_to_eid(self.port_list[num_port])[0]
resource = self.local_realm.name_to_eid(self.port_list[num_port])[1]
port = self.local_realm.name_to_eid(self.port_list[num_port])[2]
req_url = "/cli-json/set_port"
data = {
"shelf": shelf,
"resource": resource,
"port": port,
"ip_addr": self.ip_list[num_port],
"netmask": self.netmask,
"gateway": self.gateway
}
self.local_realm.json_post(req_url, data)
self.created_ports.append("%s.%s.%s" % (shelf, resource, port))
else:
raise ValueError("Netmask and gateway must be specified")
# print("++++++++++++++++\n", self.ip_list, "++++++++++++++++\n")
for num_port in range(len(self.port_list)):
if self.ip_list[num_port] != 0:
if self.gateway is not None and self.netmask is not None:
shelf = self.local_realm.name_to_eid(self.port_list[num_port])[0]
resource = self.local_realm.name_to_eid(self.port_list[num_port])[1]
port = self.local_realm.name_to_eid(self.port_list[num_port])[2]
req_url = "/cli-json/set_port"
data = {
"shelf": shelf,
"resource": resource,
"port": port,
"ip_addr": self.ip_list[num_port],
"netmask": self.netmask,
"gateway": self.gateway
}
self.local_realm.json_post(req_url, data)
self.created_ports.append("%s.%s.%s" % (shelf, resource, port))
else:
raise ValueError("Netmask and gateway must be specified")
# if use test groups and test group does not exist, create cxs, create test group, assign to test group
# if use test groups and test group exists and no cxs, create cxs, assign to test group
@@ -310,7 +314,8 @@ class FileIOTest(LFCliBase):
if self.wo_tg_exists:
if not self.wo_tg_cx_exists:
print("Creating Write Only CXs")
self.wo_profile.create(ports=self.created_ports, sleep_time=.5, debug_=self.debug,
self.wo_profile.create(ports=self.created_ports, connections_per_port=self.connections_per_port,
sleep_time=.5, debug_=self.debug,
suppress_related_commands_=None)
time.sleep(1)
print("Adding cxs to %s" % self.wo_tg_profile.group_name)
@@ -318,7 +323,8 @@ class FileIOTest(LFCliBase):
self.wo_tg_profile.add_cx(cx)
else:
print("Creating Write Only CXs")
self.wo_profile.create(ports=self.created_ports, sleep_time=.5, debug_=self.debug,
self.wo_profile.create(ports=self.created_ports, connections_per_port=self.connections_per_port,
sleep_time=.5, debug_=self.debug,
suppress_related_commands_=None)
time.sleep(1)
print("Creating Write Only test group")
@@ -330,7 +336,8 @@ class FileIOTest(LFCliBase):
if self.ro_tg_exists:
if not self.ro_tg_cx_exists:
print("Creating Read Only CXs")
self.ro_profile.create(ports=self.created_ports, sleep_time=.5, debug_=self.debug,
self.ro_profile.create(ports=self.created_ports, connections_per_port=self.connections_per_port,
sleep_time=.5, debug_=self.debug,
suppress_related_commands_=None)
time.sleep(1)
print("Adding cxs to %s" % self.ro_tg_profile.group_name)
@@ -338,7 +345,8 @@ class FileIOTest(LFCliBase):
self.ro_tg_profile.add_cx(cx)
else:
print("Creating Read Only CXs")
self.ro_profile.create(ports=self.created_ports, sleep_time=.5, debug_=self.debug,
self.ro_profile.create(ports=self.created_ports, connections_per_port=self.connections_per_port,
sleep_time=.5, debug_=self.debug,
suppress_related_commands_=None)
time.sleep(1)
print("Creating Read Only test group")
@@ -350,7 +358,8 @@ class FileIOTest(LFCliBase):
if self.wo_tg_exists:
if not self.wo_tg_cx_exists:
print("Creating Write Only CXs")
self.wo_profile.create(ports=self.created_ports, sleep_time=.5, debug_=self.debug,
self.wo_profile.create(ports=self.created_ports, connections_per_port=self.connections_per_port,
sleep_time=.5, debug_=self.debug,
suppress_related_commands_=None)
time.sleep(1)
print("Adding cxs to %s" % self.wo_tg_profile.group_name)
@@ -358,7 +367,8 @@ class FileIOTest(LFCliBase):
self.wo_tg_profile.add_cx(cx)
else:
print("Creating Write Only CXs")
self.wo_profile.create(ports=self.created_ports, sleep_time=.5, debug_=self.debug,
self.wo_profile.create(ports=self.created_ports, connections_per_port=self.connections_per_port,
sleep_time=.5, debug_=self.debug,
suppress_related_commands_=None)
time.sleep(1)
print("Creating Write Only test group")
@@ -369,7 +379,8 @@ class FileIOTest(LFCliBase):
if self.ro_tg_exists:
if not self.ro_tg_cx_exists:
print("Creating Read Only CXs")
self.ro_profile.create(ports=self.created_ports, sleep_time=.5, debug_=self.debug,
self.ro_profile.create(ports=self.created_ports, connections_per_port=self.connections_per_port,
sleep_time=.5, debug_=self.debug,
suppress_related_commands_=None)
time.sleep(1)
print("Adding cxs to %s" % self.ro_tg_profile.group_name)
@@ -377,7 +388,8 @@ class FileIOTest(LFCliBase):
self.ro_tg_profile.add_cx(cx)
else:
print("Creating Read Only CXs")
self.ro_profile.create(ports=self.created_ports, sleep_time=.5, debug_=self.debug,
self.ro_profile.create(ports=self.created_ports, connections_per_port=self.connections_per_port,
sleep_time=.5, debug_=self.debug,
suppress_related_commands_=None)
time.sleep(1)
print("Creating Read Only test group")
@@ -390,21 +402,25 @@ class FileIOTest(LFCliBase):
else:
if self.mode == "write":
print("Creating Write Only CXs")
self.wo_profile.create(ports=self.created_ports, sleep_time=.5, debug_=self.debug,
self.wo_profile.create(ports=self.created_ports, connections_per_port=self.connections_per_port,
sleep_time=.5, debug_=self.debug,
suppress_related_commands_=None)
elif self.mode == "read":
print("Creating Read Only CXs")
self.ro_profile.create(ports=self.created_ports, sleep_time=.5, debug_=self.debug,
self.ro_profile.create(ports=self.created_ports, connections_per_port=self.connections_per_port,
sleep_time=.5, debug_=self.debug,
suppress_related_commands_=None)
elif self.mode == "both":
print("Creating Write Only CXs")
self.wo_profile.create(ports=self.created_ports, sleep_time=.5, debug_=self.debug,
self.wo_profile.create(ports=self.created_ports, connections_per_port=self.connections_per_port,
sleep_time=.5, debug_=self.debug,
suppress_related_commands_=None)
print("Creating Read Only CXs")
self.ro_profile.create(ports=self.created_ports, sleep_time=.5, debug_=self.debug,
self.ro_profile.create(ports=self.created_ports, connections_per_port=self.connections_per_port,
sleep_time=.5, debug_=self.debug,
suppress_related_commands_=None)
else:
raise ValueError("Uknown mode used, must be (read, write, both)")
raise ValueError("Unknown mode used, must be (read, write, both)")
else:
raise ValueError("Mode must be set (read, write, both)")
@@ -548,11 +564,24 @@ def main():
test_fileio.py:
--------------------
Generic command layout:
./test_fileio.py --macvlan_parent <port> --num_ports <num ports> --use_macvlans
./test_fileio.py --macvlan_parent <port> --num_ports <num ports> --use_macvlans
--first_mvlan_ip <first ip in series> --netmask <netmask to use> --gateway <gateway ip addr>
./test_fileio.py --macvlan_parent eth2 --num_ports 3 --use_macvlans --first_mvlan_ip 192.168.92.13
--netmask 255.255.255.0 --gateway 192.168.92.1
./test_fileio.py --radio 1.wiphy0 --test_duration 1m --macvlan_parent eth1 --num_ports 3 --use_macvlans
--use_ports eth1#0,eth1#1,eth1#2 --connections_per_port 2 --mode write
./test_fileio.py --radio 1.wiphy0 --test_duration 1m --macvlan_parent eth1 --num_ports 3 --use_macvlans
--first_mvlan_ip 10.40.3.100 --netmask 255.255.240.0 --gateway 10.40.0.1
--use_test_groups --write_only_test_group test_wo --read_only_test_group test_ro
--add_to_group test_wo --cxs test_wo0000,test_wo0001,test_wo0002
./test_fileio.py --radio 1.wiphy0 --test_duration 1m --macvlan_parent eth1 --num_ports 3 --use_macvlans
--use_ports eth1#0=10.40.3.103,eth1#1,eth1#2 --connections_per_port 2
--netmask 255.255.240.0 --gateway 10.40.0.1
''')
parser.add_argument('--num_stations', help='Number of stations to create', default=0)
parser.add_argument('--radio', help='radio EID, e.g: 1.wiphy2')
@@ -576,9 +605,15 @@ Generic command layout:
default="AUTO")
parser.add_argument('--server_mount', help='--server_mount The server to mount, ex: 192.168.100.5/exports/test1',
default="10.40.0.1:/var/tmp/test")
parser.add_argument('--macvlan_parent', help='specifies parent port for macvlan creation', default=None)
parser.add_argument('--first_port', help='specifies name of first port to be used', default=None)
parser.add_argument('--num_ports', help='number of ports to create', default=1)
parser.add_argument('--connections_per_port', help='specifies number of connections to be used per port', default=1,
type=int)
parser.add_argument('--use_ports', help='list of comma separated ports to use with ips, \'=\' separates name and ip'
'{ port_name1=ip_addr1,port_name1=ip_addr2 }. '
'Ports without ips will be left alone', default=None)
parser.add_argument('--use_macvlans', help='will create macvlans', action='store_true', default=False)
parser.add_argument('--first_mvlan_ip', help='specifies first static ip address to be used or dhcp', default=None)
parser.add_argument('--netmask', help='specifies netmask to be used with static ip addresses', default=None)
@@ -588,12 +623,11 @@ Generic command layout:
parser.add_argument('--read_only_test_group', help='specifies name to use for read only test group', default=None)
parser.add_argument('--write_only_test_group', help='specifies name to use for write only test group', default=None)
parser.add_argument('--mode', help='write,read,both', default='both', type=str)
parser.add_argument('--use_ports', help='list of comma separated ports to use with ips, \'=\' separates name and ip'
'{ port_name1=ip_addr1,port_name1=ip_addr2 }', default=None)
tg_group = parser.add_mutually_exclusive_group()
tg_group.add_argument('--add_to_group', help='name of test group to add cxs to', default=None)
tg_group.add_argument('--del_from_group', help='name of test group to delete cxs from', default=None)
parser.add_argument('--cxs', help='list of cxs to add/remove depending on use of --add_to_group or --del_from_group', default=None)
parser.add_argument('--cxs', help='list of cxs to add/remove depending on use of --add_to_group or --del_from_group'
, default=None)
args = parser.parse_args()
update_group_args = {
@@ -647,7 +681,10 @@ Generic command layout:
temp_list = args.use_ports.split(',')
for port in temp_list:
port_list.append(port.split('=')[0])
ip_list.append(port.split('=')[1])
if '=' in port:
ip_list.append(port.split('=')[1])
else:
ip_list.append(0)
if len(port_list) != len(ip_list):
raise ValueError(temp_list, " ports must have matching ip addresses!")
@@ -695,16 +732,16 @@ Generic command layout:
write_only_test_group=args.write_only_test_group,
read_only_test_group=args.read_only_test_group,
update_group_args = update_group_args,
connections_per_port=args.connections_per_port,
mode=args.mode
# want a mount options param
)
ip_test.cleanup(port_list)
ip_test.build()
# exit(1)
if not ip_test.passes():
print(ip_test.get_fail_message())
# ip_test.start(False, False)
ip_test.start(False, False)
ip_test.stop()
if not ip_test.passes():
print(ip_test.get_fail_message())

72
py-scripts/test_ipv4_connection.py
View File

@@ -20,35 +20,42 @@ import pprint
class IPv4Test(LFCliBase):
def __init__(self,
ssid,
security,
password,
host,
port,
sta_list=None,
number_template="00000",
radio="wiphy0",
def __init__(self,
_ssid=None,
_security=None,
_password=None,
_host=None,
_port=None,
_sta_list=None,
_number_template="00000",
_radio="wiphy0",
_proxy_str=None,
_debug_on=False,
_exit_on_error=False,
_exit_on_fail=False):
super().__init__(host,
port,
super().__init__(_host,
_port,
_proxy_str=_proxy_str,
_local_realm=realm.Realm(lfclient_host=_host,
lfclient_port=_port,
halt_on_error_=_exit_on_error,
_exit_on_error=_exit_on_error,
_exit_on_fail=_exit_on_fail,
_proxy_str=_proxy_str,
debug_=_debug_on),
_debug=_debug_on,
_halt_on_error=_exit_on_error,
_exit_on_fail=_exit_on_fail)
self.host = host
self.port = port
self.ssid = ssid
self.security = security
self.password = password
self.sta_list = sta_list
self.radio = radio
self.host = _host
self.port = _port
self.ssid = _ssid
self.security = _security
self.password = _password
self.sta_list = _sta_list
self.radio = _radio
self.timeout = 120
self.number_template = number_template
self.number_template = _number_template
self.debug = _debug_on
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.station_profile = self.local_realm.new_station_profile()
self.station_profile.lfclient_url = self.lfclient_url
self.station_profile.ssid = self.ssid
@@ -137,7 +144,7 @@ def main():
description='''\
test_ipv4_connection.py
--------------------
Generic command example:
Command example:
./test_ipv4_connection.py
--upstream_port eth1
--radio wiphy0
@@ -151,6 +158,9 @@ Generic command example:
required.add_argument('--security', help='WiFi Security protocol: < open | wep | wpa | wpa2 | wpa3 >', required=True)
args = parser.parse_args()
#if args.debug:
# pprint.pprint(args)
# time.sleep(5)
if (args.radio is None):
raise ValueError("--radio required")
@@ -164,14 +174,18 @@ Generic command example:
end_id=num_sta-1,
padding_number=10000,
radio=args.radio)
ip_test = IPv4Test(host=args.mgr, port=args.mgr_port,
ssid=args.ssid,
password=args.passwd,
security=args.security,
sta_list=station_list,
radio=args.radio,
if args.debug:
print("args.proxy: %s" % args.proxy)
ip_test = IPv4Test(_host=args.mgr,
_port=args.mgr_port,
_ssid=args.ssid,
_password=args.passwd,
_security=args.security,
_sta_list=station_list,
_radio=args.radio,
_proxy_str=args.proxy,
_debug_on=args.debug)
ip_test.cleanup(station_list)
ip_test.build()
if not ip_test.passes():

62
py-scripts/test_ipv6_connection.py
View File

@@ -18,7 +18,7 @@ import pprint
class IPv6Test(LFCliBase):
def __init__(self, ssid, security, password, sta_list=None, num_stations=0, prefix="00000", host="localhost", port=8080,
def __init__(self, ssid, security, password,ap=None, mode=0, sta_list=None, num_stations=0, prefix="00000", host="localhost", port=8080,
_debug_on=False, timeout=120, radio="wiphy0",
_exit_on_error=False,
_exit_on_fail=False,
@@ -30,6 +30,8 @@ class IPv6Test(LFCliBase):
self.radio = radio
self.security = security
self.password = password
self.ap=ap
self.mode=mode
self.num_stations = num_stations
self.sta_list = sta_list
self.timeout = timeout
@@ -41,12 +43,16 @@ class IPv6Test(LFCliBase):
self.station_profile.lfclient_url = self.lfclient_url
self.station_profile.ssid = self.ssid
self.station_profile.ssid_pass = self.password,
self.station_profile.ssid_pass = self.password
if mode is not None:
self.station_profile.mode = mode
self.station_profile.security = self.security
self.station_profile.number_template_ = self.number_template
self.station_profile.mode = 0
def build(self):
def build(self):
if self.ap is not None:
self.station_profile.set_command_param("add_sta", "ap", self.ap)
self.station_profile.use_security(self.security, self.ssid, self.password)
self.station_profile.set_number_template(self.prefix)
print("Creating stations")
@@ -104,7 +110,6 @@ class IPv6Test(LFCliBase):
return self.passes()
def stop(self):
# Bring stations down
self.station_profile.admin_down()
def cleanup(self, sta_list):
@@ -125,26 +130,39 @@ def main():
description='''\
test_ipv6_connection.py:
--------------------
--------------------------------------------------
Generic command example:
python3 ./test_ipv6_connection.py --upstream_port eth1 \\
--radio wiphy0 \\
--num_stations 3 \\
--security {open|wep|wpa|wpa2|wpa3} \\
--ssid netgear \\
--passwd admin123 \\
--dest 10.40.0.1 \\
--test_duration 2m \\
--interval 1s \\
-- timeout 120 \\
--debug
python3 ./test_ipv6_connection.py
--upstream_port eth1
--radio wiphy0
--num_stations 3
--proxy
--security {open|wep|wpa|wpa2|wpa3}
--ssid netgear
--passwd admin123
--mode 1
--ap "00:0e:8e:78:e1:76"
--test_id
-- timeout 120
--debug
''')
optional = parser.add_argument_group('optional arguments')
required = parser.add_argument_group('required arguments')
required.add_argument('--security', help='WiFi Security protocol: < open | wep | wpa | wpa2 | wpa3 >', required=True)
parser.add_argument('--timeout', help='--timeout sets the length of time to wait until a connection is successful', default=30)
required = None
for agroup in parser._action_groups:
if agroup.title == "required arguments":
required = agroup
#if required is not None:
optional = None
for agroup in parser._action_groups:
if agroup.title == "optional arguments":
optional = agroup
if optional is not None:
optional.add_argument("--ap", help="Add BSSID of access point to connect to")
optional.add_argument('--mode', help=LFCliBase.Help_Mode)
optional.add_argument('--timeout', help='--timeout sets the length of time to wait until a connection is successful', default=30)
args = parser.parse_args()
num_sta=2
@@ -159,6 +177,8 @@ python3 ./test_ipv6_connection.py --upstream_port eth1 \\
ssid=args.ssid,
password=args.passwd,
security=args.security,
ap=args.ap,
mode=args.mode,
sta_list=station_list,
_debug_on=args.debug)
ipv6_test.cleanup(station_list)