diff --git a/py-scripts/test_generic.py b/py-scripts/test_generic.py index 69ff3e61..4ca32543 100755 --- a/py-scripts/test_generic.py +++ b/py-scripts/test_generic.py @@ -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)