scottk/wlan-lanforge-scripts
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-lanforge-scripts.git synced 2025-10-30 02:12:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

JSON: test_generic:

Browse Source 
- fixes endpoint creation by waiting for stations
- uses new member names
- uses new method names
This commit is contained in:
Jed Reynolds
2020-10-21 00:05:11 -07:00
parent 42ceac85ee
commit 48b6c40e2a
1 changed files with 40 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
py-scripts/test_generic.py
View File

@@ -1,5 +1,5 @@
#!/usr/bin/env python3
import pprint
import sys
import os
@@ -20,7 +20,7 @@ import datetime
class GenTest(LFCliBase):
def __init__(self, host, port, ssid, security, password, sta_list, name_prefix, upstream,
number_template="00000", test_duration="5m", type="lfping", dest="127.0.0.1",
number_template="000", test_duration="5m", type="lfping", dest="127.0.0.1",
interval=1, radio="wiphy0",
_debug_on=False,
_exit_on_error=False,
@@ -39,7 +39,7 @@ class GenTest(LFCliBase):
self.test_duration = test_duration
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_generic_cx_profile()
self.generic_endps_profile = self.local_realm.new_generic_endp_profile()
self.station_profile.lfclient_url = self.lfclient_url
self.station_profile.ssid = self.ssid
@@ -48,12 +48,15 @@ class GenTest(LFCliBase):
self.station_profile.number_template_ = self.number_template
self.station_profile.mode = 0
self.cx_profile.type = type
self.cx_profile.dest = dest
self.cx_profile.interval = interval
self.generic_endps_profile.name = name_prefix
self.generic_endps_profile.type = type
self.generic_endps_profile.dest = dest
self.generic_endps_profile.interval = interval
def start(self, print_pass=False, print_fail=False):
self.station_profile.admin_up()
pprint.pprint(self.station_profile.station_names)
LFUtils.wait_until_ports_admin_up(base_url=self.lfclient_url, port_list=self.station_profile.station_names)
temp_stas = self.sta_list.copy()
temp_stas.append(self.upstream)
if self.local_realm.wait_for_ip(temp_stas):
@@ -64,7 +67,7 @@ class GenTest(LFCliBase):
cur_time = datetime.datetime.now()
passes = 0
expected_passes = 0
self.cx_profile.start_cx()
self.generic_endps_profile.start_cx()
time.sleep(15)
end_time = self.local_realm.parse_time("30s") + cur_time
print("Starting Test...")
@@ -74,12 +77,12 @@ class GenTest(LFCliBase):
if gen_results['endpoints'] is not None:
for name in gen_results['endpoints']:
for k, v in name.items():
if v['name'] in self.cx_profile.created_endp and not v['name'].endswith('1'):
if v['name'] in self.generic_endps_profile.created_endp and not v['name'].endswith('1'):
expected_passes += 1
if v['last results'] != "" and "Unreachable" not in v['last results']:
passes += 1
else:
self._fail("%s Failed to ping %s " % (v['name'], self.cx_profile.dest), print_fail)
self._fail("%s Failed to ping %s " % (v['name'], self.generic_endps_profile.dest), print_fail)
break
# print(cur_time)
# print(end_time)
@@ -89,7 +92,7 @@ class GenTest(LFCliBase):
self._pass("PASS: All tests passed", print_pass)
def stop(self):
self.cx_profile.stop_cx()
self.generic_endps_profile.stop_cx()
self.station_profile.admin_down()
def build(self):
@@ -99,15 +102,16 @@ class GenTest(LFCliBase):
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=self.radio, sta_names_=self.sta_list, debug=self.debug)
self.cx_profile.create(ports=self.station_profile.station_names, sleep_time=.5)
self.generic_endps_profile.create(ports=self.station_profile.station_names, sleep_time=.5)
self._pass("PASS: Station build finished")
def cleanup(self, sta_list):
self.cx_profile.cleanup()
self.generic_endps_profile.cleanup()
self.station_profile.cleanup(sta_list)
LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=sta_list,
debug=self.debug)
LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=sta_list, debug=self.debug)
def main():
@@ -139,15 +143,29 @@ python3 ./test_generic.py --upstream_port eth1 \\
parser.add_argument('--interval', help='interval to use when running lfping (1s, 1m)', default=1)
args = parser.parse_args()
num_sta = 2
if (args.num_stations is not None) and (int(args.num_stations) > 0):
num_sta = int(args.num_stations)
station_list = LFUtils.portNameSeries(radio=args.radio,
prefix_="sta",
start_id_=0,
end_id_=num_sta-1,
padding_number_=100)
station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=1, padding_number_=10000,
radio=args.radio)
generic_test = GenTest(args.mgr, lfjson_port, number_template="00", sta_list=station_list,
name_prefix="GT", type=args.type, dest=args.dest, interval=1,
ssid=args.ssid, upstream=args.upstream_port,
password=args.passwd, security=args.security, test_duration=args.test_duration,
_debug_on=args.debug, radio=args.radio)
generic_test = GenTest(args.mgr, lfjson_port,
number_template="00",
radio=args.radio,
sta_list=station_list,
name_prefix="GT",
type=args.type,
dest=args.dest,
interval=1,
ssid=args.ssid,
upstream=args.upstream_port,
password=args.passwd,
security=args.security,
test_duration=args.test_duration,
_debug_on=args.debug)
generic_test.cleanup(station_list)
generic_test.build()