scottk/wlan-lanforge-scripts
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-lanforge-scripts.git synced 2025-10-30 02:12:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' of https://github.com/greearb/lanforge-scripts

Browse Source
This commit is contained in:
Logan Lipke
2020-06-02 12:26:19 -07:00
parent 8dfe5b6f10 68647efc9d
commit 65033e1d2b
2 changed files with 78 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

135
py-json/realm.py
View File

@@ -1,55 +1,51 @@
#!/usr/bin/env python3
import os
import sys
import re
import time
sys.path.append('py-json')
import json
import pprint
from LANforge import LFRequest
from LANforge import LFUtils
import re
class Realm:
def __init__(self, mgrURL="http://localhost:8080"):
self.mgrURL = mgrURL
def __init__(self, mgr_url="http://localhost:8080"):
self.mgrURL = mgr_url
def cxList(self):
def cx_list(self):
#Returns json response from webpage of all layer 3 cross connects
lf_r = LFRequest.LFRequest(self.mgrURL + "/cx")
response = lf_r.getAsJson(True)
return response
def stationList(self):
def station_list(self):
#Returns list of all stations with "sta" in their name
list = []
sta_list = []
lf_r = LFRequest.LFRequest(self.mgrURL + "/port/list?fields=_links,alias,device,port+type")
response = lf_r.getAsJson(True)
for x in range(len(response['interfaces'])):
for k,v in response['interfaces'][x].items():
if "sta" in v['device']:
list.append(response['interfaces'][x])
sta_list.append(response['interfaces'][x])
return list
return sta_list
def vapList(self):
def vap_list(self):
#Returns list of all VAPs with "vap" in their name
list = []
sta_list = []
lf_r = LFRequest.LFRequest(self.mgrURL + "/port/list?fields=_links,alias,device,port+type")
response = lf_r.getAsJson(True)
for x in range(len(response['interfaces'])):
for k,v in response['interfaces'][x].items():
if "vap" in v['device']:
list.append(response['interfaces'][x])
sta_list.append(response['interfaces'][x])
return list
return sta_list
def findPortsLike(self, pattern=""):
def find_ports_like(self, pattern=""):
#Searches for ports that match a given pattern and returns a list of names
list = []
device_name_list = []
# alias is possible but device is gauranteed
lf_r = LFRequest.LFRequest(self.mgrURL + "/port/list?fields=_links,alias,device,port+type")
response = lf_r.getAsJson(True)
@@ -57,144 +53,143 @@ class Realm:
for x in range(len(response['interfaces'])):
for k,v in response['interfaces'][x].items():
if v['device'] != "NA":
list.append(v['device'])
device_name_list.append(v['device'])
matchedList = []
matched_list = []
prefix = ""
for portname in list:
for port_name in device_name_list:
try:
if (pattern.index("+") > 0):
if pattern.index("+") > 0:
match = re.search(r"^([^+]+)[+]$", pattern)
if match.group(1):
#print("name:", portname, " Group 1: ",match.group(1))
prefix = match.group(1)
if (portname.index(prefix) == 0):
matchedList.append(portname)
if port_name.index(prefix) == 0:
matched_list.append(port_name)
elif (pattern.index("*") > 0):
elif pattern.index("*") > 0:
match = re.search(r"^([^\*]+)[\*]$", pattern)
if match.group(1):
prefix = match.group(1)
#print("group 1: ",prefix)
if (portname.index(prefix) == 0):
matchedList.append(portname)
if port_name.index(prefix) == 0:
matched_list.append(port_name)
elif (pattern.index("[") > 0):
elif pattern.index("[") > 0:
match = re.search(r"^([^\[]+)\[(\d+)\.\.(\d+)\]$", pattern)
if match.group(0):
#print("[group1]: ", match.group(1))
prefix = match.group(1)
if (portname.index(prefix)):
matchedList.append(portname) # wrong but better
if (port_name.index(prefix)):
matched_list.append(port_name) # wrong but better
except ValueError as e:
print(e)
return matchedList
return matched_list
class CXProfile:
def __init__(self, mgrURL="http://localhost:8080"):
self.mgrURL = mgrURL
self.postData = []
def __init__(self, mgr_url="http://localhost:8080"):
self.mgr_url = mgr_url
self.post_data = []
def addPorts(self, side, endpType, ports=[]):
def add_ports(self, side, endp_type, ports=[]):
#Adds post data for a cross-connect between eth1 and specified list of ports, appends to array
side = side.upper()
endpSideA = {
endp_side_a = {
"alias":"",
"shelf":1,
"resource":1,
"port":"",
"type":endpType,
"type":endp_type,
"min_rate":0,
"max_rate":0,
"min_pkt":-1,
"max_pkt":0
}
endpSideB = {
endp_side_b = {
"alias":"",
"shelf":1,
"resource":1,
"port":"",
"type":endpType,
"type":endp_type,
"min_rate":0,
"max_rate":0,
"min_pkt":-1,
"max_pkt":0
}
for portName in ports:
for port_name in ports:
if side == "A":
endpSideA["alias"] = portName+"CX-A"
endpSideA["port"] = portName
endpSideB["alias"] = portName+"CX-B"
endpSideB["port"] = "eth1"
endp_side_a["alias"] = port_name+"CX-A"
endp_side_a["port"] = port_name
endp_side_b["alias"] = port_name+"CX-B"
endp_side_b["port"] = "eth1"
elif side == "B":
endpSideA["alias"] = portName+"CX-A"
endpSideA["port"] = "eth1"
endpSideB["alias"] = portName+"CX-B"
endpSideB["port"] = portName
endp_side_a["alias"] = port_name+"CX-A"
endp_side_a["port"] = "eth1"
endp_side_b["alias"] = port_name+"CX-B"
endp_side_b["port"] = port_name
lf_r = LFRequest.LFRequest(self.mgrURL + "/cli-json/add_endp")
lf_r.addPostData(endpSideA)
lf_r = LFRequest.LFRequest(self.mgr_url + "/cli-json/add_endp")
lf_r.addPostData(endp_side_a)
json_response = lf_r.jsonPost(True)
lf_r.addPostData(endpSideB)
lf_r.addPostData(endp_side_b)
json_response = lf_r.jsonPost(True)
#LFUtils.debug_printer.pprint(json_response)
time.sleep(.5)
data = {
"alias":portName+"CX",
"alias":port_name+"CX",
"test_mgr":"default_tm",
"tx_endp":portName + "CX-A",
"rx_endp":portName + "CX-B"
"tx_endp":port_name + "CX-A",
"rx_endp":port_name + "CX-B"
}
self.postData.append(data)
self.post_data.append(data)
def create(self, sleepTime=.5):
def create(self, sleep_time=.5):
#Creates cross-connect for each port specified in the addPorts function
for data in self.postData:
lf_r = LFRequest.LFRequest(self.mgrURL + "/cli-json/add_cx")
for data in self.post_data:
lf_r = LFRequest.LFRequest(self.mgr_url + "/cli-json/add_cx")
lf_r.addPostData(data)
json_response = lf_r.jsonPost(True)
#LFUtils.debug_printer.pprint(json_response)
time.sleep(sleepTime)
time.sleep(sleep_time)
class StationProfile:
def __init__(self, mgrURL="localhost:8080", ssid="NA", ssidPass="NA", security="open", startID="", mode=0, up=True, dhcp=True):
self.mgrURL = mgrURL
def __init__(self, mgr_url="localhost:8080", ssid="NA", ssid_pass="NA", security="open", start_id="", mode=0, up=True, dhcp=True):
self.mgrURL = mgr_url
self.ssid = ssid
self.ssidPass = ssidPass
self.ssid_pass = ssid_pass
self.mode = mode
self.up = up
self.dhcp = dhcp
self.security = security
def build(self, resourceRadio, numStations):
def build(self, resource_radio, num_stations):
#Checks for errors in initialization values and creates specified number of stations using init parameters
try:
resource = port_name[0 : resourceRadio.index(".")]
name = port_name[resourceRadio.index(".")+1 : ]
if (name.index(".") >= 0):
resource = port_name[0: resource_radio.index(".")]
name = port_name[resource_radio.index(".") + 1:]
if name.index(".") >= 0:
name = name[name.index(".")+1 : ]
except ValueError as e:
print(e)
lf_r = LFRequest.LFRequest(self.mgrURL + "/cli-json/add_sta")
flags = 0
for num in range(numStations):
for num in range(num_stations):
data = {
"shelf":1,
"resource":1,
"radio":radio,
"sta_name":f"sta{num:05}",
"ssid":self.ssid,
"key":self.ssidPass,
"key":self.ssid_pass,
"mode":1,
"mac":"xx:xx:xx:xx:*:xx",
"flags":

26
py-json/realm_test.py
View File

@@ -1,26 +1,26 @@
#!/usr/bin/env python3
import Realm
import realm
test = Realm.Realm("http://localhost:8080")
test = realm.Realm("http://localhost:8080")
staList = test.stationList()
cxList = test.cxList()
vapList = test.vapList()
sta_list = test.station_list()
cx_list = test.cx_list()
vap_list = test.vap_list()
print(f"CXs: {cxList}\n")
print(f"Stations: {staList}\n")
print(f"VAPs: {vapList}\n")
print(f"CXs: {cx_list}\n")
print(f"Stations: {sta_list}\n")
print(f"VAPs: {vap_list}\n")
cxTest = Realm.CXProfile()
cxTest = realm.CXProfile()
cxTest.addPorts("A", "lf_udp", test.findPortsLike("sta+"))
cxTest.add_ports("A", "lf_udp", test.find_ports_like("sta+"))
cxTest.create()
print(test.findPortsLike("sta+"))
print(test.find_ports_like("sta+"))
print(test.findPortsLike("sta0*"))
print(test.find_ports_like("sta0*"))
print(test.findPortsLike("sta[0000..0002]"))
print(test.find_ports_like("sta[0000..0002]"))