mirror of
https://github.com/Telecominfraproject/wlan-lanforge-scripts.git
synced 2025-10-31 18:58:01 +00:00
336 lines
12 KiB
Python
Executable File
336 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
'''
|
|
|
|
This Script Loads the Existing Scenario and Run the Simultaenous Throughput over time and Generate Report and Plot the Graph
|
|
This Scrip has three classes :
|
|
1. LoadScenario : It will load the existing saved scenario to the Lanforge (Here used for Loading Bridged VAP)
|
|
2. FindPorts : Fetch the L3CX Throughput and VAP Throughput
|
|
3. Login_DUT : This class is specifically used to test the Linux based DUT that has SSH Server. It is used to read the CPU Core temperature during testing
|
|
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
|
|
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"
|
|
|
|
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
|
|
'''
|
|
|
|
import sys
|
|
if sys.version_info[0] != 3:
|
|
print("This script requires Python 3")
|
|
exit(1)
|
|
if 'py-json' not in sys.path:
|
|
sys.path.append('../py-json')
|
|
|
|
|
|
import argparse
|
|
import time
|
|
from LANforge import LFUtils
|
|
from LANforge import lfcli_base
|
|
from LANforge.lfcli_base import LFCliBase
|
|
from LANforge.LFUtils import *
|
|
import realm
|
|
|
|
from realm import Realm
|
|
import logging
|
|
|
|
import paramiko as pm
|
|
from paramiko.ssh_exception import NoValidConnectionsError as exception
|
|
import xlsxwriter
|
|
from bokeh.io import output_file, show
|
|
from bokeh.plotting import figure
|
|
from bokeh.models import LinearAxis, Range1d
|
|
from bokeh.models import HoverTool, Range1d
|
|
from bokeh.layouts import row
|
|
from datetime import datetime
|
|
|
|
# Specifically for Measuring CPU Core Temperatures
|
|
class Login_DUT:
|
|
|
|
def __init__(self, threadID, name, HOST):
|
|
self.threadID = threadID
|
|
self.name = name
|
|
self.host=HOST
|
|
self.USERNAME = "lanforge"
|
|
self.PASSWORD = "lanforge"
|
|
self.CLIENT= pm.SSHClient()
|
|
self.LF1= self.Connect()
|
|
self.data_core1=[]
|
|
self.data_core2=[]
|
|
if self.CLIENT == 0:
|
|
exit()
|
|
print("Connected to " +HOST+" DUT to Measure the Core Temperature")
|
|
def run(self):
|
|
stdin, stdout, stderr= self.CLIENT.exec_command("sensors")
|
|
out_lines = stdout.readlines()
|
|
err_lines = stderr.readlines()
|
|
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
|
|
except exception as error:
|
|
self.CLIENT = 0;
|
|
return None
|
|
|
|
|
|
|
|
# Class to Load a Scenario that has been Created in Chamber View saved under DB/[Database_Name]
|
|
class LoadScenario(LFCliBase):
|
|
def __init__(self, host, port, db_name, 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)
|
|
self.host = host
|
|
self.port = port
|
|
self.json_post("/cli-json/load", { "name": db_name, "action": 'overwrite' })
|
|
print(host+ " : Scenario Loaded...")
|
|
time.sleep(2)
|
|
|
|
|
|
|
|
|
|
# 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()
|
|
worksheet.write('A1',"Scenario Runned: "+scenario+"\n Scenario Details: "+detail)
|
|
worksheet.write('A2', 'ABSOLUTE TIME')
|
|
worksheet.write('B2', 'RELATIVE TIME (ms)')
|
|
worksheet.write('C2', 'THROUGHPUT STATION SIDE (Mbps)')
|
|
worksheet.write('D2', 'THROUGHPUT VAP SIDE (Mbps)')
|
|
worksheet.write('E2', 'CORE 0 TEMP (Degree Celsius)')
|
|
worksheet.write('F2', 'CORE 1 TEMP (Degree Celsius)')
|
|
core1=[]
|
|
core2=[]
|
|
|
|
|
|
j=3
|
|
for i in absolute_time:
|
|
worksheet.write('A'+str(j),i)
|
|
j=j+1
|
|
|
|
j=3
|
|
for i in relative_time:
|
|
worksheet.write('B'+str(j),i)
|
|
j=j+1
|
|
|
|
sta_throu=[]
|
|
vap_throu=[]
|
|
j=3
|
|
for i in throughput_sta:
|
|
print(i)
|
|
sta_throu.append(i/1000000)
|
|
worksheet.write('C'+str(j), str(i/1000000)+" Mbps")
|
|
j=j+1
|
|
j=3
|
|
for i in throughput_vap:
|
|
print(i)
|
|
vap_throu.append(i/1000000)
|
|
worksheet.write('D'+str(j), str(i/1000000)+" Mbps")
|
|
j=j+1
|
|
j=3
|
|
for i in core1_temp:
|
|
core1.append(int(str(i).split(':')[1].split('(')[0].split('.')[0].split('+')[1]))
|
|
worksheet.write('E'+str(j),str(i).split(':')[1].split('(')[0] )
|
|
j=j+1
|
|
j=3
|
|
for i in core2_temp:
|
|
core2.append(int(str(i).split(':')[1].split('(')[0].split('.')[0].split('+')[1]))
|
|
worksheet.write('F'+str(j), str(i).split(':')[1].split('(')[0])
|
|
j=j+1
|
|
|
|
Time =[]
|
|
for i in range(0,int(duration)*5):
|
|
Time.append(i)
|
|
plot(sta_throu, vap_throu, core1, core2, Time)
|
|
workbook.close()
|
|
|
|
|
|
# Plotting Function for Parameters
|
|
def plot(throughput_sta, throughput_vap, core1_temp, core2_temp, Time):
|
|
|
|
print(throughput_vap)
|
|
s1 = figure(plot_width=1000, plot_height=600)
|
|
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')
|
|
|
|
s1.line(Time, core2_temp, y_range_name='Temperature', color='green', legend_label ="CPU CORE 1 TEMPERATURE ")
|
|
#s1.circle(Time, core2_temp, y_range_name='Temperature', color='blue')
|
|
|
|
show(s1)
|
|
|
|
|
|
# Creates the Instance for LFCliBase
|
|
class VAP_Measure(LFCliBase):
|
|
def __init__(self, lfclient_host, lfclient_port):
|
|
super().__init__(lfclient_host, lfclient_port)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 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)
|
|
self.host = host
|
|
self.port = port
|
|
|
|
#Creating a Realm Object
|
|
self.local_realm = Realm(lfclient_host=host, lfclient_port=port)
|
|
|
|
|
|
def FindExistingCX(self):
|
|
|
|
return self.local_realm.cx_list()
|
|
|
|
def FindVAP(self):
|
|
return self.local_realm.vap_list()
|
|
|
|
# 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")
|
|
|
|
#print(vap_list)
|
|
vap_names=[]
|
|
for i in vap_list:
|
|
vap_names.append(str(i.keys()).split('.')[2].split('\'')[0])
|
|
print(vap_names[0])
|
|
|
|
cx_names = list(cx_data.keys())
|
|
cx_names.remove('handler')
|
|
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'))
|
|
for i in cx_names:
|
|
while(lf_utils.local_realm.json_get("/cx/"+cx_names[0]).get(cx_names[0]).get('state') != 'Run'):
|
|
continue
|
|
offset=int(round(time.time() * 1000))
|
|
for i in range(0,int(duration)):
|
|
temp=0
|
|
for i in cx_names:
|
|
temp=temp+int(lf_utils.local_realm.json_get("/cx/"+i).get(i).get('bps rx a'))
|
|
#temp=temp+lf_utils.local_realm.json_get("/cx/"+i).get(i).get('bps rx b')
|
|
for i in vap_names:
|
|
Total_Throughput_VAP_Side.append(int(vap_measure_obj.json_get("/port/1/1/"+str(i)).get('interface').get('bps rx')))
|
|
absolute_time.append(datetime.now().strftime("%H:%M:%S"))
|
|
temp_time.append(int(round(time.time() * 1000)-offset))
|
|
Total_Throughput_CX_Side.append(temp)
|
|
dut_temp_obj.run()
|
|
time.sleep(5)
|
|
relative_time=[]
|
|
relative_time.append(0)
|
|
for i in range (0,len(temp_time)-1):
|
|
relative_time.append(temp_time[i+1]-temp_time[i])
|
|
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
|
|
if (args.manager is None):
|
|
manager = "localhost"
|
|
if (args.manager is not None):
|
|
manager = args.manager
|
|
if (args.scenario is not None):
|
|
scenario = args.scenario
|
|
if (args.duration is not None):
|
|
duration = (args.duration * 60)/5
|
|
if (args.report_name is not None):
|
|
report_name = args.report_name
|
|
if (args.duration is None):
|
|
duration = (1 * 60)/5
|
|
if (args.report_name is None):
|
|
report_name = "report.xlsx"
|
|
if (args.test_detail is not None):
|
|
test_detail = args.test_detail
|
|
if (args.test_detail is None):
|
|
test_detail = "Blank test"
|
|
except Exception as e:
|
|
logging.exception(e)
|
|
exit(2)
|
|
|
|
|
|
|
|
# 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()
|
|
|