mirror of
https://github.com/Telecominfraproject/wlan-lanforge-scripts.git
synced 2025-11-01 03:07:56 +00:00
Switched to using Realm's methods for creating new profiles, made various changes to keep similarity between tests
This commit is contained in:
@@ -17,7 +17,7 @@ import pprint
|
||||
|
||||
|
||||
class IPv4Test(LFCliBase):
|
||||
def __init__(self, host, port, ssid, security, password, resource=1, sta_list=None, num_stations=0, prefix="00000", _debug_on=False,
|
||||
def __init__(self, host, port, ssid, security, password, resource=1, sta_list=None, number_template="00000", _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)
|
||||
@@ -26,30 +26,34 @@ class IPv4Test(LFCliBase):
|
||||
self.ssid = ssid
|
||||
self.security = security
|
||||
self.password = password
|
||||
self.num_stations = num_stations
|
||||
self.sta_list = sta_list
|
||||
self.resource = resource
|
||||
self.timeout = 120
|
||||
self.prefix = prefix
|
||||
self.number_template = number_template
|
||||
self.debug = _debug_on
|
||||
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
|
||||
self.profile = realm.StationProfile(self.lfclient_url, ssid=self.ssid, ssid_pass=self.password,
|
||||
security=self.security, number_template_=self.prefix, mode=0, up=False, dhcp=True,
|
||||
debug_=False, local_realm=self.local_realm)
|
||||
self.station_profile = self.local_realm.new_station_profile()
|
||||
|
||||
self.station_profile.lfclient_url = self.lfclient_url
|
||||
self.station_profile.ssid = self.ssid
|
||||
self.station_profile.ssid_pass = self.password,
|
||||
self.station_profile.security = self.security
|
||||
self.station_profile.number_template_ = self.number_template
|
||||
self.station_profile.mode = 0
|
||||
|
||||
def build(self):
|
||||
# Build stations
|
||||
self.profile.use_wpa2(True, self.ssid, self.password)
|
||||
self.profile.set_number_template(self.prefix)
|
||||
self.station_profile.use_wpa2(True, self.ssid, self.password)
|
||||
self.station_profile.set_number_template(self.number_template)
|
||||
print("Creating stations")
|
||||
self.profile.set_command_flag("add_sta", "create_admin_down", 1)
|
||||
self.profile.set_command_param("set_port", "report_timer", 1500)
|
||||
self.profile.set_command_flag("set_port", "rpt_timer", 1)
|
||||
self.profile.create(resource=1, radio="wiphy0", sta_names_=self.sta_list, debug=False)
|
||||
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(resource=1, radio="wiphy0", sta_names_=self.sta_list, debug=False)
|
||||
self._pass("PASS: Station build finished")
|
||||
|
||||
def start(self, sta_list, print_pass, print_fail):
|
||||
self.profile.admin_up(1)
|
||||
self.station_profile.admin_up(1)
|
||||
associated_map = {}
|
||||
ip_map = {}
|
||||
print("Starting test...")
|
||||
@@ -93,7 +97,7 @@ class IPv4Test(LFCliBase):
|
||||
self.json_post(url, data)
|
||||
|
||||
def cleanup(self, sta_list):
|
||||
self.profile.cleanup(self.resource, sta_list)
|
||||
self.station_profile.cleanup(self.resource, sta_list)
|
||||
LFUtils.wait_until_ports_disappear(resource_id=self.resource, base_url=self.lfclient_url, port_list=sta_list,
|
||||
debug=self.debug)
|
||||
|
||||
|
||||
@@ -19,7 +19,7 @@ import datetime
|
||||
|
||||
|
||||
class IPV4L4(LFCliBase):
|
||||
def __init__(self, host, port, ssid, security, password, url, requests_per_ten, station_list, prefix="00000",
|
||||
def __init__(self, host, port, ssid, security, password, url, requests_per_ten, station_list, number_template="00000",
|
||||
resource=1,
|
||||
test_duration="5m",
|
||||
_debug_on=False,
|
||||
@@ -33,18 +33,22 @@ class IPV4L4(LFCliBase):
|
||||
self.password = password
|
||||
self.url = url
|
||||
self.requests_per_ten = requests_per_ten
|
||||
self.prefix = prefix
|
||||
self.number_template = number_template
|
||||
self.sta_list = station_list
|
||||
self.resource = resource
|
||||
self.test_duration = test_duration
|
||||
|
||||
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
|
||||
self.profile = realm.StationProfile(self.lfclient_url, ssid=self.ssid, ssid_pass=self.password,
|
||||
security=self.security, number_template_=self.prefix, mode=0, up=False,
|
||||
dhcp=True,
|
||||
debug_=False, local_realm=self.local_realm)
|
||||
self.cx_profile = realm.L4CXProfile(lfclient_host=self.host, lfclient_port=self.port,
|
||||
local_realm=self.local_realm, debug_=False)
|
||||
self.station_profile = self.local_realm.new_station_profile()
|
||||
self.cx_profile = self.local_realm.new_l4_cx_profile()
|
||||
|
||||
self.station_profile.lfclient_url = self.lfclient_url
|
||||
self.station_profile.ssid = self.ssid
|
||||
self.station_profile.ssid_pass = self.password,
|
||||
self.station_profile.security = self.security
|
||||
self.station_profile.number_template_ = self.number_template
|
||||
self.station_profile.mode = 0
|
||||
|
||||
self.cx_profile.url = self.url
|
||||
self.cx_profile.requests_per_ten = self.requests_per_ten
|
||||
|
||||
@@ -79,13 +83,13 @@ class IPV4L4(LFCliBase):
|
||||
|
||||
def build(self):
|
||||
# Build stations
|
||||
self.profile.use_wpa2(True, self.ssid, self.password)
|
||||
self.profile.set_number_template(self.prefix)
|
||||
self.station_profile.use_wpa2(True, self.ssid, self.password)
|
||||
self.station_profile.set_number_template(self.number_template)
|
||||
print("Creating stations")
|
||||
self.profile.set_command_flag("add_sta", "create_admin_down", 1)
|
||||
self.profile.set_command_param("set_port", "report_timer", 1500)
|
||||
self.profile.set_command_flag("set_port", "rpt_timer", 1)
|
||||
self.profile.create(resource=1, radio="wiphy0", sta_names_=self.sta_list, debug=self.debug)
|
||||
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(resource=1, radio="wiphy0", sta_names_=self.sta_list, debug=self.debug)
|
||||
self._pass("PASS: Station build finished")
|
||||
temp_sta_list = []
|
||||
for station in range(len(self.sta_list)):
|
||||
@@ -97,7 +101,7 @@ class IPV4L4(LFCliBase):
|
||||
cur_time = datetime.datetime.now()
|
||||
old_rx_values = self.__get_values()
|
||||
end_time = self.local_realm.parse_time(self.test_duration) + cur_time
|
||||
self.profile.admin_up(1)
|
||||
self.station_profile.admin_up(1)
|
||||
self.local_realm.wait_for_ip()
|
||||
self.cx_profile.start_cx()
|
||||
passes = 0
|
||||
@@ -133,7 +137,7 @@ class IPV4L4(LFCliBase):
|
||||
self.json_post(url, data)
|
||||
|
||||
def cleanup(self, sta_list):
|
||||
self.profile.cleanup(self.resource, sta_list)
|
||||
self.station_profile.cleanup(self.resource, sta_list)
|
||||
self.cx_profile.cleanup()
|
||||
LFUtils.wait_until_ports_disappear(resource_id=self.resource, base_url=self.lfclient_url, port_list=sta_list,
|
||||
debug=self.debug)
|
||||
|
||||
@@ -20,7 +20,7 @@ import datetime
|
||||
|
||||
class IPV4L4(LFCliBase):
|
||||
def __init__(self, host, port, ssid, security, password, url, requests_per_ten, station_list,
|
||||
target_requests_per_ten=600, prefix="00000", resource=1, num_tests=1,
|
||||
target_requests_per_ten=600, number_template="00000", resource=1, num_tests=1,
|
||||
_debug_on=False,
|
||||
_exit_on_error=False,
|
||||
_exit_on_fail=False):
|
||||
@@ -32,19 +32,22 @@ class IPV4L4(LFCliBase):
|
||||
self.password = password
|
||||
self.url = url
|
||||
self.requests_per_ten = requests_per_ten
|
||||
self.prefix = prefix
|
||||
self.number_template = number_template
|
||||
self.sta_list = station_list
|
||||
self.resource = resource
|
||||
self.num_tests = num_tests
|
||||
self.target_requests_per_ten = target_requests_per_ten
|
||||
|
||||
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
|
||||
self.profile = realm.StationProfile(self.lfclient_url, ssid=self.ssid, ssid_pass=self.password,
|
||||
security=self.security, number_template_=self.prefix, mode=0, up=False,
|
||||
dhcp=True,
|
||||
debug_=False, local_realm=self.local_realm)
|
||||
self.cx_profile = realm.L4CXProfile(lfclient_host=self.host, lfclient_port=self.port,
|
||||
local_realm=self.local_realm, debug_=False)
|
||||
self.station_profile = self.local_realm.new_station_profile()
|
||||
self.cx_profile = self.local_realm.new_l4_cx_profile()
|
||||
|
||||
self.station_profile.lfclient_url = self.lfclient_url
|
||||
self.station_profile.ssid = self.ssid
|
||||
self.station_profile.ssid_pass = self.password,
|
||||
self.station_profile.security = self.security
|
||||
self.station_profile.number_template_ = self.number_template
|
||||
self.station_profile.mode = 0
|
||||
self.cx_profile.url = self.url
|
||||
self.cx_profile.requests_per_ten = self.requests_per_ten
|
||||
|
||||
@@ -76,13 +79,13 @@ class IPV4L4(LFCliBase):
|
||||
|
||||
def build(self):
|
||||
# Build stations
|
||||
self.profile.use_wpa2(True, self.ssid, self.password)
|
||||
self.profile.set_number_template(self.prefix)
|
||||
self.station_profile.use_wpa2(True, self.ssid, self.password)
|
||||
self.station_profile.set_number_template(self.number_template)
|
||||
print("Creating stations")
|
||||
self.profile.set_command_flag("add_sta", "create_admin_down", 1)
|
||||
self.profile.set_command_param("set_port", "report_timer", 1500)
|
||||
self.profile.set_command_flag("set_port", "rpt_timer", 1)
|
||||
self.profile.create(resource=1, radio="wiphy0", sta_names_=self.sta_list, debug=self.debug)
|
||||
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(resource=1, radio="wiphy0", sta_names_=self.sta_list, debug=self.debug)
|
||||
self._pass("PASS: Station build finished")
|
||||
temp_sta_list = []
|
||||
for station in range(len(self.sta_list)):
|
||||
@@ -96,7 +99,7 @@ class IPV4L4(LFCliBase):
|
||||
interval_time = cur_time + datetime.timedelta(minutes=10)
|
||||
passes = 0
|
||||
expected_passes = 0
|
||||
self.profile.admin_up(1)
|
||||
self.station_profile.admin_up(1)
|
||||
self.local_realm.wait_for_ip()
|
||||
self.cx_profile.start_cx()
|
||||
for test in range(self.num_tests):
|
||||
@@ -126,7 +129,7 @@ class IPV4L4(LFCliBase):
|
||||
self.json_post(url, data)
|
||||
|
||||
def cleanup(self, sta_list):
|
||||
self.profile.cleanup(self.resource, sta_list)
|
||||
self.station_profile.cleanup(self.resource, sta_list)
|
||||
self.cx_profile.cleanup()
|
||||
LFUtils.wait_until_ports_disappear(resource_id=self.resource, base_url=self.lfclient_url, port_list=sta_list,
|
||||
debug=self.debug)
|
||||
|
||||
@@ -18,7 +18,7 @@ import datetime
|
||||
|
||||
|
||||
class IPV4VariableTime(LFCliBase):
|
||||
def __init__(self, host, port, ssid, security, password, sta_list, name_prefix,resource=1,
|
||||
def __init__(self, host, port, ssid, security, password, sta_list, name_prefix, resource=1,
|
||||
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",
|
||||
@@ -35,16 +35,26 @@ class IPV4VariableTime(LFCliBase):
|
||||
self.number_template = number_template
|
||||
self.resource = resource
|
||||
self.name_prefix = name_prefix
|
||||
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
|
||||
self.station_profile = realm.StationProfile(self.lfclient_url, ssid=self.ssid, ssid_pass=self.password,
|
||||
security=self.security, number_template_=self.number_template,
|
||||
mode=0, up=True, dhcp=True, debug_=False,
|
||||
local_realm=self.local_realm)
|
||||
self.cx_profile = realm.L3CXProfile(self.host, self.port, self.local_realm, name_prefix_=self.name_prefix,
|
||||
side_a_min_bps=side_a_min_rate, side_a_max_bps=side_a_max_rate,
|
||||
side_b_min_bps=side_b_min_rate, side_b_max_bps=side_b_max_rate,
|
||||
debug_=False)
|
||||
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,
|
||||
self.station_profile.security = self.security
|
||||
self.station_profile.number_template_ = self.number_template
|
||||
self.station_profile.mode = 0
|
||||
|
||||
self.cx_profile.host = self.host
|
||||
self.cx_profile.port = self.port
|
||||
self.cx_profile.name_prefix = self.name_prefix
|
||||
self.cx_profile.side_a_min_bps = side_a_min_rate
|
||||
self.cx_profile.side_a_max_bps = side_a_max_rate
|
||||
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_=True)
|
||||
|
||||
@@ -17,7 +17,8 @@ import pprint
|
||||
|
||||
|
||||
class IPv6Test(LFCliBase):
|
||||
def __init__(self, host, port, ssid, security, password, resource=1, sta_list=None, num_stations=0, prefix="00000", _debug_on=False,
|
||||
def __init__(self, host, port, ssid, security, password, resource=1, sta_list=None, num_stations=0, prefix="00000",
|
||||
_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)
|
||||
@@ -33,12 +34,16 @@ class IPv6Test(LFCliBase):
|
||||
self.prefix = prefix
|
||||
self.debug = _debug_on
|
||||
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
|
||||
self.profile = realm.StationProfile(self.lfclient_url, ssid=self.ssid, ssid_pass=self.password,
|
||||
security=self.security, number_template_=self.prefix, mode=0, up=False, dhcp=True,
|
||||
debug_=False, local_realm=self.local_realm)
|
||||
self.station_profile = self.local_realm.new_station_profile()
|
||||
|
||||
self.station_profile.lfclient_url = self.lfclient_url
|
||||
self.station_profile.ssid = self.ssid
|
||||
self.station_profile.ssid_pass = self.password,
|
||||
self.station_profile.security = self.security
|
||||
self.station_profile.number_template_ = self.number_template
|
||||
self.station_profile.mode = 0
|
||||
|
||||
def build(self):
|
||||
# Build stations
|
||||
self.profile.use_wpa2(True, self.ssid, self.password)
|
||||
self.profile.set_number_template(self.prefix)
|
||||
print("Creating stations")
|
||||
|
||||
@@ -36,16 +36,26 @@ class IPV6VariableTime(LFCliBase):
|
||||
self.password = password
|
||||
self.number_template = number_template
|
||||
self.resource = resource
|
||||
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
|
||||
self.station_profile = realm.StationProfile(self.lfclient_url, ssid=self.ssid, ssid_pass=self.password,
|
||||
security=self.security, number_template_=self.number_template,
|
||||
mode=0, up=True, dhcp=True, debug_=False,
|
||||
local_realm=self.local_realm)
|
||||
self.cx_profile = realm.L3CXProfile(self.host, self.port, self.local_realm, name_prefix_=self.name_prefix,
|
||||
side_a_min_bps=side_a_min_rate, side_a_max_bps=side_a_max_rate,
|
||||
side_b_min_bps=side_b_min_rate, side_b_max_bps=side_b_max_rate,
|
||||
debug_=False)
|
||||
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,
|
||||
self.station_profile.security = self.security
|
||||
self.station_profile.number_template_ = self.number_template
|
||||
self.station_profile.mode = 0
|
||||
|
||||
self.cx_profile.host = self.host
|
||||
self.cx_profile.port = self.port
|
||||
self.cx_profile.name_prefix = self.name_prefix
|
||||
self.cx_profile.side_a_min_bps = side_a_min_rate
|
||||
self.cx_profile.side_a_max_bps = side_a_max_rate
|
||||
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_=True)
|
||||
|
||||
Reference in New Issue
Block a user