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

- integrated cv_commands.pywith cv_test_manager.py

Browse Source 
- create_chamberview_dut.py to create a dut
- cv_dut_profile.py as library to create a dut
- cv test manager can be used to implement chamber view also

Signed-off-by: SushantBawiskar <sushant.bawiskar@candelatech.com>
This commit is contained in:
SushantBawiskar
2021-04-24 01:07:38 +05:30
parent 9ade91f3de
commit 810377b16e
5 changed files with 316 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
py-json/cv_commands.py
View File

@@ -1,104 +0,0 @@
"""
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.lfcli_base import LFCliBase
class chamberview(LFCliBase):
def __init__(self,
lfclient_host="localhost",
lfclient_port=8080,
):
super().__init__(_lfjson_host=lfclient_host,
_lfjson_port=lfclient_port)
#behaves same as chamberview manage scenario
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" + " " + Resources + " " + Profile + " " + Amount + " " + "\'DUT:" + " " + DUT\
+ " " + Dut_Radio + "\' " + Traffic + " " + Uses1 + ","+Uses2 + " " + Freq + " " + VLAN
data = {
"type": "Network-Connectivity",
"name": scenario_name,
"text": text_blob
}
rsp = self.json_post(req_url, data)
time.sleep(2)
def pass_raw_lines_to_cv(self,
scenario_name="Automation",
Rawline=""):
req_url = "/cli-json/add_text_blob"
data = {
"type": "Network-Connectivity",
"name": scenario_name,
"text": Rawline
}
rsp = self.json_post(req_url, data)
time.sleep(2)
def show_text_blob(self, config_name, blob_test_name, brief):
req_url = "/cli-json/show_text_blob"
data = {"type": "Plugin-Settings"}
if config_name and blob_test_name:
data["name"] = "%s%s" % (blob_test_name, config_name) # config name
else:
data["name"] = "ALL"
if brief:
data["brief"] = "brief"
return self.json_post(req_url, data)
#This is for chamber view buttons
def apply_cv_scenario(self, cv_scenario):
cmd = "cv apply '%s'" % cv_scenario #To apply scenario
self.run_cv_cmd(cmd)
print("Applying %s scenario" % cv_scenario)
def build_cv_scenario(self):#build chamber view scenario
cmd = "cv build"
self.run_cv_cmd(cmd)
print("Building scenario")
def is_cv_build(self):#check if scenario is build
cmd = "cv is_built"
response = self.run_cv_cmd(cmd)
return self.check_reponse(response)
def sync_cv(self):#sync
cmd = "cv sync"
self.run_cv_cmd(cmd)
def run_cv_cmd(self, command):#Send chamber view commands
response_json = []
req_url = "/gui-json/cmd"
data = {
"cmd": command
}
rsp = self.json_post(req_url, data, debug_=False, response_json_list_=response_json)
return response_json
def check_reponse(self, response):
d1 = {k: v for e in response for (k, v) in e.items()}
return d1["LAST"]["response"]

91
py-json/cv_dut_profile.py Normal file
View File

@@ -0,0 +1,91 @@
import time
# !/usr/bin/env python3
# ---- ---- ---- ---- LANforge Base Imports ---- ---- ---- ----
from LANforge.lfcli_base import LFCliBase
class cv_dut(LFCliBase):
def __init__(self,
lfclient_host="localhost",
lfclient_port=8080,
):
super().__init__(_lfjson_host=lfclient_host,
_lfjson_port=lfclient_port)
def create_dut(self,
dut_name="DUT",
flags="4098",
ssid1="[BLANK]",
pass1="[BLANK]",
ssid2="[BLANK]",
pass2="[BLANK]",
ssid3="[BLANK]",
pass3="[BLANK]",
bssid1="00:00:00:00:00:00",
bssid2="00:00:00:00:00:00",
bssid3="00:00:00:00:00:00",
mgt_ip="0.0.0.0",
eap_id="[BLANK]"):
response_json = []
req_url = "/cli-json/add_dut"
data = {
"name": dut_name,
"flags": flags,
"img_file": "NONE",
"sw_version": "[BLANK]",
"hw_version": "[BLANK]",
"model_num": "[BLANK]",
"serial_num": "[BLANK]",
"serial_port": "[BLANK]",
"wan_port": "[BLANK]",
"lan_port": "[BLANK]",
"ssid1": ssid1,
"passwd1": pass1,
"ssid2": ssid2,
"passwd2": pass2,
"ssid3": ssid3,
"passwd3": pass3,
"mgt_ip": mgt_ip,
"api_id": "0",
"flags_mask": "NA",
"antenna_count1": "0",
"antenna_count2": "0",
"antenna_count3": "0",
"bssid1": bssid1,
"bssid2": bssid2,
"bssid3": bssid3,
"top_left_x": "0",
"top_left_y": "0",
"eap_id": eap_id,
}
rsp = self.json_post(req_url, data, debug_=False, response_json_list_=response_json)
return rsp
def add_ssid(self,
dut_name="DUT",
ssid_idx=0,
ssid='[BLANK]',
passwd='[BLANK]',
bssid='00:00:00:00:00:00',
ssid_flags=0,
ssid_flags_mask=0xFFFFFFFF):
req_url = "/cli-json/add_dut_ssid"
print( "name:" + dut_name,
"ssid_idx:"+ ssid_idx,
"ssid:"+ ssid,
"passwd:" + passwd,
"bssid:" + bssid,
"ssid_flags:" + str(ssid_flags),
"ssid_flags_mask:"+ str(ssid_flags_mask))
self.json_post(req_url, {
"name": dut_name,
"ssid_idx": ssid_idx,
"ssid": ssid,
"passwd": passwd,
"bssid": bssid,
"ssid_flags": ssid_flags,
"ssid_flags_mask": ssid_flags_mask,
})

72
py-json/cv_test_manager.py
View File

@@ -383,3 +383,75 @@ class cv_test(Realm):
csvtoinflux.post_to_influx()
print("All done posting to influx.\n")
#************************** chamber view **************************
def add_text_blob_line(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" + " " + Resources + " " + Profile + " " + Amount + " " + "\'DUT:" + " " + DUT \
+ " " + Dut_Radio + "\' " + Traffic + " " + Uses1 + "," + Uses2 + " " + Freq + " " + VLAN
data = {
"type": "Network-Connectivity",
"name": scenario_name,
"text": text_blob
}
rsp = self.json_post(req_url, data)
def pass_raw_lines_to_cv(self,
scenario_name="Automation",
Rawline=""):
req_url = "/cli-json/add_text_blob"
data = {
"type": "Network-Connectivity",
"name": scenario_name,
"text": Rawline
}
rsp = self.json_post(req_url, data)
time.sleep(2)
# This is for chamber view buttons
def apply_cv_scenario(self, cv_scenario):
cmd = "cv apply '%s'" % cv_scenario # To apply scenario
self.run_cv_cmd(cmd)
print("Applying %s scenario" % cv_scenario)
def build_cv_scenario(self): # build chamber view scenario
cmd = "cv build"
self.run_cv_cmd(cmd)
print("Building scenario")
def get_cv_build_status(self): # check if scenario is build
cmd = "cv is_built"
response = self.run_cv_cmd(cmd)
return self.check_reponse(response)
def sync_cv(self): # sync
cmd = "cv sync"
print(self.run_cv_cmd(cmd))
def run_cv_cmd(self, command): # Send chamber view commands
response_json = []
req_url = "/gui-json/cmd"
data = {
"cmd": command
}
rsp = self.json_post(req_url, data, debug_=False, response_json_list_=response_json)
return response_json
def check_reponse(self, response):
d1 = {k: v for e in response for (k, v) in e.items()}
return d1["LAST"]["response"]

15
py-scripts/create_chamberview.py
View File

@@ -46,7 +46,7 @@ if sys.version_info[0] != 3:
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
from cv_test_manager import cv_test as cv
class CreateChamberview(cv):
def __init__(self,
@@ -126,7 +126,7 @@ class CreateChamberview(cv):
else:
continue
self.manage_cv_scenario(scenario_name,
self.add_text_blob_line(scenario_name,
Resource,
Profile,
Amount,
@@ -151,7 +151,16 @@ class CreateChamberview(cv):
self.show_text_blob(None, None, False) # Show changes on GUI
self.apply_cv_scenario(scenario_name) # Apply scenario
self.build_cv_scenario() # build scenario
while self.is_cv_build() == "NO": continue # wait for cv to build
tries = 0
while (True):
if not self.get_cv_is_built():
print("Waiting %i/60 for Chamber-View to be built." % (tries))
tries += 1
if (tries > 60):
break
time.sleep(1)
else:
break
print("completed building %s scenario" %scenario_name)

141
py-scripts/create_chamberview_dut.py Normal file
View File

@@ -0,0 +1,141 @@
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_dut_profile import cv_dut as dut
from cv_test_manager import cv_test as cvtest
class DUT(dut):
def __init__(self,
lfmgr="localhost",
port="8080",
dut_name="DUT",
ssid=[],
password=[],
bssid=[],
security=[]
):
super().__init__(
lfclient_host=lfmgr,
lfclient_port=port,
)
self.cv = cvtest(lfmgr, port)
self.dut_name = dut_name
self.ssid = ssid
self.password = password
self.bssid = bssid
self.security = security
def setup(self):
self.create_dut(dut_name=self.dut_name)
def add_ssids(self):
if self.ssid:
for i in range(len(self.ssid)):
if " " in self.ssid[i][0]:
self.ssid[i][0] = (re.split(' ', self.ssid[i][0]))
elif "," in self.ssid[i][0]:
self.ssid[i][0] = (re.split(',', self.ssid[i][0]))
elif ", " in self.ssid[i][0]:
self.ssid[i][0] = (re.split(',', self.ssid[i][0]))
elif " ," in self.ssid[i][0]:
self.ssid[i][0] = (re.split(',', self.ssid[i][0]))
else:
print("Wrong arguments entered !")
exit(1)
ssid_idx = 0
ssid = "[BLANK]"
passwd = "[BLANK]"
bssid = "00:00:00:00:00:00"
flag = 0x0
for j in range(len(self.ssid[i][0])):
self.ssid[i][0][j] = self.ssid[i][0][j].split("=")
for k in range(len(self.ssid[i][0][j])):
name = self.ssid[i][0][j][k]
if str(name) == "SSID" or str(name) == "ssid" or str(name) == "s":
ssid = self.ssid[i][0][j][k + 1]
elif str(name) == "PASSWORD" or str(name) == "password" or str(name) == "pass":
passwd = self.ssid[i][0][j][k + 1]
elif str(name) == "ssid_idx" or str(name) == "no" or str(name) == "N":
ssid_idx = self.ssid[i][0][j][k + 1]
elif str(name) == "security" or str(name) == "sec":
if self.ssid[i][0][j][k + 1]:
all_flags = self.ssid[i][0][j][k + 1].split("|")
for flags in all_flags:
if flags == "WEP" or flags == "wep":
flag += 0x8
if flags == "WPA" or flags == "wpa":
flag += 0x10
if flags == "WPA2" or flags == "wpa2":
flag += 0x20
if flags == "WPA3" or flags == "wpa3":
flag += 0x100
if flags == "11r":
flag += 0x200
if flags == "EAP-TTLS":
flag += 0x400
if flags == "EAP-PEAP":
flag += 0x800
elif str(name) == "BSSID" or str(name) == "bssid" or str(name) == "B":
bssid = self.ssid[i][0][j][k + 1]
else:
continue
self.add_ssid(dut_name=self.dut_name,
ssid_idx=ssid_idx,
ssid=ssid,
passwd=passwd,
bssid=bssid,
ssid_flags=flag,
ssid_flags_mask=0xFFFFFFFF
)
def main():
parser = argparse.ArgumentParser(
description="""
./create_chamberview_dut -m "localhost" -o "8080" -d "dut_name"
-s "ssid_idx=0 ssid=NET1 security=WPA|WEP|11r|EAP-PEAP bssid=78:d2:94:bf:16:41"
-s "ssid_idx=1 ssid=NET1 security=WPA password=test bssid=78:d2:94:bf:16:40"
""")
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=str, default="8080",
help="IP Port the LANforge GUI is listening on (8080 is default)")
parser.add_argument("-d", "--dut_name", type=str, default="DUT",
help="set dut name")
parser.add_argument("-s", "--ssid", action='append', nargs=1,
help="SSID", default=[])
args = parser.parse_args()
new_dut = DUT(args.lfmgr,
args.port,
args.dut_name,
args.ssid,
)
new_dut.setup()
new_dut.add_ssids()
cv =cvtest(args.lfmgr,
args.port)
cv.show_text_blob(None, None, False) # Show changes on GUI
cv.sync_cv()
time.sleep(2)
if __name__ == "__main__":
main()