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

sta_connect: sta_connect works now

Browse Source 
Signed-off-by: Matthew Stidham <stidmatt@gmail.com>
This commit is contained in:
Matthew Stidham
2021-11-17 17:34:43 -08:00
parent 81bbc64dbc
commit 4d7451a287
1 changed files with 34 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

92
py-scripts/sta_connect.py
View File

@@ -19,8 +19,6 @@ sys.path.append(os.path.join(os.path.abspath(__file__ + "../../../")))
LFUtils = importlib.import_module("py-json.LANforge.LFUtils")
removeCX = LFUtils.removeCX
removeEndps = LFUtils.removeEndps
lfcli_base = importlib.import_module("py-json.LANforge.lfcli_base")
LFCliBase = lfcli_base.LFCliBase
realm = importlib.import_module("py-json.realm")
Realm = realm.Realm
@@ -31,16 +29,20 @@ WPA2 = "wpa2"
MODE_AUTO = 0
class StaConnect(LFCliBase):
class StaConnect(Realm):
def __init__(self, host, port, _dut_ssid="MyAP", _dut_passwd="NA", _dut_bssid="",
_user="", _passwd="", _sta_mode="0", _radio="wiphy0",
_user="lanforge", _passwd="lanforge", _sta_mode="0", _radio="wiphy0",
_resource=1, _upstream_resource=1, _upstream_port="eth2",
_sta_name=None, _debugOn=False, _dut_security=OPEN, _exit_on_error=False,
_cleanup_on_exit=True, _runtime_sec=60, _exit_on_fail=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.
super().__init__(host, port, _debug=_debugOn, _exit_on_fail=_exit_on_fail)
super().__init__(lfclient_host=host, lfclient_port=port, debug_=_debugOn, _exit_on_fail=_exit_on_fail)
fields = "_links,port,alias,ip,ap,port+type"
self.station_results = self.find_ports_like("sta*", fields, debug_=False)
self.host = host
self.port = port
self.debugOn = _debugOn
self.dut_security = ""
self.dut_ssid = _dut_ssid
@@ -61,14 +63,12 @@ class StaConnect(LFCliBase):
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
self.localrealm = Realm(lfclient_host=host, lfclient_port=port) # py > 3.6
self.resulting_stations = {}
self.resulting_endpoints = {}
# def get_realm(self) -> Realm: # py > 3.6
def get_realm(self):
return self.localrealm
self.cx_profile = self.new_l3_cx_profile()
self.cx_profile.host = self.host
self.cx_profile.port = self.port
def get_station_url(self, sta_name_=None):
if sta_name_ is None:
@@ -100,7 +100,6 @@ class StaConnect(LFCliBase):
counter = 0
# print("there are %d results" % len(self.station_results))
fields = "_links,port,alias,ip,ap,port+type"
self.station_results = self.localrealm.find_ports_like("sta*", fields, debug_=False)
if (self.station_results is None) or (len(self.station_results) < 1):
self.get_failed_result_list()
for eid,record in self.station_results.items():
@@ -152,8 +151,7 @@ class StaConnect(LFCliBase):
if response is not None:
if response["interface"] is not None:
print("removing old station")
for sta_name in self.station_names:
LFUtils.removePort(self.resource, sta_name, self.lfclient_url)
LFUtils.removePort(self.resource, sta_name, self.lfclient_url, debug=False)
LFUtils.waitUntilPortsDisappear(self.resource, self.lfclient_url, self.station_names)
# Create stations and turn dhcp on
@@ -242,7 +240,6 @@ class StaConnect(LFCliBase):
# make a copy of the connected stations for test records
for sta_name in self.station_names:
sta_url = self.get_station_url(sta_name)
station_info = self.json_get(sta_url) # + "?fields=port,ip,ap")
@@ -275,7 +272,6 @@ class StaConnect(LFCliBase):
# create endpoints and cxs
# Create UDP endpoints
self.cx_names = {}
for sta_name in self.station_names:
self.cx_names["testUDP-"+sta_name] = { "a": "testUDP-%s-A" % sta_name,
@@ -333,8 +329,8 @@ class StaConnect(LFCliBase):
self.json_post("/cli-json/set_cx_report_timer", data, suppress_related_commands_=True)
# Create TCP endpoints
self.cx_names["testTCP-"+sta_name] = { "a": "testUDP-%s-A" % sta_name,
"b": "testUDP-%s-B" % sta_name}
self.cx_names["testTCP-"+sta_name] = {"a": "testTCP-%s-A" % sta_name,
"b": "testTCP-%s-B" % sta_name}
data = {
"alias": "testTCP-%s-A" % sta_name,
"shelf": 1,
@@ -373,6 +369,7 @@ class StaConnect(LFCliBase):
}
self.json_post("/cli-json/set_cx_report_timer", data, suppress_related_commands_=True)
self.wait_until_endps_appear(self.cx_names)
return True
def start(self):
@@ -396,19 +393,9 @@ class StaConnect(LFCliBase):
}
self.json_post("/cli-json/show_cxe", data)
return True
def stop(self):
# stop cx traffic
print("Stopping CX Traffic")
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
self.cx_profile.stop_cx()
def finish(self):
# Refresh stats
@@ -436,7 +423,7 @@ class StaConnect(LFCliBase):
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" % self.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
@@ -458,59 +445,46 @@ class StaConnect(LFCliBase):
# self.test_results.append("FAILED message will fail")
# print("\n")
def cleanup(self):
for sta_name in self.station_names:
LFUtils.removePort(self.resource, sta_name, self.lfclient_url)
LFUtils.removePort(self.resource, sta_name, self.lfclient_url, debug=False)
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
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
if self.debug:
print("Removing endpoints %s" % self.cx_names.values())
removeEndps(self.lfclient_url, endp_names, debug=False)
def main():
lfjson_host = "localhost"
lfjson_port = 8080
parser = argparse.ArgumentParser(
parser = Realm.create_basic_argparse(
prog="sta_connect.py",
formatter_class=argparse.RawTextHelpFormatter,
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
./sta_connect.py --mgr 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.")
parser.add_argument('--test_duration', help='--test_duration sets the duration of the test', default="2m")
args = parser.parse_args()
if args.dest is not None:
lfjson_host = args.dest
monitor_interval = Realm.parse_time(args.test_duration).total_seconds()
if args.mgr is not None:
lfjson_host = args.mgr
if args.port is not None:
lfjson_port = args.port
staConnect = StaConnect(lfjson_host, lfjson_port)
staConnect.station_names = [ "sta0000" ]
if args.user is not None:
staConnect.user = args.user
if args.passwd is not None:
staConnect.passwd = args.passwd
staConnect = StaConnect(lfjson_host, lfjson_port, _runtime_sec=monitor_interval)
staConnect.station_names = ["sta0000"]
if args.sta_mode is not None:
staConnect.sta_mode = args.sta_mode
if args.upstream_resource is not None:
@@ -521,12 +495,14 @@ Example:
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.passwd is not None:
staConnect.dut_passwd = args.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
if args.ssid is not None:
staConnect.dut_ssid = args.ssid
if args.security is not None:
staConnect.dut_security = args.security
staConnect.run()