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

fix

Browse Source
This commit is contained in:
Matthew Stidham
2020-12-30 17:00:52 -08:00
parent cacfc15766
commit 9c94287fb5
4 changed files with 249 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
py-json/realm.py
View File

@@ -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", {

209
py-scripts/test_ipv4_connection.py Normal file
View File

@@ -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()

4
py-scripts/test_ipv4_ps.py
View File

@@ -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

75
py-scripts/test_l3_scenario_throughput.py
View File

@@ -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()