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

addition of test output logic for urls_per_ten , arg parsing changes to var_time

Browse Source
This commit is contained in:
Dipti
2021-01-22 01:33:24 -08:00
committed by Jed Reynolds
parent ac86b2bd45
commit 96d445b496
2 changed files with 50 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
py-scripts/test_ipv4_l4_urls_per_ten.py
View File

@@ -34,6 +34,7 @@ class IPV4L4(LFCliBase):
_exit_on_error=False, _exit_on_error=False,
_exit_on_fail=False): _exit_on_fail=False):
super().__init__(host, port, _debug=_debug_on, _halt_on_error=_exit_on_error, _exit_on_fail=_exit_on_fail) super().__init__(host, port, _debug=_debug_on, _halt_on_error=_exit_on_error, _exit_on_fail=_exit_on_fail)
self.host = host self.host = host
self.port = port self.port = port
self.radio = radio self.radio = radio
@@ -52,6 +53,8 @@ class IPV4L4(LFCliBase):
self.target_requests_per_ten = int(target_requests_per_ten) self.target_requests_per_ten = int(target_requests_per_ten)
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port) self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.l4cxprofile=realm.L4CXProfile(lfclient_host=host,
lfclient_port=port,local_realm=self.local_realm)
self.station_profile = self.local_realm.new_station_profile() self.station_profile = self.local_realm.new_station_profile()
self.cx_profile = self.local_realm.new_l4_cx_profile() self.cx_profile = self.local_realm.new_l4_cx_profile()
@@ -67,20 +70,6 @@ class IPV4L4(LFCliBase):
self.cx_profile.url = self.url self.cx_profile.url = self.url
self.cx_profile.requests_per_ten = self.requests_per_ten self.cx_profile.requests_per_ten = self.requests_per_ten
# def __check_request_rate(self):
# endp_list = self.json_get("layer4/list?fields=urls/s")
# expected_passes = 0
# passes = 0
# if endp_list is not None and endp_list['endpoint'] is not None:
# endp_list = endp_list['endpoint']
# for item in endp_list:
# for name, info in item.items():
# if name in self.cx_profile.created_cx.keys():
# expected_passes += 1
# if info['urls/s'] * self.requests_per_ten >= self.target_requests_per_ten * .9:
# print(name, info['urls/s'], info['urls/s'] * self.requests_per_ten, self.target_requests_per_ten * .9)
# passes += 1
# return passes == expected_passes
def build(self): def build(self):
# Build stations # Build stations
@@ -105,32 +94,8 @@ class IPV4L4(LFCliBase):
self._fail("Stations failed to get IPs", print_fail) self._fail("Stations failed to get IPs", print_fail)
exit(1) exit(1)
self.cx_profile.start_cx() self.cx_profile.start_cx()
print("Starting test") print("Starting test...")
curr_time = datetime.datetime.now()
end_time = self.local_realm.parse_time(self.test_duration) + curr_time
sleep_interval = self.local_realm.parse_time(self.test_duration) // 5
passes = 0
expected_passes = 0
for test in range(self.num_tests):
expected_passes += 1
while curr_time < end_time:
time.sleep(sleep_interval.total_seconds())
if self.debug:
print(".",end="")
curr_time = datetime.datetime.now()
if self.cx_profile.check_errors(self.debug):
if self.__check_request_rate():
passes += 1
else:
self._fail("FAIL: Request rate did not exceed 90% target rate", print_fail)
break
else:
self._fail("FAIL: Errors found getting to %s " % self.url, print_fail)
break
if passes == expected_passes:
self._pass("PASS: All tests passes", print_pass)
def stop(self): def stop(self):
self.cx_profile.stop_cx() self.cx_profile.stop_cx()
self.station_profile.admin_down() self.station_profile.admin_down()
@@ -202,12 +167,31 @@ python3 ./test_ipv4_l4_urls_per_ten.py
optional.add_argument('--target_per_ten', help='--target_per_ten target number of request per ten minutes. test will check for 90 percent this value',default=600) optional.add_argument('--target_per_ten', help='--target_per_ten target number of request per ten minutes. test will check for 90 percent this value',default=600)
optional.add_argument('--mode',help='Used to force mode of stations') 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') optional.add_argument('--ap',help='Used to force a connection to a particular AP')
optional.add_argument('--report_file',help='where you want to store results')
optional.add_argument('--output_format', help='choose either csv or xlsx')
args = parser.parse_args() args = parser.parse_args()
num_sta = 2 num_sta = 2
if (args.num_stations is not None) and (int(args.num_stations) > 0): if (args.num_stations is not None) and (int(args.num_stations) > 0):
num_stations_converted = int(args.num_stations) num_stations_converted = int(args.num_stations)
num_sta = num_stations_converted num_sta = num_stations_converted
if args.report_file is None:
if args.output_format in ['csv','json','html','hdf','stata','pickle','pdf','parquet','xlsx']:
output_form=args.output_format
print("Defaulting file output placement to /home/lanforge.")
rpt_file='/home/lanforge/data.' + output_form
else:
print("Defaulting data file output type to Excel")
rpt_file='/home/lanforge/data.xlsx'
output_form='xlsx'
else:
rpt_file=args.report_file
if args.output_format is None:
output_form=str(args.report_file).split('.')[-1]
else:
output_form=args.output_format
station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=num_sta-1, padding_number_=10000, station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=num_sta-1, padding_number_=10000,
@@ -230,6 +214,16 @@ python3 ./test_ipv4_l4_urls_per_ten.py
ip_test.cleanup(station_list) ip_test.cleanup(station_list)
ip_test.build() ip_test.build()
ip_test.start() ip_test.start()
try:
layer4traffic=','.join([[*x.keys()][0] for x in ip_test.local_realm.json_get('layer4')['endpoint']])
except:
pass
ip_test.l4cxprofile.monitor(report_file=rpt_file, duration_sec=ip_test.local_realm.parse_time(args.test_duration).total_seconds(),
created_cx=layer4traffic,
output_format=output_form,
script_name='test_ipv4_l4_urls_per_ten',
arguments=args)
ip_test.stop() ip_test.stop()
if not ip_test.passes(): if not ip_test.passes():
print(ip_test.get_fail_message()) print(ip_test.get_fail_message())

26
py-scripts/test_ipv4_variable_time.py
View File

@@ -199,14 +199,17 @@ python3 ./test_ipv4_variable_time.py
if (args.num_stations is not None) and (int(args.num_stations) > 0): if (args.num_stations is not None) and (int(args.num_stations) > 0):
num_sta = int(args.num_stations) num_sta = int(args.num_stations)
#Create directory
homedir=str(datetime.datetime.now()).replace(':','-')+'test_ipv4_variable_time'
os.mkdir('/home/lanforge/report_data/'+homedir)
if args.report_file is None: if args.report_file is None:
if args.output_format in ['csv','json','html','hdf','stata','pickle','pdf','parquet']: if args.output_format in ['csv','json','html','hdf','stata','pickle','pdf','parquet','xlsx']:
report_f='/home/lanforge/report-data/'+str(datetime.datetime.now()).replace(':','-')+'test_ipv4_variable_time.' + args.output_format report_f='/home/lanforge/report-data/'+homedir+'/data.' + args.output_format
output=args.output_format output=args.output_format
else: else:
print('Defaulting to Excel') print('Defaulting data file output type to Excel')
report_f='/home/lanforge/report-data/'+str(datetime.datetime.now()).replace(':','-')+'test_ipv4_variable_time.xlsx' report_f='/home/lanforge/report-data/'+homedir+'/data.xlsx'
output='excel' output='xlsx'
else: else:
report_f=args.report_file report_f=args.report_file
if args.output_format is None: if args.output_format is None:
@@ -240,16 +243,19 @@ python3 ./test_ipv4_variable_time.py
ip_var_test.exit_fail() ip_var_test.exit_fail()
ip_var_test.start(False, False) ip_var_test.start(False, False)
try:
layer3connections=','.join([[*x.keys()][0] for x in ip_var_test.l3cxprofile.json_get('endp')['endpoint']]) layer3connections=','.join([[*x.keys()][0] for x in ip_var_test.local_realm.json_get('endp')['endpoint']])
ip_var_test.l3cxprofile.monitor(col_names=['Name','Tx Rate','Rx Rate','Tx PDUs','Rx PDUs'], except:
raise ValueError('Try setting the upstream port flag if your device does not have an eth1 port')
ip_var_test.l3cxprofile.monitor(col_names=['Name','Tx Rate','Rx Rate','Tx PDUs','Rx PDUs','Rx Drop % A', 'Rx Drop % B', 'Bps Rx A', 'Bps Rx B', 'Rx Rate', 'Cx Estab'],
report_file=report_f, report_file=report_f,
duration_sec=ip_var_test.local_realm.parse_time(args.test_duration).seconds, duration_sec=ip_var_test.local_realm.parse_time(args.test_duration).total_seconds(),
created_cx= layer3connections, created_cx= layer3connections,
output_format=output, output_format=output,
script_name='test_ipv4_variable_time', script_name='test_ipv4_variable_time',
show=show,
arguments=args) arguments=args)
ip_var_test.stop() ip_var_test.stop()
if not ip_var_test.passes(): if not ip_var_test.passes():
print(ip_var_test.get_fail_message()) print(ip_var_test.get_fail_message())