From 621a665ab4a347d65a6833d26db7c2d5d027a324 Mon Sep 17 00:00:00 2001 From: Logan Lipke Date: Fri, 10 Jul 2020 10:11:41 -0700 Subject: [PATCH] Updated to use new methods in realm --- py-scripts/test_ipv4_connection.py | 35 ++++-------- py-scripts/test_ipv4_l4.py | 14 +---- py-scripts/test_ipv4_l4_urls_per_ten.py | 57 ++++--------------- py-scripts/test_ipv6_connection.py | 35 ++++-------- py-scripts/test_ipv6_variable_time.py | 75 +++++-------------------- 5 files changed, 44 insertions(+), 172 deletions(-) diff --git a/py-scripts/test_ipv4_connection.py b/py-scripts/test_ipv4_connection.py index 6eff996d..5c00fdd6 100755 --- a/py-scripts/test_ipv4_connection.py +++ b/py-scripts/test_ipv4_connection.py @@ -17,7 +17,7 @@ import pprint class IPv4Test(LFCliBase): - def __init__(self, host, port, ssid, security, password, 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) @@ -28,13 +28,14 @@ class IPv4Test(LFCliBase): self.password = password self.num_stations = num_stations self.sta_list = sta_list + self.resource = resource self.timeout = 120 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) + debug_=False, local_realm=self.local_realm) def build(self): # Build stations @@ -48,7 +49,6 @@ class IPv4Test(LFCliBase): self._pass("PASS: Station build finished") def start(self, sta_list, print_pass, print_fail): - # Bring stations up self.profile.admin_up(1) associated_map = {} ip_map = {} @@ -92,33 +92,18 @@ class IPv4Test(LFCliBase): # print(sta_name) self.json_post(url, data) - def cleanup(self, resource): - port_list = self.local_realm.station_list() - sta_list = [] - for item in list(port_list): - # print(list(item)) - if "sta" in list(item)[0]: - sta_list.append(self.local_realm.name_to_eid(list(item)[0])[2]) - - for sta_name in sta_list: - req_url = "cli-json/rm_vlan" - data = { - "shelf": 1, - "resource": 1, - "port": sta_name - } - self.json_post(req_url, data, self.debug) - time.sleep(.05) - LFUtils.wait_until_ports_disappear(resource_id=resource, base_url=self.lfclient_url, port_list=sta_list, debug=self.debug) - + def cleanup(self, sta_list): + self.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) def main(): lfjson_host = "localhost" lfjson_port = 8080 - station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=29, padding_number_=10000) + station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=1, padding_number_=10000) ip_test = IPv4Test(lfjson_host, lfjson_port, ssid="jedway-wpa2-x2048-4-4", password="jedway-wpa2-x2048-4-4", security="open", sta_list=station_list) - ip_test.cleanup(1) + ip_test.cleanup(station_list) ip_test.timeout = 60 ip_test.build() if not ip_test.passes(): @@ -130,7 +115,7 @@ def main(): print(ip_test.get_fail_message()) exit(1) time.sleep(30) - ip_test.cleanup(1) + ip_test.cleanup(station_list) if ip_test.passes(): print("Full test passed, all stations associated and got IP") diff --git a/py-scripts/test_ipv4_l4.py b/py-scripts/test_ipv4_l4.py index 1a7d6566..a171b779 100755 --- a/py-scripts/test_ipv4_l4.py +++ b/py-scripts/test_ipv4_l4.py @@ -48,18 +48,6 @@ class IPV4L4(LFCliBase): self.cx_profile.url = self.url self.cx_profile.requests_per_ten = self.requests_per_ten - def __set_all_cx_state(self, state, sleep_time=5): - print("Setting CX States to %s" % state) - for sta_name in self.sta_list: - req_url = "cli-json/set_cx_state" - data = { - "test_mgr": "default_tm", - "cx_name": "CX_" + sta_name + "_l4", - "cx_state": state - } - self.json_post(req_url, data) - time.sleep(sleep_time) - def __compare_vals(self, old_list, new_list): passes = 0 expected_passes = 0 @@ -138,7 +126,7 @@ class IPV4L4(LFCliBase): # new script; desired minimum urls for 10 min def stop(self): - self.__set_all_cx_state("STOPPED") + self.cx_profile.stop_cx() for sta_name in self.sta_list: data = LFUtils.portDownRequest(1, sta_name) url = "json-cli/set_port" diff --git a/py-scripts/test_ipv4_l4_urls_per_ten.py b/py-scripts/test_ipv4_l4_urls_per_ten.py index 85b26c1e..0525d5d1 100755 --- a/py-scripts/test_ipv4_l4_urls_per_ten.py +++ b/py-scripts/test_ipv4_l4_urls_per_ten.py @@ -42,7 +42,7 @@ class IPV4L4(LFCliBase): 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) + 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.cx_profile.url = self.url @@ -91,13 +91,14 @@ class IPV4L4(LFCliBase): self.cx_profile.create(ports=temp_sta_list, sleep_time=.5, debug_=self.debug, suppress_related_commands_=None) def start(self, print_pass=False, print_fail=False): + print("Starting test") cur_time = datetime.datetime.now() interval_time = cur_time + datetime.timedelta(minutes=10) passes = 0 expected_passes = 0 self.profile.admin_up(1) self.local_realm.wait_for_ip() - self.__set_all_cx_state("RUNNING") + self.cx_profile.start_cx() for test in range(self.num_tests): expected_passes += 1 while cur_time < interval_time: @@ -118,53 +119,15 @@ class IPV4L4(LFCliBase): self._pass("PASS: All tests passes", print_pass) def stop(self): - self.__set_all_cx_state("STOPPED") + self.cx_profile.stop_cx() 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): - layer4_list = self.json_get("layer4/list?fields=name") - print(layer4_list) - - if layer4_list is not None and 'endpoint' in layer4_list: - if layer4_list['endpoint'] is not None: - - for name in self.sta_list: - req_url = "cli-json/rm_cx" - data = { - "test_mgr": "default_tm", - "cx_name": "CX_" + name + "_l4" - } - self.json_post(req_url, data, True) - - time.sleep(5) - for endps in list(layer4_list['endpoint']): - for name, info in endps.items(): - print(name) - - req_url = "cli-json/rm_endp" - data = { - "endp_name": name - } - self.json_post(req_url, data, True) - - port_list = self.local_realm.station_list() - sta_list = [] - for item in list(port_list): - if "sta" in list(item)[0]: - sta_list.append(self.local_realm.name_to_eid(list(item)[0])[2]) - - for sta_name in sta_list: - req_url = "cli-json/rm_vlan" - data = { - "shelf": 1, - "resource": self.resource, - "port": sta_name - } - self.json_post(req_url, data, self.debug) - time.sleep(.05) + def cleanup(self, sta_list): + self.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) @@ -172,12 +135,12 @@ class IPV4L4(LFCliBase): def main(): lfjson_host = "localhost" lfjson_port = 8080 - station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=9, padding_number_=10000) + station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=1, padding_number_=10000) ip_test = IPV4L4(lfjson_host, lfjson_port, ssid="jedway-wpa2-x2048-4-4", password="jedway-wpa2-x2048-4-4", security="open", station_list=station_list, url="dl http://10.40.0.1 /dev/null", num_tests=1, target_requests_per_ten=600, requests_per_ten=600) - ip_test.cleanup() + ip_test.cleanup(station_list) ip_test.build() ip_test.start() ip_test.stop() @@ -185,7 +148,7 @@ def main(): print(ip_test.get_fail_message()) exit(1) time.sleep(30) - ip_test.cleanup() + ip_test.cleanup(station_list) if ip_test.passes(): print("Full test passed, all endpoints met or exceeded 90% of the target rate") diff --git a/py-scripts/test_ipv6_connection.py b/py-scripts/test_ipv6_connection.py index 682257eb..abc69f87 100755 --- a/py-scripts/test_ipv6_connection.py +++ b/py-scripts/test_ipv6_connection.py @@ -17,7 +17,7 @@ import pprint class IPv6Test(LFCliBase): - def __init__(self, host, port, ssid, security, password, 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) @@ -29,12 +29,13 @@ class IPv6Test(LFCliBase): self.num_stations = num_stations self.sta_list = sta_list self.timeout = 120 + self.resource = resource 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) + debug_=False, local_realm=self.local_realm) def build(self): # Build stations @@ -48,7 +49,6 @@ class IPv6Test(LFCliBase): self._pass("PASS: Station build finished") def start(self, sta_list, print_pass, print_fail): - # Bring stations up self.profile.admin_up(1) associated_map = {} ip_map = {} @@ -94,34 +94,19 @@ class IPv6Test(LFCliBase): # print(sta_name) self.json_post(url, data) - def cleanup(self, resource): - port_list = self.local_realm.station_list() - sta_list = [] - print("Cleaning up...") - for item in list(port_list): - # print(list(item)) - if "sta" in list(item)[0]: - sta_list.append(self.local_realm.name_to_eid(list(item)[0])[2]) - - for sta_name in sta_list: - req_url = "cli-json/rm_vlan" - data = { - "shelf": 1, - "resource": 1, - "port": sta_name - } - self.json_post(req_url, data, self.debug) - time.sleep(.05) - LFUtils.wait_until_ports_disappear(resource_id=resource, base_url=self.lfclient_url, port_list=sta_list, debug=self.debug) + def cleanup(self, sta_list): + self.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) def main(): lfjson_host = "localhost" lfjson_port = 8080 - station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=29, padding_number_=10000) + station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=1, padding_number_=10000) ipv6_test = IPv6Test(lfjson_host, lfjson_port, ssid="jedway-wpa2-x2048-4-4", password="jedway-wpa2-x2048-4-4", security="open", sta_list=station_list) - ipv6_test.cleanup(1) + ipv6_test.cleanup(station_list) ipv6_test.timeout = 60 ipv6_test.build() if not ipv6_test.passes(): @@ -133,7 +118,7 @@ def main(): print(ipv6_test.get_fail_message()) exit(1) time.sleep(30) - ipv6_test.cleanup(1) + ipv6_test.cleanup(station_list) if ipv6_test.passes(): print("Full test passed, all stations associated and got IP") diff --git a/py-scripts/test_ipv6_variable_time.py b/py-scripts/test_ipv6_variable_time.py index cae13a17..fdfa642c 100755 --- a/py-scripts/test_ipv6_variable_time.py +++ b/py-scripts/test_ipv6_variable_time.py @@ -39,27 +39,14 @@ class IPV6VariableTime(LFCliBase): 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) + 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 - def __set_all_cx_state(self, state, sleep_time=5): - print("Setting CX States to %s" % state) - cx_list = list(self.local_realm.cx_list()) - for cx_name in cx_list: - if cx_name != 'handler' or cx_name != 'uri': - req_url = "cli-json/set_cx_state" - data = { - "test_mgr": "default_tm", - "cx_name": cx_name, - "cx_state": state - } - self.json_post(req_url, data) - time.sleep(sleep_time) - def __get_rx_values(self): cx_list = self.json_get("endp?fields=name,rx+bytes", debug_=True) #print("==============\n", cx_list, "\n==============") @@ -90,12 +77,13 @@ class IPV6VariableTime(LFCliBase): return False def start(self, print_pass=False, print_fail=False): + print("Starting test") self.station_profile.admin_up(self.resource) self.local_realm.wait_for_ip(ipv6=True) 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 - self.__set_all_cx_state("RUNNING") + self.cx_profile.start_cx() passes = 0 expected_passes = 0 while cur_time < end_time: @@ -122,54 +110,17 @@ class IPV6VariableTime(LFCliBase): self._pass("PASS: All tests passed", print_pass) def stop(self): - self.__set_all_cx_state("STOPPED") + self.cx_profile.stop_cx() 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") - port_list = self.local_realm.station_list() - sta_list = [] - for item in list(port_list): - # print(list(item)) - if "sta" in list(item)[0]: - sta_list.append(self.local_realm.name_to_eid(list(item)[0])[2]) - - for sta_name in sta_list: - req_url = "cli-json/rm_vlan" - data = { - "shelf": 1, - "resource": self.resource, - "port": sta_name - } - # print(data) - self.json_post(req_url, data) - - cx_list = list(self.local_realm.cx_list()) - if cx_list is not None: - print("Cleaning up cxs") - for cx_name in cx_list: - if cx_name != 'handler' or cx_name != 'uri': - req_url = "cli-json/rm_cx" - data = { - "test_mgr": "default_tm", - "cx_name": cx_name - } - self.json_post(req_url, data) - - print("Cleaning up endps") - endp_list = self.json_get("/endp") - if endp_list is not None: - endp_list = list(endp_list['endpoint']) - for endp_name in range(len(endp_list)): - name = list(endp_list[endp_name])[0] - req_url = "cli-json/rm_endp" - data = { - "endp_name": name - } - self.json_post(req_url, data) + def cleanup(self, 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) def build(self): self.station_profile.use_wpa2(True, self.ssid, self.password) @@ -189,14 +140,14 @@ class IPV6VariableTime(LFCliBase): def main(): lfjson_host = "localhost" lfjson_port = 8080 - station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=4, padding_number_=10000) + station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=1, 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", resource=1, security="open", test_duration="5m", side_a_min_rate=256, side_b_min_rate=256) - ip_var_test.cleanup() + ip_var_test.cleanup(station_list) ip_var_test.build() if not ip_var_test.passes(): print(ip_var_test.get_fail_message()) @@ -207,7 +158,7 @@ def main(): print(ip_var_test.get_fail_message()) exit(1) time.sleep(30) - ip_var_test.cleanup() + ip_var_test.cleanup(station_list) if ip_var_test.passes(): print("Full test passed, all connections increased rx bytes")