mirror of
https://github.com/Telecominfraproject/wlan-lanforge-scripts.git
synced 2025-11-01 19:28:00 +00:00
- Changes args parser for chamberview
- added Resource, VLAN, uses-2 - Changes import in lf_wifi_capacity_test.py Signed-off-by: SushantBawiskar <sushant.bawiskar@candelatech.com>
This commit is contained in:
SushantBawiskar
@@ -1,10 +1,14 @@
|
||||
"""
|
||||
Note: This is a library file used to create a chamber view scenario.
|
||||
import this file as showed in create_chamberview.py to create a scenario
|
||||
"""
|
||||
|
||||
import time
|
||||
|
||||
# !/usr/bin/env python3
|
||||
# ---- ---- ---- ---- LANforge Base Imports ---- ---- ---- ----
|
||||
from LANforge import LFRequest
|
||||
from LANforge import LFUtils
|
||||
from LANforge.lfcli_base import LFCliBase
|
||||
import datetime
|
||||
import time
|
||||
|
||||
|
||||
class chamberview(LFCliBase):
|
||||
def __init__(self,
|
||||
@@ -15,11 +19,22 @@ class chamberview(LFCliBase):
|
||||
_lfjson_port=lfclient_port)
|
||||
|
||||
#behaves same as chamberview manage scenario
|
||||
def manage_cv_scenario(self, scenario_name="scenario-python", profile="STA-AC", amount="1", dutname="ASUS", dutradio="Radio-1", traffic="http", uses="wiphy0"):
|
||||
def manage_cv_scenario(self,
|
||||
scenario_name="Automation",
|
||||
Resources="1.1",
|
||||
Profile="STA-AC",
|
||||
Amount="1",
|
||||
DUT="DUT",
|
||||
Dut_Radio="Radio-1" ,
|
||||
Uses1="wiphy0",
|
||||
Uses2="AUTO",
|
||||
Traffic="http",
|
||||
Freq="-1",
|
||||
VLAN=""):
|
||||
req_url = "/cli-json/add_text_blob"
|
||||
|
||||
text_blob = "profile_link 1.1" + " " + profile + " " + amount + " " + "\'DUT:" + " " + dutname + " " + dutradio\
|
||||
+ "\' " + traffic + " " + uses + ",AUTO -1"
|
||||
text_blob = "profile_link" + " " + Resources + " " + Profile + " " + Amount + " " + "\'DUT:" + " " + DUT\
|
||||
+ " " + Dut_Radio + "\' " + Traffic + " " + Uses1 + ","+Uses2 + " " + Freq + " " + VLAN
|
||||
|
||||
print(text_blob)
|
||||
data = {
|
||||
@@ -27,7 +42,7 @@ class chamberview(LFCliBase):
|
||||
"name": scenario_name,
|
||||
"text": text_blob
|
||||
}
|
||||
print(data)
|
||||
|
||||
rsp = self.json_post(req_url, data)
|
||||
time.sleep(2)
|
||||
|
||||
@@ -1,80 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
"""
|
||||
Script for creating a chamberview scenario.
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import argparse
|
||||
import time
|
||||
|
||||
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'))
|
||||
|
||||
from chamberview import chamberview as cv
|
||||
|
||||
def main():
|
||||
|
||||
parser = argparse.ArgumentParser(
|
||||
description="""use build_chamberview to create a lanforge chamberview scenario
|
||||
|
||||
""")
|
||||
parser.add_argument("-m", "--lfmgr", type=str,
|
||||
help="address of the LANforge GUI machine (localhost is default)")
|
||||
parser.add_argument("-o", "--port", type=int,
|
||||
help="IP Port the LANforge GUI is listening on (8080 is default)")
|
||||
parser.add_argument("-cs", "--create_scenario", "--create_lf_scenario", type=str,
|
||||
help="name of scenario to be created")
|
||||
parser.add_argument("-p", "--profile", type=str, required=True,
|
||||
help="name of profile")
|
||||
parser.add_argument("-n", "--no_stations", type=str, required=True,
|
||||
help="Number of stations")
|
||||
parser.add_argument("-d", "--dut", "--DUT", type=str, required=True,
|
||||
help="Name of the DUT")
|
||||
parser.add_argument("-dr", "--dr", "--dut_radio", type=str, required=True,
|
||||
help="Select DUT Radio ex. \"Radio-1\", \"Radio-2\"")
|
||||
parser.add_argument("-t", "--t", "--traffic", type=str, required=True,
|
||||
help="Select traffic ex. \"tcp-dl-6m-vi\"")
|
||||
parser.add_argument("-r", "--r", "--radio", type=str, required=True,
|
||||
help="Select traffic ex. \"wiphy0\"")
|
||||
|
||||
|
||||
args = parser.parse_args()
|
||||
if args.lfmgr is not None:
|
||||
lfjson_host = args.lfmgr
|
||||
if args.port is not None:
|
||||
lfjson_port = args.port
|
||||
|
||||
scenario_name = args.create_scenario
|
||||
profile_name = args.profile
|
||||
create_stations = args.no_stations
|
||||
dut_name = args.dut
|
||||
dut_radio = args.dr
|
||||
traffic_type = args.t
|
||||
radio = args.r
|
||||
|
||||
createCV = cv(lfjson_host, lfjson_port); #Create a object
|
||||
createCV.manage_cv_scenario(scenario_name, profile_name, create_stations, dut_name, dut_radio
|
||||
, traffic_type, radio); #To manage scenario
|
||||
createCV.sync_cv() #chamberview sync
|
||||
|
||||
time.sleep(2)
|
||||
createCV.apply_cv_scenario(scenario_name) #Apply scenario
|
||||
createCV.apply_cv_scenario(scenario_name)
|
||||
createCV.apply_cv_scenario(scenario_name)
|
||||
createCV.apply_cv_scenario(scenario_name)
|
||||
createCV.apply_cv_scenario(scenario_name)
|
||||
createCV.apply_cv_scenario(scenario_name)
|
||||
|
||||
time.sleep(2)
|
||||
createCV.build_cv_scenario() #build scenario
|
||||
print("End")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
146
py-scripts/create_chamberview.py
Normal file
146
py-scripts/create_chamberview.py
Normal file
@@ -0,0 +1,146 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
"""
|
||||
Note: Script for creating a chamberview scenario.
|
||||
Run this script to set/create a chamber view scenario.
|
||||
ex. on how to run this script:
|
||||
create_chamberview.py -m "localhost" -o "8080" -cs "scenario_name"
|
||||
--line "Resource=1.1 Profile=STA-AC Amount=1 Uses-1=wiphy0 Uses-2=AUTO Freq=-1
|
||||
DUT=Test DUT_Radio=Radio-1 Traffic=http VLAN=VLAN"
|
||||
--line "Resource=1.1 Profile=upstream Amount=1 Uses-1=wiphy1 Uses-2=AUTO 2 Freq=-1
|
||||
DUT=Test DUT_Radio=Radio-2 Traffic=http VLAN=VLAN"
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import argparse
|
||||
import time
|
||||
import re
|
||||
|
||||
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'))
|
||||
|
||||
from cv_commands import chamberview as cv
|
||||
|
||||
def main():
|
||||
global Resource, Amount, DUT, DUT_Radio, Profile, Uses1, Uses2, Traffic, Freq, VLAN
|
||||
|
||||
|
||||
parser = argparse.ArgumentParser(
|
||||
description="""
|
||||
For Two line scenario use --line twice as shown in example, for multi line scenario
|
||||
use --line argument to create multiple lines
|
||||
\n
|
||||
create_chamberview.py -m "localhost" -o "8080" -cs "scenario_name"
|
||||
--line "Resource=1.1 Profile=STA-AC Amount=1 Uses-1=wiphy0 Uses-2=AUTO Freq=-1
|
||||
DUT=Test DUT_Radio=Radio-1 Traffic=http VLAN=VLAN"
|
||||
--line "Resource=1.1 Profile=upstream Amount=1 Uses-1=wiphy1 Uses-2=AUTO Freq=-1
|
||||
DUT=Test DUT_Radio=Radio-2 Traffic=http VLAN=VLAN"
|
||||
""")
|
||||
parser.add_argument("-m", "--lfmgr", type=str,
|
||||
help="address of the LANforge GUI machine (localhost is default)")
|
||||
parser.add_argument("-o", "--port", type=int,
|
||||
help="IP Port the LANforge GUI is listening on (8080 is default)")
|
||||
parser.add_argument("-cs", "--create_scenario", "--create_lf_scenario", type=str,
|
||||
help="name of scenario to be created")
|
||||
parser.add_argument("-l", "--line", action='append', nargs='+', type=str, required=True,
|
||||
help="line number")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.lfmgr is not None:
|
||||
lfjson_host = args.lfmgr
|
||||
if args.port is not None:
|
||||
lfjson_port = args.port
|
||||
|
||||
createCV = cv(lfjson_host, lfjson_port); # Create a object
|
||||
scenario_name = args.create_scenario
|
||||
line = args.line
|
||||
|
||||
Resource = "1.1"
|
||||
Profile = "STA-AC"
|
||||
Amount = "1"
|
||||
DUT = "DUT"
|
||||
DUT_Radio = "Radio-1"
|
||||
Uses1 = "wiphy0"
|
||||
Uses2 = "AUTO"
|
||||
Traffic = "http"
|
||||
Freq = "-1"
|
||||
VLAN = ""
|
||||
|
||||
for i in range(len(line)):
|
||||
if " " in line[i][0]:
|
||||
line[i][0] = (re.split(' ', line[i][0]))
|
||||
elif "," in line[i][0]:
|
||||
line[i][0] = (re.split(',', line[i][0]))
|
||||
print("in second")
|
||||
elif ", " in line[i][0]:
|
||||
line[i][0] = (re.split(',', line[i][0]))
|
||||
print("in third")
|
||||
elif " ," in line[i][0]:
|
||||
line[i][0] = (re.split(',', line[i][0]))
|
||||
print("in forth")
|
||||
else:
|
||||
print("Wrong arguments entered !")
|
||||
exit(1)
|
||||
|
||||
for j in range(len(line[i][0])):
|
||||
line[i][0][j] = line[i][0][j].split("=")
|
||||
for k in range(len(line[i][0][j])):
|
||||
name = line[i][0][j][k]
|
||||
if str(name) == "Resource" or str(name) == "Res" or str(name) == "R":
|
||||
Resource = line[i][0][j][k + 1]
|
||||
elif str(name) == "Profile" or str(name) == "Prof" or str(name) == "P":
|
||||
Profile = line[i][0][j][k + 1]
|
||||
elif str(name) == "Amount" or str(name) == "Sta" or str(name) == "A":
|
||||
Amount = line[i][0][j][k + 1]
|
||||
elif str(name) == "Uses-1" or str(name) == "U1" or str(name) == "U-1":
|
||||
Uses1 = line[i][0][j][k + 1]
|
||||
elif str(name) == "Uses-2" or str(name) == "U2" or str(name) == "U-2":
|
||||
Uses2 = line[i][0][j][k + 1]
|
||||
elif str(name) == "Freq" or str(name) == "Freq" or str(name) == "F":
|
||||
Freq = line[i][0][j][k + 1]
|
||||
elif str(name) == "DUT" or str(name) == "dut" or str(name) == "D":
|
||||
DUT = line[i][0][j][k + 1]
|
||||
elif str(name) == "DUT_Radio" or str(name) == "dr" or str(name) == "DR":
|
||||
DUT_Radio = line[i][0][j][k + 1]
|
||||
elif str(name) == "Traffic" or str(name) == "Traf" or str(name) == "T":
|
||||
Traffic = line[i][0][j][k + 1]
|
||||
elif str(name) == "VLAN" or str(name) == "Vlan" or str(name) == "V":
|
||||
VLAN = line[i][0][j][k + 1]
|
||||
else:
|
||||
continue
|
||||
|
||||
createCV.manage_cv_scenario(scenario_name,
|
||||
Resource,
|
||||
Profile,
|
||||
Amount,
|
||||
DUT,
|
||||
DUT_Radio,
|
||||
Uses1,
|
||||
Uses2,
|
||||
Traffic,
|
||||
Freq,
|
||||
VLAN
|
||||
); # To manage scenario
|
||||
|
||||
|
||||
createCV.sync_cv() #chamberview sync
|
||||
time.sleep(2)
|
||||
createCV.apply_cv_scenario(scenario_name) #Apply scenario
|
||||
time.sleep(2)
|
||||
createCV.sync_cv()
|
||||
time.sleep(2)
|
||||
createCV.apply_cv_scenario(scenario_name) # Apply scenario
|
||||
|
||||
time.sleep(2)
|
||||
createCV.build_cv_scenario() #build scenario
|
||||
print("End")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,93 +0,0 @@
|
||||
import sys
|
||||
import os
|
||||
import argparse
|
||||
import time
|
||||
import json
|
||||
from os import path
|
||||
|
||||
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'))
|
||||
|
||||
from cv_test_manager import cv_test as cvtest
|
||||
from chamberview import chamberview as cv
|
||||
from cv_test_reports import lanforge_reports as lf_rpt
|
||||
|
||||
def main():
|
||||
config_name = "WFC_scenario1" # Test Config Name (new)
|
||||
instance_name = "wfc_instance" # Test Instance name
|
||||
test_name = "WiFi Capacity" # Test name
|
||||
|
||||
# Test related settings
|
||||
batch_size = "5"
|
||||
loop_iter = "1"
|
||||
protocol = "TCP-IPv4"
|
||||
duration = " 5000"
|
||||
dict = {"batch_size": "batch_size:" + " " + str(batch_size),
|
||||
"loop_iter": "loop_iter:" + " " + str(loop_iter),
|
||||
"protocol": "protocol:" + " " + str(protocol),
|
||||
"duration": "duration:" + " " + str(duration)}
|
||||
|
||||
|
||||
|
||||
run_test = cvtest("192.168.200.21", "8080")
|
||||
createCV = cv("192.168.200.21", "8080"); # Create a object
|
||||
|
||||
|
||||
port_list = []
|
||||
|
||||
response = run_test.check_ports();
|
||||
port_size = json.dumps(len(response["interfaces"]))
|
||||
|
||||
for i in range(int(port_size)):
|
||||
list_val = json.dumps(response["interfaces"][i])
|
||||
list_val_ = json.loads(list_val).keys()
|
||||
list_val_ = str(list_val_).replace("dict_keys(['", "")
|
||||
list_val_ = str(list_val_).replace("'])", "")
|
||||
if (list_val_.__contains__("sta") or list_val_.__contains__("eth1")):
|
||||
port_list.append(list_val_)
|
||||
|
||||
for i in range(len(port_list)):
|
||||
add_port = "sel_port-" + str(i) + ":" + " " + port_list[i]
|
||||
run_test.create_test_config(config_name, add_port)
|
||||
time.sleep(0.2)
|
||||
|
||||
for key, value in dict.items():
|
||||
run_test.create_test_config(config_name, value)
|
||||
time.sleep(0.2)
|
||||
|
||||
|
||||
run_test.create_test(test_name, instance_name)
|
||||
time.sleep(5)
|
||||
createCV.sync_cv()
|
||||
time.sleep(2)
|
||||
run_test.load_test_config(config_name, instance_name)
|
||||
time.sleep(2)
|
||||
run_test.auto_save_report(instance_name)
|
||||
time.sleep(4)
|
||||
run_test.start_test(instance_name)
|
||||
|
||||
while (True):
|
||||
check = run_test.get_report_location(instance_name)
|
||||
location = json.dumps(check[0]["LAST"]["response"])
|
||||
print("WiFi Capacity Test Running...")
|
||||
if location != "\"Report Location:::\"":
|
||||
location = location.replace("Report Location:::","")
|
||||
print(location)
|
||||
time.sleep(1)
|
||||
run_test.close_instance(instance_name)
|
||||
time.sleep(1)
|
||||
run_test.cancel_instance(instance_name)
|
||||
time.sleep(4)
|
||||
break
|
||||
|
||||
time.sleep(60)
|
||||
report = lf_rpt()
|
||||
report.pull_reports(hostname="192.168.200.21", username="lanforge", password="lanforge",
|
||||
report_location=location)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
134
py-scripts/lf_wifi_capacity_test.py
Normal file
134
py-scripts/lf_wifi_capacity_test.py
Normal file
@@ -0,0 +1,134 @@
|
||||
"""
|
||||
Note: This is a test file which will run a wifi capacity test.
|
||||
ex. on how to run this script:
|
||||
./lf_wifi_capacity_test.py -lfmgr "localhost -port "8080" -test_name "WiFi Capacity" -instance_name
|
||||
"test_instance" -config_name "config_file" -pull_report y -batch_size "1" -loop_iter 1 -protocol "TCP-IPv4" -duration 5000
|
||||
if -r flag is set to y. at end of the script you will receive a report dir. from lanforge to your local machine
|
||||
"""
|
||||
import sys
|
||||
import os
|
||||
import argparse
|
||||
import time
|
||||
import json
|
||||
from os import path
|
||||
|
||||
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'))
|
||||
|
||||
from cv_test_manager import cv_test as cvtest
|
||||
from cv_commands import chamberview as cv
|
||||
from cv_test_reports import lanforge_reports as lf_rpt
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="""
|
||||
./lf_wifi_capacity_test.py -lfmgr "localhost -port "8080" -test_name "WiFi Capacity" -instance_name
|
||||
"test_instance" -config_name "config_file" -pull_report y -batch_size "1" -loop_iter 1 -protocol "TCP-IPv4" -duration 5000""")
|
||||
parser.add_argument("-m", "--lfmgr", type=str, default="localhost",
|
||||
help="address of the LANforge GUI machine (localhost is default)")
|
||||
parser.add_argument("-o", "--port", type=int, default="8080",
|
||||
help="IP Port the LANforge GUI is listening on (8080 is default)")
|
||||
parser.add_argument("-t", "--test_name", type=str, default="WiFi Capacity",
|
||||
help="name of test to be run ex. \"WiFi Capacity\"")
|
||||
parser.add_argument("-i", "--instance_name", type=str, required=True,
|
||||
help="name of test instance (by default: test_ref)")
|
||||
parser.add_argument("-c", "--config_name", type=str, required=True,
|
||||
help="Test config name (by default: DEFAULT)")
|
||||
parser.add_argument("-r", "--pull_report", type=str, required=True,
|
||||
help="pull reports from lanforge (by default: y)")
|
||||
parser.add_argument("-b", "--batch_size", type=str, required=True,
|
||||
help="config batch size (by default: 1)")
|
||||
parser.add_argument("-l", "--loop_iter", type=str, required=True,
|
||||
help="config loop iter (by default: 1)")
|
||||
parser.add_argument("-p", "--protocol", type=str, required=True,
|
||||
help="config protocol (by default: TCP-IPv4)")
|
||||
parser.add_argument("-d", "--duration", type=str, required=True,
|
||||
help="config duration (by default: 5000)")
|
||||
|
||||
args = parser.parse_args()
|
||||
if args.lfmgr is not None:
|
||||
lf_host = args.lfmgr
|
||||
if args.port is not None:
|
||||
lf_hostport = args.port
|
||||
|
||||
try:
|
||||
test_name = args.test_name
|
||||
instance_name = args.instance_name
|
||||
config_name = args.config_name
|
||||
batch_size = args.batch_size
|
||||
loop_iter = args.loop_iter
|
||||
protocol = args.protocol
|
||||
duration = args.duration
|
||||
pull_report = args.pull_report
|
||||
except:
|
||||
print("Wrong arguments entered")
|
||||
exit(1)
|
||||
|
||||
|
||||
# Test related settings
|
||||
dict = {"batch_size": "batch_size:" + " " + str(batch_size),
|
||||
"loop_iter": "loop_iter:" + " " + str(loop_iter),
|
||||
"protocol": "protocol:" + " " + str(protocol),
|
||||
"duration": "duration:" + " " + str(duration)}
|
||||
|
||||
run_test = cvtest(lf_host, lf_hostport)
|
||||
createCV = cv(lf_host, lf_hostport); # Create a object
|
||||
|
||||
port_list = []
|
||||
|
||||
response = run_test.check_ports();
|
||||
port_size = json.dumps(len(response["interfaces"]))
|
||||
|
||||
for i in range(int(port_size)):
|
||||
list_val = json.dumps(response["interfaces"][i])
|
||||
list_val_ = json.loads(list_val).keys()
|
||||
list_val_ = str(list_val_).replace("dict_keys(['", "")
|
||||
list_val_ = str(list_val_).replace("'])", "")
|
||||
if (list_val_.__contains__("sta") or list_val_.__contains__("eth1")):
|
||||
port_list.append(list_val_)
|
||||
|
||||
for i in range(len(port_list)):
|
||||
add_port = "sel_port-" + str(i) + ":" + " " + port_list[i]
|
||||
run_test.create_test_config(config_name,"Wifi-Capacity-",add_port)
|
||||
time.sleep(0.2)
|
||||
|
||||
for key, value in dict.items():
|
||||
run_test.create_test_config(config_name,"Wifi-Capacity-",value)
|
||||
time.sleep(0.2)
|
||||
|
||||
run_test.create_test(test_name, instance_name)
|
||||
time.sleep(5)
|
||||
createCV.sync_cv()
|
||||
time.sleep(2)
|
||||
run_test.load_test_config(config_name, instance_name)
|
||||
time.sleep(2)
|
||||
run_test.auto_save_report(instance_name)
|
||||
time.sleep(4)
|
||||
run_test.start_test(instance_name)
|
||||
|
||||
while (True):
|
||||
check = run_test.get_report_location(instance_name)
|
||||
location = json.dumps(check[0]["LAST"]["response"])
|
||||
print("WiFi Capacity Test Running...")
|
||||
if location != "\"Report Location:::\"":
|
||||
location = location.replace("Report Location:::", "")
|
||||
print(location)
|
||||
time.sleep(1)
|
||||
run_test.close_instance(instance_name)
|
||||
time.sleep(1)
|
||||
run_test.cancel_instance(instance_name)
|
||||
time.sleep(4)
|
||||
location = location.strip("\"")
|
||||
report = lf_rpt()
|
||||
print(location)
|
||||
if (pull_report == "yes" ) or (pull_report == "y") or (pull_report == "Y"):
|
||||
report.pull_reports(hostname=lf_host, username="lanforge", password="lanforge",
|
||||
report_location=location)
|
||||
break
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user