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

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

Browse Source
This commit is contained in:
Dipti Dhond
2020-06-26 15:01:45 -07:00
parent 52f3f6778b 7c1503d2fd
commit 8525229f23
10 changed files with 674 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
gui/kpi.java
View File

@@ -612,6 +612,7 @@ public class kpi {
BufferedWriter bw = new BufferedWriter(new FileWriter(ofile));
bw.write("CASE_ID " + caseid);
bw.write(System.lineSeparator());
bw.write("RUN_ID " + i);
bw.write(System.lineSeparator());
bw.write(testrails_msg);

37
py-json/LANforge/LFRequest.py
View File

@@ -23,8 +23,9 @@ class LFRequest:
default_headers = {
'Accept': 'application/json'}
def __init__(self, url, uri=None, debug_=False):
def __init__(self, url, uri=None, debug_=False, die_on_error_=False):
self.debug = debug_
self.die_on_error = die_on_error_;
if not url.startswith("http://") and not url.startswith("https://"):
print("No http:// or https:// found, prepending http:// to "+url)
@@ -45,7 +46,7 @@ class LFRequest:
raise Exception("Bad LFRequest of url[%s] uri[%s] -> None" % url, uri)
# request first url on stack
def formPost(self, show_error=True, debug=False):
def formPost(self, show_error=True, debug=False, die_on_error_=False):
if (debug == False) and (self.debug == True):
debug = True;
responses = []
@@ -72,7 +73,7 @@ class LFRequest:
return responses[0]
except urllib.error.HTTPError as error:
if (show_error):
print("----- formPost() HTTPError: --------------------------------------------")
print("----- LFRequest::formPost:75 HTTPError: --------------------------------------------")
print("%s: %s; URL: %s"%(error.code, error.reason, request.get_full_url()))
LFUtils.debug_printer.pprint(error.headers)
#print("Error: ", sys.exc_info()[0])
@@ -86,15 +87,19 @@ class LFRequest:
LFUtils.debug_printer.pprint(responses[0].reason)
print("------------------------------------------------------------------------")
if (die_on_error_ == True) or (self.die_on_error == True):
exit(1)
except urllib.error.URLError as uerror:
if (show_error):
print("----- formPost() URLError: ---------------------------------------------")
print("----- LFRequest::formPost:91 URLError: ---------------------------------------------")
print("Reason: %s; URL: %s"%(uerror.reason, request.get_full_url()))
print("------------------------------------------------------------------------")
if (die_on_error_ == True) or (self.die_on_error == True):
exit(1)
return None
def jsonPost(self, show_error=True, debug=False):
def jsonPost(self, show_error=True, debug=False, die_on_error_=False):
if (debug == False) and (self.debug == True):
debug = True
responses = []
@@ -113,7 +118,7 @@ class LFRequest:
return responses[0]
except urllib.error.HTTPError as error:
if show_error:
print("----- jsonPost() HTTPError: --------------------------------------------")
print("----- LFRequest::jsonPost:116 HTTPError: --------------------------------------------")
print("<%s> HTTP %s: %s"%(request.get_full_url(), error.code, error.reason, ))
print("Error: ", sys.exc_info()[0])
@@ -133,14 +138,18 @@ class LFRequest:
print("----- Response: --------------------------------------------------------")
LFUtils.debug_printer.pprint(responses[0].reason)
print("------------------------------------------------------------------------")
if (die_on_error_ == True) or (self.die_on_error == True):
exit(1)
except urllib.error.URLError as uerror:
if show_error:
print("----- jsonPost() URLError: ---------------------------------------------")
print("----- LFRequest::jsonPost:138 URLError: ---------------------------------------------")
print("Reason: %s; URL: %s"%(uerror.reason, request.get_full_url()))
print("------------------------------------------------------------------------")
if (die_on_error_ == True) or (self.die_on_error == True):
exit(1)
return None
def get(self, show_error=True, debug=False):
def get(self, show_error=True, debug=False, die_on_error_=False):
if (debug == False) and (self.debug == True):
debug = True
if (debug):
@@ -152,7 +161,7 @@ class LFRequest:
return myresponses[0]
except urllib.error.HTTPError as error:
if show_error:
print("----- get() HTTPError: --------------------------------------------")
print("----- LFRequest::get:155 HTTPError: --------------------------------------------")
print("<%s> HTTP %s: %s"%(myrequest.get_full_url(), error.code, error.reason, ))
if error.code != 404:
print("Error: ", sys.exc_info()[0])
@@ -172,16 +181,20 @@ class LFRequest:
print("----- Response: --------------------------------------------------------")
LFUtils.debug_printer.pprint(myresponses[0].reason)
print("------------------------------------------------------------------------")
if (die_on_error_ == True) or (self.die_on_error == True):
exit(1)
except urllib.error.URLError as uerror:
if show_error:
print("----- get() URLError: ---------------------------------------------")
print("----- LFRequest::get:177 URLError: ---------------------------------------------")
print("Reason: %s; URL: %s"%(uerror.reason, myrequest.get_full_url()))
print("------------------------------------------------------------------------")
if (die_on_error_ == True) or (self.die_on_error == True):
exit(1)
return None
def getAsJson(self, show_error=True):
def getAsJson(self, show_error=True, die_on_error_=False):
responses = []
responses.append(self.get(show_error))
responses.append(self.get(show_error, die_on_error_=die_on_error_))
if (len(responses) < 1):
return None
if (responses[0] == None):

43
py-json/LANforge/lfcli_base.py
View File

@@ -19,9 +19,10 @@ class LFCliBase:
self.lfclient_host = _lfjson_host
self.lfclient_port = _lfjson_port
self.debug = _debug
self.haltOnError = _halt_on_error
self.lfclient_url = "http://%s:%s" % (self.lfclient_host, self.lfclient_port)
self.test_results = []
self.halt_on_error = _halt_on_error
self.exit_on_error = _exit_on_error
self.exit_on_fail = _exit_on_fail
# toggle using preexec_cli, preexec_method; the preexec_X parameters are useful
@@ -34,47 +35,47 @@ class LFCliBase:
def clear_test_results(self):
self.test_results.clear()
def json_post(self, _req_url, _data):
def json_post(self, _req_url, _data, debug_=False, use_preexec_=True):
json_response = None
try:
lf_r = LFRequest.LFRequest(self.lfclient_url, _req_url, debug_=self.debug)
if self.use_preexec == True:
del _data['suppress_preexec_cli']
del _data['suppress_preexec_method']
lf_r = LFRequest.LFRequest(self.lfclient_url, _req_url, debug_=self.debug, die_on_error_=self.exit_on_error)
if self.use_preexec or use_preexec_:
_data['suppress_preexec_cli'] = False
_data['suppress_preexec_method'] = False
else:
_data['suppress_preexec_cli'] = True
_data['suppress_preexec_method'] = True
lf_r.addPostData(_data)
if (self.debug):
if debug_ or self.debug:
LANforge.LFUtils.debug_printer.pprint(_data)
json_response = lf_r.jsonPost(show_error=self.debug, debug=self.debug)
json_response = lf_r.jsonPost(show_error=self.debug, debug=(self.debug or debug_), die_on_error_=self.exit_on_error)
except Exception as x:
if self.debug or self.haltOnError:
if self.debug or self.halt_on_error or self.exit_on_error:
print("jsonPost posted to %s" % _req_url)
pprint(_data)
pprint.pprint(_data)
print("Exception %s:" % x)
traceback.print_exception(Exception, x, x.__traceback__, chain=True)
if self.haltOnError:
if self.halt_on_error or self.exit_on_error:
exit(1)
return json_response
def json_get(self, _req_url):
if self.debug:
def json_get(self, _req_url, debug_=False):
if self.debug or debug_:
print("URL: "+_req_url)
json_response = None
try:
lf_r = LFRequest.LFRequest(self.lfclient_url, _req_url)
lf_r = LFRequest.LFRequest(self.lfclient_url, _req_url, debug_=(self.debug or debug_), die_on_error_=self.exit_on_error)
json_response = lf_r.getAsJson(self.debug)
#debug_printer.pprint(json_response)
if (json_response is None) and self.debug:
if (json_response is None) and (self.debug or debug_):
raise ValueError(json_response)
except ValueError as ve:
if self.debug or self.haltOnError:
if self.debug or self.halt_on_error or self.exit_on_error:
print("jsonGet asked for " + _req_url)
print("Exception %s:" % ve)
traceback.print_exception(ValueError, ve, ve.__traceback__, chain=True)
if self.haltOnError:
if self.halt_on_error or self.exit_on_error:
sys.exit(1)
return json_response
@@ -115,7 +116,7 @@ class LFCliBase:
# print("lfcli_base error: %s" % exception)
pprint.pprint(exception)
traceback.print_exception(Exception, exception, exception.__traceback__, chain=True)
if self.haltOnError:
if self.halt_on_error:
print("halting on error")
sys.exit(1)
# else:
@@ -124,13 +125,13 @@ class LFCliBase:
def check_connect(self):
if self.debug:
print("Checking for LANforge GUI connection: %s" % self.lfclient_url)
response = self.json_get("/")
response = self.json_get("/", debug_=self.debug)
duration = 0
while (response is None) and (duration < 300):
print("LANforge GUI connection not found sleeping 5 seconds, tried: %s" % self.lfclient_url)
duration += 2
time.sleep(2)
response = self.json_get("")
response = self.json_get("", debug_=self.debug)
if duration >= 300:
print("Could not connect to LANforge GUI")
@@ -170,7 +171,7 @@ class LFCliBase:
self.test_results.append(self.fail_pref + message)
if print_:
print(self.fail_pref + message)
if self.exit_on_error:
if self.exit_on_fail:
sys.exit(1)
# use this inside the class to log a pass result

121
py-json/README.md
View File

@@ -2,24 +2,126 @@
Similar to the JSON sibling directory that provides Perl JSON adapters
with the LANforge client, this directory provides Python adapters to
the LANforge client. The LANforge client is the LANforge Java GUI
the LANforge client. The LANforge client is the LANforge Java GUI
package running with the `-http` switch (by default) and possibly in the *headless*
mode using the `-daemon` switch.
Follow our [getting started cookbook](http://www.candelatech.com/cookbook.php?vol=cli&book=Querying+the+LANforge+GUI+for+JSON+Data)
Follow our [getting started cookbook](http://www.candelatech.com/cookbook.php?vol=cli&book=Querying+the+LANforge+GUI+for+JSON+Data)
to learn more about how to operate your LANforge client.
## Getting Started ##
New automation tests and JSON client scripts should go in `../py-scripts`. This directory
is intended for utility and library scripts. To use this module, make sure your include path
captures this module by adding it to your `sys.path`. We recommend your scripts in `../py-scripts`
begin with these imports:
`if 'py-json' not in sys.path:
sys.path.append('../py-json')
from LANforge import LFUtils
from LANforge import lfcli_base
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
import realm
from realm import Realm
## These Scripts ##
* `show_ports.py`: this simple example shows how to gather a digest of ports
* `__init__.py`: this is a module header and it defines its relationship to sub-module LANforge,
requiring LFRequest.
* `LANforge`: this module is for our json library. Use gain access to these by using:
`import LANforge`
`from LANforge import LFUtils`
`from LANforge import LFRequest`
* `create_sta.py`: Please follow though `create_sta.py` to see how you can
utilize the JSON API provided by the LANforge client. It
is possible to use similar commands to create virtual Access points.
* `create_wanlink.py`: example that creates a WANlink
* `generic_cx.py`: example that creates a cross connect
* `realm.py`: module defining the Realm class. `Realm` is a toolbox class that also serves as a facade
for finer-grained methods in LFUtils and LFRequest:
* `__init__`: our constructor
* `load()`: load a test scenario database, as you would find in the GUI Status tab
* `cx_list()`: request json list of cross connects
* `station_map()`: request a map of stations via `/port/list` and alter the list to name based map of only stations
* `station_list()`: request a list of stations
* `vap_list()`: request a list of virtual APs
* `remove_vlan_by_eid()`: a way of deleting a port/station/vAP
* `find_ports_like()`: returns a list of ports matching a string prefix, like:
* `sta\*` matches names starting with `sta`
* `sta10+` matches names with port numbers 10 or greater
* `sta[10..20]` matches a range of stations including the range sta10 -- sta20
* `name_to_eid()`: takes a name like `1.1.eth1` and returns it split into an array `[1, 1, "eth1"]`
* `parse_time()`: returns numeric seconds when given strings like `1d`, `2h`, or `3m` or `4s`
* `parse_link()`:
* `new_station_profile()`: creates a blank station profile, configure station properties in this profile
and then use its `create()` method to create a series of stations
* `new_l3_cx_profile()`: creates a blank Layer-3 profile, configure this connection profile and
then use its `create()` method to create a series of endpoints and cross connects
* `new_l4_cx_profile()`: creates a blank Layer-4 (http/ftp) profile, configure it then call `create()`
* `new_generic_cx_profile()`: creates a blank Generic connection profile (for lfping/iperf3/curl-post/speedtest.net)
then configure and call `create()`
* class `L3CXProfile`: this class is the Layer-3 connection profile **unfinished**
* `__init__`: should be called by `Realm::new_l3_cx_profile()`
* `create()`: pass endpoint-type, side-a list, side-b list, and sleep_time between creating endpoints and connections
* Parameters for this profile include:
* prefix
* txbps
* class `L4CXProfile`: this class is the Layer-4 connection profile **unfinished**
* `__init__`: should be called by `Realm::new_l4_cx_profile()`
* `create()`: pass a list of ports to create endpoints on, note that resulting cross connects are prefixed with `CX_`
* Parameters for this profile include:
* url
* requests_per_ten: number of requests to make in ten minutes
* class `GenCXProfile`: this class is the Generic connection profile **unfinished**
* `__init__`: should be called by `Realm::new_gen_cx_profile()`
* `create()`: pass a list of ports to create connections on
* Parameters for this profile include:
* type: includes lfping, iperf3, speedtest, lfcurl or cmd
* dest: IP address of destination for command
* class `StationProfile`: configure instances of this class for creating series of ports
* `__init__`: should be called by `Realm::new_station_profile()`
* `use_wpa2()`: pass on=True,ssid=a,passwd,b to set station_command_param add_sta/ssid, add_sta_key
pass on=False,ssid=a to turn off command_flag add_sta/flags/wpa2_enable
* `set_command_param()`
* `set_command_flag()`
* `set_prefix()`
* `add_named_flags()`
* `create()`: you can use either an integer number of stations or a list of station names, if you want to create a
specific range of ports, create the names first and do not specify `num_stations`
* resource: the resource number for the radio
* radio: name of the radio, like 'wiphy0'
* num_stations: `value > 0` indicates creating station series `sta0000..sta$value`
* sta_names_: a list of station names to create, please use `LFUtils.port_name_series()`
* dry_run: True avoids posting commands
* debug:
* `realm_test.py`: exercises realm.py
* `show_ports.py`: this simple example shows how to gather a digest of ports
* `test_l4.py`: example of how to use LFRequest to create a L4 endpoint
* `wct-example.py`: example of using expect on port 3990 to operate a WiFi Capacity Test
* `ws-sta-monitor.py`: websocket 8081 client that filters interesting station events from the lfclient websocket
## LANforge ##
This directory defines the LANforge module holding:
This directory defines the LANforge module holding the following classes:
* lfcli_base.py / class **LFCliBase**: This is a base class we encourage using for creating tests and
other automation scripts. It provides a centralized manner for making uniform JSON GET and POST
calls.
* `__init__`: call this from your classes __init__ method as super().__init__(...) like below:
* LFRequest: provides default mechanism to make API queries, use this
class MyScript(LFCliBase):
def __init__(self, host, port, debug_=False, _exit_on_error=False, _exit_on_fail=False):
super().__init__(host, port, _debug=debug_, _halt_on_error=_exit_on_error, _exit_on_fail=_exit_on_fail)
Those parameters provide base functionality:
* host: lfclient host running the LANforge GUI or headless LANforgeGUI -daemon
* port: lfclient HTTP port, typically 8080
* _debug: provides verbose mode behavior
* _halt_on_error: if a HTTP 400 or HTTP 500 occurs or some execeptions are raised, exit
* _exit_on_fail: if a test calls _fail(), exit
* LFRequest.py / class **LFRequest**: provides default mechanism to make API queries, use this
to create most of your API requests, but you may also use the normal
`urllib.request` library on simple GET requests if you wish.
* formPost(): post data in url-encoded format
@@ -28,7 +130,7 @@ This directory defines the LANforge module holding:
* getAsJson(): converts get() JSON results into python objects
* addPostData(): provide a dictionary to this method before calling formPost() or jsonPost()
* LFUtils: defines constants and utility methods
* LFUtils.py / class **LFUtils**: defines constants and utility methods
* class PortEID: convenient handle for port objects
* newStationDownRequest(): create POST data object for station down
* portSetDhcpDownRequest(): create POST data object for station down, apply `use_dhcp` flags
@@ -42,8 +144,13 @@ This directory defines the LANforge module holding:
* findPortEids(): returns EIDs of ports
* waitUntilPortsAdminDown(): watch ports until they report admin down
* waitUntilPortsAdminUp(): watch ports until they report admin up
* waitUntilPortsDisappear(): use this after deleting ports
* wait_until_ports_disappear(): use this after deleting ports
* ~~waitUntilPortsDisappear()~~: use this after deleting ports, **deprecated**
* waitUntilPortsAppear(): use this after `add_sta` or `set_port`
* removePort(): remove a port using rm_vlan command
* removeCX(): request a list of CX names be removed
* removeEndps(): request a list of endpoint names be removed
* execWrap(): hair trigger method that exits when a command fails when called by os.system()
Have fun coding!

122
py-json/realm.py
View File

@@ -10,6 +10,7 @@ from LANforge import add_sta
from LANforge import lfcli_base
from LANforge.lfcli_base import LFCliBase
from generic_cx import GenericCx
import datetime
class Realm(LFCliBase):
@@ -21,7 +22,7 @@ class Realm(LFCliBase):
# loads a database
def load(self, name):
if (name is None) or (name is ""):
if (name is None) or (name == ""):
raise ValueError("Realm::load: wants a test scenario database name, please find one in the Status tab of the GUI")
data = {
@@ -30,7 +31,7 @@ class Realm(LFCliBase):
"clean_dut":"yes",
"clean_chambers": "yes"
}
self.json_post("/cli-json/load")
self.json_post("/cli-json/load", debug_=self.debug)
time.sleep(1)
# Returns json response from webpage of all layer 3 cross connects
@@ -174,48 +175,61 @@ class Realm(LFCliBase):
info = eid.split('.')
return info
def parse_time(self, time_string):
if isinstance(time_string, str):
pattern = re.compile("^(\d+)([dhms]$)")
td = pattern.match(time_string)
if td is not None:
dur_time = int(td.group(1))
dur_measure = str(td.group(2))
if dur_measure == "d":
duration_time = datetime.timedelta(days=dur_time)
elif dur_measure == "h":
duration_time = datetime.timedelta(hours=dur_time)
elif dur_measure == "m":
duration_time = datetime.timedelta(minutes=dur_time)
else:
duration_time = datetime.timedelta(seconds=dur_time)
else:
raise ValueError("Unknown value for time_string: %s" % time_string)
else:
raise ValueError("time_string must be of type str. Type %s provided" % type(time_string))
return duration_time
def parse_link(self, link):
link = self.lfclient_url + link
info = ()
def new_station_profile(self):
station_prof = StationProfile(self.lfclient_url, debug_=self.debug)
return station_prof
def new_l3_cx_profile(self):
cx_prof = L3CXProfile(self.lfclient_host, self.lfclient_port, debug_=self.debug)
cx_prof = L3CXProfile(self.lfclient_host, self.lfclient_port, local_realm=self, debug_=self.debug)
return cx_prof
def new_l4_cx_profile(self):
cx_prof = L4CXProfile(self.lfclient_host, self.lfclient_port, debug_=self.debug)
cx_prof = L4CXProfile(self.lfclient_host, self.lfclient_port, local_realm=self, debug_=self.debug)
return cx_prof
def new_generic_cx_profile(self):
cx_prof = GenCXProfile(self.lfclient_host, self.lfclient_port, debug_=self.debug)
cx_prof = GenCXProfile(self.lfclient_host, self.lfclient_port,local_realm=self, debug_=self.debug)
return cx_prof
class L3CXProfile(LFCliBase):
def __init__(self, lfclient_host, lfclient_port, debug_=False):
def __init__(self, lfclient_host, lfclient_port, local_realm, debug_=False):
super().__init__(lfclient_host, lfclient_port, debug_, _halt_on_error=True)
self.lfclient_url = "http://%s:%s" % (lfclient_host, lfclient_port)
self.debug = debug_
def name_to_eid(self, eid):
info = []
if eid is None or eid == "":
raise ValueError("name_to_eid wants eid like 1.1.sta0 but given[%s]" % eid)
else:
if '.' in eid:
info = eid.split('.')
return info
self.local_realm = local_realm
def create(self, endp_type, side_a, side_b, sleep_time=.5):
post_data = []
if type(side_a) == list and type(side_b) != list:
side_b_info = self.name_to_eid(side_b)
side_b_info = self.local_realm.name_to_eid(side_b)
if len(side_b_info) == 3:
side_b_shelf = side_b_info[0]
side_b_resource = side_b_info[1]
@@ -228,7 +242,7 @@ class L3CXProfile(LFCliBase):
raise ValueError("side_b must have a shelf and/or resource number")
for port_name in side_a:
side_a_info = self.name_to_eid(port_name)
side_a_info = self.local_realm.name_to_eid(port_name)
if len(side_a_info) == 3:
side_a_shelf = side_a_info[0]
side_a_resource = side_a_info[1]
@@ -239,7 +253,7 @@ class L3CXProfile(LFCliBase):
side_a_name = side_a_info[1]
endp_side_a = {
"alias": port_name + "-A",
"alias": side_a_name + "-A",
"shelf": side_a_shelf,
"resource": side_a_resource,
"port": side_a_name,
@@ -250,7 +264,7 @@ class L3CXProfile(LFCliBase):
"max_pkt": 0
}
endp_side_b = {
"alias": port_name + "-B",
"alias": side_a_name + "-B",
"shelf": side_b_shelf,
"resource": side_b_resource,
"port": side_b_name,
@@ -261,21 +275,21 @@ class L3CXProfile(LFCliBase):
"max_pkt": 0
}
url = self.lfclient_url + "/cli-json/add_endp"
LFCliBase.json_post(self, url, endp_side_a)
LFCliBase.json_post(self, url, endp_side_b)
url = "/cli-json/add_endp"
self.local_realm.json_post(url, endp_side_a)
self.local_realm.json_post(url, endp_side_b)
time.sleep(sleep_time)
data = {
"alias": port_name + "CX",
"alias": self.local_realm.name_to_eid(port_name)[-1] + "CX",
"test_mgr": "default_tm",
"tx_endp": port_name + "CX-A",
"rx_endp": port_name + "CX-B"
"tx_endp": side_a_name + "-A",
"rx_endp": side_a_name + "-B"
}
post_data.append(data)
elif type(side_b) == list and type(side_a) != list:
side_a_info = self.name_to_eid(side_a)
side_a_info = self.local_realm.name_to_eid(side_a)
if len(side_a_info) == 3:
side_a_shelf = side_a_info[0]
side_a_resource = side_a_info[1]
@@ -288,8 +302,7 @@ class L3CXProfile(LFCliBase):
raise ValueError("side_a must have a shelf and/or resource number")
for port_name in side_b:
side_b_info = self.name_to_eid(port_name)
print(side_b_info)
side_b_info = self.local_realm.name_to_eid(port_name)
if len(side_b_info) == 3:
side_b_shelf = side_b_info[0]
side_b_resource = side_b_info[1]
@@ -299,7 +312,7 @@ class L3CXProfile(LFCliBase):
side_b_resource = side_b_info[0]
side_b_name = side_b_info[1]
endp_side_a = {
"alias": port_name + "-A",
"alias": side_b_name + "-A",
"shelf": side_a_shelf,
"resource": side_a_resource,
"port": side_a_name,
@@ -310,7 +323,7 @@ class L3CXProfile(LFCliBase):
"max_pkt": 0
}
endp_side_b = {
"alias": port_name + "-B",
"alias": side_b_name + "-B",
"shelf": side_b_shelf,
"resource": side_b_resource,
"port": side_b_name,
@@ -321,24 +334,25 @@ class L3CXProfile(LFCliBase):
"max_pkt": 0
}
url = self.lfclient_url + "/cli-json/add_endp"
LFCliBase.json_post(self, url, endp_side_a)
LFCliBase.json_post(self, url, endp_side_b)
url = "/cli-json/add_endp"
self.local_realm.json_post(url, endp_side_a)
self.local_realm.json_post(url, endp_side_b)
time.sleep(sleep_time)
data = {
"alias": port_name + "CX",
"alias": self.local_realm.name_to_eid(port_name)[-1] + "CX",
"test_mgr": "default_tm",
"tx_endp": port_name + "CX-A",
"rx_endp": port_name + "CX-B"
"tx_endp": side_b_name + "-A",
"rx_endp": side_b_name + "-B"
}
post_data.append(data)
else:
raise ValueError("side_a or side_b must be of type list but not both: side_a is type %s side_b is type %s" % (type(side_a), type(side_b)))
print("post_data", post_data)
for data in post_data:
url = self.lfclient_url + "/cli-json/add_cx"
LFCliBase.json_post(self, url, data)
url = "/cli-json/add_cx"
self.local_realm.json_post(url, data)
time.sleep(sleep_time)
def to_string(self):
@@ -348,18 +362,19 @@ class L3CXProfile(LFCliBase):
class L4CXProfile(LFCliBase):
def __init__(self, lfclient_host, lfclient_port, debug_=False):
def __init__(self, lfclient_host, lfclient_port, local_realm,debug_=False):
super().__init__(lfclient_host, lfclient_port, debug_, _halt_on_error=True)
self.lfclient_url = "http://%s:%s" % (lfclient_host, lfclient_port)
self.debug = debug_
self.url = "http://localhost/"
self.requests_per_ten = 600
self.local_realm = local_realm
def create(self, ports=[], sleep_time=.5):
post_data = []
for port_name in ports:
data = {
"alias": port_name + "_l4",
"alias": self.local_realm.name_to_eid(port_name)[-1] + "_l4",
"shelf": 1,
"resource": 1,
"port": port_name,
@@ -368,26 +383,26 @@ class L4CXProfile(LFCliBase):
"url_rate": self.requests_per_ten,
"url": self.url
}
url = self.lfclient_url + "cli-json/add_l4_endp"
LFCliBase.json_post(self, url, data)
url = "cli-json/add_l4_endp"
self.local_realm.json_post(url, data)
time.sleep(sleep_time)
data = {
"alias": port_name + "_l4CX",
"alias": self.local_realm.name_to_eid(port_name)[-1] + "_l4CX",
"test_mgr": "default_tm",
"tx_endp": port_name + "_l4",
"tx_endp": self.local_realm.name_to_eid(port_name)[-1] + "_l4",
"rx_endp": "NA"
}
post_data.append(data)
for data in post_data:
url = self.lfclient_url + "/cli-json/add_cx"
LFCliBase.json_post(self, url, data)
url = "/cli-json/add_cx"
self.local_realm.json_post(url, data)
time.sleep(sleep_time)
class GenCXProfile(LFCliBase):
def __init__(self, lfclient_host, lfclient_port, debug_=False):
def __init__(self, lfclient_host, lfclient_port, local_realm, debug_=False):
super().__init__(lfclient_host, lfclient_port, debug_, _halt_on_error=True)
self.lfclient_host = lfclient_host
self.lfclient_port = lfclient_port
@@ -398,12 +413,13 @@ class GenCXProfile(LFCliBase):
self.interval = 1
self.count = 2
self.cmd = ""
self.local_realm = local_realm
def create(self, ports=[], sleep_time=.5):
post_data = []
for port_name in ports:
gen_name = port_name + "_gen"
gen_name2 = port_name + "_gen"
gen_name = self.local_realm.name_to_eid(port_name)[-1] + "_gen0"
gen_name2 = self.local_realm.name_to_eid(port_name)[-1] + "_gen1"
genl = GenericCx(lfclient_host=self.lfclient_host, lfclient_port=self.lfclient_port)
genl.createGenEndp(gen_name, 1, 1, port_name, "gen_generic")
genl.createGenEndp(gen_name2, 1, 1, port_name, "gen_generic")
@@ -414,7 +430,7 @@ class GenCXProfile(LFCliBase):
time.sleep(sleep_time)
data = {
"alias": port_name + "_gen_CX",
"alias": self.local_realm.name_to_eid(port_name)[-1] + "_gen_CX",
"test_mgr": "default_tm",
"tx_endp": gen_name,
"rx_endp": gen_name2
@@ -422,8 +438,8 @@ class GenCXProfile(LFCliBase):
post_data.append(data)
for data in post_data:
url = self.lfclient_url + "/cli-json/add_cx"
LFCliBase.json_post(self, url, data)
url = "/cli-json/add_cx"
self.local_realm.json_post(url, data)
time.sleep(sleep_time)

30
py-scripts/README.txt
View File

@@ -1,3 +1,33 @@
This directory will contain python scripts useful for unit-tests. It uses
libraries in ../py-json
Scripts included are:
__init__.py:
cicd_TipIntegration.py:
cicd_testrail.py:
cicd_testrailAndInfraSetup.py:
run_cv_scenario.py:
sta_connect.py:
sta_connect2.py:
sta_connect_example.py:
sta_connect_multi_example.py:
stations_connected.py:
test_ipv4_connection.py:
test_ipv4_variable_time.py:
test_wanlink.py:
vap_stations_example.py:

163
py-scripts/run_cv_scenario.py Executable file
View File

@@ -0,0 +1,163 @@
#!/usr/bin/env python3
# This script will set the LANforge to a BLANK database then it will load the specified database
# and start a graphical report
import sys
if sys.version_info[0] != 3:
print("This script requires Python 3")
exit(1)
if 'py-json' not in sys.path:
sys.path.append('../py-json')
import argparse
from LANforge import LFUtils
from LANforge import lfcli_base
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
import realm
from realm import Realm
"""
cvScenario.scenario_db = args.scenario_db
if args.cv_test is not None:
cvScenario.cv_test = args.cv_test
if args.test_scenario is not None:
cvScenario.test_scenario = args.test_scenario
"""
class RunCvScenario(LFCliBase):
def __init__(self, lfhost="localhost", lfport=8080, debug_=True, scenario_db_=None, cv_test_=None, test_scenario_=None):
super().__init__( _lfjson_host=lfhost, _lfjson_port=lfport, _debug=debug_, _halt_on_error=True, _exit_on_error=True, _exit_on_fail=True)
self.scenario_db = scenario_db_
self.cv_test = cv_test_
self.test_scenario = test_scenario_
self.localrealm = Realm(lfclient_host=lfhost, lfclient_port=lfport, debug_=debug_)
self.report_name = None
def get_report_file_name(self):
return self.report_name
def build(self):
data = {
"name": "BLANK",
"action":"overwrite",
"clean_dut":"yes",
"clean_chambers": "yes"
}
self.json_post("/cli-json/load", data)
sleep(1)
port_counter = 0;
attempts = 6
while (attempts > 0) and (port_counter > 0):
sleep(1)
attempts -= 1
port_list = self.localrealm.find_ports_like("vap+")
alias_map = LFUtils.portListToAliasMap(port_list)
port_counter = len(alias_map)
port_list = self.localrealm.find_ports_like("sta+")
alias_map = LFUtils.portListToAliasMap(port_list)
port_counter += len(alias_map)
if port_counter == 0:
break
if (port_counter != 0) and (attempts == 0):
print("There appears to be a vAP in this database, quitting.")
pprint(alias_map);
exit(1)
data = {
"name": self.scenario_db,
"action":"overwrite",
"clean_dut":"yes",
"clean_chambers": "yes"
}
self.json_post("/cli-json/load", data)
self._pass("Loaded scenario %s" % self.scenario_db, True)
def start(self):
# /gui_cli takes commands keyed on 'cmd', so we create an array of commands
commands = [
"cv apply s1101",
"cv create 'Rate vs Range' rvr_instance"
"cv click rvr_instance Start",
"cv get rvr_instance 'Report Location:'"
]
self.use_preexec = False
for command in commands:
data = {
"cmd": command
}
self.json_post("/gui-cli", data)
sleep(1)
self._fail("start unfinished", print_=True)
def stop(self):
self._fail("stop unfinished", print_=True)
def cleanup(self):
self._fail("cleanup unfinished", print_=True)
def main():
lfjson_host = "localhost"
lfjson_port = 8080
parser = argparse.ArgumentParser(
description="""LANforge Reporting Script: Load a scenario and run a RvR report
Example:
./load_ap_scenario.py --lfmgr 127.0.0.1 --scenario_db 'handsets' --cv_test --test_scenario 'test-20'
""")
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("-d", "--scenario_db", type=str, help="Name of test scenario database (see Status Tab)")
parser.add_argument("-t", "--cv_test", type=str, help="Chamber View test")
parser.add_argument("-s", "--test_scenario", type=str, help="Scenario name of the CV test")
args = parser.parse_args()
if args.lfmgr is not None:
lfjson_host = args.lfmgr
if args.port is not None:
lfjson_port = args.port
cvScenario = RunCvScenario(lfjson_host, lfjson_port)
if args.scenario_db is not None:
cvScenario.scenario_db = args.scenario_db
if args.cv_test is not None:
cvScenario.cv_test = args.cv_test
if args.test_scenario is not None:
cvScenario.test_scenario = args.test_scenario
if (cvScenario.scenario_db is None) or (cvScenario.scenario_db == ""):
raise ValueError("Please specificy scenario database name with --scenario_db")
cvScenario.build()
if cvScenario.passes() != True:
print(cvScenario.get_fail_message())
exit(1)
cvScenario.start()
if cvScenario.passes() != True:
print(cvScenario.get_fail_message())
exit(1)
cvScenario.stop()
if cvScenario.passes() != True:
print(cvScenario.get_fail_message())
exit(1)
cvScenario.cleanup()
if cvScenario.passes() != True:
print(cvScenario.get_fail_message())
exit(1)
report_file = cvScenario.get_report_file_name()
print("Report file saved to "+report_file)
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
if __name__ == "__main__":
main()

28
py-scripts/sta_connect.py
View File

@@ -158,7 +158,8 @@ class StaConnect(LFCliBase):
for sta_name in self.station_names:
add_sta_data["sta_name"] = sta_name
print(" %s," % sta_name, end="")
self.json_post("/cli-json/add_sta", add_sta_data)
self.json_post("/cli-json/add_sta", add_sta_data, use_preexec_=False)
time.sleep(0.01)
set_port_data = {
"shelf": 1,
@@ -170,13 +171,14 @@ class StaConnect(LFCliBase):
for sta_name in self.station_names:
set_port_data["port"] = sta_name
print(" %s," % sta_name, end="")
self.json_post("/cli-json/set_port", set_port_data)
self.json_post("/cli-json/set_port", set_port_data, use_preexec_=False)
time.sleep(0.01)
print("\nBringing ports up...")
data = {"shelf": 1,
"resource": self.resource,
"port": "ALL",
"probe_flags": 1}
self.json_post("/cli-json/nc_show_ports", data)
self.json_post("/cli-json/nc_show_ports", data, use_preexec_=False)
LFUtils.waitUntilPortsAdminUp(self.resource, self.lfclient_url, self.station_names)
# station_info = self.jsonGet(self.mgr_url, "%s?fields=port,ip,ap" % (self.getStaUrl()))
@@ -216,7 +218,7 @@ class StaConnect(LFCliBase):
"port": "ALL",
"probe_flags": 1
}
self.json_post("/cli-json/nc_show_ports", data)
self.json_post("/cli-json/nc_show_ports", data, use_preexec_=False)
# make a copy of the connected stations for test records
@@ -267,14 +269,14 @@ class StaConnect(LFCliBase):
"ip_port": "-1",
"min_rate": 1000000
}
self.json_post("/cli-json/add_endp", data)
self.json_post("/cli-json/add_endp", data, use_preexec_=False)
data = {
"name" : "testUDP-%s-A" % sta_name,
"flag" : "UseAutoNAT",
"val" : 1
}
self.json_post("/cli-json/set_endp_flag", data)
self.json_post("/cli-json/set_endp_flag", data, use_preexec_=False)
data = {
"alias": "testUDP-%s-B" % sta_name,
@@ -285,14 +287,14 @@ class StaConnect(LFCliBase):
"ip_port": "-1",
"min_rate": 1000000
}
self.json_post("/cli-json/add_endp", data)
self.json_post("/cli-json/add_endp", data, use_preexec_=False)
data = {
"name" : "testUDP-%s-B" % sta_name,
"flag" : "UseAutoNAT",
"val" : 1
}
self.json_post("/cli-json/set_endp_flag", data)
self.json_post("/cli-json/set_endp_flag", data, use_preexec_=False)
# Create CX
data = {
@@ -301,14 +303,14 @@ class StaConnect(LFCliBase):
"tx_endp": "testUDP-%s-A" % sta_name,
"rx_endp": "testUDP-%s-B" % sta_name,
}
self.json_post("/cli-json/add_cx", data)
self.json_post("/cli-json/add_cx", data, use_preexec_=False)
data = {
"test_mgr": "default_tm",
"cx_name": "testUDP-%s" % sta_name,
"milliseconds": 1000
}
self.json_post("/cli-json/set_cx_report_timer", data)
self.json_post("/cli-json/set_cx_report_timer", data, use_preexec_=False)
# Create TCP endpoints
cx_names["testTCP-"+sta_name] = { "a": "testUDP-%s-A" % sta_name,
@@ -322,7 +324,7 @@ class StaConnect(LFCliBase):
"ip_port": "0",
"min_rate": 1000000
}
self.json_post("/cli-json/add_endp", data)
self.json_post("/cli-json/add_endp", data, use_preexec_=False)
data = {
"alias": "testTCP-%s-B" % sta_name,
@@ -333,7 +335,7 @@ class StaConnect(LFCliBase):
"ip_port": "-1",
"min_rate": 1000000
}
self.json_post("/cli-json/add_endp", data)
self.json_post("/cli-json/add_endp", data, use_preexec_=False)
# Create CX
data = {
@@ -349,7 +351,7 @@ class StaConnect(LFCliBase):
"cx_name": "testTCP-%s" % sta_name,
"milliseconds": 1000
}
self.json_post("/cli-json/set_cx_report_timer", data)
self.json_post("/cli-json/set_cx_report_timer", data, use_preexec_=False)
# start cx traffic
print("\nStarting CX Traffic")

140
py-scripts/test_ipv4_variable_time.py Executable file
View File

@@ -0,0 +1,140 @@
#!/usr/bin/env python3
import sys
if sys.version_info[0] != 3:
print("This script requires Python 3")
exit(1)
if 'py-json' not in sys.path:
sys.path.append('../py-json')
import argparse
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
import realm
import time
import datetime
class IPV4VariableTime(LFCliBase):
def __init__(self, host, port, ssid, security, password, num_stations, prefix="00000", test_duration="5m",
_debug_on=False,
_exit_on_error=False,
_exit_on_fail=False):
super().__init__(host, port, _debug=_debug_on, _halt_on_error=_exit_on_error, _exit_on_fail=_exit_on_fail)
self.host = host
self.port = port
self.ssid = ssid
self.security = security
self.password = password
self.num_stations = num_stations
self.prefix = prefix
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.station_profile = realm.StationProfile(self.lfclient_url, ssid=self.ssid, ssid_pass=self.password,
security=self.security, prefix=self.prefix, mode=0, up=True,
dhcp=True,
debug_=False)
self.cx_profile = self.local_realm.new_l3_cx_profile()
self.test_duration = test_duration
def __set_all_cx_state(self, state, sleep_time=5):
print("Setting CX States to %s" % state)
cx_list = list(self.local_realm.cx_list())
for cx_name in cx_list:
if cx_name != 'handler' or cx_name != 'uri':
req_url = "cli-json/set_cx_state"
data = {
"test_mgr": "default_tm",
"cx_name": cx_name,
"cx_state": state
}
super().json_post(req_url, data)
time.sleep(sleep_time)
def run_test(self):
cur_time = datetime.datetime.now()
end_time = self.local_realm.parse_time(self.test_duration) + cur_time
self.__set_all_cx_state("RUNNING")
while cur_time < end_time:
#print(cur_time, end_time)
cur_time = datetime.datetime.now()
# Run test
time.sleep(1)
self.__set_all_cx_state("STOPPED")
def cleanup(self):
print("Cleaning up stations")
port_list = self.local_realm.station_list()
sta_list = []
for item in list(port_list):
# print(list(item))
if "sta" in list(item)[0]:
sta_list.append(self.local_realm.name_to_eid(list(item)[0])[2])
for sta_name in sta_list:
req_url = "cli-json/rm_vlan"
data = {
"shelf": 1,
"resource": 1,
"port": sta_name
}
# print(data)
super().json_post(req_url, data)
cx_list = list(self.local_realm.cx_list())
print("Cleaning up cxs")
for cx_name in cx_list:
if cx_name != 'handler' or cx_name != 'uri':
req_url = "cli-json/rm_cx"
data = {
"test_mgr": "default_tm",
"cx_name": cx_name
}
super().json_post(req_url, data)
print("Cleaning up endps")
endp_list = super().json_get("/endp")
if endp_list is not None:
endp_list = list(endp_list['endpoint'])
for endp_name in range(len(endp_list)):
name = list(endp_list[endp_name])[0]
req_url = "cli-json/rm_endp"
data = {
"endp_name": name
}
super().json_post(req_url, data)
def run(self):
super().clear_test_results()
print("Cleaning up old stations")
self.cleanup()
sta_list = []
self.station_profile.use_wpa2(True, self.ssid, self.password)
self.station_profile.set_prefix(self.prefix)
print("Creating stations")
self.station_profile.create(resource=1, radio="wiphy0", num_stations=self.num_stations, debug=False)
for name in list(self.local_realm.station_list()):
if "sta" in list(name)[0]:
sta_list.append(list(name)[0])
print("sta_list", sta_list)
self.cx_profile.create(endp_type="lf_udp", side_a=sta_list, side_b="1.eth1", sleep_time=.5)
def main():
lfjson_host = "localhost"
lfjson_port = 8080
ip_var_test = IPV4VariableTime(lfjson_host, lfjson_port, prefix="00", ssid="jedway-wpa2-x2048-4-4",
password="jedway-wpa2-x2048-4-4",
security="open", num_stations=10, test_duration="1m")
ip_var_test.run()
ip_var_test.run_test()
ip_var_test.cleanup()
if __name__ == "__main__":
main()

95
py-scripts/test_wanlink.py Normal file
View File

@@ -0,0 +1,95 @@
#!/usr/bin/env python3
import sys
if sys.version_info[0] != 3:
print("This script requires Python 3")
exit(1)
if 'py-json' not in sys.path:
sys.path.append('../py-json')
import argparse
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
import realm
import time
class LANtoWAN(LFCliBase):
def __init__(self, host, port, ssid, security, password, lan_port="eth2", wan_port="eth3", _debug_on=False,
_exit_on_error=False,
_exit_on_fail=False):
super().__init__(host, port, _debug=_debug_on, _halt_on_error=_exit_on_error, _exit_on_fail=_exit_on_fail)
self.host = host
self.port = port
self.ssid = ssid
self.security = security
self.password = password
self.timeout = 120
self.lan_port = lan_port
self.wan_port = wan_port
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.profile = realm.StationProfile(self.lfclient_url, ssid=self.ssid, ssid_pass=self.password,
security=self.security, prefix=self.prefix, mode=0, up=True, dhcp=True,
debug_=False)
self.cxProfile = realm.new_l3_cx_profile()
def run_test(self): pass
def create_wanlinks(self, shelf=1, resource=1, latency=20, max_rate=1544000):
print("Creating wanlinks")
# create redirects for wanlink
url = "/cli-json/add_rdd"
data = {
"shelf": shelf,
"resource": resource,
"port": "rdd0",
"peer_ifname": "rdd1"
}
self.json_post(url, data)
url = "/cli-json/add_rdd"
data = {
"shelf": shelf,
"resource": resource,
"port": "rdd1",
"peer_ifname": "rdd0"
}
self.json_post(url, data)
time.sleep(.05)
# create wanlink endpoints
url = "/cli-json/add_wl_endp"
data = {
"alias": "wlan0",
"shelf": shelf,
"resource": resource,
"port": "rdd0",
"latency": latency,
"max_rate": max_rate
}
self.json_post(url, data)
url = "/cli-json/add_wl_endp"
data = {
"alias": "wlan1",
"shelf": shelf,
"resource": resource,
"port": "rdd1",
"latency": latency,
"max_rate": max_rate
}
self.json_post(url, data)
time.sleep(.05)
def run(self):
self.profile.use_wpa2(True, self.ssid, self.password)
self.profile.create(resource=1, radio="wiphy0", num_stations=3, debug=False)
def cleanup(self): pass
def main():
if __name__ == "__main__":
main()