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

Added speedtest test and cli arguments

Browse Source
This commit is contained in:
Logan Lipke
2020-10-28 16:44:28 -07:00
parent 81dfade8d2
commit 7bb0c0818b
1 changed files with 72 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

86
py-scripts/test_generic.py
View File

@@ -16,15 +16,15 @@ from LANforge import LFUtils
import realm
import time
import datetime
import json
class GenTest(LFCliBase):
def __init__(self, host, port, ssid, security, password, sta_list, name_prefix, upstream,
number_template="000", test_duration="5m", type="lfping", dest="127.0.0.1", cmd ="",
interval=1, radio="wiphy0",
interval=1, radio="wiphy0", speedtest_min_up=None, speedtest_min_dl=None, speedtest_max_ping=None,
_debug_on=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)
self.host = host
self.port = port
@@ -37,6 +37,9 @@ class GenTest(LFCliBase):
self.number_template = number_template
self.name_prefix = name_prefix
self.test_duration = test_duration
self.speedtest_min_up = float(speedtest_min_up)
self.speedtest_min_dl = float(speedtest_min_dl)
self.speedtest_max_ping = float(speedtest_max_ping)
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.station_profile = self.local_realm.new_station_profile()
self.generic_endps_profile = self.local_realm.new_generic_endp_profile()
@@ -54,6 +57,45 @@ class GenTest(LFCliBase):
self.generic_endps_profile.cmd = cmd
self.generic_endps_profile.interval = interval
def choose_ping_command(self):
gen_results = self.json_get("generic/list?fields=name,last+results", debug_=self.debug)
if gen_results['endpoints'] is not None:
for name in gen_results['endpoints']:
for k, v in name.items():
if v['name'] in self.generic_endps_profile.created_endp and not v['name'].endswith('1'):
if v['last results'] != "" and "Unreachable" not in v['last results']:
return True, v['name']
else:
return False, v['name']
def choose_lfcurl_command(self):
return False, ''
def choose_speedtest_command(self):
gen_results = self.json_get("generic/list?fields=name,last+results", debug_=self.debug)
if gen_results['endpoints'] is not None:
for name in gen_results['endpoints']:
for k, v in name.items():
if v['last results'] is not None and v['name'] in self.generic_endps_profile.created_endp and v['last results'] != '':
last_results = json.loads(v['last results'])
if last_results['download'] is None and last_results['upload'] is None and last_results['ping'] is None:
return False, v['name']
elif last_results['download'] >= self.speedtest_min_dl and \
last_results['upload'] >= self.speedtest_min_up and \
last_results['ping'] <= self.speedtest_max_ping:
return True, v['name']
def choose_generic_command(self):
gen_results = self.json_get("generic/list?fields=name,last+results", debug_=self.debug)
if gen_results['endpoints'] is not None:
for name in gen_results['endpoints']:
for k, v in name.items():
if v['name'] in self.generic_endps_profile.created_endp and not v['name'].endswith('1'):
if v['last results'] != "" and "not known" not in v['last results']:
return True, v['name']
else:
return False, v['name']
def start(self, print_pass=False, print_fail=False):
self.station_profile.admin_up()
temp_stas = []
@@ -73,19 +115,29 @@ class GenTest(LFCliBase):
time.sleep(15)
end_time = self.local_realm.parse_time("30s") + cur_time
print("Starting Test...")
result = False
while cur_time < end_time:
cur_time = datetime.datetime.now()
gen_results = self.json_get("generic/list?fields=name,last+results", debug_=self.debug)
if gen_results['endpoints'] is not None:
for name in gen_results['endpoints']:
for k, v in name.items():
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.generic_endps_profile.dest), print_fail)
break
if self.generic_endps_profile.type == "lfping":
result = self.choose_ping_command()
elif self.generic_endps_profile.type == "generic":
result = self.choose_generic_command()
elif self.generic_endps_profile.type == "lfcurl":
result = self.choose_lfcurl_command()
elif self.generic_endps_profile.type == "speedtest":
result = self.choose_speedtest_command()
else:
continue
expected_passes += 1
# pprint.pprint(result)
if result is not None:
if result[0]:
passes += 1
else:
self._fail("%s Failed to ping %s " % (result[1], self.generic_endps_profile.dest),
print_fail)
break
# print(cur_time)
# print(end_time)
time.sleep(1)
@@ -144,6 +196,9 @@ python3 ./test_generic.py --upstream_port eth1 \\
parser.add_argument('--dest', help='destination IP for command', default="10.40.0.1")
parser.add_argument('--test_duration', help='duration of the test eg: 30s, 2m, 4h', default="2m")
parser.add_argument('--interval', help='interval to use when running lfping (1s, 1m)', default=1)
parser.add_argument('--speedtest_min_up', help='sets the minimum upload threshold for the speedtest type', default=None)
parser.add_argument('--speedtest_min_dl', help='sets the minimum download threshold for the speedtest type', default=None)
parser.add_argument('--speedtest_max_ping', help='sets the minimum ping threshold for the speedtest type', default=None)
args = parser.parse_args()
num_sta = 2
@@ -169,6 +224,9 @@ python3 ./test_generic.py --upstream_port eth1 \\
password=args.passwd,
security=args.security,
test_duration=args.test_duration,
speedtest_min_up=args.speedtest_min_up,
speedtest_min_dl=args.speedtest_min_dl,
speedtest_max_ping=args.speedtest_max_ping,
_debug_on=args.debug)
generic_test.cleanup(station_list)