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

added sniffer & supplicant logs for client-connect

Browse Source 
Signed-off-by: anil-tegala <anil.tegala@candelatech.com>
This commit is contained in:
anil-tegala
2022-12-06 09:14:40 +05:30
parent 3ff11952d7
commit 202a348ca8
1 changed files with 49 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
lf_libs/lf_tests.py
View File

@@ -582,7 +582,6 @@ class lf_tests(lf_libs):
logging.info("Setup interface data:\n" + json.dumps(str(data), indent=2))
# query and fetch vlan Ip Address
port_data = self.json_get(_req_url="/port?fields=alias,port+type,ip,mac")['interfaces']
# Fail if Vlan don't have IP
@@ -737,6 +736,33 @@ class lf_tests(lf_libs):
client_connect_obj = []
station_data_all = {}
for radio in data[identifier]["station_data"]:
if band == "twog":
if dict(dut_data.get(identifier)["radio_data"]).keys().__contains__("2G") and \
dict(dut_data.get(identifier)["radio_data"])["2G"] is not None:
sniffer_channel = dict(dut_data.get(identifier)["radio_data"])["2G"]["channel"]
if data[identifier]["sniff_radio_2g"] is not None and sniffer_channel is not None:
self.start_sniffer(radio_channel=sniffer_channel, test_name=f'{data[identifier]["station_data"][radio][0]}',
radio=data[identifier]["sniff_radio_2g"],
duration=120)
logging.info("started-sniffer")
if band == "fiveg":
if dict(dut_data.get(identifier)["radio_data"]).keys().__contains__("5G") and \
dict(dut_data.get(identifier)["radio_data"])["5G"] is not None:
sniffer_channel = dict(dut_data.get(identifier)["radio_data"])["5G"]["channel"]
if data[identifier]["sniff_radio_5g"] is not None and sniffer_channel is not None:
self.start_sniffer(radio_channel=sniffer_channel,
radio=data[identifier]["sniff_radio_5g"],
duration=120)
logging.info("started-sniffer")
if band == "sixg":
if dict(dut_data.get(identifier)["radio_data"]).keys().__contains__("6G") and \
dict(dut_data.get(identifier)["radio_data"])["6G"] is not None:
sniffer_channel = dict(dut_data.get(identifier)["radio_data"])["6G"]["channel"]
if data[identifier]["sniff_radio_6g"] is not None and sniffer_channel is not None:
self.start_sniffer(radio_channel=sniffer_channel,
radio=data[identifier]["sniff_radio_6g"],
duration=120)
logging.info("started-sniffer")
client_connect = CreateStation(_host=self.manager_ip, _port=self.manager_http_port,
_sta_list=data[identifier]["station_data"][radio],
_password=data[identifier]["passkey"],
@@ -777,12 +803,26 @@ class lf_tests(lf_libs):
if allure_attach:
self.attach_table_allure(data=sta_table_dict, allure_name=allure_name)
# stop sniffer if active
logging.info(msg=str("Cleaning up sniffer interface If available on PORT Manager"))
port_data = self.json_get(_req_url="/port?fields=alias,parent+dev,port+type,ip,mac")['interfaces']
# for i in port_data:
# for item in i:
# if i[item]['port type'] == '802.1Q VLAN' and i[item]['ip'] == '0.0.0.0':
# logging.info('VLAN do not have IP')
if self.start_sniffer:
self.stop_sniffer()
logging.info("pass_fail result: " + str(pass_fail))
if False in pass_fail:
logging.info("Station did not get an ip")
for radio in data[identifier]["station_data"]:
self.get_supplicant_logs(radio=str(radio))
pytest.fail("Station did not get an ip")
else:
logging.info("ALL Stations got IP's")
for radio in data[identifier]["station_data"]:
self.get_supplicant_logs(radio=str(radio))
return station_data_all
def dfs_test(self, ssid=None, security=None, passkey=None, mode=None,
@@ -1309,9 +1349,11 @@ class lf_tests(lf_libs):
val = [['ul_rate_sel: Per-Station Upload Rate:']]
thr1 = threading.Thread(target=thread_fun, args=(sta_list[8:16],))
thr1.start()
wct_obj = self.wifi_capacity(instance_name=instance_name, mode=mode, vlan_id=vlan, download_rate=downld_rate,
wct_obj = self.wifi_capacity(instance_name=instance_name, mode=mode, vlan_id=vlan,
download_rate=downld_rate,
stations=sel_stations, raw_lines=val, batch_size="8", upload_rate=upld_rate,
protocol="UDP-IPv4", duration="120000", create_stations=False, dut_data=dut_data,
protocol="UDP-IPv4", duration="120000", create_stations=False,
dut_data=dut_data,
sort="interleave", )
report_name = wct_obj[0].report_name[0]['LAST']["response"].split(":::")[1].split("/")[-1] + "/"