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

Json: create_wanlink.py example script fully working demonstrating listing creation and toggling

Browse Source
This commit is contained in:
Jed Reynolds
2019-12-31 12:58:38 -08:00
parent 8a8211fdad
commit f8f91e52a4
1 changed files with 121 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

149
py-json/create_wanlink.py
View File

@@ -4,12 +4,15 @@
# -
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
import sys
import urllib
if sys.version_info[0] != 3:
print("This script requires Python 3")
exit()
import time
from time import sleep
from urllib import error
import pprint
import LANforge
from LANforge import LFRequest
@@ -24,30 +27,41 @@ def main():
base_url = "http://localhost:8080"
json_post = ""
json_response = ""
num_wanlinks = -1
# see if there are old wanlinks to remove
lf_r = LFRequest.LFRequest(base_url+"/wl/list")
try:
json_response = lf_r.getAsJson()
LFUtils.debug_printer.pprint(json_response)
for key,value in json_response.items():
if (isinstance(value, dict) and "_links" in value):
num_wanlinks =+ 1
except urllib.error.HTTPError as error:
num_wanlinks = 0;
# remove old wanlinks
lf_r = LFRequest.LFRequest(base_url+"/cli-json/rm_cx")
lf_r.addPostData({
'test_mgr': 'all',
'cx_name': 'wl_eg1'
})
lf_r.jsonPost()
sleep(0.05)
if (num_wanlinks > 0):
lf_r = LFRequest.LFRequest(base_url+"/cli-json/rm_cx")
lf_r.addPostData({
'test_mgr': 'all',
'cx_name': 'wl_eg1'
})
lf_r.jsonPost()
sleep(0.05)
lf_r = LFRequest.LFRequest(base_url+"/cli-json/rm_endp")
lf_r.addPostData({
'endp_name': 'wl_eg1-A'
})
lf_r.jsonPost()
sleep(0.05)
lf_r = LFRequest.LFRequest(base_url+"/cli-json/rm_endp")
lf_r.addPostData({
'endp_name': 'wl_eg1-A'
})
lf_r.jsonPost()
sleep(0.05)
lf_r = LFRequest.LFRequest(base_url+"/cli-json/rm_endp")
lf_r.addPostData({
'endp_name': 'wl_eg1-B'
})
lf_r.jsonPost()
sleep(0.05)
lf_r = LFRequest.LFRequest(base_url+"/cli-json/rm_endp")
lf_r.addPostData({
'endp_name': 'wl_eg1-B'
})
lf_r.jsonPost()
sleep(0.05)
# create wanlink 1a
lf_r = LFRequest.LFRequest(base_url+"/cli-json/add_wl_endp")
@@ -88,14 +102,62 @@ def main():
lf_r.jsonPost()
sleep(0.05)
# start wanlink
# start wanlink once we see it
seen = 0
while (seen < 1):
sleep(1)
lf_r = LFRequest.LFRequest(base_url+"/wl/wl_eg1?fields=name,eid,state,_links")
try:
json_response = lf_r.getAsJson()
if (json_response is None):
continue
LFUtils.debug_printer.pprint(json_response)
for key,value in json_response.items():
if (isinstance(value, dict)):
if ("_links" in value):
if (value["name"] == "wl_eg1"):
seen = 1
#else:
# print(" name was not wl_eg1")
#else:
# print("value lacks _links")
#else:
# print("value not a dict")
except urllib.error.HTTPError as error:
print("Error code "+error.code)
continue
print("starting wanlink:")
lf_r = LFRequest.LFRequest(base_url+"/cli-json/set_cx_state")
lf_r.addPostData({
'test_mgr' = 'all',
'cx_name' = 'wl_eg1',
'cx_state' = 'RUNNING'
'test_mgr': 'all',
'cx_name': 'wl_eg1',
'cx_state': 'RUNNING'
})
lf_r.jsonPost()
running = 0
while (running < 1):
sleep(1)
lf_r = LFRequest.LFRequest(base_url+"/wl/wl_eg1?fields=name,eid,state,_links")
try:
json_response = lf_r.getAsJson()
if (json_response is None):
continue
for key,value in json_response.items():
if (isinstance(value, dict)):
if ("_links" in value):
if (value["name"] == "wl_eg1"):
if (value["state"].startswith("Run")):
LFUtils.debug_printer.pprint(json_response)
running = 1
except urllib.error.HTTPError as error:
print("Error code "+error.code)
continue
print("Wanlink is running, wait one sec...")
sleep(1)
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
@@ -119,15 +181,46 @@ def main():
# stop wanlink
lf_r = LFRequest.LFRequest(base_url+"/cli-json/set_cx_state")
lf_r.addPostData({
'test_mgr' = 'all',
'cx_name' = 'wl_eg1',
'cx_state' = 'STOPPED'
'test_mgr': 'all',
'cx_name': 'wl_eg1',
'cx_state': 'STOPPED'
})
lf_r.jsonPost()
running = 1
while (running > 1):
sleep(1)
lf_r = LFRequest.LFRequest(base_url+"/wl/wl_eg1?fields=name,eid,state,_links")
LFUtils.debug_printer.pprint(json_response)
try:
json_response = lf_r.getAsJson()
if (json_response is None):
continue
for key,value in json_response.items():
if (isinstance(value, dict)):
if ("_links" in value):
if (value["name"] == "wl_eg1"):
if (value["state"].startswith("Stop")):
LFUtils.debug_printer.pprint(json_response)
running = 0
except urllib.error.HTTPError as error:
print("Error code "+error.code)
continue
print("Wanlink is stopped.")
print("Wanlink info:")
json_response = LFResponse.LFRequest(base_url+"/")
lf_r = LFRequest.LFRequest(base_url+"/wl/wl_eg1")
json_response = lf_r.getAsJson()
LFUtils.debug_printer.pprint(json_response)
lf_r = LFRequest.LFRequest(base_url+"/wl_ep/wl_eg1-A")
json_response = lf_r.getAsJson()
LFUtils.debug_printer.pprint(json_response)
lf_r = LFRequest.LFRequest(base_url+"/wl_ep/wl_eg1-B")
json_response = lf_r.getAsJson()
LFUtils.debug_printer.pprint(json_response)
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
if __name__ == '__main__':