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

Merge remote-tracking branch 'origin/master'

Browse Source
This commit is contained in:
Logan Lipke
2020-07-02 11:07:47 -07:00
parent d14de26a66 013093c47a
commit ebc001831f
9 changed files with 362 additions and 269 deletions
Download Patch File Download Diff File Expand all files Collapse all files

281
py-scripts/sta_connect2.py
View File

@@ -33,14 +33,14 @@ class StaConnect2(LFCliBase):
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=None, _debugOn=False, _dut_security=OPEN, _exit_on_error=False,
_sta_name=None, debug_=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, _halt_on_error=_exit_on_error, _exit_on_fail=_exit_on_fail)
self.debugOn = _debugOn
self.dut_security = ""
super().__init__(host, port, _debug=debug_, _halt_on_error=_exit_on_error, _exit_on_fail=_exit_on_fail)
self.debug = debug_
self.dut_security = _dut_security
self.dut_ssid = _dut_ssid
self.dut_passwd = _dut_passwd
self.dut_bssid = _dut_bssid
@@ -62,6 +62,9 @@ class StaConnect2(LFCliBase):
self.localrealm = Realm(lfclient_host=host, lfclient_port=port) # py > 3.6
self.resulting_stations = {}
self.resulting_endpoints = {}
self.sta_profile = None
self.l3_udp_profile = None
self.l3_tcp_profile = None
# def get_realm(self) -> Realm: # py > 3.6
def get_realm(self):
@@ -125,64 +128,68 @@ class StaConnect2(LFCliBase):
return False
if upstream_json['interface']['ip'] == "0.0.0.0":
pprint.pprint(upstream_json)
if self.debug:
pprint.pprint(upstream_json)
self._fail("Warning: %s lacks ip address" % self.get_upstream_url(), print_=True)
return False
# remove old stations
print("removing old station")
print("Removing old stations")
for sta_name in self.station_names:
sta_url = self.get_station_url(sta_name)
response = self.json_get(sta_url)
if response is not None:
if response["interface"] is not None:
for sta_name in self.station_names:
LFUtils.removePort(self.resource, sta_name, self.lfclient_url)
LFUtils.wait_until_ports_disappear(self.resource, self.lfclient_url, self.station_names)
if (response is not None) and (response["interface"] is not None):
for sta_name in self.station_names:
LFUtils.removePort(self.resource, sta_name, self.lfclient_url)
LFUtils.wait_until_ports_disappear(self.resource, self.lfclient_url, self.station_names)
# Create stations and turn dhcp on
sta_profile = self.localrealm.new_station_profile()
self.station_profile = self.localrealm.new_station_profile()
if self.dut_security == WPA2:
sta_profile.use_wpa2(on=True, ssid=self.dut_ssid, passwd=self.dut_passwd)
self.station_profile.use_wpa2(on=True, ssid=self.dut_ssid, passwd=self.dut_passwd)
elif self.dut_security == OPEN:
sta_profile.use_wpa2(on=False, ssid=self.dut_ssid)
sta_profile.set_command_flag("add_sta", "create_admin_down", 1)
self.station_profile.use_wpa2(on=False, ssid=self.dut_ssid)
self.station_profile.set_command_flag("add_sta", "create_admin_down", 1)
print("Adding new stations ", end="")
sta_profile.create(resource=self.resource, radio=self.radio, sta_names_=self.station_names)
self.station_profile.create(resource=self.resource, radio=self.radio, sta_names_=self.station_names, up_=False, debug=False)
LFUtils.wait_until_ports_appear(self.resource, self.lfclient_url, self.station_names)
# Create endpoints and cxs
# Create UDP endpoints
l3_udp_profile = self.localrealm.new_l3_cx_profile()
l3_udp_profile.side_a_min_bps = 128000
l3_udp_profile.side_b_min_bps = 128000
l3_udp_profile.side_a_min_pdu = 1200
l3_udp_profile.side_b_min_pdu = 1500
l3_udp_profile.report_timer = 1000
l3_udp_profile.prefix = "udp_"
l3_udp_profile.create(endp_type="lf_udp",
side_a=list(self.localrealm.find_ports_like("sta+")),
side_b="%d.%s" % (self.resource, self.upstream_port))
self.l3_udp_profile = self.localrealm.new_l3_cx_profile()
self.l3_udp_profile.side_a_min_bps = 128000
self.l3_udp_profile.side_b_min_bps = 128000
self.l3_udp_profile.side_a_min_pdu = 1200
self.l3_udp_profile.side_b_min_pdu = 1500
self.l3_udp_profile.report_timer = 1000
self.l3_udp_profile.name_prefix = "udp"
self.l3_udp_profile.create(endp_type="lf_udp",
side_a=list(self.localrealm.find_ports_like("sta+")),
side_b="%d.%s" % (self.resource, self.upstream_port),
suppress_related_commands=True)
# Create TCP endpoints
l3_tcp_profile = self.localrealm.new_l3_cx_profile()
l3_tcp_profile.side_a_min_bps = 128000
l3_tcp_profile.side_b_min_bps = 56000
l3_tcp_profile.prefix = "tcp_"
l3_tcp_profile.report_timer = 1000
self.l3_tcp_profile = self.localrealm.new_l3_cx_profile()
self.l3_tcp_profile.side_a_min_bps = 128000
self.l3_tcp_profile.side_b_min_bps = 56000
self.l3_tcp_profile.name_prefix = "tcp"
self.l3_tcp_profile.report_timer = 1000
self.l3_tcp_profile.create(endp_type="lf_tcp",
side_a=list(self.localrealm.find_ports_like("sta+")),
side_b="%d.%s" % (self.resource, self.upstream_port),
suppress_related_commands=True)
def start(self):
# print("\nBringing ports up...")
# data = {"shelf": 1,
# "resource": self.resource,
# "port": "ALL",
# "probe_flags": 1}
# self.json_post("/cli-json/nc_show_ports", data)
# LFUtils.waitUntilPortsAdminUp(self.resource, self.lfclient_url, self.station_names)
if not self.station_profile.up:
print("\nBringing ports up...")
data = {"shelf": 1,
"resource": self.resource,
"port": "ALL",
"probe_flags": 1}
self.json_post("/cli-json/nc_show_ports", data)
self.station_profile.admin_up(self.resource)
LFUtils.waitUntilPortsAdminUp(self.resource, self.lfclient_url, self.station_names)
# station_info = self.jsonGet(self.mgr_url, "%s?fields=port,ip,ap" % (self.getStaUrl()))
duration = 0
@@ -191,40 +198,37 @@ class StaConnect2(LFCliBase):
ap = ""
print("Waiting for %s stations to associate to AP: " % len(self.station_names), end="")
connected_stations = {}
# while (len(connected_stations.keys()) < len(self.station_names)) and (duration < maxTime):
# duration += 3
# time.sleep(3)
# print(".", end="")
# 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")
#
# # 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 == ""):
# if self.debugOn:
# print(" -%s," % sta_name, end="")
# else:
# if ip == "0.0.0.0":
# if self.debugOn:
# print(" %s (0.0.0.0)" % sta_name, end="")
# else:
# connected_stations[sta_name] = sta_url
# data = {
# "shelf":1,
# "resource": self.resource,
# "port": "ALL",
# "probe_flags": 1
# }
# self.json_post("/cli-json/nc_show_ports", data)
while (len(connected_stations.keys()) < len(self.station_names)) and (duration < maxTime):
duration += 3
time.sleep(3)
print(".", end="")
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")
# make a copy of the connected stations for test records
# 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 == ""):
if self.debug:
print(" -%s," % sta_name, end="")
else:
if ip == "0.0.0.0":
if self.debug:
print(" %s (0.0.0.0)" % sta_name, end="")
else:
connected_stations[sta_name] = sta_url
data = {
"shelf":1,
"resource": self.resource,
"port": "ALL",
"probe_flags": 1
}
self.json_post("/cli-json/nc_show_ports", data)
for sta_name in self.station_names:
sta_url = self.get_station_url(sta_name)
@@ -258,82 +262,61 @@ class StaConnect2(LFCliBase):
self.remove_stations()
return False
# start cx traffic
print("\nStarting CX Traffic")
# for cx_name in cx_names.keys():
# data = {
# "test_mgr": "ALL",
# "cx_name": cx_name,
# "cx_state": "RUNNING"
# }
# self.json_post("/cli-json/set_cx_state", data)
self.l3_udp_profile.start_cx()
self.l3_tcp_profile.start_cx()
time.sleep(1)
# Refresh stats
self.l3_udp_profile.refresh_cx()
self.l3_tcp_profile.refresh_cx()
def collect_endp_stats(self, endp_map):
print("Collecting Data")
fields="?fields=name,tx+bytes,rx+bytes"
for (cx_name, endps) in endp_map.items():
try:
endp_url = "/endp/%s%s" % (endps[0], fields)
endp_json = self.json_get(endp_url)
self.resulting_endpoints[endp_url] = endp_json
ptest_a_tx = endp_json['endpoint']['tx bytes']
ptest_a_rx = endp_json['endpoint']['rx bytes']
#ptest = self.json_get("/endp/%s?fields=tx+bytes,rx+bytes" % cx_names[cx_name]["b"])
endp_url = "/endp/%s%s" % (endps[1], fields)
endp_json = self.json_get(endp_url)
self.resulting_endpoints[endp_url] = endp_json
ptest_b_tx = endp_json['endpoint']['tx bytes']
ptest_b_rx = endp_json['endpoint']['rx bytes']
self.compare_vals("testTCP-A TX", ptest_a_tx)
self.compare_vals("testTCP-A RX", ptest_a_rx)
self.compare_vals("testTCP-B TX", ptest_b_tx)
self.compare_vals("testTCP-B RX", ptest_b_rx)
except Exception as e:
self.error(e)
print("Refresh CX stats")
# for cx_name in cx_names.keys():
# data = {
# "test_mgr": "ALL",
# "cross_connect": cx_name
# }
# self.json_post("/cli-json/show_cxe", data)
def stop(self):
# stop cx traffic
print("Stopping CX Traffic")
# for cx_name in cx_names.keys():
# data = {
# "test_mgr": "ALL",
# "cx_name": cx_name,
# "cx_state": "STOPPED"
# }
# self.json_post("/cli-json/set_cx_state", data)
self.l3_udp_profile.stop_cx()
self.l3_tcp_profile.stop_cx()
# Refresh stats
print("\nRefresh CX stats")
# for cx_name in cx_names.keys():
# data = {
# "test_mgr": "ALL",
# "cross_connect": cx_name
# }
# self.json_post("/cli-json/show_cxe", data)
self.l3_udp_profile.refresh_cx()
self.l3_tcp_profile.refresh_cx()
# print("Sleeping for 5 seconds")
print("Sleeping for 5 seconds")
time.sleep(5)
# get data for endpoints JSON
print("Collecting Data")
# for cx_name in cx_names.keys():
# try:
# # ?fields=tx+bytes,rx+bytes
# endp_url = "/endp/%s" % cx_names[cx_name]["a"]
# ptest = self.json_get(endp_url)
# self.resulting_endpoints[endp_url] = ptest
#
# 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" % cx_names[cx_name]["b"])
# endp_url = "/endp/%s" % cx_names[cx_name]["b"]
# ptest = self.json_get(endp_url)
# self.resulting_endpoints[endp_url] = ptest
#
# ptest_b_tx = ptest['endpoint']['tx bytes']
# ptest_b_rx = ptest['endpoint']['rx bytes']
#
# self.compare_vals("testTCP-A TX", ptest_a_tx)
# self.compare_vals("testTCP-A RX", ptest_a_rx)
#
# self.compare_vals("testTCP-B TX", ptest_b_tx)
# self.compare_vals("testTCP-B RX", ptest_b_rx)
#
# except Exception as e:
# self.error(e)
# Examples of what happens when you add test results that do not begin with PASS
# self.test_results.append("Neutral message will fail")
# self.test_results.append("FAILED message will fail")
self.collect_endp_stats(self.l3_udp_profile.created_cx)
self.collect_endp_stats(self.l3_tcp_profile.created_cx)
# print("\n")
def cleanup(self):
@@ -342,10 +325,12 @@ class StaConnect2(LFCliBase):
for sta_name in self.station_names:
LFUtils.removePort(self.resource, sta_name, self.lfclient_url)
endp_names = []
removeCX(self.lfclient_url, cx_names.keys())
for cx_name in cx_names:
endp_names.append(cx_names[cx_name]["a"])
endp_names.append(cx_names[cx_name]["b"])
removeCX(self.lfclient_url, self.l3_udp_profile.get_cx_names())
removeCX(self.lfclient_url, self.l3_tcp_profile.get_cx_names())
for (cx_name, endp_names) in self.l3_udp_profile.created_cx.items():
endp_names.append(endp_names[0])
endp_names.append(endp_names[1])
removeEndps(self.lfclient_url, endp_names)
# ~class
@@ -373,8 +358,10 @@ Example:
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_security", type=str, help="DUT security: openLF, wpa, wpa2, wpa3")
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("--debug", type=str, help="enable debugging")
args = parser.parse_args()
if args.dest is not None:
@@ -382,7 +369,13 @@ Example:
if args.port is not None:
lfjson_port = args.port
staConnect = StaConnect2(lfjson_host, lfjson_port)
on_flags = [ 1, "1", "on", "yes", "true" ]
debug_v = False
if args.debug is not None:
if args.debug in on_flags:
debug_v = True
staConnect = StaConnect2(lfjson_host, lfjson_port, debug_=debug_v)
staConnect.station_names = [ "sta0000" ]
if args.user is not None:
staConnect.user = args.user
@@ -398,16 +391,18 @@ Example:
staConnect.radio = args.radio
if args.resource is not None:
staConnect.resource = args.resource
if args.dut_ssid is not None:
staConnect.dut_ssid = args.dut_ssid
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
if args.dut_security is not None:
staConnect.dut_security = args.dut_security
staConnect.setup()
staConnect.start()
print("napping %f sec" % staConnect.runtime_secs)
time.sleep(staConnect.runtime_secs)
staConnect.stop()
run_results = staConnect.get_result_list()

4
py-scripts/test_ipv4_connection.py
View File

@@ -31,13 +31,13 @@ class IPv4Test(LFCliBase):
self.prefix = prefix
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.profile = realm.StationProfile(self.lfclient_url, ssid=self.ssid, ssid_pass=self.password,
security=self.security, prefix=self.prefix, mode=0, up=True, dhcp=True,
security=self.security, number_template_=self.prefix, mode=0, up=True, dhcp=True,
debug_=False)
def build(self):
# Build stations
self.profile.use_wpa2(True, self.ssid, self.password)
self.profile.set_prefix(self.prefix)
self.profile.set_number_template(self.prefix)
print("Creating stations")
self.profile.set_command_flag("add_sta", "create_admin_down", 1)
self.profile.create(resource=1, radio="wiphy0", sta_names_=self.sta_list, debug=False)

4
py-scripts/test_ipv4_variable_time.py
View File

@@ -33,7 +33,7 @@ class IPV4VariableTime(LFCliBase):
self.prefix = prefix
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.station_profile = realm.StationProfile(self.lfclient_url, ssid=self.ssid, ssid_pass=self.password,
security=self.security, prefix=self.prefix, mode=0, up=True,
security=self.security, number_template_=self.prefix, mode=0, up=True,
dhcp=True,
debug_=False)
self.cx_profile = realm.L3CXProfile(self.host, self.port, self.local_realm, side_a_min_bps=side_a_min_rate,
@@ -163,7 +163,7 @@ class IPV4VariableTime(LFCliBase):
sta_list = []
self.station_profile.use_wpa2(True, self.ssid, self.password)
self.station_profile.set_prefix(self.prefix)
self.station_profile.set_number_template(self.prefix)
print("Creating stations")
self.station_profile.create(resource=1, radio="wiphy0", num_stations=self.num_stations, debug=False)

4
py-scripts/test_ipv6_variable_time.py
View File

@@ -33,7 +33,7 @@ class IPV6VariableTime(LFCliBase):
self.prefix = prefix
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.station_profile = realm.StationProfile(self.lfclient_url, ssid=self.ssid, ssid_pass=self.password,
security=self.security, prefix=self.prefix, mode=0, up=True,
security=self.security, number_template_=self.prefix, mode=0, up=True,
dhcp=True,
debug_=False)
self.cx_profile = realm.L3CXProfile(self.host, self.port, self.local_realm, side_a_min_rate=side_a_min_rate,
@@ -163,7 +163,7 @@ class IPV6VariableTime(LFCliBase):
sta_list = []
self.station_profile.use_wpa2(True, self.ssid, self.password)
self.station_profile.set_prefix(self.prefix)
self.station_profile.set_number_template(self.prefix)
print("Creating stations")
self.station_profile.create(resource=1, radio="wiphy0", num_stations=self.num_stations, debug=False)

2
py-scripts/test_wanlink.py
View File

@@ -30,7 +30,7 @@ class LANtoWAN(LFCliBase):
self.wan_port = wan_port
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.profile = realm.StationProfile(self.lfclient_url, ssid=self.ssid, ssid_pass=self.password,
security=self.security, prefix=self.prefix, mode=0, up=True, dhcp=True,
security=self.security, number_template_=self.prefix, mode=0, up=True, dhcp=True,
debug_=False)
self.cxProfile = realm.new_l3_cx_profile()