diff --git a/py-json/README.md b/py-json/README.md index 553843b7..39da9750 100644 --- a/py-json/README.md +++ b/py-json/README.md @@ -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! diff --git a/py-scripts/README.md b/py-scripts/README.md new file mode 100644 index 00000000..fa50361a --- /dev/null +++ b/py-scripts/README.md @@ -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: + + diff --git a/py-scripts/README.txt b/py-scripts/README.txt deleted file mode 100644 index f55c9577..00000000 --- a/py-scripts/README.txt +++ /dev/null @@ -1,3 +0,0 @@ -This directory will contain python scripts useful for unit-tests. It uses -libraries in ../py-json - diff --git a/py-scripts/sta_connect.py b/py-scripts/sta_connect.py index 9327375a..d58a3e02 100755 --- a/py-scripts/sta_connect.py +++ b/py-scripts/sta_connect.py @@ -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,11 +273,11 @@ 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, - "b": "testUDP-%s-B" % 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, "shelf": 1, @@ -313,8 +331,8 @@ 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, - "b": "testUDP-%s-B" % 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, "shelf": 1, @@ -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,16 +456,16 @@ class StaConnect(LFCliBase): # self.test_results.append("FAILED message will fail") # print("\n") - # remove all endpoints and cxs - if self.cleanup_on_exit: - 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"]) - removeEndps(self.lfclient_url, endp_names) + + 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, 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 diff --git a/py-scripts/sta_connect_example.py b/py-scripts/sta_connect_example.py index 993403fb..7b09b3a1 100755 --- a/py-scripts/sta_connect_example.py +++ b/py-scripts/sta_connect_example.py @@ -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()