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

cleanup method and wait_for_ip implemented

Browse Source
This commit is contained in:
Dipti
2020-12-07 01:13:05 -08:00
parent 744ac3b4ff
commit d99b3a9dac
1 changed files with 58 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
py-scripts/test_1k_clients_jedtest.py
View File

@@ -14,13 +14,14 @@ from LANforge import LFUtils
import argparse
import time
import datetime
import pprint
class Test1KClients(LFCliBase):
def __init__(self,
host,
port,
num_sta_=200,
_debug_on=False,
_debug_on=True,
_exit_on_error=False,
_exit_on_fail=False):
super().__init__(host,
@@ -41,7 +42,6 @@ class Test1KClients(LFCliBase):
if num_sta_ is None:
raise ValueError("need a number of stations per radio")
self.num_sta = int(num_sta_)
self.station_radio_map = {
# port_name_series(prefix=prefix_, start_id=start_id_, end_id=end_id_, padding_number=padding_number_, radio=radio)
"1.1.wiphy0" : LFUtils.port_name_series(start_id=0, end_id=self.num_sta-1, padding_number=10000, radio="1.1.wiphy0"),
@@ -52,8 +52,21 @@ class Test1KClients(LFCliBase):
"1.2.wiphy1" : LFUtils.port_name_series(start_id=4000, end_id=4000+self.num_sta-1, padding_number=10000, radio="1.2.wiphy1"),
"1.2.wiphy2" : LFUtils.port_name_series(start_id=5000, end_id=5000+self.num_sta-1, padding_number=10000, radio="1.2.wiphy2")
}
self.station_profile_map = {}
self.name_prefix = "ONEA"
side_a_max_rate = 56000
side_a_min_rate = side_a_max_rate
side_b_max_rate = 56000
side_b_min_rate = side_b_max_rate
self.cx_profile = self.local_realm.new_l3_cx_profile()
self.cx_profile.name_prefix = self.name_prefix
self.cx_profile.side_a_min_bps = side_a_min_rate
self.cx_profile.side_a_max_bps = side_a_max_rate
self.cx_profile.side_b_min_bps = side_b_min_rate
self.cx_profile.side_b_max_bps = side_b_max_rate
self.station_profile_map = {}
def build(self):
for (radio, name_series) in self.station_radio_map.items():
print("building stations for %s"%radio)
@@ -65,7 +78,8 @@ class Test1KClients(LFCliBase):
self.ssid_radio_map[radio][1],
self.ssid_radio_map[radio][2])
self.station_profile_map[radio] = station_profile
#creating phantom stations here
self._pass("defined %s station profiles" % len(self.station_radio_map))
for (radio, station_profile) in self.station_profile_map.items():
station_profile.create(radio=radio,
@@ -78,57 +92,55 @@ class Test1KClients(LFCliBase):
hs20_enable=False,
sleep_time=2)
self.local_realm.wait_until_ports_appear(self.station_radio_map[radio])
self._pass("built stations on %s radios" % len(self.station_radio_map))
def start(self):
print("bringing stations up")
prev_ip_num=0
for (radio, station_profile) in self.station_profile_map.items():
station_profile.admin_up()
if self.local_realm.wait_for_ip(station_list=self.station_radio_map[radio], debug=self.debug, timeout_sec=90):
self._pass("stations on radio %s up" % radio)
else:
self._fail("stations on radio %s are still down" % radio)
exit(1)
total_num_sta=6*self.num_sta
self.local_realm.wait_for_ip(station_list=self.station_radio_map[radio], debug=self.debug, timeout_sec=30)
curr_ip_num = self.local_realm.check_for_num_curr_ips(num_sta_with_ips=prev_ip_num,station_list=self.station_radio_map[radio], debug=self.debug)
print(curr_ip_num)
print(prev_ip_num)
print(total_num_sta)
while ((prev_ip_num < curr_ip_num) and (curr_ip_num < total_num_sta)):
self.local_realm.wait_for_ip(station_list=self.station_radio_map[radio], debug=self.debug, timeout_sec=60)
prev_ip_num = curr_ip_num
curr_ip_num = self.local_realm.check_for_num_curr_ips(num_sta_with_ips=prev_ip_num,station_list=self.station_radio_map[radio], debug=self.debug)
if curr_ip_num == total_num_sta:
self._pass("stations on radio %s up" % radio)
else:
self._fail("FAIL: Not all stations on radio %s up" % radio)
exit(1)
cur_time = datetime.datetime.now()
old_cx_rx_values = self.__get_rx_values() #
end_time = self.local_realm.parse_time("3m") + cur_time
self.cx_profile.start_cx() #
passes = 0
expected_passes = 0
while cur_time < end_time:
interval_time = cur_time + datetime.timedelta(minutes=1)
while cur_time < interval_time:
cur_time = datetime.datetime.now()
time.sleep(1)
new_cx_rx_values = self.__get_rx_values() #
# print(old_cx_rx_values, new_cx_rx_values)
# print("\n-----------------------------------")
# print(cur_time, end_time, cur_time + datetime.timedelta(minutes=1))
# print("-----------------------------------\n")
expected_passes += 1
if self.__compare_vals(old_cx_rx_values, new_cx_rx_values): #
passes += 1
else:
self._fail("FAIL: Not all stations increased traffic", print_fail)
break
old_cx_rx_values = new_cx_rx_values #
cur_time = datetime.datetime.now()
# old_cx_rx_values = self.__get_rx_values() #
end_time = self.local_realm.parse_time("3m") + cur_time
self.cx_profile.start_cx() #
passes = 0
expected_passes = 0
while cur_time < end_time:
interval_time = cur_time + datetime.timedelta(minutes=1)
while cur_time < interval_time:
cur_time = datetime.datetime.now()
time.sleep(1)
cur_time = datetime.datetime.now()
if passes == expected_passes:
self._pass("PASS: All tests passed", print_pass)
if passes == expected_passes:
self._pass("PASS: All tests passed", print_pass)
def stop(self):
pass
def cleanup(self):
#for (radio, station_list) in self.station_radio_map.items():
self.local_realm.remove_all_stations(1)
self.local_realm.remove_all_stations(2)
#self.cx_profile.cleanup_prefix()
for (radio, name_series) in self.station_radio_map.items():
sta_list= self.station_radio_map[radio]
for sta in sta_list:
self.local_realm.rm_port(sta, check_exists=True)
def main():
num_sta=200
@@ -141,12 +153,17 @@ def main():
argparser.add_argument("--sta_per_radio",
type=int,
help="number of stations per radio")
argparser.add_argument("--test_duration",
type=int,
help="length of test duration")
args = argparser.parse_args()
kilo_test = Test1KClients(lfjson_host,
lfjson_port,
num_sta_=args.sta_per_radio)
num_sta_=args.sta_per_radio,
_debug_on=args.debug)
kilo_test.cleanup()
kilo_test.build()
if not kilo_test.passes():