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

Merge remote-tracking branch 'origin/master'

Browse Source
This commit is contained in:
Logan Lipke
2020-06-26 18:00:14 -07:00
parent 2a2bf8a5b6 d0a830745c
commit 38b424e8b3
5 changed files with 118 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
py-json/README.md
View File

@@ -104,9 +104,24 @@ begin with these imports:
## 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
@@ -115,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
@@ -129,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!

35
py-scripts/README.md Normal file
View File

@@ -0,0 +1,35 @@
# LANForge Python Scripts
This directory contains python scripts useful for unit-tests. It uses
libraries in ../py-json.
#### Scripts included are:
* 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:

3
py-scripts/README.txt
View File

@@ -1,3 +0,0 @@
This directory will contain python scripts useful for unit-tests. It uses
libraries in ../py-json

60
py-scripts/sta_connect.py
View File

@@ -56,6 +56,7 @@ class StaConnect(LFCliBase):
self.sta_url_map = None # defer construction
self.upstream_url = None # defer construction
self.station_names = []
self.cx_names = {}
if _sta_name is not None:
self.station_names = [ _sta_name ]
# self.localrealm :Realm = Realm(lfclient_host=host, lfclient_port=port) # py > 3.6
@@ -116,6 +117,23 @@ class StaConnect(LFCliBase):
#super(StaConnect, self).clear_test_results().test_results.clear()
def run(self):
if not self.setup():
return False
if not self.start():
return False
time.sleep(self.runtime_secs)
if not self.stop():
return False
if not self.finish():
return False
# remove all endpoints and cxs
if self.cleanup_on_exit:
if not self.cleanup():
return False
return True
def setup(self):
self.clear_test_results()
self.check_connect()
eth1IP = self.json_get(self.get_upstream_url())
@@ -255,10 +273,10 @@ class StaConnect(LFCliBase):
# create endpoints and cxs
# Create UDP endpoints
cx_names = {}
self.cx_names = {}
for sta_name in self.station_names:
cx_names["testUDP-"+sta_name] = { "a": "testUDP-%s-A" % sta_name,
self.cx_names["testUDP-"+sta_name] = { "a": "testUDP-%s-A" % sta_name,
"b": "testUDP-%s-B" % sta_name}
data = {
"alias": "testUDP-%s-A" % sta_name,
@@ -313,7 +331,7 @@ class StaConnect(LFCliBase):
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,
self.cx_names["testTCP-"+sta_name] = { "a": "testUDP-%s-A" % sta_name,
"b": "testUDP-%s-B" % sta_name}
data = {
"alias": "testTCP-%s-A" % sta_name,
@@ -353,9 +371,12 @@ class StaConnect(LFCliBase):
}
self.json_post("/cli-json/set_cx_report_timer", data, use_preexec_=False)
return True
def start(self):
# start cx traffic
print("\nStarting CX Traffic")
for cx_name in cx_names.keys():
for cx_name in self.cx_names.keys():
data = {
"test_mgr": "ALL",
"cx_name": cx_name,
@@ -366,28 +387,31 @@ class StaConnect(LFCliBase):
# Refresh stats
print("Refresh CX stats")
for cx_name in cx_names.keys():
for cx_name in self.cx_names.keys():
data = {
"test_mgr": "ALL",
"cross_connect": cx_name
}
self.json_post("/cli-json/show_cxe", data)
return True
time.sleep(self.runtime_secs)
def stop(self):
# stop cx traffic
print("Stopping CX Traffic")
for cx_name in cx_names.keys():
for cx_name in self.cx_names.keys():
data = {
"test_mgr": "ALL",
"cx_name": cx_name,
"cx_state": "STOPPED"
}
self.json_post("/cli-json/set_cx_state", data)
return True
def finish(self):
# Refresh stats
print("\nRefresh CX stats")
for cx_name in cx_names.keys():
for cx_name in self.cx_names.keys():
data = {
"test_mgr": "ALL",
"cross_connect": cx_name
@@ -399,19 +423,19 @@ class StaConnect(LFCliBase):
# get data for endpoints JSON
print("Collecting Data")
for cx_name in cx_names.keys():
for cx_name in self.cx_names.keys():
try:
# ?fields=tx+bytes,rx+bytes
endp_url = "/endp/%s" % cx_names[cx_name]["a"]
endp_url = "/endp/%s" % self.cx_names[cx_name]["a"]
ptest = self.json_get(endp_url)
self.resulting_endpoints[endp_url] = ptest
ptest_a_tx = ptest['endpoint']['tx bytes']
ptest_a_rx = ptest['endpoint']['rx bytes']
#ptest = self.json_get("/endp/%s?fields=tx+bytes,rx+bytes" % cx_names[cx_name]["b"])
endp_url = "/endp/%s" % cx_names[cx_name]["b"]
#ptest = self.json_get("/endp/%s?fields=tx+bytes,rx+bytes" % self.cx_names[cx_name]["b"])
endp_url = "/endp/%s" % self.cx_names[cx_name]["b"]
ptest = self.json_get(endp_url)
self.resulting_endpoints[endp_url] = ptest
@@ -432,15 +456,15 @@ class StaConnect(LFCliBase):
# self.test_results.append("FAILED message will fail")
# print("\n")
# remove all endpoints and cxs
if self.cleanup_on_exit:
def cleanup(self):
for sta_name in self.station_names:
LFUtils.removePort(self.resource, sta_name, self.lfclient_url)
endp_names = []
removeCX(self.lfclient_url, cx_names.keys())
for cx_name in cx_names:
endp_names.append(cx_names[cx_name]["a"])
endp_names.append(cx_names[cx_name]["b"])
removeCX(self.lfclient_url, self.cx_names.keys())
for cx_name in self.cx_names:
endp_names.append(self.cx_names[cx_name]["a"])
endp_names.append(self.cx_names[cx_name]["b"])
removeEndps(self.lfclient_url, endp_names)
# ~class

15
py-scripts/sta_connect_example.py
View File

@@ -13,20 +13,25 @@ if 'py-json' not in sys.path:
# if you lack __init__.py in this directory you will not find sta_connect module
import sta_connect
from sta_connect import StaConnect
import time
def main():
staConnect = StaConnect("localhost", 8080, _debugOn=False)
staConnect.sta_mode = 0
staConnect.upstream_resource = 1
staConnect.upstream_port = "eth1"
staConnect.upstream_port = "eth2"
staConnect.radio = "wiphy0"
staConnect.resource = 1
staConnect.dut_security = sta_connect.WPA2
staConnect.dut_ssid = "jedway-wpa2-x2048-5-1"
staConnect.dut_passwd = "jedway-wpa2-x2048-5-1"
staConnect.dut_ssid = "Default-SSID-2g"
staConnect.dut_passwd = "12345678"
staConnect.station_names = [ "sta000" ]
staConnect.run()
staConnect.setup()
staConnect.start()
time.sleep(20)
staConnect.stop()
staConnect.finish()
staConnect.cleanup()
is_passing = staConnect.passes()
if is_passing == False:
# run_results = staConnect.get_failed_result_list()