mirror of
https://github.com/Telecominfraproject/wlan-lanforge-scripts.git
synced 2025-11-01 03:07:56 +00:00
added option to run traffic on already created stations
Signed-off-by: shivamcandela <shivam.thakur@candelatech.com>
This commit is contained in:
shivamcandela
@@ -1,11 +1,20 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
"""test_ipv4_variable_time.py will create stations and endpoints to generate and verify layer-3 traffic.
|
||||
This Sdript has two working modes:
|
||||
Mode 1:
|
||||
When station is not available,
|
||||
|
||||
This script will create a variable number of stations each with their own set of cross-connects and endpoints.
|
||||
It will then create layer 3 traffic over a specified amount of time, testing for increased traffic at regular intervals.
|
||||
This test will pass if all stations increase traffic over the full test duration.
|
||||
|
||||
Mode 2:
|
||||
|
||||
When station is already available This script will create layer3 cross-connects and endpoints It will then
|
||||
create layer 3 traffic over a specified amount of time, testing for increased traffic at regular intervals.
|
||||
This test will pass if all stations increase traffic over the full test duration.
|
||||
|
||||
Use './test_ipv4_variable_time.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.
|
||||
@@ -34,6 +43,7 @@ class IPV4VariableTime(Realm):
|
||||
security=None,
|
||||
password=None,
|
||||
sta_list=[],
|
||||
create_sta=True,
|
||||
name_prefix=None,
|
||||
upstream=None,
|
||||
radio=None,
|
||||
@@ -57,6 +67,7 @@ class IPV4VariableTime(Realm):
|
||||
self.port = port
|
||||
self.ssid = ssid
|
||||
self.sta_list = sta_list
|
||||
self.create_sta = create_sta
|
||||
self.security = security
|
||||
self.password = password
|
||||
self.radio = radio
|
||||
@@ -99,6 +110,7 @@ class IPV4VariableTime(Realm):
|
||||
self.cx_profile.side_b_max_bps = side_b_max_rate
|
||||
|
||||
def start(self, print_pass=False, print_fail=False):
|
||||
if self.create_sta:
|
||||
self.station_profile.admin_up()
|
||||
# to-do- check here if upstream port got IP
|
||||
temp_stas = self.station_profile.station_names.copy()
|
||||
@@ -121,12 +133,13 @@ class IPV4VariableTime(Realm):
|
||||
|
||||
def cleanup(self):
|
||||
self.cx_profile.cleanup()
|
||||
if self.create_sta:
|
||||
self.station_profile.cleanup()
|
||||
LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=self.station_profile.station_names,
|
||||
debug=self.debug)
|
||||
|
||||
def build(self):
|
||||
|
||||
if self.create_sta:
|
||||
self.station_profile.use_security(self.security, self.ssid, self.password)
|
||||
self.station_profile.set_number_template(self.number_template)
|
||||
print("Creating stations")
|
||||
@@ -134,10 +147,12 @@ class IPV4VariableTime(Realm):
|
||||
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=self.traffic_type, side_a=self.station_profile.station_names, side_b=self.upstream,
|
||||
sleep_time=0)
|
||||
self._pass("PASS: Station build finished")
|
||||
|
||||
self.cx_profile.create(endp_type=self.traffic_type, side_a=self.sta_list,
|
||||
side_b=self.upstream,
|
||||
sleep_time=0)
|
||||
print("Some")
|
||||
|
||||
def main():
|
||||
parser = Realm.create_basic_argparse(
|
||||
@@ -185,6 +200,15 @@ python3 ./test_ipv4_variable_time.py
|
||||
--layer3_cols 'name','tx bytes','rx bytes','dropped' (column names from the GUI to print on report - please read below to know what to put here according to preferences)
|
||||
--port_mgr_cols 'ap','ip' (column names from the GUI to print on report - please read below to know what to put here according to preferences)
|
||||
--debug
|
||||
|
||||
python3 ./test_ipv4_variable_time.py
|
||||
--upstream_port eth1 (upstream POrt)
|
||||
--traffic_type lf_udp (traffic type, lf_udp | lf_tcp)
|
||||
--test_duration 5m (duration to run traffic 5m --> 5 Minutes)
|
||||
--create_sta False (False, means it will not create stations and use the sta_names specified below)
|
||||
--sta_names sta000,sta001,sta002 (used if --create_sta False, comma separated names of stations)
|
||||
|
||||
|
||||
===============================================================================
|
||||
** FURTHER INFORMATION **
|
||||
Using the layer3_cols flag:
|
||||
@@ -272,10 +296,17 @@ python3 ./test_ipv4_variable_time.py
|
||||
parser.add_argument('--influx_bucket', help='Password for your Influx database')
|
||||
parser.add_argument('--influx_org', help='Name of your Influx database')
|
||||
parser.add_argument('--influx_port', help='Port where your influx database is located', default=8086)
|
||||
parser.add_argument('--influx_tag', action='append', nargs=2, help='--influx_tag <key> <val> Can add more than one of these.')
|
||||
parser.add_argument('--influx_mgr', help='IP address of the server your Influx database is hosted if different from your LANforge Manager', default=None)
|
||||
|
||||
parser.add_argument('--influx_tag', action='append', nargs=2,
|
||||
help='--influx_tag <key> <val> Can add more than one of these.')
|
||||
parser.add_argument('--influx_mgr',
|
||||
help='IP address of the server your Influx database is hosted if different from your LANforge Manager',
|
||||
default=None)
|
||||
parser.add_argument('--create_sta', help='Used to force a connection to a particular AP', default=True)
|
||||
parser.add_argument('--sta_names', help='Used to force a connection to a particular AP', default="sta0000")
|
||||
args = parser.parse_args()
|
||||
create_sta = True
|
||||
if args.create_sta == "False":
|
||||
create_sta = False
|
||||
|
||||
num_sta = 2
|
||||
if (args.num_stations is not None) and (int(args.num_stations) > 0):
|
||||
@@ -329,12 +360,16 @@ python3 ./test_ipv4_variable_time.py
|
||||
else:
|
||||
compared_rept = args.compared_report
|
||||
|
||||
if create_sta:
|
||||
station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=num_sta - 1, padding_number_=10000,
|
||||
radio=args.radio)
|
||||
else:
|
||||
station_list = args.sta_names.split(",")
|
||||
ip_var_test = IPV4VariableTime(host=args.mgr,
|
||||
port=args.mgr_port,
|
||||
number_template="0000",
|
||||
sta_list=station_list,
|
||||
create_sta=create_sta,
|
||||
name_prefix="VT",
|
||||
upstream=args.upstream_port,
|
||||
ssid=args.ssid,
|
||||
@@ -350,8 +385,12 @@ python3 ./test_ipv4_variable_time.py
|
||||
traffic_type=args.traffic_type,
|
||||
_debug_on=args.debug)
|
||||
|
||||
if create_sta:
|
||||
ip_var_test.pre_cleanup()
|
||||
|
||||
ip_var_test.build()
|
||||
# exit()
|
||||
if create_sta:
|
||||
if not ip_var_test.passes():
|
||||
print(ip_var_test.get_fail_message())
|
||||
ip_var_test.exit_fail()
|
||||
@@ -433,15 +472,16 @@ python3 ./test_ipv4_variable_time.py
|
||||
debug=args.debug)
|
||||
|
||||
ip_var_test.stop()
|
||||
if create_sta:
|
||||
if not ip_var_test.passes():
|
||||
print(ip_var_test.get_fail_message())
|
||||
ip_var_test.exit_fail()
|
||||
LFUtils.wait_until_ports_admin_up(port_list=station_list)
|
||||
|
||||
ip_var_test.cleanup()
|
||||
|
||||
if ip_var_test.passes():
|
||||
ip_var_test.success()
|
||||
|
||||
ip_var_test.cleanup()
|
||||
print("IPv4 Variable Time Test Report Data: {}".format(report_f))
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user