mirror of
https://github.com/Telecominfraproject/wlan-testing.git
synced 2025-11-03 20:38:05 +00:00
343 lines
17 KiB
Python
343 lines
17 KiB
Python
#########################################################################################################
|
|
# Used by Nightly_Sanity
|
|
# This has different types of old_pytest like Single client connectivity, Single_Client_EAP, testrail_retest
|
|
#########################################################################################################
|
|
import sys
|
|
import os
|
|
|
|
import allure
|
|
|
|
sys.path.append(
|
|
os.path.dirname(
|
|
os.path.realpath(__file__)
|
|
)
|
|
)
|
|
if "libs" not in sys.path:
|
|
sys.path.append(f'../libs')
|
|
for folder in 'py-json', 'py-scripts':
|
|
if folder not in sys.path:
|
|
sys.path.append(f'../../lanforge/lanforge-scripts/{folder}')
|
|
|
|
sys.path.append(f"../lanforge/lanforge-scripts/py-scripts/tip-cicd-sanity")
|
|
|
|
sys.path.append(f'../libs')
|
|
sys.path.append(f'../libs/lanforge/')
|
|
from sta_connect2 import StaConnect2
|
|
import time
|
|
# from eap_connect import EAPConnect
|
|
from test_ipv4_ttls import TTLSTest
|
|
from lf_wifi_capacity_test import WiFiCapacityTest
|
|
from create_station import CreateStation
|
|
import lf_dataplane_test
|
|
from lf_dataplane_test import DataplaneTest
|
|
from csv_to_influx import CSVtoInflux
|
|
from influx2 import RecordInflux
|
|
|
|
|
|
class RunTest:
|
|
|
|
def __init__(self, lanforge_data=None, local_report_path="../reports/", influx_params=None, debug=False):
|
|
self.lanforge_ip = lanforge_data["ip"]
|
|
self.lanforge_port = lanforge_data["port"]
|
|
self.twog_radios = lanforge_data["2.4G-Radio"]
|
|
self.fiveg_radios = lanforge_data["5G-Radio"]
|
|
self.ax_radios = lanforge_data["AX-Radio"]
|
|
self.upstream_port = lanforge_data["upstream"].split(".")[2]
|
|
self.twog_prefix = lanforge_data["2.4G-Station-Name"]
|
|
self.fiveg_prefix = lanforge_data["5G-Station-Name"]
|
|
self.ax_prefix = lanforge_data["AX-Station-Name"]
|
|
self.debug = debug
|
|
self.lf_ssh_port = lanforge_data["ssh_port"]
|
|
self.staConnect = None
|
|
self.dataplane_obj = None
|
|
self.wificapacity_obj = None
|
|
self.influx_params = influx_params
|
|
self.influxdb = RecordInflux(_lfjson_host=self.lanforge_ip,
|
|
_lfjson_port=self.lanforge_port,
|
|
_influx_host=influx_params["influx_host"],
|
|
_influx_port=influx_params["influx_port"],
|
|
_influx_org=influx_params["influx_org"],
|
|
_influx_token=influx_params["influx_token"],
|
|
_influx_bucket=influx_params["influx_bucket"])
|
|
self.local_report_path = local_report_path
|
|
if not os.path.exists(self.local_report_path):
|
|
os.mkdir(self.local_report_path)
|
|
# self.staConnect = StaConnect2(self.lanforge_ip, self.lanforge_port, debug_=self.debug)
|
|
|
|
def Client_Connectivity(self, ssid="[BLANK]", passkey="[BLANK]", security="open", extra_securities=[],
|
|
station_name=[], mode="BRIDGE", vlan_id=1, band="twog", cleanup=True):
|
|
"""SINGLE CLIENT CONNECTIVITY using test_connect2.py"""
|
|
self.staConnect = StaConnect2(self.lanforge_ip, self.lanforge_port, debug_=self.debug)
|
|
self.staConnect.sta_mode = 0
|
|
self.staConnect.upstream_resource = 1
|
|
if mode == "BRIDGE":
|
|
self.staConnect.upstream_port = self.upstream_port
|
|
elif mode == "NAT":
|
|
self.staConnect.upstream_port = self.upstream_port
|
|
else:
|
|
self.staConnect.upstream_port = self.upstream_port + "." + str(vlan_id)
|
|
if band == "twog":
|
|
self.staConnect.radio = self.twog_radios[0]
|
|
self.staConnect.sta_prefix = self.twog_prefix
|
|
if band == "fiveg":
|
|
self.staConnect.radio = self.fiveg_radios[0]
|
|
self.staConnect.sta_prefix = self.fiveg_prefix
|
|
self.staConnect.resource = 1
|
|
self.staConnect.dut_ssid = ssid
|
|
self.staConnect.dut_passwd = passkey
|
|
self.staConnect.dut_security = security
|
|
self.staConnect.station_names = station_name
|
|
self.staConnect.runtime_secs = 40
|
|
self.staConnect.bringup_time_sec = 80
|
|
self.staConnect.cleanup_on_exit = True
|
|
# self.staConnect.cleanup()
|
|
self.staConnect.setup(extra_securities=extra_securities)
|
|
self.staConnect.start()
|
|
print("napping %f sec" % self.staConnect.runtime_secs)
|
|
time.sleep(self.staConnect.runtime_secs)
|
|
for sta_name in self.staConnect.station_names:
|
|
try:
|
|
station_data_str = ""
|
|
sta_url = self.staConnect.get_station_url(sta_name)
|
|
station_info = self.staConnect.json_get(sta_url)
|
|
for i in station_info["interface"]:
|
|
try:
|
|
station_data_str = station_data_str + i + " : " + str(station_info["interface"][i]) + "\n"
|
|
except Exception as e:
|
|
print(e)
|
|
allure.attach(name=str(sta_name), body=str(station_data_str))
|
|
except Exception as e:
|
|
print(e)
|
|
self.staConnect.stop()
|
|
if cleanup:
|
|
self.staConnect.cleanup()
|
|
run_results = self.staConnect.get_result_list()
|
|
for result in run_results:
|
|
print("test result: " + result)
|
|
result = True
|
|
print("Client Connectivity :", self.staConnect.passes)
|
|
if self.staConnect.passes():
|
|
print("client connection to", self.staConnect.dut_ssid, "successful. Test Passed")
|
|
else:
|
|
print("client connection to", self.staConnect.dut_ssid, "unsuccessful. Test Failed")
|
|
result = False
|
|
time.sleep(3)
|
|
return self.staConnect.passes(), result
|
|
|
|
def EAP_Connect(self, ssid="[BLANK]", passkey="[BLANK]", security="wpa2", extra_securities=[],
|
|
mode="BRIDGE", band="twog", vlan_id=100,
|
|
station_name=[], key_mgmt="WPA-EAP",
|
|
pairwise="NA", group="NA", wpa_psk="DEFAULT",
|
|
ttls_passwd="nolastart", ieee80211w=1,
|
|
wep_key="NA", ca_cert="NA", eap="TTLS", identity="nolaradius"):
|
|
self.eap_connect = TTLSTest(host=self.lanforge_ip, port=self.lanforge_port,
|
|
sta_list=station_name, vap=False, _debug_on=self.debug)
|
|
|
|
self.eap_connect.station_profile.sta_mode = 0
|
|
self.eap_connect.upstream_resource = 1
|
|
if mode == "BRIDGE":
|
|
self.eap_connect.l3_cx_obj_udp.upstream = self.upstream_port
|
|
self.eap_connect.l3_cx_obj_tcp.upstream = self.upstream_port
|
|
elif mode == "NAT":
|
|
self.eap_connect.l3_cx_obj_udp.upstream = self.upstream_port
|
|
self.eap_connect.l3_cx_obj_tcp.upstream = self.upstream_port
|
|
else:
|
|
self.eap_connect.l3_cx_obj_udp.upstream = self.upstream_port + "." + str(vlan_id)
|
|
self.eap_connect.l3_cx_obj_tcp.upstream = self.upstream_port + "." + str(vlan_id)
|
|
if band == "twog":
|
|
self.eap_connect.radio = self.twog_radios[0]
|
|
# self.eap_connect.sta_prefix = self.twog_prefix
|
|
if band == "fiveg":
|
|
self.eap_connect.radio = self.fiveg_radios[0]
|
|
# self.eap_connect.sta_prefix = self.fiveg_prefix
|
|
# self.eap_connect.resource = 1
|
|
if eap == "TTLS":
|
|
self.eap_connect.ieee80211w = ieee80211w
|
|
self.eap_connect.key_mgmt = key_mgmt
|
|
self.eap_connect.station_profile.set_command_flag("add_sta", "80211u_enable", 0)
|
|
self.eap_connect.identity = identity
|
|
self.eap_connect.ttls_passwd = ttls_passwd
|
|
self.eap_connect.pairwise = pairwise
|
|
self.eap_connect.group = group
|
|
if eap == "TLS":
|
|
self.eap_connect.key_mgmt = "WPA-EAP-SUITE-B"
|
|
self.eap_connect.station_profile.set_command_flag("add_sta", "80211u_enable", 0)
|
|
self.eap_connect.pairwise = "TKIP"
|
|
self.eap_connect.group = "TKIP"
|
|
self.eap_connect.eap = "EAP-TLS"
|
|
|
|
# self.eap_connect.hs20_enable = False
|
|
self.eap_connect.ssid = ssid
|
|
self.eap_connect.password = passkey
|
|
self.eap_connect.security = security
|
|
self.eap_connect.sta_list = station_name
|
|
self.eap_connect.build(extra_securities=extra_securities)
|
|
self.eap_connect.start(station_name, True, True)
|
|
for sta_name in station_name:
|
|
# try:
|
|
station_data_str = ""
|
|
# sta_url = self.eap_connect.get_station_url(sta_name)
|
|
# station_info = self.eap_connect.json_get(sta_url)
|
|
station_info = self.eap_connect.json_get("port/1/1/" + sta_name)
|
|
for i in station_info["interface"]:
|
|
try:
|
|
station_data_str = station_data_str + i + " : " + str(station_info["interface"][i]) + "\n"
|
|
except Exception as e:
|
|
print(e)
|
|
allure.attach(name=str(sta_name), body=str(station_data_str))
|
|
# except Exception as e:
|
|
# print(e)
|
|
|
|
self.eap_connect.stop()
|
|
endp_data = []
|
|
for i in self.eap_connect.resulting_endpoints:
|
|
endp_data.append(self.eap_connect.resulting_endpoints[i]["endpoint"])
|
|
cx_data = ""
|
|
for i in endp_data:
|
|
for j in i:
|
|
cx_data = cx_data + str(j) + " : " + str(i[j]) + "\n"
|
|
cx_data = cx_data + "\n"
|
|
allure.attach(name="cx_data", body=str(cx_data))
|
|
self.eap_connect.cleanup(station_name)
|
|
return self.eap_connect.passes()
|
|
|
|
def wifi_capacity(self, mode="BRIDGE", vlan_id=100, instance_name="wct_instance", stations=None):
|
|
if mode == "BRIDGE":
|
|
self.client_connect.upstream_port = self.upstream_port
|
|
elif mode == "NAT":
|
|
self.client_connect.upstream_port = self.upstream_port
|
|
elif mode == "VLAN":
|
|
self.client_connect.upstream_port = self.upstream_port + "." + str(vlan_id)
|
|
self.wificapacity_obj = WiFiCapacityTest(lfclient_host=self.lanforge_ip,
|
|
lf_port=self.lanforge_port,
|
|
lf_user="lanforge",
|
|
lf_password="lanforge",
|
|
report_dir=self.local_report_path,
|
|
instance_name=instance_name,
|
|
config_name="wifi_config",
|
|
upstream="1.1." + self.upstream_port,
|
|
batch_size="1",
|
|
loop_iter="1",
|
|
protocol="UDP-IPv4",
|
|
duration="3000",
|
|
pull_report=True,
|
|
load_old_cfg=False,
|
|
upload_rate="10Mbps",
|
|
download_rate="1Gbps",
|
|
sort="interleave",
|
|
stations=stations,
|
|
create_stations=False,
|
|
radio=None,
|
|
security=None,
|
|
paswd=None,
|
|
ssid=None,
|
|
enables=[],
|
|
disables=[],
|
|
raw_lines=[],
|
|
raw_lines_file="",
|
|
sets=[])
|
|
|
|
self.wificapacity_obj.setup()
|
|
self.wificapacity_obj.run()
|
|
report_name = self.wificapacity_obj.report_name[0]['LAST']["response"].split(":::")[1].split("/")[-1]
|
|
influx = CSVtoInflux(influxdb=self.influxdb, _influx_tag=self.influx_params["influx_tag"],
|
|
target_csv=self.local_report_path + report_name + "/kpi.csv")
|
|
influx.post_to_influx()
|
|
|
|
return self.wificapacity_obj
|
|
|
|
def Client_Connect(self, ssid="[BLANK]", passkey="[BLANK]", security="wpa2", mode="BRIDGE", band="twog",
|
|
vlan_id=100,
|
|
station_name=[]):
|
|
|
|
self.client_connect = CreateStation(_host=self.lanforge_ip, _port=self.lanforge_port,
|
|
_sta_list=station_name, _password=passkey, _ssid=ssid, _security=security)
|
|
|
|
self.client_connect.station_profile.sta_mode = 0
|
|
self.client_connect.upstream_resource = 1
|
|
if mode == "BRIDGE":
|
|
self.client_connect.upstream_port = self.upstream_port
|
|
elif mode == "NAT":
|
|
self.client_connect.upstream_port = self.upstream_port
|
|
else:
|
|
self.client_connect.upstream_port = self.upstream_port + "." + str(vlan_id)
|
|
if band == "twog":
|
|
self.client_connect.radio = self.twog_radios[0]
|
|
# self.client_connect.sta_prefix = self.twog_prefix
|
|
if band == "fiveg":
|
|
self.client_connect.radio = self.fiveg_radios[0]
|
|
self.client_connect.build()
|
|
|
|
self.client_connect.wait_for_ip(station_name, timeout_sec=60)
|
|
if self.client_connect.wait_for_ip(station_name, timeout_sec=60):
|
|
self.client_connect._pass("ALL Stations got IP's", print_=True)
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
def Client_disconnect(self, station_name=[]):
|
|
self.client_dis = CreateStation(_host=self.lanforge_ip, _port=self.lanforge_port,
|
|
_sta_list=station_name, _password="passkey", _ssid="ssid", _security="security")
|
|
self.client_dis.station_profile.cleanup(station_name)
|
|
return True
|
|
|
|
def dataplane(self, station_name=None, mode="BRIDGE", vlan_id=100, download_rate="85%", dut_name="TIP",
|
|
upload_rate="85%", duration="1m", instance_name="test_demo", raw_lines=None):
|
|
if mode == "BRIDGE":
|
|
self.client_connect.upstream_port = self.upstream_port
|
|
elif mode == "NAT":
|
|
self.client_connect.upstream_port = self.upstream_port
|
|
elif mode == "VLAN":
|
|
self.client_connect.upstream_port = self.upstream_port + "." + str(vlan_id)
|
|
|
|
if raw_lines is None:
|
|
raw_lines = [['pkts: MTU'], ['directions: DUT Transmit;DUT Receive'], ['traffic_types: UDP;TCP'],
|
|
["show_3s: 1"], ["show_ll_graphs: 1"], ["show_log: 1"]]
|
|
|
|
self.dataplane_obj = DataplaneTest(lf_host=self.lanforge_ip,
|
|
lf_port=self.lanforge_port,
|
|
ssh_port=self.lf_ssh_port,
|
|
local_path=self.local_report_path,
|
|
lf_user="lanforge",
|
|
lf_password="lanforge",
|
|
instance_name=instance_name,
|
|
config_name="dpt_config",
|
|
upstream="1.1." + self.upstream_port,
|
|
pull_report=True,
|
|
load_old_cfg=False,
|
|
download_speed=download_rate,
|
|
upload_speed=upload_rate,
|
|
duration=duration,
|
|
dut=dut_name,
|
|
station="1.1." + station_name[0],
|
|
raw_lines=raw_lines)
|
|
|
|
self.dataplane_obj.setup()
|
|
self.dataplane_obj.run()
|
|
report_name = self.dataplane_obj.report_name[0]['LAST']["response"].split(":::")[1].split("/")[-1]
|
|
influx = CSVtoInflux(influxdb=self.influxdb, _influx_tag=self.influx_params["influx_tag"],
|
|
target_csv=self.local_report_path + report_name + "/kpi.csv")
|
|
influx.post_to_influx()
|
|
|
|
return self.dataplane_obj
|
|
|
|
|
|
if __name__ == '__main__':
|
|
lanforge_data = {
|
|
"ip": "localhost",
|
|
"port": 8801,
|
|
"ssh_port": 22,
|
|
"2.4G-Radio": ["wiphy0"],
|
|
"5G-Radio": ["wiphy1"],
|
|
"AX-Radio": ["wiphy2"],
|
|
"upstream": "1.1.eth1",
|
|
"2.4G-Station-Name": "wlan0",
|
|
"5G-Station-Name": "wlan0",
|
|
"AX-Station-Name": "ax",
|
|
}
|
|
obj = RunTest(lanforge_data=lanforge_data, debug=False)
|
|
# a = obj.staConnect.json_get("/events/since/")
|
|
# print(a)
|
|
# print(obj.eap_connect.json_get("port/1/1/sta0000?fields=ap,ip"))
|
|
# obj.EAP_Connect(station_name=["sta0000", "sta0001"], eap="TTLS", ssid="testing_radius")
|