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

inheritance fixes

Browse Source
This commit is contained in:
Matthew Stidham
2021-01-18 13:12:27 -08:00
parent ae059322d1
commit fbce00e988
2 changed files with 447 additions and 379 deletions
Download Patch File Download Diff File Expand all files Collapse all files

213
py-json/realm.py
View File

@@ -20,6 +20,7 @@ import base64
import xlsxwriter
import pandas as pd
def wpa_ent_list():
return [
"DEFAULT",
@@ -45,6 +46,7 @@ def wpa_ent_list():
"OWE"
]
class Realm(LFCliBase):
def __init__(self,
lfclient_host="localhost",
@@ -317,7 +319,8 @@ class Realm(LFCliBase):
dbg_param = "?__debug=1"
while (last_response != "YES"):
response = self.json_post("/gui-json/cmd%s" % dbg_param, data, debug_=debug_, response_json_list_=response_json)
response = self.json_post("/gui-json/cmd%s" % dbg_param, data, debug_=debug_,
response_json_list_=response_json)
# LFUtils.debug_printer.pprint(response_json)
last_response = response_json[0]["LAST"]["response"]
if (last_response != "YES"):
@@ -331,7 +334,8 @@ class Realm(LFCliBase):
# loads a database
def load(self, name):
if (name is None) or (name == ""):
raise ValueError("Realm::load: wants a test scenario database name, please find one in the Status tab of the GUI")
raise ValueError(
"Realm::load: wants a test scenario database name, please find one in the Status tab of the GUI")
data = {
"name": name,
@@ -636,7 +640,6 @@ class Realm(LFCliBase):
num_sta_with_ips += 1
return num_sta_with_ips
def duration_time_to_seconds(self, time_string):
if isinstance(time_string, str):
pattern = re.compile("^(\d+)([dhms]$)")
@@ -793,11 +796,13 @@ class Realm(LFCliBase):
return cx_prof
def new_vap_profile(self):
vap_prof = VAPProfile(lfclient_host=self.lfclient_host, lfclient_port=self.lfclient_port, local_realm=self, debug_=self.debug)
vap_prof = VAPProfile(lfclient_host=self.lfclient_host, lfclient_port=self.lfclient_port, local_realm=self,
debug_=self.debug)
return vap_prof
def new_vr_profile(self):
vap_prof = VRProfile(lfclient_host=self.lfclient_host, lfclient_port=self.lfclient_port, local_realm=self, debug_=self.debug)
vap_prof = VRProfile(lfclient_host=self.lfclient_host, lfclient_port=self.lfclient_port, local_realm=self,
debug_=self.debug)
return vap_prof
def new_http_profile(self):
@@ -817,6 +822,7 @@ class Realm(LFCliBase):
def new_test_group_profile(self):
return TestGroupProfile(self.lfclient_host, self.lfclient_port, local_realm=self, debug_=self.debug)
class BaseProfile(LFCliBase):
def __init__(self, local_realm):
self.parent_realm = local_realm
@@ -824,16 +830,19 @@ class BaseProfile(LFCliBase):
self.exit_on_error = False
def json_get(self, target):
return self.json_get(target)
return self.parent_realm.json_get(target)
def json_post(self, url,data,debug_,suppress_related_commands_=None):
return self.json_post(url,data,debug_=False,suppress_related_commands_=suppress_related_commands_)
def json_post(self, req_url, data, debug_=False, suppress_related_commands_=None):
return self.parent_realm.json_post(req_url,
data,
suppress_related_commands_=suppress_related_commands_,
debug_=debug_)
def parse_time(self, target):
return self.parent_realm.parse_time(target)
def stop_cx(self, target):
return self.parent_realm.stop_cx(target)
def stopping_cx(self, name):
return self.parent_realm.stop_cx(name)
def cleanup_cxe_prefix(self, prefix):
return self.parent_realm.cleanup_cxe_prefix(prefix)
@@ -841,6 +850,9 @@ class BaseProfile(LFCliBase):
def rm_cx(self, target):
return self.parent_realm.rm_cx(target)
def rm_endp(self, ename):
self.parent_realm.rm_endp(ename)
def created_cx(self, target):
return self.parent_realm.created_cx(target)
@@ -856,6 +868,7 @@ class BaseProfile(LFCliBase):
def wait_until_cxs_appear(self, target, debug=False):
return self.parent_realm.wait_until_cxs_appear(target, debug=False)
class MULTICASTProfile(LFCliBase):
def __init__(self, lfclient_host, lfclient_port, local_realm,
report_timer_=3000, name_prefix_="Unset", number_template_="00000", debug_=False):
@@ -896,7 +909,8 @@ class MULTICASTProfile(LFCliBase):
"endp_name": endp_name
}
url = "cli-json/start_endp"
self.local_realm.json_post(url, json_data, debug_=debug_, suppress_related_commands_=suppress_related_commands)
self.local_realm.json_post(url, json_data, debug_=debug_,
suppress_related_commands_=suppress_related_commands)
pass
@@ -908,7 +922,8 @@ class MULTICASTProfile(LFCliBase):
"endp_name": endp_name
}
url = "cli-json/stop_endp"
self.local_realm.json_post(url, json_data, debug_=debug_, suppress_related_commands_=suppress_related_commands)
self.local_realm.json_post(url, json_data, debug_=debug_,
suppress_related_commands_=suppress_related_commands)
pass
@@ -972,7 +987,6 @@ class MULTICASTProfile(LFCliBase):
these_endp = [side_tx_name]
self.local_realm.wait_until_endps_appear(these_endp, debug=debug_)
def create_mc_rx(self, endp_type, side_rx, suppress_related_commands=None, debug_=False):
if self.debug:
debug_ = True
@@ -1007,7 +1021,8 @@ class MULTICASTProfile(LFCliBase):
}
url = "cli-json/add_endp"
self.local_realm.json_post(url, json_data, debug_=debug_, suppress_related_commands_=suppress_related_commands)
self.local_realm.json_post(url, json_data, debug_=debug_,
suppress_related_commands_=suppress_related_commands)
json_data = {
'name': side_rx_name,
@@ -1017,7 +1032,8 @@ class MULTICASTProfile(LFCliBase):
'rcv_mcast': 'Yes'
}
url = "cli-json/set_mc_endp"
self.local_realm.json_post(url, json_data, debug_=debug_, suppress_related_commands_=suppress_related_commands)
self.local_realm.json_post(url, json_data, debug_=debug_,
suppress_related_commands_=suppress_related_commands)
self.created_mc[side_rx_name] = side_rx_name
these_endp.append(side_rx_name)
@@ -1028,14 +1044,23 @@ class MULTICASTProfile(LFCliBase):
pprint.pprint(self)
class L3CXProfile(BaseProfile):
def __init__(self, lfclient_host, lfclient_port,local_realm,
side_a_min_bps=None, side_b_min_bps=None,
side_a_max_bps=0, side_b_max_bps=0,
side_a_min_pdu=-1, side_b_min_pdu=-1,
side_a_max_pdu=0, side_b_max_pdu=0,
report_timer_=3000, name_prefix_="Unset", number_template_="00000", debug_=False):
def __init__(self,
lfclient_host,
lfclient_port,
local_realm,
side_a_min_bps=None,
side_b_min_bps=None,
side_a_max_bps=0,
side_b_max_bps=0,
side_a_min_pdu=-1,
side_b_min_pdu=-1,
side_a_max_pdu=0,
side_b_max_pdu=0,
report_timer_=3000,
name_prefix_="Unset",
number_template_="00000",
debug_=False):
"""
:param lfclient_host:
:param lfclient_port:
@@ -1054,7 +1079,6 @@ class L3CXProfile(BaseProfile):
super().__init__(local_realm)
self.lfclient_url = "http://%s:%s" % (lfclient_host, lfclient_port)
self.debug = debug_
self.local_realm = local_realm
self.side_a_min_pdu = side_a_min_pdu
self.side_b_min_pdu = side_b_min_pdu
self.side_a_max_pdu = side_a_max_pdu
@@ -1191,7 +1215,8 @@ class L3CXProfile(BaseProfile):
for x in range(0, len(endpoints[0])):
endpoints2.append(list(list(endpoints[y][x].values())[0].values()))
import itertools
timestamps2=list(itertools.chain.from_iterable(itertools.repeat(x, len(created_cx.split(','))) for x in timestamps))
timestamps2 = list(
itertools.chain.from_iterable(itertools.repeat(x, len(created_cx.split(','))) for x in timestamps))
for point in range(0, len(endpoints2)):
endpoints2[point].insert(0, timestamps2[point])
# step 4 save and close
@@ -1215,7 +1240,10 @@ class L3CXProfile(BaseProfile):
df.columns = header_row
import requests
import ast
try:
systeminfo = ast.literal_eval(requests.get('http://localhost:8080').text)
except:
systeminfo = ast.literal_eval(requests.get('http://192.168.1.3:8080').text)
df['LFGUI Release'] = systeminfo['VersionInfo']['BuildVersion']
df['Script Name'] = script_name
df['Arguments'] = arguments
@@ -1251,7 +1279,7 @@ class L3CXProfile(BaseProfile):
def refresh_cx(self):
for cx_name in self.created_cx.keys():
self.local_realm.json_post("/cli-json/show_cxe", {
self.json_post("/cli-json/show_cxe", {
"test_mgr": "ALL",
"cross_connect": cx_name
}, debug_=self.debug)
@@ -1262,7 +1290,7 @@ class L3CXProfile(BaseProfile):
for cx_name in self.created_cx.keys():
if self.debug:
print("cx-name: %s" % (cx_name))
self.local_realm.json_post("/cli-json/set_cx_state", {
self.json_post("/cli-json/set_cx_state", {
"test_mgr": "default_tm",
"cx_name": cx_name,
"cx_state": "RUNNING"
@@ -1273,12 +1301,12 @@ class L3CXProfile(BaseProfile):
def stop_cx(self):
print("Stopping CXs...")
for cx_name in self.created_cx.keys():
self.stop_cx(cx_name)
self.stopping_cx(cx_name)
print(".", end='')
print("")
def cleanup_prefix(self):
self.local_realm.cleanup_cxe_prefix(self.name_prefix)
self.cleanup_cxe_prefix(self.name_prefix)
def cleanup(self):
print("Cleaning up cxs and endpoints")
@@ -1294,7 +1322,8 @@ class L3CXProfile(BaseProfile):
print("Cleaning endpoint: %s" % (ename))
self.rm_endp(self.created_cx[cx_name][side])
def create(self, endp_type, side_a, side_b, sleep_time=0.03, suppress_related_commands=None, debug_=False, tos=None):
def create(self, endp_type, side_a, side_b, sleep_time=0.03, suppress_related_commands=None, debug_=False,
tos=None):
if self.debug:
debug_ = True
@@ -1309,7 +1338,8 @@ class L3CXProfile(BaseProfile):
or (self.side_a_max_bps is None) \
or (self.side_b_min_bps is None) \
or (self.side_b_max_bps is None):
raise ValueError("side_a_min_bps, side_a_max_bps, side_b_min_bps, and side_b_max_bps must all be set to a value")
raise ValueError(
"side_a_min_bps, side_a_max_bps, side_b_min_bps, and side_b_max_bps must all be set to a value")
if type(side_a) == list and type(side_b) != list:
side_b_info = self.name_to_eid(side_b)
@@ -1359,8 +1389,10 @@ class L3CXProfile(BaseProfile):
}
url = "/cli-json/add_endp"
self.local_realm.json_post(url, endp_side_a, debug_=debug_, suppress_related_commands_=suppress_related_commands)
self.local_realm.json_post(url, endp_side_b, debug_=debug_, suppress_related_commands_=suppress_related_commands)
self.json_post(url, endp_side_a, debug_=debug_,
suppress_related_commands_=suppress_related_commands)
self.json_post(url, endp_side_b, debug_=debug_,
suppress_related_commands_=suppress_related_commands)
# print("napping %f sec"%sleep_time)
time.sleep(sleep_time)
@@ -1370,16 +1402,20 @@ class L3CXProfile(BaseProfile):
"flag": "AutoHelper",
"val": 1
}
self.local_realm.json_post(url, data, debug_=debug_, suppress_related_commands_=suppress_related_commands)
self.json_post(url, data, debug_=debug_,
suppress_related_commands_=suppress_related_commands)
data["name"] = endp_b_name
self.local_realm.json_post(url, data, debug_=debug_, suppress_related_commands_=suppress_related_commands)
self.json_post(url, data, debug_=debug_,
suppress_related_commands_=suppress_related_commands)
if (endp_type == "lf_udp") or (endp_type == "udp") or (endp_type == "lf_udp6") or (endp_type == "udp6"):
data["name"] = endp_a_name
data["flag"] = "UseAutoNAT"
self.local_realm.json_post(url, data, debug_=debug_, suppress_related_commands_=suppress_related_commands)
self.json_post(url, data, debug_=debug_,
suppress_related_commands_=suppress_related_commands)
data["name"] = endp_b_name
self.local_realm.json_post(url, data, debug_=debug_, suppress_related_commands_=suppress_related_commands)
self.json_post(url, data, debug_=debug_,
suppress_related_commands_=suppress_related_commands)
if tos != None:
self.set_endp_tos(endp_a_name, tos)
@@ -1447,8 +1483,10 @@ class L3CXProfile(BaseProfile):
}
url = "/cli-json/add_endp"
self.local_realm.json_post(url, endp_side_a, debug_=debug_, suppress_related_commands_=suppress_related_commands)
self.local_realm.json_post(url, endp_side_b, debug_=debug_, suppress_related_commands_=suppress_related_commands)
self.json_post(url, endp_side_a, debug_=debug_,
suppress_related_commands_=suppress_related_commands)
self.json_post(url, endp_side_b, debug_=debug_,
suppress_related_commands_=suppress_related_commands)
# print("napping %f sec" %sleep_time )
time.sleep(sleep_time)
@@ -1481,13 +1519,15 @@ class L3CXProfile(BaseProfile):
"milliseconds": self.report_timer
})
else:
raise ValueError("side_a or side_b must be of type list but not both: side_a is type %s side_b is type %s" % (type(side_a), type(side_b)))
raise ValueError(
"side_a or side_b must be of type list but not both: side_a is type %s side_b is type %s" % (
type(side_a), type(side_b)))
self.wait_until_endps_appear(these_endp, debug=debug_)
for data in cx_post_data:
url = "/cli-json/add_cx"
self.local_realm.json_post(url, data, debug_=debug_, suppress_related_commands_=suppress_related_commands)
self.json_post(url, data, debug_=debug_, suppress_related_commands_=suppress_related_commands)
time.sleep(0.01)
self.wait_until_cxs_appear(these_cx, debug=debug_)
@@ -1591,7 +1631,8 @@ class L4CXProfile(LFCliBase):
"proxy_auth_type": 0x200
}
url = "cli-json/add_l4_endp"
self.local_realm.json_post(url, endp_data, debug_=debug_, suppress_related_commands_=suppress_related_commands_)
self.local_realm.json_post(url, endp_data, debug_=debug_,
suppress_related_commands_=suppress_related_commands_)
time.sleep(sleep_time)
endp_data = {
@@ -1605,7 +1646,8 @@ class L4CXProfile(LFCliBase):
for cx_data in cx_post_data:
url = "/cli-json/add_cx"
self.local_realm.json_post(url, cx_data, debug_=debug_, suppress_related_commands_=suppress_related_commands_)
self.local_realm.json_post(url, cx_data, debug_=debug_,
suppress_related_commands_=suppress_related_commands_)
time.sleep(sleep_time)
@@ -1769,6 +1811,7 @@ class GenCXProfile(LFCliBase):
})
time.sleep(sleep_time)
class WifiMonitor:
def __init__(self, lfclient_url, local_realm, up=True, debug_=False, resource_=1):
self.debug = debug_
@@ -1793,7 +1836,8 @@ class WifiMonitor:
# we want to query the existing country code of the radio
# there's no reason to change it but we get hollering from server
# if we don't provide a value for the parameter
jr = self.local_realm.json_get("/radiostatus/1/%s/%s?fields=channel,frequency,country"%(resource_, radio_), debug_=self.debug)
jr = self.local_realm.json_get("/radiostatus/1/%s/%s?fields=channel,frequency,country" % (resource_, radio_),
debug_=self.debug)
if jr is None:
raise ValueError("No radio %s.%s found" % (resource_, radio_))
@@ -1861,8 +1905,6 @@ class WifiMonitor:
baseurl=self.lfclient_url,
debug=self.debug)
def admin_up(self):
up_request = LFUtils.port_up_request(resource_id=self.resource, port_name=self.monitor_name)
self.local_realm.json_post("/cli-json/set_port", up_request)
@@ -1885,9 +1927,9 @@ class WifiMonitor:
}
self.local_realm.json_post("/cli-json/sniff_port", _data=data)
# "sniff_port 1 %s %s NA %s %s.pcap %i"%(r, m, sflags, m, int(dur))
class VAPProfile(LFCliBase):
def __init__(self, lfclient_host, lfclient_port, local_realm,
vap_name="",
@@ -2026,7 +2068,9 @@ class VAPProfile(LFCliBase):
self.desired_add_vap_flags_mask.append(param_name)
elif command_name == "set_port":
if (param_name not in set_port.set_port_current_flags) and (param_name not in set_port.set_port_cmd_flags) and (param_name not in set_port.set_port_interest_flags):
if (param_name not in set_port.set_port_current_flags) and (
param_name not in set_port.set_port_cmd_flags) and (
param_name not in set_port.set_port_interest_flags):
print("Parameter name [%s] not defined in set_port.py" % param_name)
if self.debug:
pprint(set_port.set_port_cmd_flags)
@@ -2087,7 +2131,8 @@ class VAPProfile(LFCliBase):
return result
def create(self, resource, radio, channel=None, up_=None, debug=False, use_ht40=True, use_ht80=True, use_ht160=False,
def create(self, resource, radio, channel=None, up_=None, debug=False, use_ht40=True, use_ht80=True,
use_ht160=False,
suppress_related_commands_=True, use_radius=False, hs20_enable=False):
port_list = self.local_realm.json_get("port/1/1/list")
if port_list is not None:
@@ -2118,7 +2163,8 @@ class VAPProfile(LFCliBase):
# print("MODE ========= ", self.mode)
jr = self.local_realm.json_get("/radiostatus/1/%s/%s?fields=channel,frequency,country" % (resource, radio), debug_=self.debug)
jr = self.local_realm.json_get("/radiostatus/1/%s/%s?fields=channel,frequency,country" % (resource, radio),
debug_=self.debug)
if jr is None:
raise ValueError("No radio %s.%s found" % (resource, radio))
@@ -2193,7 +2239,6 @@ class VAPProfile(LFCliBase):
wifi_extra_r.addPostData(self.wifi_extra_data)
json_response = wifi_extra_r.jsonPost(debug)
port_list = self.local_realm.json_get("port/1/1/list")
if port_list is not None:
port_list = port_list['interfaces']
@@ -2221,8 +2266,6 @@ class VAPProfile(LFCliBase):
}
self.local_realm.json_post("cli-json/set_port", bridge_set_port)
if (self.up):
self.admin_up(1)
@@ -2239,6 +2282,7 @@ class VAPProfile(LFCliBase):
# And now see if they are gone
LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=desired_ports)
class VRProfile(LFCliBase):
def __init__(self, lfclient_host, lfclient_port, local_realm, ssid="NA", ssid_pass="NA", mode=0, debug_=False):
super().__init__(_lfjson_host=lfclient_host, _lfjson_port=lfclient_port, _debug=debug_)
@@ -2297,7 +2341,8 @@ class VRProfile(LFCliBase):
"peer_ifname": "rdd1"
}
# print("creating rdd0")
self.local_realm.json_post("add_rdd", rdd_data, suppress_related_commands_=suppress_related_commands_, debug_=debug_)
self.local_realm.json_post("add_rdd", rdd_data, suppress_related_commands_=suppress_related_commands_,
debug_=debug_)
rdd_data = {
"shelf": 1,
@@ -2306,24 +2351,28 @@ class VRProfile(LFCliBase):
"peer_ifname": "rdd0"
}
# print("creating rdd1")
self.local_realm.json_post("add_rdd", rdd_data, suppress_related_commands_=suppress_related_commands_, debug_=debug_)
self.local_realm.json_post("add_rdd", rdd_data, suppress_related_commands_=suppress_related_commands_,
debug_=debug_)
self.set_port_data["port"] = "rdd0"
self.set_port_data["ip_addr"] = gateway
self.set_port_data["netmask"] = netmask
self.set_port_data["gateway"] = gateway
self.local_realm.json_post("set_port", self.set_port_data, suppress_related_commands_=suppress_related_commands_, debug_=debug_)
self.local_realm.json_post("set_port", self.set_port_data,
suppress_related_commands_=suppress_related_commands_, debug_=debug_)
self.set_port_data["port"] = "rdd1"
self.set_port_data["ip_addr"] = ip_addr
self.set_port_data["netmask"] = netmask
self.set_port_data["gateway"] = gateway
self.local_realm.json_post("set_port", self.set_port_data, suppress_related_commands_=suppress_related_commands_, debug_=debug_)
self.local_realm.json_post("set_port", self.set_port_data,
suppress_related_commands_=suppress_related_commands_, debug_=debug_)
self.created_rdds.append("rdd0")
self.created_rdds.append("rdd1")
def create_vrcx(self, resource, local_dev, remote_dev, subnets, nexthop, flags, suppress_related_commands_=True, debug_=False):
def create_vrcx(self, resource, local_dev, remote_dev, subnets, nexthop, flags, suppress_related_commands_=True,
debug_=False):
if self.vr_name is not None:
self.vrcx_data["resource"] = resource
self.vrcx_data["vr_name"] = self.vr_name
@@ -2332,11 +2381,11 @@ class VRProfile(LFCliBase):
self.vrcx_data["subnets"] = subnets
self.vrcx_data["nexthop"] = nexthop
self.vrcx_data["flags"] = flags
self.local_realm.json_post("add_vrcx", self.vrcx_data, suppress_related_commands_=suppress_related_commands_, debug_=debug_)
self.local_realm.json_post("add_vrcx", self.vrcx_data,
suppress_related_commands_=suppress_related_commands_, debug_=debug_)
else:
raise ValueError("vr_name must be set. Current name: %s" % self.vr_name)
def create(self, resource, upstream_port="eth1", debug=False,
upstream_subnets="20.20.20.0/24", upstream_nexthop="20.20.20.1",
local_subnets="10.40.0.0/24", local_nexthop="10.40.3.198",
@@ -2350,18 +2399,22 @@ class VRProfile(LFCliBase):
else:
raise ValueError("vr_name must be set. Current name: %s" % self.vr_name)
# Create 1 rdd pair
self.create_rdd(resource=resource, ip_addr=rdd_ip, gateway=rdd_gateway, netmask=rdd_netmask) # rdd0, rdd1; rdd0 gateway, rdd1 connected to network
self.create_rdd(resource=resource, ip_addr=rdd_ip, gateway=rdd_gateway,
netmask=rdd_netmask) # rdd0, rdd1; rdd0 gateway, rdd1 connected to network
# connect rdds and upstream
self.create_vrcx(resource=resource, local_dev=upstream_port, remote_dev="NA", subnets=upstream_subnets, nexthop=upstream_nexthop,
self.create_vrcx(resource=resource, local_dev=upstream_port, remote_dev="NA", subnets=upstream_subnets,
nexthop=upstream_nexthop,
flags=257, suppress_related_commands_=suppress_related_commands_, debug_=debug)
self.create_vrcx(resource=resource, local_dev="rdd0", remote_dev="rdd1", subnets=local_subnets, nexthop=local_nexthop,
self.create_vrcx(resource=resource, local_dev="rdd0", remote_dev="rdd1", subnets=local_subnets,
nexthop=local_nexthop,
flags=1, suppress_related_commands_=suppress_related_commands_, debug_=debug)
def cleanup(self, resource, delay=0.03):
# TODO: Cleanup for VRProfile
pass
class DUTProfile(LFCliBase):
def __init__(self, lfclient_host, lfclient_port, local_realm, debug_=False):
super().__init__(lfclient_host, lfclient_port, debug_, _halt_on_error=True, _local_realm=local_realm)
@@ -2472,6 +2525,7 @@ class DUTProfile(LFCliBase):
"text-64": notebytes.decode('ascii')
}, self.debug)
class TestGroupProfile(LFCliBase):
def __init__(self, lfclient_host, lfclient_port, local_realm, test_group_name=None, debug_=False):
super().__init__(lfclient_host, lfclient_port, debug_, _halt_on_error=True)
@@ -2554,6 +2608,7 @@ class FIOEndpProfile(LFCliBase):
Very often you will create the FileIO writer profile first so that it creates the data
that a reader profile will subsequently use.
"""
def __init__(self, lfclient_host, lfclient_port, local_realm, io_direction="write", debug_=False):
super().__init__(lfclient_host, lfclient_port, debug_, _halt_on_error=True)
self.local_realm = local_realm
@@ -2665,7 +2720,8 @@ class FIOEndpProfile(LFCliBase):
else:
raise ValueError("Unexpected name for port_name %s" % port_name)
if self.directory is None or self.server_mount is None or self.fs_type is None:
raise ValueError("directory [%s], server_mount [%s], and type [%s] must not be None" % (self.directory, self.server_mount, self.fs_type))
raise ValueError("directory [%s], server_mount [%s], and type [%s] must not be None" % (
self.directory, self.server_mount, self.fs_type))
endp_data = {
"alias": self.cx_prefix + name + "_" + str(num_connection) + "_fio",
"shelf": shelf,
@@ -2689,7 +2745,8 @@ class FIOEndpProfile(LFCliBase):
endp_data["directory"] = "/mnt/lf/wo_" + name + "_" + str(num_connection) + "_fio"
url = "cli-json/add_file_endp"
self.local_realm.json_post(url, endp_data, debug_=False, suppress_related_commands_=suppress_related_commands_)
self.local_realm.json_post(url, endp_data, debug_=False,
suppress_related_commands_=suppress_related_commands_)
time.sleep(sleep_time)
data = {
@@ -2715,11 +2772,13 @@ class FIOEndpProfile(LFCliBase):
"rx_endp": "NA"
}
cx_post_data.append(endp_data)
self.created_cx[self.cx_prefix + name + "_" + str(num_connection) + "_fio" ] = "CX_" + self.cx_prefix + name + "_" + str(num_connection) + "_fio"
self.created_cx[self.cx_prefix + name + "_" + str(
num_connection) + "_fio"] = "CX_" + self.cx_prefix + name + "_" + str(num_connection) + "_fio"
for cx_data in cx_post_data:
url = "/cli-json/add_cx"
self.local_realm.json_post(url, cx_data, debug_=debug_, suppress_related_commands_=suppress_related_commands_)
self.local_realm.json_post(url, cx_data, debug_=debug_,
suppress_related_commands_=suppress_related_commands_)
time.sleep(sleep_time)
@@ -2835,7 +2894,8 @@ class MACVLANProfile(LFCliBase):
self.desired_set_port_interest_flags.append("ip_address")
self.desired_set_port_interest_flags.append("ip_Mask")
self.desired_set_port_interest_flags.append("ip_gateway")
self.ip_list = LFUtils.gen_ip_series(ip_addr=self.first_ip_addr, netmask=self.netmask, num_ips=self.num_macvlans)
self.ip_list = LFUtils.gen_ip_series(ip_addr=self.first_ip_addr, netmask=self.netmask,
num_ips=self.num_macvlans)
self.set_port_data["current_flags"] = self.add_named_flags(self.desired_set_port_current_flags,
set_port.set_port_current_flags)
@@ -2862,7 +2922,8 @@ class MACVLANProfile(LFCliBase):
else:
data["flags"] = 0
self.created_macvlans.append("%s.%s.%s#%d" % (self.shelf, self.resource,
self.macvlan_parent, int(self.desired_macvlans[i][self.desired_macvlans[i].index('#')+1:])))
self.macvlan_parent, int(
self.desired_macvlans[i][self.desired_macvlans[i].index('#') + 1:])))
self.local_realm.json_post(req_url, data)
time.sleep(sleep_time)
@@ -2893,7 +2954,6 @@ class MACVLANProfile(LFCliBase):
# And now see if they are gone
LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=self.created_macvlans)
def admin_up(self):
for macvlan in self.created_macvlans:
self.local_realm.admin_up(macvlan)
@@ -2902,6 +2962,7 @@ class MACVLANProfile(LFCliBase):
for macvlan in self.created_macvlans:
self.local_realm.admin_down(macvlan)
class PacketFilter():
def get_filter_wlan_assoc_packets(self, ap_mac, sta_mac):
@@ -2926,6 +2987,7 @@ class PacketFilter():
return lines
class PortUtils():
def __init__(self, local_realm):
self.local_realm = local_realm
@@ -2970,6 +3032,7 @@ class PortUtils():
else:
raise ValueError("Port name required")
class HTTPProfile(LFCliBase):
def __init__(self, lfclient_host, lfclient_port, local_realm, debug_=False):
super().__init__(lfclient_host, lfclient_port, debug_, _halt_on_error=True)
@@ -3105,7 +3168,8 @@ class HTTPProfile(LFCliBase):
"proxy_auth_type": 0x200
}
url = "cli-json/add_l4_endp"
self.local_realm.json_post(url, endp_data, debug_=debug_, suppress_related_commands_=suppress_related_commands_)
self.local_realm.json_post(url, endp_data, debug_=debug_,
suppress_related_commands_=suppress_related_commands_)
time.sleep(sleep_time)
endp_data = {
@@ -3119,7 +3183,8 @@ class HTTPProfile(LFCliBase):
for cx_data in cx_post_data:
url = "/cli-json/add_cx"
self.local_realm.json_post(url, cx_data, debug_=debug_, suppress_related_commands_=suppress_related_commands_)
self.local_realm.json_post(url, cx_data, debug_=debug_,
suppress_related_commands_=suppress_related_commands_)
time.sleep(sleep_time)
@@ -3344,7 +3409,9 @@ class StationProfile:
self.desired_add_sta_flags_mask.append(param_name)
elif command_name == "set_port":
if (param_name not in set_port.set_port_current_flags) and (param_name not in set_port.set_port_cmd_flags) and (param_name not in set_port.set_port_interest_flags):
if (param_name not in set_port.set_port_current_flags) and (
param_name not in set_port.set_port_cmd_flags) and (
param_name not in set_port.set_port_interest_flags):
print("Parameter name [%s] not defined in set_port.py" % param_name)
if self.debug:
pprint(set_port.set_port_cmd_flags)
@@ -3368,6 +3435,7 @@ class StationProfile:
self.desired_set_port_interest_flags.remove(param_name)
else:
raise ValueError("Unknown param name: " + param_name)
# use this for hinting station name; stations begin with 'sta', the
# stations created with a prefix '0100' indicate value 10100 + n with
# resulting substring(1,) applied; station 900 becomes 'sta1000'
@@ -3423,7 +3491,6 @@ class StationProfile:
# And now see if they are gone
LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=desired_stations)
# Checks for errors in initialization values and creates specified number of stations using init parameters
def create(self, radio,
num_stations=0,

1
py-scripts/test_ipv4_variable_time.py
View File

@@ -259,6 +259,7 @@ python3 ./test_ipv4_variable_time.py
output_format=output,
script_name='test_ipv4_variable_time',
arguments=args)
print(ip_var_test.cx_profile.create())
ip_var_test.stop()
if not ip_var_test.passes():
print(ip_var_test.get_fail_message())