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

sta_connect: Fix whitespace

Browse Source 
Signed-off-by: Matthew Stidham <stidmatt@gmail.com>
This commit is contained in:
Matthew Stidham
2021-12-06 15:12:39 -08:00
parent ed03860677
commit 25f537a027
1 changed files with 25 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
py-scripts/sta_connect.py
View File

@@ -13,7 +13,6 @@ if sys.version_info[0] != 3:
print("This script requires Python 3") print("This script requires Python 3")
exit(1) exit(1)
sys.path.append(os.path.join(os.path.abspath(__file__ + "../../../"))) sys.path.append(os.path.join(os.path.abspath(__file__ + "../../../")))
LFUtils = importlib.import_module("py-json.LANforge.LFUtils") LFUtils = importlib.import_module("py-json.LANforge.LFUtils")
@@ -62,7 +61,7 @@ class StaConnect(Realm):
self.station_names = [] self.station_names = []
self.cx_names = {} self.cx_names = {}
if _sta_name is not None: if _sta_name is not None:
self.station_names = [ _sta_name ] self.station_names = [_sta_name]
self.resulting_stations = {} self.resulting_stations = {}
self.resulting_endpoints = {} self.resulting_endpoints = {}
@@ -99,23 +98,22 @@ class StaConnect(Realm):
def num_associated(self, bssid): def num_associated(self, bssid):
counter = 0 counter = 0
# print("there are %d results" % len(self.station_results)) # print("there are %d results" % len(self.station_results))
fields = "_links,port,alias,ip,ap,port+type"
if (self.station_results is None) or (len(self.station_results) < 1): if (self.station_results is None) or (len(self.station_results) < 1):
self.get_failed_result_list() self.get_failed_result_list()
for eid,record in self.station_results.items(): for eid, record in self.station_results.items():
#print("-- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- ") # print("-- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- ")
#pprint(eid) # pprint(eid)
#pprint(record) # pprint(record)
if record["ap"] == bssid: if record["ap"] == bssid:
counter += 1 counter += 1
#print("-- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- ") # print("-- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- ")
return counter return counter
def clear_test_results(self): def clear_test_results(self):
self.resulting_stations = {} self.resulting_stations = {}
self.resulting_endpoints = {} self.resulting_endpoints = {}
super().clear_test_results() super().clear_test_results()
#super(StaConnect, self).clear_test_results().test_results.clear() # super(StaConnect, self).clear_test_results().test_results.clear()
def run(self): def run(self):
if not self.setup(): if not self.setup():
@@ -231,7 +229,7 @@ class StaConnect(Realm):
else: else:
connected_stations[sta_name] = sta_url connected_stations[sta_name] = sta_url
data = { data = {
"shelf":1, "shelf": 1,
"resource": self.resource, "resource": self.resource,
"port": "ALL", "port": "ALL",
"probe_flags": 1 "probe_flags": 1
@@ -242,7 +240,7 @@ class StaConnect(Realm):
for sta_name in self.station_names: for sta_name in self.station_names:
sta_url = self.get_station_url(sta_name) sta_url = self.get_station_url(sta_name)
station_info = self.json_get(sta_url) # + "?fields=port,ip,ap") station_info = self.json_get(sta_url) # + "?fields=port,ip,ap")
self.resulting_stations[sta_url] = station_info self.resulting_stations[sta_url] = station_info
ap = station_info["interface"]["ap"] ap = station_info["interface"]["ap"]
ip = station_info["interface"]["ip"] ip = station_info["interface"]["ip"]
@@ -250,13 +248,14 @@ class StaConnect(Realm):
print(" %s +AP %s, " % (sta_name, ap), end="") print(" %s +AP %s, " % (sta_name, ap), end="")
if self.dut_bssid != "": if self.dut_bssid != "":
if self.dut_bssid.lower() == ap.lower(): if self.dut_bssid.lower() == ap.lower():
self._pass(sta_name+" connected to BSSID: " + ap) self._pass(sta_name + " connected to BSSID: " + ap)
# self.test_results.append("PASSED: ) # self.test_results.append("PASSED: )
# print("PASSED: Connected to BSSID: "+ap) # print("PASSED: Connected to BSSID: "+ap)
else: else:
self._fail("%s connected to wrong BSSID, requested: %s Actual: %s" % (sta_name, self.dut_bssid, ap)) self._fail(
"%s connected to wrong BSSID, requested: %s Actual: %s" % (sta_name, self.dut_bssid, ap))
else: else:
self._fail(sta_name+" did not connect to AP") self._fail(sta_name + " did not connect to AP")
return False return False
if ip == "0.0.0.0": if ip == "0.0.0.0":
@@ -274,8 +273,8 @@ class StaConnect(Realm):
# Create UDP endpoints # Create UDP endpoints
for sta_name in self.station_names: for sta_name in self.station_names:
self.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} "b": "testUDP-%s-B" % sta_name}
data = { data = {
"alias": "testUDP-%s-A" % sta_name, "alias": "testUDP-%s-A" % sta_name,
"shelf": 1, "shelf": 1,
@@ -288,9 +287,9 @@ class StaConnect(Realm):
self.json_post("/cli-json/add_endp", data, suppress_related_commands_=True) self.json_post("/cli-json/add_endp", data, suppress_related_commands_=True)
data = { data = {
"name" : "testUDP-%s-A" % sta_name, "name": "testUDP-%s-A" % sta_name,
"flag" : "UseAutoNAT", "flag": "UseAutoNAT",
"val" : 1 "val": 1
} }
self.json_post("/cli-json/set_endp_flag", data, suppress_related_commands_=True) self.json_post("/cli-json/set_endp_flag", data, suppress_related_commands_=True)
@@ -306,9 +305,9 @@ class StaConnect(Realm):
self.json_post("/cli-json/add_endp", data, suppress_related_commands_=True) self.json_post("/cli-json/add_endp", data, suppress_related_commands_=True)
data = { data = {
"name" : "testUDP-%s-B" % sta_name, "name": "testUDP-%s-B" % sta_name,
"flag" : "UseAutoNAT", "flag": "UseAutoNAT",
"val" : 1 "val": 1
} }
self.json_post("/cli-json/set_endp_flag", data, suppress_related_commands_=True) self.json_post("/cli-json/set_endp_flag", data, suppress_related_commands_=True)
@@ -329,8 +328,8 @@ class StaConnect(Realm):
self.json_post("/cli-json/set_cx_report_timer", data, suppress_related_commands_=True) self.json_post("/cli-json/set_cx_report_timer", data, suppress_related_commands_=True)
# Create TCP endpoints # Create TCP endpoints
self.cx_names["testTCP-"+sta_name] = {"a": "testTCP-%s-A" % sta_name, self.cx_names["testTCP-" + sta_name] = {"a": "testTCP-%s-A" % sta_name,
"b": "testTCP-%s-B" % sta_name} "b": "testTCP-%s-B" % sta_name}
data = { data = {
"alias": "testTCP-%s-A" % sta_name, "alias": "testTCP-%s-A" % sta_name,
"shelf": 1, "shelf": 1,
@@ -506,7 +505,7 @@ Example:
staConnect.run() staConnect.run()
run_results = staConnect.get_result_list() staConnect.get_result_list()
if not staConnect.passes(): if not staConnect.passes():
print("FAIL: Some tests failed") print("FAIL: Some tests failed")
@@ -522,6 +521,7 @@ Example:
print(staConnect.get_all_message()) print(staConnect.get_all_message())
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - # - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -