scottk/wlan-lanforge-scripts
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-lanforge-scripts.git synced 2025-11-20 12:04:52 +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 16:14:43 -07:00
parent c507db642c
commit 814f148384
1 changed files with 51 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
py-scripts/test_ipv6_variable_time.py
View File

@@ -11,15 +11,18 @@ if 'py-json' not in sys.path:
import argparse import argparse
from LANforge.lfcli_base import LFCliBase from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import * from LANforge import LFUtils
import realm import realm
import time import time
import datetime import datetime
class IPV6VariableTime(LFCliBase): class IPV6VariableTime(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, def __init__(self, host, port, ssid, security, password, name_prefix, sta_list,
side_b_max_rate=0, prefix="00000", test_duration="5m", 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",
resource=1,
_debug_on=False, _debug_on=False,
_exit_on_error=False, _exit_on_error=False,
_exit_on_fail=False): _exit_on_fail=False):
@@ -27,18 +30,20 @@ class IPV6VariableTime(LFCliBase):
self.host = host self.host = host
self.port = port self.port = port
self.ssid = ssid self.ssid = ssid
self.sta_list = sta_list
self.name_prefix = name_prefix
self.security = security self.security = security
self.password = password self.password = password
self.num_stations = num_stations self.number_template = number_template
self.prefix = prefix self.resource = resource
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port) 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, 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,
dhcp=True, mode=0, up=True, dhcp=True, 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) debug_=False)
self.cx_profile = realm.L3CXProfile(self.host, self.port, self.local_realm, side_a_min_rate=side_a_min_rate,
side_b_min_rate=side_b_min_rate, side_a_max_rate=side_a_max_rate,
side_b_max_rate=side_b_max_rate, debug_=False)
self.test_duration = test_duration self.test_duration = test_duration
def __set_all_cx_state(self, state, sleep_time=5): def __set_all_cx_state(self, state, sleep_time=5):
@@ -84,7 +89,9 @@ class IPV6VariableTime(LFCliBase):
else: else:
return False 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(self.resource)
self.local_realm.wait_for_ip(ipv6=True)
cur_time = datetime.datetime.now() cur_time = datetime.datetime.now()
old_cx_rx_values = self.__get_rx_values() old_cx_rx_values = self.__get_rx_values()
end_time = self.local_realm.parse_time(self.test_duration) + cur_time end_time = self.local_realm.parse_time(self.test_duration) + cur_time
@@ -114,7 +121,12 @@ class IPV6VariableTime(LFCliBase):
if passes == expected_passes: if passes == expected_passes:
self._pass("PASS: All tests passed", print_pass) self._pass("PASS: All tests passed", print_pass)
def stop(self):
self.__set_all_cx_state("STOPPED") 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): def cleanup(self):
print("Cleaning up stations") print("Cleaning up stations")
@@ -129,7 +141,7 @@ class IPV6VariableTime(LFCliBase):
req_url = "cli-json/rm_vlan" req_url = "cli-json/rm_vlan"
data = { data = {
"shelf": 1, "shelf": 1,
"resource": 1, "resource": self.resource,
"port": sta_name "port": sta_name
} }
# print(data) # print(data)
@@ -160,33 +172,44 @@ class IPV6VariableTime(LFCliBase):
self.json_post(req_url, data) self.json_post(req_url, data)
def build(self): def build(self):
sta_list = []
self.station_profile.use_wpa2(True, self.ssid, self.password) 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") print("Creating stations")
self.station_profile.create(resource=1, radio="wiphy0", num_stations=self.num_stations, 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)
for name in list(self.local_realm.station_list()): self.station_profile.set_command_flag("set_port", "rpt_timer", 1)
if "sta" in list(name)[0]: temp_sta_list = []
sta_list.append(list(name)[0]) for station in range(len(self.sta_list)):
temp_sta_list.append(str(self.resource) + "." + self.sta_list[station])
print("sta_list", sta_list) self.station_profile.create(resource=1, radio="wiphy0", sta_names_=self.sta_list, debug=False)
self.cx_profile.create(endp_type="lf_udp6", side_a=sta_list, side_b="1.eth1", sleep_time=.5) self.cx_profile.create(endp_type="lf_udp6", side_a=temp_sta_list, side_b="1.eth1", sleep_time=.5)
self._pass("PASS: Station build finished")
def main(): def main():
lfjson_host = "localhost" lfjson_host = "localhost"
lfjson_port = 8080 lfjson_port = 8080
ip_var_test = IPV6VariableTime(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 = IPV6VariableTime(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", password="jedway-wpa2-x2048-4-4",
security="open", num_stations=10, test_duration="5m", resource=1, security="open", test_duration="5m",
side_a_min_rate=256, side_b_min_rate=256) side_a_min_rate=256, side_b_min_rate=256)
ip_var_test.cleanup() ip_var_test.cleanup()
ip_var_test.build() ip_var_test.build()
time.sleep(5) if not ip_var_test.passes():
ip_var_test.run_test(print_pass=True, print_fail=True) 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() ip_var_test.cleanup()
if ip_var_test.passes():
print("Full test passed, all connections increased rx bytes")
if __name__ == "__main__": if __name__ == "__main__":