mirror of
https://github.com/Telecominfraproject/wlan-lanforge-scripts.git
synced 2025-10-31 18:58:01 +00:00
lf_cleanup.py iterates over the individual cxs, endp, and stations
test_l3_longevity.py : comment on channel width Signed-off-by: Chuck SmileyRekiere <chuck.smileyrekiere@candelatech.com>
This commit is contained in:
Chuck SmileyRekiere
@@ -29,95 +29,159 @@ Realm = realm.Realm
|
||||
class lf_clean(Realm):
|
||||
def __init__(self,
|
||||
host="localhost",
|
||||
port=8080):
|
||||
port=8080,
|
||||
clean_cxs=None,
|
||||
clean_endp=None,
|
||||
clean_sta=None):
|
||||
super().__init__(lfclient_host=host,
|
||||
lfclient_port=port),
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.clean_cxs = clean_cxs
|
||||
self.clean_endp = clean_endp
|
||||
self.clean_sta = clean_sta
|
||||
self.cxs_done = False
|
||||
self.endp_done = False
|
||||
self.sta_done = False
|
||||
|
||||
def cleanup(self):
|
||||
try:
|
||||
sta_json = super().json_get(
|
||||
"port/1/1/list?field=alias")['interfaces']
|
||||
except TypeError:
|
||||
sta_json = None
|
||||
cx_json = super().json_get("cx")
|
||||
endp_json = super().json_get("endp")
|
||||
|
||||
# get and remove current stations
|
||||
if sta_json is not None:
|
||||
print("Removing old stations")
|
||||
for name in list(sta_json):
|
||||
for alias in list(name):
|
||||
if 'sta' in alias:
|
||||
# print(alias)
|
||||
info = self.name_to_eid(alias)
|
||||
req_url = "cli-json/rm_vlan"
|
||||
def cxs_clean(self):
|
||||
still_looking_cxs = True
|
||||
iterations_cxs = 1
|
||||
while still_looking_cxs and iterations_cxs <= 10:
|
||||
iterations_cxs += 1
|
||||
print("cxs_clean: iterations_cxs: {iterations_cxs}".format(iterations_cxs=iterations_cxs))
|
||||
cx_json = super().json_get("cx")
|
||||
if cx_json is not None:
|
||||
print("Removing old cross connects")
|
||||
for name in list(cx_json):
|
||||
if name != 'handler' and name != 'uri':
|
||||
# print(name)
|
||||
req_url = "cli-json/rm_cx"
|
||||
data = {
|
||||
"shelf": info[0],
|
||||
"resource": info[1],
|
||||
"port": info[2]
|
||||
"test_mgr": "default_tm",
|
||||
"cx_name": name
|
||||
}
|
||||
# print(data)
|
||||
super().json_post(req_url, data)
|
||||
time.sleep(.5)
|
||||
if 'wlan' in alias:
|
||||
info = self.name_to_eid(alias)
|
||||
req_url = "cli-json/rm_vlan"
|
||||
data = {
|
||||
"shelf": info[0],
|
||||
"resource": info[1],
|
||||
"port": info[2]
|
||||
}
|
||||
# print(data)
|
||||
super().json_post(req_url, data)
|
||||
time.sleep(.5)
|
||||
if 'Unknown' in alias:
|
||||
info = self.name_to_eid(alias)
|
||||
req_url = "cli-json/rm_vlan"
|
||||
data = {
|
||||
"shelf": info[0],
|
||||
"resource": info[1],
|
||||
"port": info[2]
|
||||
}
|
||||
# print(data)
|
||||
super().json_post(req_url, data)
|
||||
time.sleep(.5)
|
||||
time.sleep(1)
|
||||
else:
|
||||
print("No cross connects found to cleanup")
|
||||
still_looking_cxs = False
|
||||
print("clean_cxs still_looking_cxs {cxs_looking}".format(cxs_looking=still_looking_cxs))
|
||||
if not still_looking_cxs:
|
||||
self.cxs_done = True
|
||||
return still_looking_cxs
|
||||
|
||||
else:
|
||||
print("No stations found to cleanup")
|
||||
|
||||
# get and remove current cxs
|
||||
if cx_json is not None:
|
||||
print("Removing old cross connects")
|
||||
for name in list(cx_json):
|
||||
if name != 'handler' and name != 'uri':
|
||||
# print(name)
|
||||
req_url = "cli-json/rm_cx"
|
||||
def endp_clean(self):
|
||||
still_looking_endp = True
|
||||
iterations_endp = 0
|
||||
while still_looking_endp and iterations_endp <= 10:
|
||||
iterations_endp += 1
|
||||
print("endp_clean: iterations_endp: {iterations_endp}".format(iterations_endp=iterations_endp))
|
||||
# get and remove current endps
|
||||
endp_json = super().json_get("endp")
|
||||
if endp_json is not None:
|
||||
print("Removing old endpoints")
|
||||
for name in list(endp_json['endpoint']):
|
||||
# print(list(name)[0])
|
||||
req_url = "cli-json/rm_endp"
|
||||
data = {
|
||||
"test_mgr": "default_tm",
|
||||
"cx_name": name
|
||||
"endp_name": list(name)[0]
|
||||
}
|
||||
# print(data)
|
||||
super().json_post(req_url, data)
|
||||
time.sleep(.5)
|
||||
else:
|
||||
print("No cross connects found to cleanup")
|
||||
time.sleep(1)
|
||||
else:
|
||||
print("No endpoints found to cleanup")
|
||||
still_looking_endp = False
|
||||
print("clean_endp still_looking_endp {ednp_looking}".format(ednp_looking=still_looking_endp))
|
||||
if not still_looking_endp:
|
||||
self.endp_done = True
|
||||
return still_looking_endp
|
||||
|
||||
# get and remove current endps
|
||||
if endp_json is not None:
|
||||
print("Removing old endpoints")
|
||||
for name in list(endp_json['endpoint']):
|
||||
# print(list(name)[0])
|
||||
req_url = "cli-json/rm_endp"
|
||||
data = {
|
||||
"endp_name": list(name)[0]
|
||||
}
|
||||
# print(data)
|
||||
super().json_post(req_url, data)
|
||||
time.sleep(.5)
|
||||
else:
|
||||
print("No endpoints found to cleanup")
|
||||
def sta_clean(self):
|
||||
still_looking_sta = True
|
||||
iterations_sta = 0
|
||||
while still_looking_sta and iterations_sta <= 10:
|
||||
iterations_sta += 1
|
||||
print("sta_clean: iterations_sta: {iterations_sta}".format(iterations_sta=iterations_sta))
|
||||
try:
|
||||
sta_json = super().json_get(
|
||||
"port/1/1/list?field=alias")['interfaces']
|
||||
except TypeError:
|
||||
sta_json = None
|
||||
|
||||
# get and remove current stations
|
||||
if sta_json is not None:
|
||||
print("Removing old stations")
|
||||
for name in list(sta_json):
|
||||
for alias in list(name):
|
||||
if 'sta' in alias:
|
||||
# print(alias)
|
||||
info = self.name_to_eid(alias)
|
||||
req_url = "cli-json/rm_vlan"
|
||||
data = {
|
||||
"shelf": info[0],
|
||||
"resource": info[1],
|
||||
"port": info[2]
|
||||
}
|
||||
# print(data)
|
||||
super().json_post(req_url, data)
|
||||
time.sleep(.5)
|
||||
if 'wlan' in alias:
|
||||
info = self.name_to_eid(alias)
|
||||
req_url = "cli-json/rm_vlan"
|
||||
data = {
|
||||
"shelf": info[0],
|
||||
"resource": info[1],
|
||||
"port": info[2]
|
||||
}
|
||||
# print(data)
|
||||
super().json_post(req_url, data)
|
||||
time.sleep(.5)
|
||||
if 'Unknown' in alias:
|
||||
info = self.name_to_eid(alias)
|
||||
req_url = "cli-json/rm_vlan"
|
||||
data = {
|
||||
"shelf": info[0],
|
||||
"resource": info[1],
|
||||
"port": info[2]
|
||||
}
|
||||
# print(data)
|
||||
super().json_post(req_url, data)
|
||||
time.sleep(.5)
|
||||
time.sleep(1)
|
||||
|
||||
else:
|
||||
print("No stations found to cleanup")
|
||||
still_looking_sta = False
|
||||
print("clean_sta still_looking_sta {sta_looking}".format(sta_looking=still_looking_sta))
|
||||
if not still_looking_sta:
|
||||
self.sta_done = True
|
||||
return still_looking_sta
|
||||
|
||||
|
||||
'''
|
||||
1: delete cx
|
||||
2: delete endp
|
||||
3: delete sta
|
||||
when deleting sta first, you will end up with phantom CX
|
||||
'''
|
||||
def cleanup(self):
|
||||
if self.clean_cxs:
|
||||
# also clean the endp when cleaning cxs
|
||||
still_looking_cxs = self.cxs_clean()
|
||||
still_looking_endp = self.endp_clean()
|
||||
print("clean_cxs: still_looking_cxs {looking_cxs} still_looking_endp {looking_endp}"\
|
||||
.format(looking_cxs=still_looking_cxs,looking_endp=still_looking_endp))
|
||||
if self.clean_endp and not clean_cxs:
|
||||
still_looking_endp = self.endp_clean()
|
||||
print("clean_endp: still_looking_endp {looking_endp}".format(looking_endp=still_looking_endp))
|
||||
|
||||
if self.clean_sta:
|
||||
still_looking_sta = self.sta_clean()
|
||||
print("clean_sta: still_looking_sta {looking_sta}".format(looking_sta=still_looking_sta))
|
||||
|
||||
def main():
|
||||
|
||||
@@ -145,15 +209,36 @@ python3 ./lf_clean.py --mgr MGR
|
||||
'--lfmgr',
|
||||
help='--mgr <hostname for where LANforge GUI is running>',
|
||||
default='localhost')
|
||||
parser.add_argument(
|
||||
'--cxs',
|
||||
help="--cxs, this will clear all the endps and cxs",
|
||||
action='store_true')
|
||||
parser.add_argument(
|
||||
'--endp',
|
||||
help="--endp, this will clear all the endps",
|
||||
action='store_true')
|
||||
parser.add_argument(
|
||||
'--sta',
|
||||
help="--sta, this will clear all the stations",
|
||||
action='store_true')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
clean = lf_clean(host=args.mgr)
|
||||
print("cleaning up stations, cxs and endpoints: start")
|
||||
clean.cleanup()
|
||||
print("cleaning up stations, cxs and endpoints:\
|
||||
done with current pass may need to iterate multiple times")
|
||||
|
||||
if args.cxs or args.endp or args.sta:
|
||||
clean = lf_clean(host=args.mgr,clean_cxs=args.cxs,clean_endp=args.endp,clean_sta=args.sta )
|
||||
print("cleaning cxs: {cxs} endpoints: {endp} stations: {sta} start".format(cxs=args.cxs,endp=args.endp,sta=args.sta))
|
||||
if args.cxs:
|
||||
print("cleaning cxs will also clean endp")
|
||||
clean.cxs_clean()
|
||||
clean.endp_clean()
|
||||
if args.endp and not args.cxs:
|
||||
clean.endp_clean()
|
||||
if args.sta:
|
||||
clean.sta_clean()
|
||||
print("Clean done")
|
||||
# print("Clean cxs_done {cxs_done} endp_done {endp_done} sta_done {sta_done}"
|
||||
# .format(cxs_done=clean.cxs_done,endp_done=clean.endp_done,sta_done=clean.sta_done))
|
||||
else:
|
||||
print("please add option of --cxs ,--endp, or --sta to clean")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
@@ -274,6 +274,9 @@ class L3VariableTime(Realm):
|
||||
self.station_profile.security = ssid_security_
|
||||
self.station_profile.number_template = self.number_template
|
||||
self.station_profile.mode = 0
|
||||
# place the enable and disable flags
|
||||
# self.station_profile.desired_add_sta_flags = self.enable_flags
|
||||
# self.station_profile.desired_add_sta_flags_mask = self.enable_flags + self.disable_flags
|
||||
self.station_profile.set_reset_extra(reset_port_enable=reset_port_enable_,
|
||||
test_duration=self.duration_time_to_seconds(self.test_duration),
|
||||
reset_port_min_time=self.duration_time_to_seconds(reset_port_time_min_),
|
||||
|
||||
Reference in New Issue
Block a user