scottk/wlan-lanforge-scripts
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-lanforge-scripts.git synced 2025-10-30 18:27:53 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
7d0a01eb326887d2b04435b2bab5d0a625b5a919
wlan-lanforge-scripts/py-scripts/test_generic.py
Matthew Stidham 9e2e4a1c73 test_generic: Fix os.path.exists error 
Signed-off-by: Matthew Stidham <stidmatt@gmail.com>
2021-11-29 16:46:34 -08:00

372 lines
18 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""
NAME: test_generic.py
PURPOSE:
test_generic.py will create stations and endpoints to generate traffic based on a command-line specified command type.
This script will create a variable number of stations to test generic endpoints. Multiple command types can be tested
including ping, speedtest, generic types. The test will check the last-result attribute for different things
depending on what test is being run. Ping will test for successful pings, speedtest will test for download
speed, upload speed, and ping time, generic will test for successful generic commands
SETUP:
Enable the generic tab in LANforge GUI
EXAMPLE:
LFPING:
./test_generic.py --radio wiphy1 --ssid ct523c --passwd ct523c --security wpa2 --num_stations 4 --type lfping --dest "192.168.0.104" --debug
LFCURL (under construction):
./test_generic.py --mgr localhost --mgr_port 4122 --radio wiphy1 --num_stations 26 --ssid jedway-wpa2-x2048-4-1 --passwd jedway-wpa2-x2048-4-1 --security wpa2 --type lfcurl --dest 10.40.0.1
GENERIC:
./test_generic.py --mgr localhost--mgr_port 4122 --radio wiphy1 --num_stations 2 --ssid jedway-wpa2-x2048-4-1 --passwd jedway-wpa2-x2048-4-1 --security wpa2 --type generic
SPEEDTEST:
./test_generic.py --mgr localhost --mgr_port 4122 --radio wiphy2 --num_stations 13 --ssid jedway-wpa2-x2048-4-1 --passwd jedway-wpa2-x2048-4-1 --type speedtest --speedtest_min_up 20
--speedtest_min_dl 20 --speedtest_max_ping 150 --security wpa2
IPERF3 (under construction):
./test_generic.py --mgr localhost --mgr_port 4122 --radio wiphy1 --num_stations 3 --ssid jedway-wpa2-x2048-4-1 --passwd jedway-wpa2-x2048-4-1 --security wpa2 --type iperf3
Use './test_generic.py --help' to see command line usage and options
Copyright 2021 Candela Technologies Inc
License: Free to distribute and modify. LANforge systems must be licensed.
"""
import sys
import os
import importlib
import pprint
import argparse
import time
import datetime
if sys.version_info[0] != 3:
print("This script requires Python 3")
exit(1)
sys.path.append(os.path.join(os.path.abspath(__file__ + "../../../")))
lfcli_base = importlib.import_module("py-json.LANforge.lfcli_base")
LFCliBase = lfcli_base.LFCliBase
LFUtils = importlib.import_module("py-json.LANforge.LFUtils")
realm = importlib.import_module("py-json.realm")
Realm = realm.Realm
class GenTest(LFCliBase):
def __init__(self, ssid, security, passwd, sta_list, client, name_prefix, upstream, host="localhost", port=8080,
number_template="000", test_duration="5m", test_type="lfping", dest=None, cmd=None,
interval=1, radio=None, speedtest_min_up=None, speedtest_min_dl=None, speedtest_max_ping=None,
file_output=None,
loop_count=None,
_debug_on=False,
_exit_on_error=False,
_exit_on_fail=False):
super().__init__(host, port, _local_realm=Realm(host, port), _debug=_debug_on, _exit_on_fail=_exit_on_fail)
self.ssid = ssid
self.radio = radio
self.upstream = upstream
self.sta_list = sta_list
self.security = security
self.passwd = passwd
self.number_template = number_template
self.name_prefix = name_prefix
self.test_duration = test_duration
self.debug = _debug_on
if client:
self.client_name = client
self.station_profile = self.local_realm.new_station_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
self.station_profile.ssid_pass = self.passwd,
self.station_profile.security = self.security
self.station_profile.number_template_ = self.number_template
self.station_profile.mode = 0
self.generic_endps_profile.name = name_prefix
self.generic_endps_profile.type = test_type
self.generic_endps_profile.dest = dest
self.generic_endps_profile.cmd = cmd
self.generic_endps_profile.interval = interval
self.generic_endps_profile.file_output = file_output
self.generic_endps_profile.loop_count = loop_count
if speedtest_min_up is not None:
self.generic_endps_profile.speedtest_min_up = float(speedtest_min_up)
if speedtest_min_dl is not None:
self.generic_endps_profile.speedtest_min_dl = float(speedtest_min_dl)
if speedtest_max_ping is not None:
self.generic_endps_profile.speedtest_max_ping = float(speedtest_max_ping)
def check_tab_exists(self):
response = self.json_get("generic")
if response is None:
return False
else:
return True
def start(self):
self.station_profile.admin_up()
temp_stas = []
for station in self.sta_list.copy():
temp_stas.append(self.local_realm.name_to_eid(station)[2])
if self.debug:
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)
if self.local_realm.wait_for_ip(station_list=temp_stas, ipv4=True):
self._pass("All stations got IPs")
else:
self._fail("Stations failed to get IPs")
self.exit_fail()
self.generic_endps_profile.start_cx()
def stop(self):
print("Stopping Test...")
self.generic_endps_profile.stop_cx()
self.station_profile.admin_down()
def build(self):
self.station_profile.use_security(self.security, self.ssid, self.passwd)
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.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.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)
def main():
optional = []
optional.append({'name': '--mode', 'help': 'Used to force mode of stations'})
optional.append({'name': '--ap', 'help': 'Used to force a connection to a particular AP'})
optional.append({'name': '--output_format', 'help': 'choose either csv or xlsx'})
optional.append({'name': '--report_file', 'help': 'where you want to store results', 'default': None})
optional.append({'name': '--a_min', 'help': '--a_min bps rate minimum for side_a', 'default': 256000})
optional.append({'name': '--b_min', 'help': '--b_min bps rate minimum for side_b', 'default': 256000})
optional.append({'name': '--gen_cols', 'help': 'Columns wished to be monitored from layer 3 endpoint tab',
'default': ['name', 'tx bytes', 'rx bytes']})
optional.append({'name': '--port_mgr_cols', 'help': 'Columns wished to be monitored from port manager tab',
'default': ['ap', 'ip', 'parent dev']})
optional.append(
{'name': '--compared_report', 'help': 'report path and file which is wished to be compared with new report',
'default': None})
optional.append({'name': '--monitor_interval',
'help': 'how frequently do you want your monitor function to take measurements; 250ms, 35s, 2h',
'default': '2s'})
parser = LFCliBase.create_basic_argparse(
prog='test_generic.py',
formatter_class=argparse.RawTextHelpFormatter,
epilog='''Create generic endpoints and test for their ability to execute chosen commands\n''',
description='''
test_generic.py
--------------------
Generic command example:
python3 ./test_generic.py
--mgr localhost (optional)
--mgr_port 4122 (optional)
--upstream_port eth1 (optional)
--radio wiphy0 (required)
--num_stations 3 (optional)
--security {open|wep|wpa|wpa2|wpa3} (required)
--ssid netgear (required)
--passwd admin123 (required)
--type lfping {generic|lfping|iperf3-client | speedtest | lf_curl} (required)
--dest 10.40.0.1 (required - also target for iperf3)
--test_duration 2m
--interval 1s
--debug
Example commands:
LFPING:
./test_generic.py --mgr localhost --mgr_port 4122 --radio wiphy0 --num_stations 7 --ssid jedway-wpa2-x2048-4-1 --passwd jedway-wpa2-x2048-4-1 --type lfping --dest 10.40.0.1 --security wpa2
LFCURL (under construction):
./test_generic.py --mgr localhost --mgr_port 4122 --radio wiphy1 --num_stations 26 --ssid jedway-wpa2-x2048-4-1 --passwd jedway-wpa2-x2048-4-1 --security wpa2 --type lfcurl --dest 10.40.0.1
GENERIC:
./test_generic.py --mgr localhost--mgr_port 4122 --radio wiphy1 --num_stations 2 --ssid jedway-wpa2-x2048-4-1 --passwd jedway-wpa2-x2048-4-1 --security wpa2 --type generic
SPEEDTEST:
./test_generic.py --mgr localhost --mgr_port 4122 --radio wiphy2 --num_stations 13 --ssid jedway-wpa2-x2048-4-1 --passwd jedway-wpa2-x2048-4-1 --type speedtest --speedtest_min_up 20
--speedtest_min_dl 20 --speedtest_max_ping 150 --security wpa2
IPERF3 (under construction):
./test_generic.py --mgr localhost --mgr_port 4122 --radio wiphy1 --num_stations 3 --ssid jedway-wpa2-x2048-4-1 --passwd jedway-wpa2-x2048-4-1 --security wpa2 --type iperf3
''',
more_optional=optional)
parser.add_argument('--type', help='type of command to run: generic, lfping, iperf3-client, iperf3-server, lfcurl',
default="lfping")
parser.add_argument('--cmd', help='specifies command to be run by generic type endp', default='')
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)
parser.add_argument('--client', help='client to the iperf3 server', default=None)
parser.add_argument('--file_output', help='location to output results of lf_curl, absolute path preferred',
default=None)
parser.add_argument('--loop_count', help='determines the number of loops to use in lf_curl', default=None)
args = parser.parse_args()
num_sta = 2
if (args.num_stations is not None) and (int(args.num_stations) > 0):
num_stations_converted = int(args.num_stations)
num_sta = num_stations_converted
# Create directory
# if file path with output file extension is not given...
# check if home/lanforge/report-data exists. if not, save
# in new folder based in current file's directory
systeminfopath = None
if args.report_file is None:
new_file_path = str(datetime.datetime.now().strftime("%Y-%m-%d-%H-h-%M-m-%S-s")).replace(':',
'-') + '-test_generic' # create path name
if os.path.exists('/home/lanforge/report-data/'):
path = os.path.join('/home/lanforge/report-data/', new_file_path)
os.mkdir(path)
else:
curr_dir_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
path = os.path.join(curr_dir_path, new_file_path)
os.mkdir(path)
systeminfopath = str(path) + '/systeminfo.txt'
if args.output_format in ['csv', 'json', 'html', 'hdf', 'stata', 'pickle', 'pdf', 'png', 'parquet',
'xlsx']:
report_f = str(path) + '/data.' + args.output_format
output = args.output_format
else:
print(
'Not supporting this report format or cannot find report format provided. Defaulting to csv data file output type, naming it data.csv.')
report_f = str(path) + '/data.csv'
output = 'csv'
else:
systeminfopath = str(args.report_file).split('/')[-1]
report_f = args.report_file
if args.output_format is None:
output = str(args.report_file).split('.')[-1]
else:
output = args.output_format
print("Saving final report data in ... " + report_f)
# Retrieve last data file
compared_rept = None
if args.compared_report:
compared_report_format = args.compared_report.split('.')[-1]
# if compared_report_format not in ['csv', 'json', 'dta', 'pkl','html','xlsx','parquet','h5']:
if compared_report_format != 'csv':
print(ValueError("Cannot process this file type. Please select a different file and re-run script."))
exit(1)
else:
compared_rept = args.compared_report
station_list = LFUtils.portNameSeries(radio=args.radio,
prefix_="sta",
start_id_=0,
end_id_=num_sta - 1,
padding_number_=100)
generic_test = GenTest(host=args.mgr, port=args.mgr_port,
number_template="00",
radio=args.radio,
sta_list=station_list,
name_prefix="GT",
test_type=args.type,
dest=args.dest,
cmd=args.cmd,
interval=1,
ssid=args.ssid,
upstream=args.upstream_port,
passwd=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,
file_output=args.file_output,
loop_count=args.loop_count,
client=args.client,
_debug_on=args.debug)
if not generic_test.check_tab_exists():
raise ValueError("Error received from GUI, please ensure generic tab is enabled")
generic_test.cleanup(station_list)
generic_test.build()
if not generic_test.passes():
print(generic_test.get_fail_message())
generic_test.exit_fail()
generic_test.start()
if not generic_test.passes():
print(generic_test.get_fail_message())
generic_test.exit_fail()
try:
genconnections = ','.join([[*x.keys()][0] for x in generic_test.json_get('generic')['endpoints']])
except ValueError as error:
raise ValueError(
'1. Enable the generic tab in LANforge GUI , if still fails 2. Try setting the upstream port flag if your device does not have an eth1 port \n'
'%s' % error)
if type(args.gen_cols) is not list:
generic_cols = list(args.gen_cols.split(","))
# send col names here to file to reformat
else:
generic_cols = args.gen_cols
# send col names here to file to reformat
if type(args.port_mgr_cols) is not list:
port_mgr_cols = list(args.port_mgr_cols.split(","))
# send col names here to file to reformat
else:
port_mgr_cols = args.port_mgr_cols
# send col names here to file to reformat
if args.debug:
print("Generic Endp column names are...")
print(generic_cols)
print("Port Manager column names are...")
print(port_mgr_cols)
try:
monitor_interval = Realm.parse_time(args.monitor_interval).total_seconds()
except ValueError as error:
print(ValueError(
"The time string provided for monitor_interval argument is invalid. Please see supported time stamp increments and inputs for monitor_interval in --help. \n"
"%s" % error))
exit(1)
generic_test.start()
generic_test.generic_endps_profile.monitor(generic_cols=generic_cols,
sta_list=station_list,
# port_mgr_cols=port_mgr_cols,
report_file=report_f,
systeminfopath=systeminfopath,
duration_sec=Realm.parse_time(args.test_duration).total_seconds(),
monitor_interval_ms=monitor_interval,
created_cx=genconnections,
output_format=output,
compared_report=compared_rept,
script_name='test_generic',
arguments=args,
debug=args.debug)
generic_test.stop()
time.sleep(30)
generic_test.cleanup(station_list)
if generic_test.passes():
generic_test.exit_success()
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink