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

test_l3_unicast_traffic_gen: Cleanup whitespace

Browse Source 
Signed-off-by: Matthew Stidham <stidmatt@gmail.com>
This commit is contained in:
Matthew Stidham
2021-11-29 14:53:18 -08:00
parent 5b0f6cfc59
commit ea79a485e0
1 changed files with 84 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
py-scripts/test_l3_unicast_traffic_gen.py
View File

@@ -10,7 +10,6 @@ if sys.version_info[0] != 3:
print("This script requires Python 3")
exit(1)
sys.path.append(os.path.join(os.path.abspath(__file__ + "../../../")))
lfcli_base = importlib.import_module("py-json.LANforge.lfcli_base")
@@ -40,7 +39,7 @@ class L3VariableTimeLongevity(LFCliBase):
self.security = security
self.number_template = number_template
self.resource = resource
self.debug=_debug_on
self.debug = _debug_on
self.name_prefix = name_prefix
self.test_duration = test_duration
self.cx_stations_lists = station_lists
@@ -107,17 +106,15 @@ class L3VariableTimeLongevity(LFCliBase):
if self.debug:
print("Bringing up station {}".format(station_profile))
station_profile.admin_up(self.resource)
if self.local_realm.wait_for_ip(self.resource, station_list,timeout_sec=10*len(station_list)):
if self.local_realm.wait_for_ip(self.resource, station_list, timeout_sec=10 * len(station_list)):
if self.debug:
print("ip's aquired {}".format(station_list))
else:
print("print failed to get IP's: {}".format(station_list))
if self.local_realm.wait_for_ip(self.resource, station_list,timeout_sec=120):
if self.local_realm.wait_for_ip(self.resource, station_list, timeout_sec=120):
print("tried again: print failed to get IP's: {}".format(station_list))
exit(1)
self.cx_profile.start_cx()
cur_time = datetime.datetime.now()
@@ -162,8 +159,8 @@ class L3VariableTimeLongevity(LFCliBase):
def cleanup(self, resource):
resource = 1
data = {
"name":"BLANK",
"action":"overwrite"
"name": "BLANK",
"action": "overwrite"
}
url = "cli-json/load"
self.json_post(url, data)
@@ -171,7 +168,7 @@ class L3VariableTimeLongevity(LFCliBase):
timeout = 20
done = False
while timeout > 0 and not done:
time.sleep( 1)
time.sleep(1)
port_r = self.json_get("/port/1/1/list?fields=alias")
if self.debug:
print("port interfaces {}".format(port_r["interfaces"]))
@@ -191,27 +188,27 @@ class L3VariableTimeLongevity(LFCliBase):
# refactor in LFUtils.port_zero_request()
resource = 1
data ={
'shelf':1,
'resource':1,
'port':'eth1',
'ip_addr':'0.0.0.0',
'netmask':'0.0.0.0',
'gateway':'0.0.0.0',
'current_flags':0,
'interest':402653212
data = {
'shelf': 1,
'resource': 1,
'port': 'eth1',
'ip_addr': '0.0.0.0',
'netmask': '0.0.0.0',
'gateway': '0.0.0.0',
'current_flags': 0,
'interest': 402653212
}
url = "cli-json/set_port"
self.json_post(url, data)
# refactor into LFUtils
data ={
"shelf":1,
data = {
"shelf": 1,
"resource": resource,
"port":"br0",
"network_devs":"eth1",
"br_flags":1
"port": "br0",
"network_devs": "eth1",
"br_flags": 1
}
url = "cli-json/add_br"
self.json_post(url, data)
@@ -221,7 +218,8 @@ class L3VariableTimeLongevity(LFCliBase):
self.json_post("/cli-json/set_port", data)
except:
print("LFUtils.port_dhcp_up_request didn't complete ")
print("or the json_post failed either way {} did not set up dhcp so test may not pass data ".format(self.side_b))
print("or the json_post failed either way {} did not set up dhcp so test may not pass data ".format(
self.side_b))
resource = 1
index = 0
@@ -229,31 +227,35 @@ class L3VariableTimeLongevity(LFCliBase):
station_profile.use_security(station_profile.security, station_profile.ssid, station_profile.ssid_pass)
station_profile.set_number_template(station_profile.number_template)
if self.debug:
print("radio: {} station_profile: {} Creating stations: {} ".format(self.radio_list[index],station_profile, station_list))
print("radio: {} station_profile: {} Creating stations: {} ".format(self.radio_list[index],
station_profile, station_list))
temp_station_list = []
for station in range(len(station_list)):
temp_station_list.append(str(self.resource) + "." + station_list[station])
station_profile.create(radio=self.radio_list[index], sta_names_=station_list, debug=False )
station_profile.create(radio=self.radio_list[index], sta_names_=station_list, debug=False)
index += 1
self.cx_profile.create(endp_type=self.endp_type, side_a=temp_station_list, side_b='1.'+self.side_b, sleep_time=.5)
self.cx_profile.create(endp_type=self.endp_type, side_a=temp_station_list, side_b='1.' + self.side_b,
sleep_time=.5)
self._pass("PASS: Stations build finished")
def valid_endp_type(endp_type):
valid_endp_type=['lf_udp','lf_udp6','lf_tcp','lf_tcp6']
valid_endp_type = ['lf_udp', 'lf_udp6', 'lf_tcp', 'lf_tcp6']
if str(endp_type) in valid_endp_type:
return endp_type
else:
print('invalid endp_type. Valid types lf_udp, lf_udp6, lf_tcp, lf_tcp6')
exit(1)
def main():
lfjson_host = "localhost"
lfjson_port = 8080
parser = argparse.ArgumentParser(
prog='test_l3_longevity.py',
#formatter_class=argparse.RawDescriptionHelpFormatter,
# formatter_class=argparse.RawDescriptionHelpFormatter,
formatter_class=argparse.RawTextHelpFormatter,
epilog='''\
Useful Information:
@@ -317,16 +319,21 @@ python3 .\\test_l3_longevity.py --test_duration 4m --endp_type lf_tcp --upstream
''')
parser.add_argument('-d','--test_duration', help='--test_duration <how long to run> example --time 5d (5 days) default: 3m options: number followed by d, h, m or s',default='3m')
parser.add_argument('-t', '--endp_type', help='--endp_type <type of traffic> example --endp_type lf_udp, default: lf_udp , options: lf_udp, lf_udp6, lf_tcp, lf_tcp6',
default='lf_udp',type=valid_endp_type)
parser.add_argument('-u', '--upstream_port', help='--upstream_port <upstream_port> example: --upstream_port eth1',default='eth1')
parser.add_argument('-d', '--test_duration',
help='--test_duration <how long to run> example --time 5d (5 days) default: 3m options: number followed by d, h, m or s',
default='3m')
parser.add_argument('-t', '--endp_type',
help='--endp_type <type of traffic> example --endp_type lf_udp, default: lf_udp , options: lf_udp, lf_udp6, lf_tcp, lf_tcp6',
default='lf_udp', type=valid_endp_type)
parser.add_argument('-u', '--upstream_port', help='--upstream_port <upstream_port> example: --upstream_port eth1',
default='eth1')
parser.add_argument('--debug', help='Enable debugging', default=False, action="store_true")
requiredNamed = parser.add_argument_group('required arguments')
requiredNamed.add_argument('-r','--radio', action='append', nargs=4, metavar=('<wiphyX>', '<number last station>','<ssid>','<ssid password>'),
help ='--radio <number_of_wiphy> <number of last station> <ssid> <ssid password> ',required=True)
requiredNamed.add_argument('-r', '--radio', action='append', nargs=4,
metavar=('<wiphyX>', '<number last station>', '<ssid>', '<ssid password>'),
help='--radio <number_of_wiphy> <number of last station> <ssid> <ssid password> ',
required=True)
args = parser.parse_args()
if args.test_duration:
@@ -371,13 +378,14 @@ python3 .\\test_l3_longevity.py --test_duration 4m --endp_type lf_tcp --upstream
if number_of_stations > MAX_NUMBER_OF_STATIONS:
print("number of stations per radio exceeded max of : {}".format(MAX_NUMBER_OF_STATIONS))
quit(1)
station_list = LFUtils.portNameSeries(prefix_="sta", start_id_= 1 + index*1000, end_id_= number_of_stations + index*1000, padding_number_=10000)
station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=1 + index * 1000,
end_id_=number_of_stations + index * 1000, padding_number_=10000)
station_lists.append(station_list)
index += 1
ip_var_test = L3VariableTimeLongevity(lfjson_host,
lfjson_port,
number_template="00",
station_lists= station_lists,
station_lists=station_lists,
name_prefix="var_time",
endp_type=endp_type,
side_b=side_b,