From 2bc6a2a541959dbddfb9d43a67217f4bc172ff13 Mon Sep 17 00:00:00 2001 From: Jed Reynolds Date: Wed, 16 Dec 2020 23:13:08 -0800 Subject: [PATCH] test_ipv4_variable_time.py: multiple fixes: - whitespace - calls super()._init_ with _local_realm parameter - fixes missing mode parameter check - updates optional and required arguments - adds help for --mode --- py-scripts/test_ipv4_variable_time.py | 123 ++++++++++++++++++-------- 1 file changed, 87 insertions(+), 36 deletions(-) diff --git a/py-scripts/test_ipv4_variable_time.py b/py-scripts/test_ipv4_variable_time.py index 26a08ab3..5f264f83 100755 --- a/py-scripts/test_ipv4_variable_time.py +++ b/py-scripts/test_ipv4_variable_time.py @@ -10,6 +10,7 @@ if sys.version_info[0] != 3: if 'py-json' not in sys.path: sys.path.append(os.path.join(os.path.abspath('..'), 'py-json')) +import pprint import argparse from LANforge.lfcli_base import LFCliBase from LANforge import LFUtils @@ -20,15 +21,35 @@ import datetime class IPV4VariableTime(LFCliBase): def __init__(self, - ssid, security, password, sta_list, name_prefix, upstream, radio, - host="localhost", port=8080, mode = 0, ap=None, - side_a_min_rate=56, side_a_max_rate=0, - side_b_min_rate=56, side_b_max_rate=0, - number_template="00000", test_duration="5m", use_ht160=False, + ssid, + security, + password, + sta_list, + name_prefix, + upstream, + radio, + host="localhost", + port=8080, + mode=0, + ap=None, + side_a_min_rate=56000, + side_a_max_rate=0, # indicates 'same' + side_b_min_rate=56000, + side_b_max_rate=0, # indicates 'same' + number_template="00000", + test_duration="5m", + use_ht160=False, _debug_on=False, _exit_on_error=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, + _local_realm = realm.Realm(lfclient_host=host, + lfclient_port=port, + debug_=_debug_on, + halt_on_error_=_exit_on_error), + _debug=_debug_on, + _halt_on_error=_exit_on_error, + _exit_on_fail=_exit_on_fail) self.upstream = upstream self.host = host self.port = port @@ -43,10 +64,8 @@ class IPV4VariableTime(LFCliBase): self.debug = _debug_on self.name_prefix = name_prefix 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_l3_cx_profile() - self.station_profile.lfclient_url = self.lfclient_url self.station_profile.ssid = self.ssid self.station_profile.ssid_pass = self.password @@ -56,10 +75,10 @@ class IPV4VariableTime(LFCliBase): self.station_profile.use_ht160 = use_ht160 if self.station_profile.use_ht160: self.station_profile.mode = 9 - self.station_profile.mode = mode + if mode is not None: + self.station_profile.mode = mode if self.ap is not None: - self.station_profile.set_command_param("add_sta", "ap",self.ap) - + self.station_profile.set_command_param("add_sta", "ap", self.ap) self.cx_profile.host = self.host self.cx_profile.port = self.port @@ -69,7 +88,6 @@ class IPV4VariableTime(LFCliBase): self.cx_profile.side_b_min_bps = side_b_min_rate self.cx_profile.side_b_max_bps = side_b_max_rate - def __get_rx_values(self): cx_list = self.json_get("endp?fields=name,rx+bytes", debug_=self.debug) if self.debug: @@ -80,8 +98,8 @@ class IPV4VariableTime(LFCliBase): if cx_name != 'uri' and cx_name != 'handler': for item, value in cx_name.items(): for value_name, value_rx in value.items(): - if value_name == 'rx bytes' and item in self.cx_profile.created_cx.values(): - cx_rx_map[item] = value_rx + if (value_name == 'rx bytes') and (item in self.cx_profile.created_cx.values()): + cx_rx_map[item] = value_rx return cx_rx_map def __compare_vals(self, old_list, new_list): @@ -140,7 +158,6 @@ class IPV4VariableTime(LFCliBase): if passes == expected_passes: self._pass("PASS: All tests passed") - def stop(self): self.cx_profile.stop_cx() self.station_profile.admin_down() @@ -153,19 +170,24 @@ class IPV4VariableTime(LFCliBase): def cleanup(self): self.cx_profile.cleanup() self.station_profile.cleanup() - LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=self.station_profile.station_names, + LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, + port_list=self.station_profile.station_names, debug=self.debug) def build(self): - self.station_profile.use_security(self.security, self.ssid, self.password) self.station_profile.set_number_template(self.number_template) print("Creating stations") 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(endp_type="lf_udp", side_a=self.station_profile.station_names, side_b=self.upstream, sleep_time=0) + self.station_profile.create(radio=self.radio, + sta_names_=self.sta_list, + debug=self.debug) + self.cx_profile.create(endp_type="lf_udp", + side_a=self.station_profile.station_names, + side_b=self.upstream, + sleep_time=0) self._pass("PASS: Station build finished") @@ -174,16 +196,13 @@ def main(): prog='test_ipv4_variable_time.py', formatter_class=argparse.RawTextHelpFormatter, epilog='''\ - Create stations to test connection and traffic on VAPs of varying security types (WEP, WPA, WPA2, WPA3, Open) - ''', +Create stations to test connection and traffic on VAPs of varying security types (WEP, WPA, WPA2, WPA3, Open) +''', description='''\ test_ipv4_variable_time.py: -------------------- -Generic command layout: - - python3 ./test_ipv4_variable_time.py - --upstream_port eth1 +Options: --upstream_port eth1 --radio wiphy0 --num_stations 32 --security {open|wep|wpa|wpa2|wpa3} \\ @@ -208,17 +227,46 @@ Generic command layout: --a_min 1000 --b_min 1000 --ap "00:0e:8e:78:e1:76" - --debug - ''') + --debug''') + + #optional = parser.add_argument_group('optional arguments') + #required = parser.add_argument_group('required arguments') + required = None + for agroup in parser._action_groups: + if agroup.title == "required arguments": + required = agroup + # pprint.pprint(required) + if required is not None: + required.add_argument('--a_min', help='bps rate minimum for endpoint A', type=int, default=256000) + required.add_argument('--b_min', help='bps rate minimum for endpoint B', type=int, default=256000) + required.add_argument('--test_duration', help='duration of the test', default="2m") + + optional = None + for agroup in parser._action_groups: + if agroup.title == "optional arguments": + optional = agroup + # pprint.pprint(optional) + if optional is not None: + optional.add_argument('--monitor', help='print CSV output of connection status', default=False, action='store_true') + optional.add_argument('--save_to', help='save CSV monitor output to file', default=None) + optional.add_argument('--monitor_interval_sec', help='seconds between polling endpoints', default=5) + optional.add_argument('--mode' help="""Station WiFi modes: use the number value below: + auto : 0, + a : 1, + b : 2, + g : 3, + abg : 4, + abgn : 5, + bgn : 6, + bg : 7, + abgnAC : 8, + anAC : 9, + an : 10, + bgnAC : 11, + abgnAX : 12, + bgnAX : 13, +""") - optional = parser.add_argument_group('optional arguments') - required = parser.add_argument_group('required arguments') - parser.add_argument('--a_min', help='--a_min bps rate minimum for side_a', default=256000) - parser.add_argument('--b_min', help='--b_min bps rate minimum for side_b', default=256000) - parser.add_argument('--test_duration', help='--test_duration sets the duration of the test', default="2m") - required.add_argument('--security', help='WiFi Security protocol: < open | wep | wpa | wpa2 | wpa3 >', required=True) - 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') args = parser.parse_args() num_sta = 2 @@ -226,7 +274,10 @@ Generic command layout: num_sta = int(args.num_stations) station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=num_sta-1, padding_number_=10000, radio=args.radio) - ip_var_test = IPV4VariableTime(host=args.mgr, port=args.mgr_port, number_template="00", sta_list=station_list, + ip_var_test = IPV4VariableTime(host=args.mgr, + port=args.mgr_port, + number_template="00", + sta_list=station_list, name_prefix="VT", upstream=args.upstream_port, ssid=args.ssid,