From 9c94287fb5e82ef8761c2b08e0d460af163c8e14 Mon Sep 17 00:00:00 2001 From: Matthew Stidham Date: Wed, 30 Dec 2020 17:00:52 -0800 Subject: [PATCH] fix --- py-json/realm.py | 1 - py-scripts/test_ipv4_connection.py | 209 ++++++++++++++++++++++ py-scripts/test_ipv4_ps.py | 4 +- py-scripts/test_l3_scenario_throughput.py | 75 ++++---- 4 files changed, 249 insertions(+), 40 deletions(-) create mode 100644 py-scripts/test_ipv4_connection.py diff --git a/py-json/realm.py b/py-json/realm.py index f3c650d4..d6e2406e 100755 --- a/py-json/realm.py +++ b/py-json/realm.py @@ -1208,7 +1208,6 @@ class L3CXProfile(LFCliBase): pass - def refresh_cx(self): for cx_name in self.created_cx.keys(): self.json_post("/cli-json/show_cxe", { diff --git a/py-scripts/test_ipv4_connection.py b/py-scripts/test_ipv4_connection.py new file mode 100644 index 00000000..89a5e5c6 --- /dev/null +++ b/py-scripts/test_ipv4_connection.py @@ -0,0 +1,209 @@ +#!/usr/bin/env python3 + + +import sys +import os +import argparse + +if sys.version_info[0] != 3: + print("This script requires Python 3") + exit(1) + +if 'py-json' not in sys.path: + sys.path.append(os.path.join(os.path.abspath('..'), 'py-json')) +import LANforge +from LANforge.lfcli_base import LFCliBase +from LANforge import LFUtils +import realm +import time +import pprint + + +class IPv4Test(LFCliBase): + def __init__(self, + _ssid=None, + _security=None, + _password=None, + _host=None, + _port=None, + _sta_list=None, + _number_template="00000", + _radio="wiphy0", + _proxy_str=None, + _debug_on=False, + _exit_on_error=False, + _exit_on_fail=False): + super().__init__(_host, + _port, + _proxy_str=_proxy_str, + _local_realm=realm.Realm(lfclient_host=_host, + lfclient_port=_port, + halt_on_error_=_exit_on_error, + _exit_on_error=_exit_on_error, + _exit_on_fail=_exit_on_fail, + _proxy_str=_proxy_str, + debug_=_debug_on), + _debug=_debug_on, + _halt_on_error=_exit_on_error, + _exit_on_fail=_exit_on_fail) + self.host = _host + self.port = _port + self.ssid = _ssid + self.security = _security + self.password = _password + self.sta_list = _sta_list + self.radio = _radio + self.timeout = 120 + self.number_template = _number_template + self.debug = _debug_on + 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 + if self.debug: + print("----- Station List ----- ----- ----- ----- ----- ----- \n") + pprint.pprint(self.sta_list) + print("---- ~Station List ----- ----- ----- ----- ----- ----- \n") + + + def build(self): + # Build stations + self.station_profile.use_security(self.security, self.ssid, self.password) + self.station_profile.set_number_template(self.number_template) + + print("Creating stations") + 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(radio=self.radio, sta_names_=self.sta_list, debug=self.debug) + self._pass("PASS: Station build finished") + + def start(self, sta_list, print_pass, print_fail): + self.station_profile.admin_up() + associated_map = {} + ip_map = {} + print("Starting test...") + for sec in range(self.timeout): + for sta_name in sta_list: + eidn = self.local_realm.name_to_eid(sta_name) + url = "/port/1/%s/%s" % (eidn[1], eidn[2]) + sta_status = self.json_get(url + "?fields=port,alias,ip,ap", debug_=self.debug) + # print(sta_status) + if (sta_status is None) or (sta_status['interface'] is None) or (sta_status['interface']['ap'] is None): + continue + if (len(sta_status['interface']['ap']) == 17) and (sta_status['interface']['ap'][-3] == ':'): + if self.debug: + print("Associated", sta_name, sta_status['interface']['ap'], sta_status['interface']['ip']) + associated_map[sta_name] = 1 + if sta_status['interface']['ip'] != '0.0.0.0': + if self.debug: + print("IP", sta_name, sta_status['interface']['ap'], sta_status['interface']['ip']) + ip_map[sta_name] = 1 + if (len(sta_list) == len(ip_map)) and (len(sta_list) == len(associated_map)): + break + else: + time.sleep(1) + + if self.debug: + print("sta_list", len(sta_list), sta_list) + print("ip_map", len(ip_map), ip_map) + print("associated_map", len(associated_map), associated_map) + if (len(sta_list) == len(ip_map)) and (len(sta_list) == len(associated_map)): + self._pass("PASS: All stations associated with IP", print_pass) + else: + self._fail("FAIL: Not all stations able to associate/get IP", print_fail) + print("sta_list", sta_list) + print("ip_map", ip_map) + print("associated_map", associated_map) + + return self.passes() + + def stop(self): + # Bring stations down + self.station_profile.admin_down() + + def cleanup(self, sta_list): + self.station_profile.cleanup(sta_list, debug_=self.debug) + LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, + port_list=sta_list, + debug=self.debug) + time.sleep(1) + +def main(): + parser = LFCliBase.create_basic_argparse( + prog='test_ipv4_connection.py', + formatter_class=argparse.RawTextHelpFormatter, + epilog='''\ + Create stations that attempt to authenticate, associate, and receive IP addresses on the + chosen SSID + ''', + + description='''\ + test_ipv4_connection.py +-------------------- +Command example: +./test_ipv4_connection.py + --upstream_port eth1 + --radio wiphy0 + --num_stations 3 + --security open + --ssid netgear + --passwd BLANK + --debug + ''') + required = parser.add_argument_group('required arguments') + #required.add_argument('--security', help='WiFi Security protocol: < open | wep | wpa | wpa2 | wpa3 >', required=True) + + args = parser.parse_args() + #if args.debug: + # pprint.pprint(args) + # time.sleep(5) + if (args.radio is None): + raise ValueError("--radio required") + + num_sta = 2 + if (args.num_stations is not None) and (int(args.num_stations) > 0): + num_stations_converted = int(args.num_stations) + num_sta = num_stations_converted + + station_list = LFUtils.port_name_series(prefix="sta", + start_id=0, + end_id=num_sta-1, + padding_number=10000, + radio=args.radio) + if args.debug: + print("args.proxy: %s" % args.proxy) + ip_test = IPv4Test(_host=args.mgr, + _port=args.mgr_port, + _ssid=args.ssid, + _password=args.passwd, + _security=args.security, + _sta_list=station_list, + _radio=args.radio, + _proxy_str=args.proxy, + _debug_on=args.debug) + + ip_test.cleanup(station_list) + ip_test.build() + if not ip_test.passes(): + print(ip_test.get_fail_message()) + ip_test.add_event(name="test_ipv4_connection.py", message=ip_test.get_fail_message()) + ip_test.exit_fail() + ip_test.start(station_list, False, False) + ip_test.stop() + if not ip_test.passes(): + print(ip_test.get_fail_message()) + ip_test.add_event(name="test_ipv4_connection.py", message=ip_test.get_fail_message()) + ip_test.exit_fail() + time.sleep(30) + ip_test.cleanup(station_list) + if ip_test.passes(): + ip_test.add_event(name="test_ipv4_connection.py", message="Full test passed, all stations associated and got IP") + ip_test.exit_success() + + +if __name__ == "__main__": + main() diff --git a/py-scripts/test_ipv4_ps.py b/py-scripts/test_ipv4_ps.py index 0659be61..a903bd94 100755 --- a/py-scripts/test_ipv4_ps.py +++ b/py-scripts/test_ipv4_ps.py @@ -184,8 +184,8 @@ Generic command layout: --upstream_port eth1 --radio wiphy3 --num_stations 4 --ssid jedway-wpa2-x2048-4-1 --passwd jedway-wpa2-x2048-4-1 +--ssid jedway-wpa2-x2048-4-1 +--passwd jedway-wpa2-x2048-4-1 --security {wpa2|open|wpa|wpa3} --a_min 250000 --b_min 260000 diff --git a/py-scripts/test_l3_scenario_throughput.py b/py-scripts/test_l3_scenario_throughput.py index c3770bd9..bdc9ab19 100755 --- a/py-scripts/test_l3_scenario_throughput.py +++ b/py-scripts/test_l3_scenario_throughput.py @@ -9,19 +9,19 @@ In this example, Another Lanforge is used as DUT It also have a function : GenerateReport that generates the report in xlsx format as well as it plots the Graph of throughput over time with temperature It also have Plot function that generates a html page that contains the plot - - - Prerequisite + + + Prerequisite Start the Lanforge Manager both Sides - + Installation pip install paramiko pip install bokeh pip install XlsxWriter Example - ./test_l3_scenario_throughput.py --manager 192.168.200.18 --scenario Test_Scenario --report_name test_Report --duration 5 --test_detail "Single Station Test" - + ./test_l3_scenario_throughput.py --manager 192.168.200.18 --scenario Test_Scenario --report_name test_Report --duration 5 --test_detail "Single Station Test" + This Script is intended to automate the testing of DUT That has stations as well as AP. To automate the simultaenous testing and check the DUT Temperature ''' @@ -54,6 +54,7 @@ from bokeh.models import LinearAxis, Range1d from bokeh.models import HoverTool, Range1d from bokeh.layouts import row from datetime import datetime +import socket # Specifically for Measuring CPU Core Temperatures class Login_DUT: @@ -78,14 +79,14 @@ class Login_DUT: print(out_lines[len(out_lines)-3], out_lines[len(out_lines)-2]) self.data_core1.append(out_lines[len(out_lines)-3]) self.data_core2.append(out_lines[len(out_lines)-2]) - + def Connect(self): self.CLIENT.load_system_host_keys() self.CLIENT.set_missing_host_key_policy(pm.AutoAddPolicy()) try: self.CLIENT.connect(self.host, username=self.USERNAME, password=self.PASSWORD,timeout=10) - return None + return None except exception as error: self.CLIENT = 0; return None @@ -105,7 +106,7 @@ class LoadScenario(LFCliBase): -# Generates XLSX Report +# Generates XLSX Report def GenerateReport(scenario, detail, throughput_sta, throughput_vap, absolute_time, relative_time, core1_temp, core2_temp, duration, name): workbook = xlsxwriter.Workbook(name) worksheet = workbook.add_worksheet() @@ -118,8 +119,8 @@ def GenerateReport(scenario, detail, throughput_sta, throughput_vap, absolute_ti worksheet.write('F2', 'CORE 1 TEMP (Degree Celsius)') core1=[] core2=[] - - + + j=3 for i in absolute_time: worksheet.write('A'+str(j),i) @@ -129,7 +130,7 @@ def GenerateReport(scenario, detail, throughput_sta, throughput_vap, absolute_ti for i in relative_time: worksheet.write('B'+str(j),i) j=j+1 - + sta_throu=[] vap_throu=[] j=3 @@ -170,16 +171,16 @@ def plot(throughput_sta, throughput_vap, core1_temp, core2_temp, Time): s1.title.text = "WIFI Throughput vs Temperature Plot" s1.xaxis.axis_label = "Time " s1.yaxis.axis_label = "Throughput in Mbps" - + s1.line( Time, throughput_sta, color='black', legend_label ="Throughput Over Station Connections ") #s1.circle(Time, throughput_sta, color='red') s1.line( Time, throughput_vap, color='blue', legend_label ="Throughput Over VAP ") #s1.circle(Time, throughput_vap, color='blue') - + s1.extra_y_ranges = {"Temperature": Range1d(start=0, end=150)} s1.add_layout(LinearAxis(y_range_name="Temperature", axis_label="Temperature in Degree Celsius"), 'right') - + s1.line(Time, core1_temp, y_range_name='Temperature', color='red', legend_label ="CPU CORE 0 TEMPERATURE ") #s1.circle(Time, core1_temp, y_range_name='Temperature', color='red') @@ -188,7 +189,7 @@ def plot(throughput_sta, throughput_vap, core1_temp, core2_temp, Time): show(s1) - + # Creates the Instance for LFCliBase class VAP_Measure(LFCliBase): def __init__(self, lfclient_host, lfclient_port): @@ -199,7 +200,7 @@ class VAP_Measure(LFCliBase): -# Added Standard Function to Fetch L3 CX and VAP Directly +# Added Standard Function to Fetch L3 CX and VAP Directly class FindPorts(LFCliBase): def __init__(self, host, port, security_debug_on=False, _exit_on_error=False,_exit_on_fail=False): super().__init__(host, port, _debug=security_debug_on, _halt_on_error=_exit_on_error, _exit_on_fail=_exit_on_fail) @@ -217,20 +218,21 @@ class FindPorts(LFCliBase): def FindVAP(self): return self.local_realm.vap_list() -# Utility to Find the Traffic Running on Existing CX and VAP +# Utility to Find the Traffic Running on Existing CX and VAP def PortUtility(host, port, duration, report_name, scenario, detail): lf_utils = FindPorts(host, port) - + # cx data will be having all parameters of L3 Connections available in the Realm. It is needed to get the names of all L3 CX, which is stored in cx_names. It is required so as we can extract the real time data running on that CX cx_data = lf_utils.FindExistingCX() #print(cx_data) - + # vap_list will have the List of all the vap ports available, This is required to get the VAP names in order to fetch the throughput over that vap vap_list =lf_utils.FindVAP() vap_measure_obj=VAP_Measure(host,port) - - dut_temp_obj = Login_DUT(1, "Thread-1", "192.168.200.18") + + hostname=socket.gethostbyname(socket.gethostname()) + dut_temp_obj = Login_DUT(1, "Thread-1", hostname) #print(vap_list) vap_names=[] @@ -243,7 +245,7 @@ def PortUtility(host, port, duration, report_name, scenario, detail): cx_names.remove('uri') absolute_time=[] temp_time =[] - + Total_Throughput_CX_Side =[] Total_Throughput_VAP_Side =[] print(lf_utils.local_realm.json_get("/cx/"+cx_names[0]).get(cx_names[0]).get('state')) @@ -270,22 +272,22 @@ def PortUtility(host, port, duration, report_name, scenario, detail): print(Total_Throughput_CX_Side) print(Total_Throughput_VAP_Side) GenerateReport(scenario, detail, Total_Throughput_CX_Side, Total_Throughput_VAP_Side, absolute_time, relative_time, dut_temp_obj.data_core1, dut_temp_obj.data_core2, duration, report_name) - + # main method def main(): parser = argparse.ArgumentParser(description="Test Scenario of DUT Temperature measurement along with simultaneous throughput on VAP as well as stations") - + parser.add_argument("-m", "--manager", type=str, help="Enter the address of Lanforge Manager (By default localhost)") parser.add_argument("-sc", "--scenario", type=str, help="Enter the Name of the Scenario you want to load (by Default DFLT)") parser.add_argument("-t", "--duration", type=int, help="Enter the Time for which you want to run test (In Minutes)") parser.add_argument("-o", "--report_name", type=str, help="Enter the Name of the Output file ('Report.xlsx')") parser.add_argument("-td", "--test_detail", type=str, help="Enter the Test Detail in Quotes ") - + args = None - + try: args = parser.parse_args() # Lanforge Manager IP Address @@ -310,26 +312,25 @@ def main(): except Exception as e: logging.exception(e) exit(2) - - + + hostname=socket.gethostbyname(socket.gethostname()) # Loading DUT Scenario Scenario_1 = LoadScenario("192.168.200.18", 8080, "Lexus_Dut") - - + + # Loading LF Scenario DB_Lanforge_2 = "LF_Device" Scenario_2 = LoadScenario(manager, 8080, scenario) #Wait for Sometime time.sleep(10) - + # Port Utility function for reading CX and VAP PortUtility(manager,8080, duration, report_name, scenario, test_detail) - - - - + + + + if __name__ == '__main__': main() -