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

test_l3_scenario_throughut: Make file pip compliant

Browse Source 
Signed-off-by: Matthew Stidham <stidmatt@gmail.com>
This commit is contained in:
Matthew Stidham
2021-11-24 13:44:54 -08:00
parent eb6d4a4e70
commit 661e427128
1 changed files with 108 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
py-scripts/test_l3_scenario_throughput.py
View File

@@ -1,5 +1,5 @@
#!/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 Script has three classes :
1. LoadScenario : It will load the existing saved scenario to the Lanforge (Here used for Loading Bridged VAP)
@@ -23,7 +23,7 @@
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
import os
import importlib
@@ -31,7 +31,6 @@ import argparse
import time
import logging
import paramiko as pmgo
from paramiko.ssh_exception import NoValidConnectionsError as exception
import xlsxwriter
from bokeh.io import show
from bokeh.plotting import figure
@@ -43,7 +42,6 @@ if sys.version_info[0] != 3:
print("This script requires Python 3")
exit(1)
sys.path.append(os.path.join(os.path.abspath(__file__ + "../../../")))
lfcli_base = importlib.import_module("py-json.LANforge.lfcli_base")
@@ -63,12 +61,13 @@ class Login_DUT:
self.USERNAME = "lanforge"
self.PASSWORD = "lanforge"
self.CLIENT = pmgo.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")
self.Connect()
def run(self):
stdin, stdout, stderr = self.CLIENT.exec_command("sensors")
out_lines = stdout.readlines()
@@ -77,17 +76,10 @@ class Login_DUT:
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(pmgo.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]
@@ -101,10 +93,9 @@ class LoadScenario(LFCliBase):
time.sleep(2)
# Generates XLSX Report
def GenerateReport(scenario, detail, throughput_sta, throughput_vap, absolute_time, relative_time, core1_temp, core2_temp, duration, name):
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)
@@ -117,7 +108,6 @@ def GenerateReport(scenario, detail, throughput_sta, throughput_vap, absolute_ti
core1 = []
core2 = []
j = 3
for i in absolute_time:
worksheet.write('A' + str(j), i)
@@ -162,7 +152,6 @@ def GenerateReport(scenario, detail, throughput_sta, throughput_vap, absolute_ti
# 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"
@@ -193,10 +182,6 @@ class VAP_Measure(LFCliBase):
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):
@@ -207,17 +192,15 @@ class FindPorts(LFCliBase):
# 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
@@ -246,89 +229,63 @@ def PortUtility(host, port, duration, report_name, scenario, detail):
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'):
for _ 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)):
for _ in range(0, int(duration)):
temp = 0
for i in cx_names:
for _ 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')))
for _ 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)
relative_time = [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)
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(
prog="test_l3_scenario_throughput.py",
formatter_class=argparse.RawTextHelpFormatter,
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("-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=str, help="Enter the Time for which you want to run test")
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 ")
parser.add_argument("-o", "--report_name", type=str, help="Enter the Name of the Output file ('Report.xlsx')", default='report.xlsx')
parser.add_argument("-td", "--test_detail", type=str, help="Enter the Test Detail in Quotes ", default='Blank test')
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.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)
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)
Scenario_2 = LoadScenario(args.manager, 8080, args.scenario)
# Wait for Sometime
time.sleep(10)
duration_sec = Realm.parse_time(args.duration).total_seconds() * 60
# Port Utility function for reading CX and VAP
PortUtility(manager,8080, duration_sec, report_name, scenario, test_detail)
PortUtility(args.manager, 8080, duration_sec, args.report_name, args.scenario, args.test_detail)
if __name__ == '__main__':