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

test_ipv4_variable_time.py: multiple fixes:

Browse Source 
- whitespace
- calls super()._init_ with _local_realm parameter
- fixes missing mode parameter check
- updates optional and required arguments
- adds help for --mode
This commit is contained in:
Jed Reynolds
2020-12-16 23:13:08 -08:00
parent 535f15e6c3
commit 2bc6a2a541
1 changed files with 87 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

123
py-scripts/test_ipv4_variable_time.py
View File

@@ -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,