From 6da7ef8afcd904c8e4e02404f5d3ce8edfc40631 Mon Sep 17 00:00:00 2001 From: Jed Reynolds Date: Mon, 1 Jun 2020 13:59:07 -0700 Subject: [PATCH 1/8] sta_connect: refactoring leading to a base class LFCliBase and class structure for sta_connect --- py-json/LANforge/LFCliBase.py | 32 ++ py-scripts/sta_connect.py | 730 +++++++++++++++++----------------- 2 files changed, 400 insertions(+), 362 deletions(-) create mode 100644 py-json/LANforge/LFCliBase.py diff --git a/py-json/LANforge/LFCliBase.py b/py-json/LANforge/LFCliBase.py new file mode 100644 index 00000000..d3dfec7a --- /dev/null +++ b/py-json/LANforge/LFCliBase.py @@ -0,0 +1,32 @@ +#!env /usr/bin/python + +# Extend this class to use common set of debug and request features for your script + +from LANforge.LFUtils import * + + +class LFCliBase: + def __init__(self, lfjson_host="localhost", lfjson_port=8080, _debug=False): + self.debugOn = _debug + self.lfjson_host = lfjson_host + self.lfjson_port = lfjson_port + self.mgr_url = f"http://{self.lfjson_host}:{self.lfjson_port}/" + + def jsonPost(self, _req_url, _data): + lf_r = LFRequest.LFRequest(self.mgr_url + _req_url) + _data['suppress_preexec_cli'] = True + _data['suppress_preexec_method'] = True + lf_r.addPostData(_data) + json_response = lf_r.jsonPost(self.debugOn) + # Debugging + # if (json_response != None): + # print("jsonReq: response: ") + # LFUtils.debug_printer.pprint(vars(json_response)) + return json_response + + def jsonGet(self, _req_url): + lf_r = LFRequest.LFRequest(self.mgr_url + _req_url) + json_response = lf_r.getAsJson(self.debugOn) + return json_response + +# ~class diff --git a/py-scripts/sta_connect.py b/py-scripts/sta_connect.py index d996d191..46144919 100755 --- a/py-scripts/sta_connect.py +++ b/py-scripts/sta_connect.py @@ -5,412 +5,418 @@ # to the requested BSSID if bssid is specified as an argument. # The script will clean up the station and connections at the end of the test. -import os -import time import sys import argparse if 'py-json' not in sys.path: - sys.path.append('../py-json') + sys.path.append('../py-json') -import subprocess -import json -import pprint -from LANforge import LFRequest +# from LANforge import LFRequest from LANforge import LFUtils +# from LANforge import LFCliBase +from LANforge.LFCliBase import LFCliBase from LANforge.LFUtils import * -import create_genlink as genl - -debugOn = True if sys.version_info[0] != 3: print("This script requires Python 3") exit(1) -dest = "localhost"; -port = "8080"; -dut_ssid = "MyAP" -dut_passwd = "NA" -dut_bssid = "" -user = "" -passwd = "" -sta_mode = "0" # See add_sta LANforge CLI users guide entry -radio = "wiphy0" -resource = "1" -upstream_resource = "1" -upstream_port = "eth2" -sta_name = "sta001" -parser = argparse.ArgumentParser(description="LANforge Unit Test: Connect Station to AP\nExample:\n./sta_connect.py --dest 192.168.100.209 --dut_ssid OpenWrt-2 --dut_bssid 24:F5:A2:08:21:6C") -parser.add_argument("-d", "--dest", 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("-u", "--user", type=str, help="TBD: credential login/username") -parser.add_argument("-p", "--passwd", type=str, help="TBD: credential password") -parser.add_argument("--resource", type=str, help="LANforge Station resource ID to use, default is 1") -parser.add_argument("--upstream_resource", type=str, help="LANforge Ethernet port resource ID to use, default is 1") -parser.add_argument("--upstream_port", type=str, help="LANforge Ethernet port name, default is eth2") -parser.add_argument("--radio", type=str, help="LANforge radio to use, default is wiphy0") -parser.add_argument("--sta_mode", type=str, help="LANforge station-mode setting (see add_sta LANforge CLI documentation, default is 0 (auto))") -parser.add_argument("--dut_ssid", type=str, help="DUT SSID") -parser.add_argument("--dut_passwd", type=str, help="DUT PSK password. Do not set for OPEN auth") -parser.add_argument("--dut_bssid", type=str, help="DUT BSSID to which we expect to connect.") +class StaConnect(LFCliBase): + def __init__(self, lfjson_host, lfjson_port, _dut_ssid="MyAP", _dut_passwd="NA", _dut_bssid="", + _user="", _passwd="", _sta_mode="0", _radio="wiphy0", + _resource=1, _upstream_resource=1, _upstream_port="eth2", + _sta_name="sta001", _debugOn=False): + # self.dest = "localhost" # in super + # self.port = "8080" # in super -args = None + super().__init__(self, lfjson_host, lfjson_port) + self.dut_ssid = _dut_ssid + self.dut_passwd = _dut_passwd + self.dut_bssid = _dut_bssid + self.user = _user + self.passwd = _passwd + self.sta_mode = _sta_mode # See add_sta LANforge CLI users guide entry + self.radio = _radio + self.resource = _resource + self.upstream_resource = _upstream_resource + self.upstream_port = _upstream_port + self.sta_name = _sta_name + self.sta_url = None # defer construction + self.upstream_url = None # defer construction -args = parser.parse_args() -if (args.dest != None): - dest = args.dest -if (args.port != None): - port = args.port -if (args.user != None): - user = args.user -if (args.passwd != None): - passwd = args.passwd -if (args.sta_mode != None): - sta_mode = args.sta_mode -if (args.upstream_resource != None): - upstream_resource = args.upstream_resource -if (args.upstream_port != None): - upstream_port = args.upstream_port -if (args.radio != None): - radio = args.radio -if (args.resource != None): - resource = args.resource -if (args.dut_passwd != None): - dut_passwd = args.dut_passwd -if (args.dut_bssid != None): - dut_bssid = args.dut_bssid -if (args.dut_ssid != None): - dut_ssid = args.dut_ssid + def getStaUrl(self): + if self.sta_url is None: + self.sta_url = f"port/1/{self.resource}/{self.sta_name}" + return self.sta_url -mgrURL = "http://%s:%s/"%(dest, port) -radio_url = "port/1/%s/%s"%(resource, radio) -sta_url = "port/1/%s/%s"%(resource, sta_name) -upstream_url = "port/1/%s/%s"%(upstream_resource, upstream_port) + def getUpstreamUrl(self): + if self.upstream_url is None: + self.upstream_url = f"port/1/{self.upstream_resource}/{self.upstream_port}" + return self.upstream_url -def jsonReq(mgrURL, reqURL, data, exitWhenCalled=False): - lf_r = LFRequest.LFRequest(mgrURL + reqURL) + def checkConnect(self): + print(f"Checking for LANforge GUI connection: {self.mgr_url}") + response = super().jsonGet("/") + duration = 0 + while (response is None) and (duration < 300): + print(f"LANforge GUI connection not found sleeping 5 seconds, tried: {self.mgr_url}") + duration += 2 + time.sleep(2) + response = super().jsonGet("/") - data['suppress_preexec_cli'] = True - data['suppress_preexec_method'] = True + if duration >= 300: + print("Could not connect to LANforge GUI") + sys.exit(1) - lf_r.addPostData(data) + # Compare pre-test values to post-test values + @staticmethod + def compareVals(name, postVal): + # print(f"Comparing {name}") + if postVal > 0: + print("PASSED: %s %s" % (name, postVal)) + else: + print("FAILED: %s did not report traffic: %s" % (name, postVal)) - json_response = lf_r.jsonPost(True) - # Debugging - #if (json_response != None): - # print("jsonReq: response: ") - # LFUtils.debug_printer.pprint(vars(json_response)) - if exitWhenCalled: - print("jsonReq: bye") - sys.exit(1) + def run(self): + self.checkConnect() + # Create stations and turn dhcp on + print("Creating station and turning on dhcp") -def getJsonInfo(mgrURL, reqURL, debug=False): - lf_r = LFRequest.LFRequest(mgrURL + reqURL) - json_response = lf_r.getAsJson(debug) - return json_response - #print(name) - #j_printer = pprint.PrettyPrinter(indent=2) - #j_printer.pprint(json_response) - #for record in json_response[key]: - # j_printer.pprint(record) + url = self.getStaUrl() + response = super().jsonGet(url) + if response is not None: + if response["interface"] is not None: + print("removing old station") + LFUtils.removePort(self.resource, self.sta_name, self.mgr_url) + time.sleep(5) - -print("Checking for LANforge GUI connection: %s"%(mgrURL)) -response = getJsonInfo(mgrURL, radio_url); -duration = 0 -while ((response == None) and (duration < 300)): - print("LANforge GUI connection not found sleeping 5 seconds, tried: %s"%(mgrURL)) - duration += 2 - time.sleep(2) - response = getJsonInfo(mgrURL, radio_url) - -if duration >= 300: - print("Could not connect to LANforge GUI") - sys.exit(1) - - -#Create stations and turn dhcp on -print("Creating station and turning on dhcp") - -url = sta_url -debugOn = True -response = getJsonInfo(mgrURL, url) -if (response is not None): - if (response["interface"] is not None): - print("removing old station") - LFUtils.removePort(resource, sta_name, mgrURL) - time.sleep(5) - -# See add_sta in LANforge CLI user guide -url = "cli-json/add_sta" -data = { - "shelf":1, - "resource":resource, - "radio":radio, - "sta_name":sta_name, - "ssid":dut_ssid, - "key":dut_passwd, - "mode":sta_mode, - "mac":"xx:xx:xx:xx:*:xx", - "flags":0x10000 # verbose, open -} -print("adding new station") -jsonReq(mgrURL, url, data) - -reqURL = "cli-json/set_port" -data = { - "shelf":1, - "resource":resource, - "port":sta_name, - "current_flags": 0x80000000, # use DHCP, not down - "interest":0x4002 # set dhcp, current flags -} -print("configuring port") -jsonReq(mgrURL, reqURL, data) - -time.sleep(5) - -eth1IP = getJsonInfo(mgrURL, upstream_url) -if eth1IP['interface']['ip'] == "0.0.0.0": - print("Warning: %s lacks ip address"%(upstream_url)) - -reqURL = "cli-json/nc_show_ports" -data = { "shelf":1, - "resource":resource, - "port":sta_name, - "probe_flags":1 } -jsonReq(mgrURL, reqURL, data) - -station_info = getJsonInfo(mgrURL, "%s?fields=port,ip,ap"%(sta_url)) -duration = 0 -maxTime = 300 -ip = "0.0.0.0" -ap = "" -while ((ip == "0.0.0.0") and (duration < maxTime)): - - duration += 2 - time.sleep(2) - - station_info = getJsonInfo(mgrURL, "%s?fields=port,ip,ap"%(sta_url)) - - #LFUtils.debug_printer.pprint(station_info) - if ((station_info is not None) and ("interface" in station_info)): - if ("ip" in station_info["interface"]): - ip = station_info["interface"]["ip"] - if ("ap" in station_info["interface"]): - ap = station_info["interface"]["ap"] - - if ((ap == "Not-Associated") or (ap == "")): - print("Station waiting to associate...") - else: - if (ip == "0.0.0.0"): - print("Station waiting for IP ...") - -if ((ap != "") and (ap != "Not-Associated")): - print("Connected to AP: %s"%(ap)) - if (dut_bssid != ""): - if (dut_bssid.lower() == ap.lower()): - print("PASSED: Connected to BSSID: %s"%(ap)) - else: - print("FAILED: Connected to wrong BSSID, requested: %s Actual: %s"%(dut_bssid, ap)) -else: - print("FAILED: Did not connect to AP"); - sys.exit(3) - -if (ip is "0.0.0.0"): - print("FAILED: %s did not get an ip. Ending test"%(sta_name)) - print("Cleaning up...") - removePort(resource, sta_name, mgrURL) - sys.exit(1) -else: - print("PASSED: Connected to AP: %s With IP: %s"%(ap, ip)) - -#create endpoints and cxs -# Create UDP endpoints -reqURL = "cli-json/add_endp" -data = { - "alias":"testUDP-A", - "shelf":1, - "resource":resource, - "port":sta_name, - "type":"lf_udp", - "ip_port":"-1", - "min_rate":1000000 - } -jsonReq(mgrURL, reqURL, data) - -reqURL = "cli-json/add_endp" -data = { - "alias":"testUDP-B", - "shelf":1, - "resource":upstream_resource, - "port":upstream_port, - "type":"lf_udp", - "ip_port":"-1", - "min_rate":1000000 - } -jsonReq(mgrURL, reqURL, data) - -# Create CX -reqURL = "cli-json/add_cx" -data = { - "alias":"testUDP", - "test_mgr":"default_tm", - "tx_endp":"testUDP-A", - "rx_endp":"testUDP-B", - } -jsonReq(mgrURL, reqURL, data) - -# Create TCP endpoints -reqURL = "cli-json/add_endp" -data = { - "alias":"testTCP-A", - "shelf":1, - "resource":resource, - "port":sta_name, - "type":"lf_tcp", - "ip_port":"0", - "min_rate":1000000 - } -jsonReq(mgrURL, reqURL, data) - -reqURL = "cli-json/add_endp" -data = { - "alias":"testTCP-B", - "shelf":1, - "resource":upstream_resource, - "port":upstream_port, - "type":"lf_tcp", - "ip_port":"-1", - "min_rate":1000000 - } -jsonReq(mgrURL, reqURL, data) - -# Create CX -reqURL = "cli-json/add_cx" -data = { - "alias":"testTCP", - "test_mgr":"default_tm", - "tx_endp":"testTCP-A", - "rx_endp":"testTCP-B", - } -jsonReq(mgrURL, reqURL, data) - -cxNames = ["testTCP","testUDP"] -endpNames = ["testTCP-A", "testTCP-B", - "testUDP-A", "testUDP-B"] - -#start cx traffic -print("\nStarting CX Traffic") -for name in range(len(cxNames)): - reqURL = "cli-json/set_cx_state" - data = { - "test_mgr":"ALL", - "cx_name":cxNames[name], - "cx_state":"RUNNING" + # See add_sta in LANforge CLI user guide + url = "cli-json/add_sta" + data = { + "shelf": 1, + "resource": self.resource, + "radio": self.radio, + "sta_name": self.sta_name, + "ssid": self.dut_ssid, + "key": self.dut_passwd, + "mode": self.sta_mode, + "mac": "xx:xx:xx:xx:*:xx", + "flags": 0x10000 # verbose, open } - jsonReq(mgrURL, reqURL, data) + print("adding new station") + super().jsonPost(url, data) -# Refresh stats -print("\nRefresh CX stats") -for name in range(len(cxNames)): - reqURL = "cli-json/show_cxe" - data = { - "test_mgr":"ALL", - "cross_connect":cxNames[name] + reqURL = "cli-json/set_port" + data = { + "shelf": 1, + "resource": self.resource, + "port": self.sta_name, + "current_flags": 0x80000000, # use DHCP, not down + "interest": 0x4002 # set dhcp, current flags } - jsonReq(mgrURL, reqURL, data) + print("configuring port") + super().jsonPost(reqURL, data) -#print("Sleeping for 15 seconds") -time.sleep(15) + time.sleep(5) -#stop cx traffic -print("\nStopping CX Traffic") -for name in range(len(cxNames)): - reqURL = "cli-json/set_cx_state" - data = { - "test_mgr":"ALL", - "cx_name":cxNames[name], - "cx_state":"STOPPED" + eth1IP = self.jsonGet(self.getUpstreamUrl()) + if eth1IP['interface']['ip'] == "0.0.0.0": + print(f"Warning: {self.getUpstreamUrl()} lacks ip address") + + reqURL = "cli-json/nc_show_ports" + data = {"shelf": 1, + "resource": self.resource, + "port": self.sta_name, + "probe_flags": 1} + super().jsonPost(reqURL, data) + + # station_info = self.jsonGet(self.mgr_url, "%s?fields=port,ip,ap" % (self.getStaUrl())) + duration = 0 + maxTime = 300 + ip = "0.0.0.0" + ap = "" + while (ip == "0.0.0.0") and (duration < maxTime): + + duration += 2 + time.sleep(2) + + station_info = super().jsonGet(f"{self.getStaUrl()}?fields=port,ip,ap") + + # LFUtils.debug_printer.pprint(station_info) + if (station_info is not None) and ("interface" in station_info): + if "ip" in station_info["interface"]: + ip = station_info["interface"]["ip"] + if "ap" in station_info["interface"]: + ap = station_info["interface"]["ap"] + + if (ap == "Not-Associated") or (ap == ""): + print("Station waiting to associate...") + else: + if ip == "0.0.0.0": + print("Station waiting for IP ...") + + if (ap != "") and (ap != "Not-Associated"): + print(f"Connected to AP: {ap}") + if self.dut_bssid != "": + if self.dut_bssid.lower() == ap.lower(): + print(f"PASSED: Connected to BSSID: {ap}") + else: + print("FAILED: Connected to wrong BSSID, requested: %s Actual: %s" % (self.dut_bssid, ap)) + else: + print("FAILED: Did not connect to AP") + sys.exit(3) + + if ip == "0.0.0.0": + print(f"FAILED: {self.sta_name} did not get an ip. Ending test") + print("Cleaning up...") + removePort(self.resource, self.sta_name, self.mgr_url) + sys.exit(1) + else: + print("PASSED: Connected to AP: %s With IP: %s" % (ap, ip)) + + # create endpoints and cxs + # Create UDP endpoints + reqURL = "cli-json/add_endp" + data = { + "alias": "testUDP-A", + "shelf": 1, + "resource": self.resource, + "port": self.sta_name, + "type": "lf_udp", + "ip_port": "-1", + "min_rate": 1000000 } - jsonReq(mgrURL, reqURL, data) + super().jsonPost(reqURL, data) -# Refresh stats -print("\nRefresh CX stats") -for name in range(len(cxNames)): - reqURL = "cli-json/show_cxe" - data = { - "test_mgr":"ALL", - "cross_connect":cxNames[name] + reqURL = "cli-json/add_endp" + data = { + "alias": "testUDP-B", + "shelf": 1, + "resource": self.upstream_resource, + "port": self.upstream_port, + "type": "lf_udp", + "ip_port": "-1", + "min_rate": 1000000 } - jsonReq(mgrURL, reqURL, data) + super().jsonPost(reqURL, data) -#print("Sleeping for 5 seconds") -time.sleep(5) + # Create CX + reqURL = "cli-json/add_cx" + data = { + "alias": "testUDP", + "test_mgr": "default_tm", + "tx_endp": "testUDP-A", + "rx_endp": "testUDP-B", + } + super().jsonPost(reqURL, data) -#get data for endpoints JSON -print("Collecting Data") -try: - ptestTCPA = getJsonInfo(mgrURL, "endp/testTCP-A?fields=tx+bytes,rx+bytes") - ptestTCPATX = ptestTCPA['endpoint']['tx bytes'] - ptestTCPARX = ptestTCPA['endpoint']['rx bytes'] + # Create TCP endpoints + reqURL = "cli-json/add_endp" + data = { + "alias": "testTCP-A", + "shelf": 1, + "resource": self.resource, + "port": self.sta_name, + "type": "lf_tcp", + "ip_port": "0", + "min_rate": 1000000 + } + super().jsonPost(reqURL, data) - ptestTCPB = getJsonInfo(mgrURL, "endp/testTCP-B?fields=tx+bytes,rx+bytes") - ptestTCPBTX = ptestTCPB['endpoint']['tx bytes'] - ptestTCPBRX = ptestTCPB['endpoint']['rx bytes'] + reqURL = "cli-json/add_endp" + data = { + "alias": "testTCP-B", + "shelf": 1, + "resource": self.upstream_resource, + "port": self.upstream_port, + "type": "lf_tcp", + "ip_port": "-1", + "min_rate": 1000000 + } + super().jsonPost(reqURL, data) - ptestUDPA = getJsonInfo(mgrURL, "endp/testUDP-A?fields=tx+bytes,rx+bytes") - ptestUDPATX = ptestUDPA['endpoint']['tx bytes'] - ptestUDPARX = ptestUDPA['endpoint']['rx bytes'] + # Create CX + reqURL = "cli-json/add_cx" + data = { + "alias": "testTCP", + "test_mgr": "default_tm", + "tx_endp": "testTCP-A", + "rx_endp": "testTCP-B", + } + super().jsonPost(reqURL, data) - ptestUDPB = getJsonInfo(mgrURL, "endp/testUDP-B?fields=tx+bytes,rx+bytes") - ptestUDPBTX = ptestUDPB['endpoint']['tx bytes'] - ptestUDPBRX = ptestUDPB['endpoint']['rx bytes'] -except Exception as e: - print("Something went wrong") - print(e) - print("Cleaning up...") - reqURL = "cli-json/rm_vlan" - data = { - "shelf":1, - "resource":resource, - "port":sta_name - } + cxNames = ["testTCP", "testUDP"] + endpNames = ["testTCP-A", "testTCP-B", + "testUDP-A", "testUDP-B"] - jsonReq(mgrURL, reqURL, data) + # start cx traffic + print("\nStarting CX Traffic") + for name in range(len(cxNames)): + reqURL = "cli-json/set_cx_state" + data = { + "test_mgr": "ALL", + "cx_name": cxNames[name], + "cx_state": "RUNNING" + } + super().jsonPost(reqURL, data) - removeCX(mgrURL, cxNames) - removeEndps(mgrURL, endpNames) - sys.exit(1) + # Refresh stats + print("\nRefresh CX stats") + for name in range(len(cxNames)): + reqURL = "cli-json/show_cxe" + data = { + "test_mgr": "ALL", + "cross_connect": cxNames[name] + } + super().jsonPost(reqURL, data) -#compare pre-test values to post-test values + # print("Sleeping for 15 seconds") + time.sleep(15) -def compareVals(name, postVal): - #print(f"Comparing {name}") - if postVal > 0: - print("PASSED: %s %s"%(name, postVal)) - else: - print("FAILED: %s did not report traffic: %s"%(name, postVal)) + # stop cx traffic + print("\nStopping CX Traffic") + for name in range(len(cxNames)): + reqURL = "cli-json/set_cx_state" + data = { + "test_mgr": "ALL", + "cx_name": cxNames[name], + "cx_state": "STOPPED" + } + super().jsonPost(reqURL, data) -print("\n") -compareVals("testTCP-A TX", ptestTCPATX) -compareVals("testTCP-A RX", ptestTCPARX) + # Refresh stats + print("\nRefresh CX stats") + for name in range(len(cxNames)): + reqURL = "cli-json/show_cxe" + data = { + "test_mgr": "ALL", + "cross_connect": cxNames[name] + } + super().jsonPost(reqURL, data) -compareVals("testTCP-B TX", ptestTCPBTX) -compareVals("testTCP-B RX", ptestTCPBRX) + # print("Sleeping for 5 seconds") + time.sleep(5) -compareVals("testUDP-A TX", ptestUDPATX) -compareVals("testUDP-A RX", ptestUDPARX) + # get data for endpoints JSON + print("Collecting Data") + try: + ptestTCPA = super().jsonGet("endp/testTCP-A?fields=tx+bytes,rx+bytes") + ptestTCPATX = ptestTCPA['endpoint']['tx bytes'] + ptestTCPARX = ptestTCPA['endpoint']['rx bytes'] -compareVals("testUDP-B TX", ptestUDPBTX) -compareVals("testUDP-B RX", ptestUDPBRX) -print("\n") + ptestTCPB = super().jsonGet("endp/testTCP-B?fields=tx+bytes,rx+bytes") + ptestTCPBTX = ptestTCPB['endpoint']['tx bytes'] + ptestTCPBRX = ptestTCPB['endpoint']['rx bytes'] + + ptestUDPA = super().jsonGet("endp/testUDP-A?fields=tx+bytes,rx+bytes") + ptestUDPATX = ptestUDPA['endpoint']['tx bytes'] + ptestUDPARX = ptestUDPA['endpoint']['rx bytes'] + + ptestUDPB = super().jsonGet("endp/testUDP-B?fields=tx+bytes,rx+bytes") + ptestUDPBTX = ptestUDPB['endpoint']['tx bytes'] + ptestUDPBRX = ptestUDPB['endpoint']['rx bytes'] + except Exception as e: + print("Something went wrong") + print(e) + print("Cleaning up...") + reqURL = "cli-json/rm_vlan" + data = { + "shelf": 1, + "resource": self.resource, + "port": self.sta_name + } + + self.jsonPost(reqURL, data) + + removeCX(self.mgr_url, cxNames) + removeEndps(self.mgr_url, endpNames) + sys.exit(1) + + print("\n") + self.compareVals("testTCP-A TX", ptestTCPATX) + self.compareVals("testTCP-A RX", ptestTCPARX) + + self.compareVals("testTCP-B TX", ptestTCPBTX) + self.compareVals("testTCP-B RX", ptestTCPBRX) + + self.compareVals("testUDP-A TX", ptestUDPATX) + self.compareVals("testUDP-A RX", ptestUDPARX) + + self.compareVals("testUDP-B TX", ptestUDPBTX) + self.compareVals("testUDP-B RX", ptestUDPBRX) + print("\n") + + # remove all endpoints and cxs + LFUtils.removePort(self.resource, self.sta_name, self.mgr_url) + + removeCX(self.mgr_url, cxNames) + removeEndps(self.mgr_url, endpNames) + +# ~class + + +# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + + +def main(): + lfjson_host = "http://localhost" + lfjson_port = 8080 + parser = argparse.ArgumentParser( + description="""LANforge Unit Test: Connect Station to AP +Example: +./sta_connect.py --dest 192.168.100.209 --dut_ssid OpenWrt-2 --dut_bssid 24:F5:A2:08:21:6C +""") + parser.add_argument("-d", "--dest", 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("-u", "--user", type=str, help="TBD: credential login/username") + parser.add_argument("-p", "--passwd", type=str, help="TBD: credential password") + parser.add_argument("--resource", type=str, help="LANforge Station resource ID to use, default is 1") + parser.add_argument("--upstream_resource", type=str, help="LANforge Ethernet port resource ID to use, default is 1") + parser.add_argument("--upstream_port", type=str, help="LANforge Ethernet port name, default is eth2") + parser.add_argument("--radio", type=str, help="LANforge radio to use, default is wiphy0") + parser.add_argument("--sta_mode", type=str, + help="LANforge station-mode setting (see add_sta LANforge CLI documentation, default is 0 (auto))") + parser.add_argument("--dut_ssid", type=str, help="DUT SSID") + parser.add_argument("--dut_passwd", type=str, help="DUT PSK password. Do not set for OPEN auth") + parser.add_argument("--dut_bssid", type=str, help="DUT BSSID to which we expect to connect.") + + args = parser.parse_args() + if args.dest is not None: + lfjson_host = args.dest + if args.port is not None: + lfjson_port = args.port + + staConnect = StaConnect(lfjson_host, lfjson_port) + + if args.user is not None: + staConnect.user = args.user + if args.passwd is not None: + staConnect.passwd = args.passwd + if args.sta_mode is not None: + staConnect.sta_mode = args.sta_mode + if args.upstream_resource is not None: + staConnect.upstream_resource = args.upstream_resource + if args.upstream_port is not None: + staConnect.upstream_port = args.upstream_port + if args.radio is not None: + staConnect.radio = args.radio + if args.resource is not None: + staConnect.resource = args.resource + if args.dut_passwd is not None: + staConnect.dut_passwd = args.dut_passwd + if args.dut_bssid is not None: + staConnect.dut_bssid = args.dut_bssid + if args.dut_ssid is not None: + staConnect.dut_ssid = args.dut_ssid + + +# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -#remove all endpoints and cxs -LFUtils.removePort(resource, sta_name, mgrURL) +if __name__ == "__main__": + main() + + -removeCX(mgrURL, cxNames) -removeEndps(mgrURL, endpNames) From 356a2cae4fa38c8ac88935e82bd4e998aa9811c9 Mon Sep 17 00:00:00 2001 From: Jed Reynolds Date: Mon, 1 Jun 2020 14:00:25 -0700 Subject: [PATCH 2/8] lfcli_base: renamed module to follow PEP8 reccos --- py-json/LANforge/lfcli_base.py | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) create mode 100644 py-json/LANforge/lfcli_base.py diff --git a/py-json/LANforge/lfcli_base.py b/py-json/LANforge/lfcli_base.py new file mode 100644 index 00000000..d3dfec7a --- /dev/null +++ b/py-json/LANforge/lfcli_base.py @@ -0,0 +1,32 @@ +#!env /usr/bin/python + +# Extend this class to use common set of debug and request features for your script + +from LANforge.LFUtils import * + + +class LFCliBase: + def __init__(self, lfjson_host="localhost", lfjson_port=8080, _debug=False): + self.debugOn = _debug + self.lfjson_host = lfjson_host + self.lfjson_port = lfjson_port + self.mgr_url = f"http://{self.lfjson_host}:{self.lfjson_port}/" + + def jsonPost(self, _req_url, _data): + lf_r = LFRequest.LFRequest(self.mgr_url + _req_url) + _data['suppress_preexec_cli'] = True + _data['suppress_preexec_method'] = True + lf_r.addPostData(_data) + json_response = lf_r.jsonPost(self.debugOn) + # Debugging + # if (json_response != None): + # print("jsonReq: response: ") + # LFUtils.debug_printer.pprint(vars(json_response)) + return json_response + + def jsonGet(self, _req_url): + lf_r = LFRequest.LFRequest(self.mgr_url + _req_url) + json_response = lf_r.getAsJson(self.debugOn) + return json_response + +# ~class From de7314b6a9cd9cace2c1e99fbf4828ce79cbeb23 Mon Sep 17 00:00:00 2001 From: Jed Reynolds Date: Mon, 1 Jun 2020 14:00:37 -0700 Subject: [PATCH 3/8] lfcli_base: renamed module to follow PEP8 reccos --- py-json/LANforge/LFCliBase.py | 32 -------------------------------- 1 file changed, 32 deletions(-) delete mode 100644 py-json/LANforge/LFCliBase.py diff --git a/py-json/LANforge/LFCliBase.py b/py-json/LANforge/LFCliBase.py deleted file mode 100644 index d3dfec7a..00000000 --- a/py-json/LANforge/LFCliBase.py +++ /dev/null @@ -1,32 +0,0 @@ -#!env /usr/bin/python - -# Extend this class to use common set of debug and request features for your script - -from LANforge.LFUtils import * - - -class LFCliBase: - def __init__(self, lfjson_host="localhost", lfjson_port=8080, _debug=False): - self.debugOn = _debug - self.lfjson_host = lfjson_host - self.lfjson_port = lfjson_port - self.mgr_url = f"http://{self.lfjson_host}:{self.lfjson_port}/" - - def jsonPost(self, _req_url, _data): - lf_r = LFRequest.LFRequest(self.mgr_url + _req_url) - _data['suppress_preexec_cli'] = True - _data['suppress_preexec_method'] = True - lf_r.addPostData(_data) - json_response = lf_r.jsonPost(self.debugOn) - # Debugging - # if (json_response != None): - # print("jsonReq: response: ") - # LFUtils.debug_printer.pprint(vars(json_response)) - return json_response - - def jsonGet(self, _req_url): - lf_r = LFRequest.LFRequest(self.mgr_url + _req_url) - json_response = lf_r.getAsJson(self.debugOn) - return json_response - -# ~class From 29c6c7c7b869771d9ad79be97f8d28c76f8a8807 Mon Sep 17 00:00:00 2001 From: Jed Reynolds Date: Mon, 1 Jun 2020 15:21:49 -0700 Subject: [PATCH 4/8] LFRequest.py: tidies up lint errors like redundant parens --- py-json/LANforge/LFRequest.py | 36 ++++++++++++++++------------------- 1 file changed, 16 insertions(+), 20 deletions(-) diff --git a/py-json/LANforge/LFRequest.py b/py-json/LANforge/LFRequest.py index 8d8867d9..cd6a1049 100644 --- a/py-json/LANforge/LFRequest.py +++ b/py-json/LANforge/LFRequest.py @@ -7,16 +7,14 @@ if sys.version_info[0] != 3: print("This script requires Python 3") exit() -import email.message -import http.client + import urllib.request import urllib.error -from urllib import error import urllib.parse import json -import LANforge from LANforge import LFUtils + class LFRequest: Default_Base_URL = "http://localhost:8080" No_Data = {'No Data':0} @@ -45,14 +43,14 @@ class LFRequest: request.headers['Content-type'] = 'application/x-www-form-urlencoded' resp = '' try: - resp = urllib.request.urlopen(request); + resp = urllib.request.urlopen(request) responses.append(resp) return responses[0] except urllib.error.HTTPError as error: if (show_error): print("----- formPost() HTTPError: --------------------------------------------") print("%s: %s; URL: %s"%(error.code, error.reason, request.get_full_url())) - LFUtils.debug_printer.pprint(error.headers()) + LFUtils.debug_printer.pprint(error.headers) #print("Error: ", sys.exc_info()[0]) #print("Request URL:", request.get_full_url()) print("Request Content-type:", request.get_header('Content-type')) @@ -84,11 +82,11 @@ class LFRequest: request.headers['Content-type'] = 'application/json' try: - resp = urllib.request.urlopen(request); + resp = urllib.request.urlopen(request) responses.append(resp) return responses[0] except urllib.error.HTTPError as error: - if (show_error): + if show_error: print("----- jsonPost() HTTPError: --------------------------------------------") print("<%s> HTTP %s: %s"%(request.get_full_url(), error.code, error.reason, )) @@ -99,24 +97,23 @@ class LFRequest: print("Request Data:") LFUtils.debug_printer.pprint(request.data) - if (error.headers): + if error.headers: # the HTTPError is of type HTTPMessage a subclass of email.message - #print(type(error.keys())) + # print(type(error.keys())) for headername in sorted(error.headers.keys()): print ("Response %s: %s "%(headername, error.headers.get(headername))) - if (len(responses) > 0): + if len(responses) > 0: print("----- Response: --------------------------------------------------------") LFUtils.debug_printer.pprint(responses[0].reason) print("------------------------------------------------------------------------") except urllib.error.URLError as uerror: - if (show_error): + if show_error: print("----- jsonPost() URLError: ---------------------------------------------") print("Reason: %s; URL: %s"%(uerror.reason, request.get_full_url())) print("------------------------------------------------------------------------") return None - def get(self, show_error=True): myrequest = urllib.request.Request(url=self.requested_url, headers=self.default_headers) myresponses = [] @@ -124,10 +121,10 @@ class LFRequest: myresponses.append(urllib.request.urlopen(myrequest)) return myresponses[0] except urllib.error.HTTPError as error: - if (show_error): + if show_error: print("----- get() HTTPError: --------------------------------------------") print("<%s> HTTP %s: %s"%(myrequest.get_full_url(), error.code, error.reason, )) - if (error.code != 404): + if error.code != 404: print("Error: ", sys.exc_info()[0]) print("Request URL:", myrequest.get_full_url()) print("Request Content-type:", myrequest.get_header('Content-type')) @@ -135,24 +132,23 @@ class LFRequest: print("Request Data:") LFUtils.debug_printer.pprint(myrequest.data) - if (error.headers): + if error.headers: # the HTTPError is of type HTTPMessage a subclass of email.message - #print(type(error.keys())) + # print(type(error.keys())) for headername in sorted(error.headers.keys()): print ("Response %s: %s "%(headername, error.headers.get(headername))) - if (len(myresponses) > 0): + if len(myresponses) > 0: print("----- Response: --------------------------------------------------------") LFUtils.debug_printer.pprint(myresponses[0].reason) print("------------------------------------------------------------------------") except urllib.error.URLError as uerror: - if (show_error): + if show_error: print("----- get() URLError: ---------------------------------------------") print("Reason: %s; URL: %s"%(uerror.reason, myrequest.get_full_url())) print("------------------------------------------------------------------------") return None - def getAsJson(self, show_error=True): responses = [] responses.append(self.get(show_error)) From f69d4cdb909f4930ca3778f3c3ba763e38670fa5 Mon Sep 17 00:00:00 2001 From: Jed Reynolds Date: Mon, 1 Jun 2020 15:22:49 -0700 Subject: [PATCH 5/8] sta_connect.py, lfcli_base.py: fixes inheritance mistake, continues refactoring, test now runs --- py-json/LANforge/lfcli_base.py | 25 +++++++++++++++++++++---- py-scripts/sta_connect.py | 33 +++++++++------------------------ 2 files changed, 30 insertions(+), 28 deletions(-) diff --git a/py-json/LANforge/lfcli_base.py b/py-json/LANforge/lfcli_base.py index d3dfec7a..ac70efbb 100644 --- a/py-json/LANforge/lfcli_base.py +++ b/py-json/LANforge/lfcli_base.py @@ -6,11 +6,14 @@ from LANforge.LFUtils import * class LFCliBase: - def __init__(self, lfjson_host="localhost", lfjson_port=8080, _debug=False): + # do not use `super(LFCLiBase,self).__init__(self, host, port, _debugOn) + # that is py2 era syntax and will force self into the host variable, making you + # very confused. + def __init__(self, _lfjson_host, _lfjson_port, _debug=False): + self.lfjson_host = _lfjson_host + self.lfjson_port = _lfjson_port self.debugOn = _debug - self.lfjson_host = lfjson_host - self.lfjson_port = lfjson_port - self.mgr_url = f"http://{self.lfjson_host}:{self.lfjson_port}/" + self.mgr_url = "http://%s:%s/" % (self.lfjson_host, self.lfjson_port) def jsonPost(self, _req_url, _data): lf_r = LFRequest.LFRequest(self.mgr_url + _req_url) @@ -29,4 +32,18 @@ class LFCliBase: json_response = lf_r.getAsJson(self.debugOn) return json_response + def checkConnect(self): + print(f"Checking for LANforge GUI connection: {self.mgr_url}") + response = self.jsonGet("") + duration = 0 + while (response is None) and (duration < 300): + print(f"LANforge GUI connection not found sleeping 5 seconds, tried: {self.mgr_url}") + duration += 2 + time.sleep(2) + response = self.jsonGet("") + + if duration >= 300: + print("Could not connect to LANforge GUI") + sys.exit(1) + # ~class diff --git a/py-scripts/sta_connect.py b/py-scripts/sta_connect.py index 46144919..4fa38408 100755 --- a/py-scripts/sta_connect.py +++ b/py-scripts/sta_connect.py @@ -14,7 +14,7 @@ if 'py-json' not in sys.path: # from LANforge import LFRequest from LANforge import LFUtils # from LANforge import LFCliBase -from LANforge.LFCliBase import LFCliBase +from LANforge.lfcli_base import LFCliBase from LANforge.LFUtils import * if sys.version_info[0] != 3: @@ -23,14 +23,15 @@ if sys.version_info[0] != 3: class StaConnect(LFCliBase): - def __init__(self, lfjson_host, lfjson_port, _dut_ssid="MyAP", _dut_passwd="NA", _dut_bssid="", + def __init__(self, host, port, _dut_ssid="MyAP", _dut_passwd="NA", _dut_bssid="", _user="", _passwd="", _sta_mode="0", _radio="wiphy0", _resource=1, _upstream_resource=1, _upstream_port="eth2", _sta_name="sta001", _debugOn=False): - # self.dest = "localhost" # in super - # self.port = "8080" # in super + # do not use `super(LFCLiBase,self).__init__(self, host, port, _debugOn) + # that is py2 era syntax and will force self into the host variable, making you + # very confused. + super().__init__(host, port, _debugOn) - super().__init__(self, lfjson_host, lfjson_port) self.dut_ssid = _dut_ssid self.dut_passwd = _dut_passwd self.dut_bssid = _dut_bssid @@ -55,20 +56,6 @@ class StaConnect(LFCliBase): self.upstream_url = f"port/1/{self.upstream_resource}/{self.upstream_port}" return self.upstream_url - def checkConnect(self): - print(f"Checking for LANforge GUI connection: {self.mgr_url}") - response = super().jsonGet("/") - duration = 0 - while (response is None) and (duration < 300): - print(f"LANforge GUI connection not found sleeping 5 seconds, tried: {self.mgr_url}") - duration += 2 - time.sleep(2) - response = super().jsonGet("/") - - if duration >= 300: - print("Could not connect to LANforge GUI") - sys.exit(1) - # Compare pre-test values to post-test values @staticmethod def compareVals(name, postVal): @@ -360,7 +347,7 @@ class StaConnect(LFCliBase): def main(): - lfjson_host = "http://localhost" + lfjson_host = "localhost" lfjson_port = 8080 parser = argparse.ArgumentParser( description="""LANforge Unit Test: Connect Station to AP @@ -410,13 +397,11 @@ Example: if args.dut_ssid is not None: staConnect.dut_ssid = args.dut_ssid + staConnect.run() + # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - if __name__ == "__main__": main() - - - From f9d8849bba27cec6ebb001d1af8093d2cd7bdc7d Mon Sep 17 00:00:00 2001 From: Jed Reynolds Date: Mon, 1 Jun 2020 21:44:50 -0700 Subject: [PATCH 6/8] LFRequest.py: adds more debugging --- py-json/LANforge/lfcli_base.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/py-json/LANforge/lfcli_base.py b/py-json/LANforge/lfcli_base.py index ac70efbb..6309b434 100644 --- a/py-json/LANforge/lfcli_base.py +++ b/py-json/LANforge/lfcli_base.py @@ -1,7 +1,8 @@ #!env /usr/bin/python # Extend this class to use common set of debug and request features for your script - +from pprint import pprint +import LANforge.LFUtils from LANforge.LFUtils import * @@ -20,6 +21,8 @@ class LFCliBase: _data['suppress_preexec_cli'] = True _data['suppress_preexec_method'] = True lf_r.addPostData(_data) + if (self.debugOn): + LANforge.LFUtils.debug_printer.pprint(_data) json_response = lf_r.jsonPost(self.debugOn) # Debugging # if (json_response != None): @@ -34,7 +37,7 @@ class LFCliBase: def checkConnect(self): print(f"Checking for LANforge GUI connection: {self.mgr_url}") - response = self.jsonGet("") + response = self.jsonGet("/") duration = 0 while (response is None) and (duration < 300): print(f"LANforge GUI connection not found sleeping 5 seconds, tried: {self.mgr_url}") From def0120da28d3e3a053ee7d2cf235ec23bece444 Mon Sep 17 00:00:00 2001 From: Jed Reynolds Date: Mon, 1 Jun 2020 21:45:38 -0700 Subject: [PATCH 7/8] LFRequest.py: uses wpa2 flag; uses waitX methods --- py-scripts/sta_connect.py | 33 ++++++++++++++++----------------- 1 file changed, 16 insertions(+), 17 deletions(-) diff --git a/py-scripts/sta_connect.py b/py-scripts/sta_connect.py index 4fa38408..28fe852d 100755 --- a/py-scripts/sta_connect.py +++ b/py-scripts/sta_connect.py @@ -16,6 +16,7 @@ from LANforge import LFUtils # from LANforge import LFCliBase from LANforge.lfcli_base import LFCliBase from LANforge.LFUtils import * +from pprint import pprint if sys.version_info[0] != 3: print("This script requires Python 3") @@ -67,8 +68,9 @@ class StaConnect(LFCliBase): def run(self): self.checkConnect() - # Create stations and turn dhcp on - print("Creating station and turning on dhcp") + eth1IP = self.jsonGet(self.getUpstreamUrl()) + if eth1IP['interface']['ip'] == "0.0.0.0": + print(f"Warning: {self.getUpstreamUrl()} lacks ip address") url = self.getStaUrl() response = super().jsonGet(url) @@ -76,10 +78,14 @@ class StaConnect(LFCliBase): if response["interface"] is not None: print("removing old station") LFUtils.removePort(self.resource, self.sta_name, self.mgr_url) - time.sleep(5) + LFUtils.waitUntilPortsDisappear(self.resource, self.mgr_url, [self.sta_name]) - # See add_sta in LANforge CLI user guide + # Create stations and turn dhcp on + print("Creating station %s and turning on dhcp..." % self.sta_name) url = "cli-json/add_sta" + flags = 0x10000 + if "" != self.dut_passwd: + flags += 0x400 data = { "shelf": 1, "resource": self.resource, @@ -89,9 +95,9 @@ class StaConnect(LFCliBase): "key": self.dut_passwd, "mode": self.sta_mode, "mac": "xx:xx:xx:xx:*:xx", - "flags": 0x10000 # verbose, open + "flags": flags # verbose, wpa2 } - print("adding new station") + print("Adding new station %s " % self.sta_name) super().jsonPost(url, data) reqURL = "cli-json/set_port" @@ -102,21 +108,16 @@ class StaConnect(LFCliBase): "current_flags": 0x80000000, # use DHCP, not down "interest": 0x4002 # set dhcp, current flags } - print("configuring port") + print("Configuring %s..." % self.sta_name) super().jsonPost(reqURL, data) - time.sleep(5) - - eth1IP = self.jsonGet(self.getUpstreamUrl()) - if eth1IP['interface']['ip'] == "0.0.0.0": - print(f"Warning: {self.getUpstreamUrl()} lacks ip address") - reqURL = "cli-json/nc_show_ports" data = {"shelf": 1, "resource": self.resource, "port": self.sta_name, "probe_flags": 1} super().jsonPost(reqURL, data) + LFUtils.waitUntilPortsAdminUp(self.resource, self.mgr_url, [self.sta_name]) # station_info = self.jsonGet(self.mgr_url, "%s?fields=port,ip,ap" % (self.getStaUrl())) duration = 0 @@ -124,10 +125,8 @@ class StaConnect(LFCliBase): ip = "0.0.0.0" ap = "" while (ip == "0.0.0.0") and (duration < maxTime): - duration += 2 time.sleep(2) - station_info = super().jsonGet(f"{self.getStaUrl()}?fields=port,ip,ap") # LFUtils.debug_printer.pprint(station_info) @@ -138,10 +137,10 @@ class StaConnect(LFCliBase): ap = station_info["interface"]["ap"] if (ap == "Not-Associated") or (ap == ""): - print("Station waiting to associate...") + print("Waiting for %s associate to AP [%s]..." % (self.sta_name, ap)) else: if ip == "0.0.0.0": - print("Station waiting for IP ...") + print("Waiting for %s to gain IP ..." % self.sta_name) if (ap != "") and (ap != "Not-Associated"): print(f"Connected to AP: {ap}") From ad1abd4bcbb44cecc756e3e362f7b643ca3c4ee5 Mon Sep 17 00:00:00 2001 From: Jed Reynolds Date: Mon, 1 Jun 2020 21:47:37 -0700 Subject: [PATCH 8/8] removes realmX scripts from py-scripts --- py-scripts/realm.py | 204 --------------------------------------- py-scripts/realm_test.py | 26 ----- 2 files changed, 230 deletions(-) delete mode 100755 py-scripts/realm.py delete mode 100755 py-scripts/realm_test.py diff --git a/py-scripts/realm.py b/py-scripts/realm.py deleted file mode 100755 index 9ada4a98..00000000 --- a/py-scripts/realm.py +++ /dev/null @@ -1,204 +0,0 @@ -#!/usr/bin/env python3 -import os -import sys -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 cxList(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): - #Returns list of all stations with "sta" in their name - 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]) - - return list - - def vapList(self): - #Returns list of all VAPs with "vap" in their name - 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]) - - return list - - - def findPortsLike(self, pattern=""): - #Searches for ports that match a given pattern and returns a list of names - 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) - #print(response) - for x in range(len(response['interfaces'])): - for k,v in response['interfaces'][x].items(): - if v['device'] != "NA": - list.append(v['device']) - - matchedList = [] - - prefix = "" - for portname in list: - try: - 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) - - 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) - - 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 - except ValueError as e: - print(e) - return matchedList - -class CXProfile: - - def __init__(self, mgrURL="http://localhost:8080"): - self.mgrURL = mgrURL - self.postData = [] - - def addPorts(self, side, endpType, ports=[]): - #Adds post data for a cross-connect between eth1 and specified list of ports, appends to array - side = side.upper() - endpSideA = { - "alias":"", - "shelf":1, - "resource":1, - "port":"", - "type":endpType, - "min_rate":0, - "max_rate":0, - "min_pkt":-1, - "max_pkt":0 - } - - endpSideB = { - "alias":"", - "shelf":1, - "resource":1, - "port":"", - "type":endpType, - "min_rate":0, - "max_rate":0, - "min_pkt":-1, - "max_pkt":0 - } - - for portName in ports: - if side == "A": - endpSideA["alias"] = portName+"CX-A" - endpSideA["port"] = portName - endpSideB["alias"] = portName+"CX-B" - endpSideB["port"] = "eth1" - elif side == "B": - endpSideA["alias"] = portName+"CX-A" - endpSideA["port"] = "eth1" - endpSideB["alias"] = portName+"CX-B" - endpSideB["port"] = portName - - lf_r = LFRequest.LFRequest(self.mgrURL + "/cli-json/add_endp") - lf_r.addPostData(endpSideA) - json_response = lf_r.jsonPost(True) - lf_r.addPostData(endpSideB) - json_response = lf_r.jsonPost(True) - #LFUtils.debug_printer.pprint(json_response) - time.sleep(.5) - - - data = { - "alias":portName+"CX", - "test_mgr":"default_tm", - "tx_endp":portName + "CX-A", - "rx_endp":portName + "CX-B" - } - - self.postData.append(data) - - def create(self, sleepTime=.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") - lf_r.addPostData(data) - json_response = lf_r.jsonPost(True) - #LFUtils.debug_printer.pprint(json_response) - time.sleep(sleepTime) - - -class StationProfile: - - def __init__(self, mgrURL="localhost:8080", ssid="NA", ssidPass="NA", security="open", startID="", mode=0, up=True, dhcp=True): - self.mgrURL = mgrURL - self.ssid = ssid - self.ssidPass = ssidPass - self.mode = mode - self.up = up - self.dhcp = dhcp - self.security = security - - def build(self, resourceRadio, numStations): - #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): - 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): - data = { - "shelf":1, - "resource":1, - "radio":radio, - "sta_name":f"sta{num:05}", - "ssid":self.ssid, - "key":self.ssidPass, - "mode":1, - "mac":"xx:xx:xx:xx:*:xx", - "flags": - } - lf_r.addPostData(data) - json_response = lf_r.jsonPost(True) - diff --git a/py-scripts/realm_test.py b/py-scripts/realm_test.py deleted file mode 100755 index bf3e376b..00000000 --- a/py-scripts/realm_test.py +++ /dev/null @@ -1,26 +0,0 @@ -#!/usr/bin/env python3 - -import Realm - - -test = Realm.Realm("http://localhost:8080") - -staList = test.stationList() -cxList = test.cxList() -vapList = test.vapList() - - -print(f"CXs: {cxList}\n") -print(f"Stations: {staList}\n") -print(f"VAPs: {vapList}\n") - -cxTest = Realm.CXProfile() - -cxTest.addPorts("A", "lf_udp", test.findPortsLike("sta+")) -cxTest.create() - -print(test.findPortsLike("sta+")) - -print(test.findPortsLike("sta0*")) - -print(test.findPortsLike("sta[0000..0002]"))