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

Updated script to use new methods in realm, refactored code to better fit with build, start, stop, and cleanup methods

Browse Source
This commit is contained in:
Logan Lipke
2020-07-08 14:59:54 -07:00
parent 6068890c32
commit c507db642c
1 changed files with 56 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
py-scripts/test_ipv4_variable_time.py
View File

@@ -11,15 +11,18 @@ if 'py-json' not in sys.path:
import argparse
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
from LANforge import LFUtils
import realm
import time
import datetime
class IPV4VariableTime(LFCliBase):
def __init__(self, host, port, ssid, security, password, num_stations, side_a_min_rate=56, side_b_min_rate=56, side_a_max_rate=0,
side_b_max_rate=0, prefix="00000", test_duration="5m",
def __init__(self, host, port, ssid, security, password, num_stations, 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",
_debug_on=False,
_exit_on_error=False,
_exit_on_fail=False):
@@ -27,18 +30,24 @@ class IPV4VariableTime(LFCliBase):
self.host = host
self.port = port
self.ssid = ssid
self.sta_list = sta_list
self.security = security
self.password = password
self.num_stations = num_stations
self.prefix = prefix
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.prefix, mode=0, up=True,
security=self.security, number_template_=self.number_template,
mode=0,
up=True,
dhcp=True,
debug_=False)
self.cx_profile = realm.L3CXProfile(self.host, self.port, self.local_realm, side_a_min_bps=side_a_min_rate,
side_b_min_bps=side_b_min_rate, side_a_max_bps=side_a_max_rate,
side_b_max_bps=side_b_max_rate, debug_=False)
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
def __set_all_cx_state(self, state, sleep_time=5):
@@ -84,7 +93,9 @@ class IPV4VariableTime(LFCliBase):
else:
return False
def run_test(self, print_pass=False, print_fail=False):
def start(self, print_pass=False, print_fail=False):
self.station_profile.admin_up(1)
self.local_realm.wait_for_ip()
cur_time = datetime.datetime.now()
old_cx_rx_values = self.__get_rx_values()
end_time = self.local_realm.parse_time(self.test_duration) + cur_time
@@ -114,7 +125,12 @@ class IPV4VariableTime(LFCliBase):
if passes == expected_passes:
self._pass("PASS: All tests passed", print_pass)
def stop(self):
self.__set_all_cx_state("STOPPED")
for sta_name in self.sta_list:
data = LFUtils.portDownRequest(1, sta_name)
url = "json-cli/set_port"
self.json_post(url, data)
def cleanup(self):
print("Cleaning up stations")
@@ -129,7 +145,7 @@ class IPV4VariableTime(LFCliBase):
req_url = "cli-json/rm_vlan"
data = {
"shelf": 1,
"resource": 1,
"resource": self.resource,
"port": sta_name
}
# print(data)
@@ -159,34 +175,49 @@ class IPV4VariableTime(LFCliBase):
}
self.json_post(req_url, data)
def run(self):
sta_list = []
LFUtils.wait_until_ports_disappear(resource_id=self.resource, base_url=self.lfclient_url, port_list=sta_list,
debug=self.debug)
def build(self):
self.station_profile.use_wpa2(True, self.ssid, self.password)
self.station_profile.set_number_template(self.prefix)
self.station_profile.set_number_template(self.number_template)
print("Creating stations")
self.station_profile.create(resource=1, radio="wiphy0", num_stations=self.num_stations, debug=False)
for name in list(self.local_realm.station_list()):
if "sta" in list(name)[0]:
sta_list.append(list(name)[0])
print("sta_list", sta_list)
self.cx_profile.create(endp_type="lf_udp", side_a=sta_list, side_b="1.eth1", sleep_time=.5)
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)
temp_sta_list = []
for station in range(len(self.sta_list)):
temp_sta_list.append(str(self.resource) + "." + self.sta_list[station])
self.station_profile.create(resource=1, radio="wiphy0", sta_names_=self.sta_list, debug=False)
self.cx_profile.create(endp_type="lf_udp", side_a=temp_sta_list, side_b="1.eth1", sleep_time=.5)
self._pass("PASS: Station build finished")
def main():
lfjson_host = "localhost"
lfjson_port = 8080
ip_var_test = IPV4VariableTime(lfjson_host, lfjson_port, prefix="00", ssid="jedway-wpa2-x2048-4-4",
station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=4, padding_number_=10000)
ip_var_test = IPV4VariableTime(lfjson_host, lfjson_port, number_template="00", sta_list=station_list,
name_prefix="var_time",
ssid="jedway-wpa2-x2048-4-4",
password="jedway-wpa2-x2048-4-4",
resource=1,
security="open", num_stations=10, test_duration="5m",
side_a_min_rate=256, side_b_min_rate=256)
ip_var_test.cleanup()
ip_var_test.run()
time.sleep(5)
ip_var_test.run_test(print_pass=True, print_fail=True)
ip_var_test.build()
if not ip_var_test.passes():
print(ip_var_test.get_fail_message())
exit(1)
ip_var_test.start(False, False)
ip_var_test.stop()
if not ip_var_test.passes():
print(ip_var_test.get_fail_message())
exit(1)
time.sleep(30)
ip_var_test.cleanup()
if ip_var_test.passes():
print("Full test passed, all connections increased rx bytes")
if __name__ == "__main__":