mirror of
https://github.com/Telecominfraproject/wlan-lanforge-scripts.git
synced 2025-10-30 18:27:53 +00:00
534 lines
26 KiB
Python
534 lines
26 KiB
Python
#!/usr/bin/env python3
|
|
import sys
|
|
import os
|
|
import importlib
|
|
import time
|
|
import datetime
|
|
import argparse
|
|
import allure
|
|
from pprint import pprint
|
|
|
|
if sys.version_info[0] != 3:
|
|
print("This script requires Python 3")
|
|
exit(1)
|
|
|
|
sys.path.append(os.path.join(os.path.abspath(__file__ + "../../../")))
|
|
|
|
LFUtils = importlib.import_module("py-json.LANforge.LFUtils")
|
|
realm = importlib.import_module("py-json.realm")
|
|
Realm = realm.Realm
|
|
l3_cxprofile = importlib.import_module("py-json.l3_cxprofile")
|
|
multicast_profile = importlib.import_module("py-json.multicast_profile")
|
|
station_profile = importlib.import_module("py-json.station_profile")
|
|
wifi_monitor = importlib.import_module("py-json.wifi_monitor_profile")
|
|
cv_test_reports = importlib.import_module("py-json.cv_test_reports")
|
|
from lf_sniff_radio import SniffRadio
|
|
lf_rpt = cv_test_reports.lanforge_reports
|
|
|
|
import pyshark as ps
|
|
|
|
'''
|
|
Usage: python3 lf_power_save_with_unicast.py --mgr 192.168.200.229 --ssid Endurance1 --radio wiphy0
|
|
--upstream_port 1.1.eth1 --monitor_radio wiphy0
|
|
--report_path /home/mahesh/Desktop/lanforge-scripts/lanforge-scripts/py-scripts
|
|
'''
|
|
|
|
|
|
class UnicastPowersaveTraffic(Realm):
|
|
|
|
def __init__(self, host, port, ssid, security, password, station_list, min_rate_multi_cast=56,
|
|
max_rate_multi_cast=56,
|
|
side_a_unicast_min_rate=56, side_b_unicast_min_rate=56, side_a_unicast_max_rate=0,
|
|
side_b_unicast_max_rate=0, pdu_size=1000,band="sixg",
|
|
upstream="1.1.eth1", interface_to_capture=None,
|
|
prefix="00000", test_duration="5m", report_path="",
|
|
station_radio="wiphy0", monitor_radio="wiphy0", remote_host_cap_ip=None,
|
|
output_file_for_cap="",
|
|
remote_host_cap_interface=None,
|
|
_debug_on=False, _exit_on_error=False, _exit_on_fail=False):
|
|
super().__init__(lfclient_host=host, lfclient_port=port, debug_=_debug_on)
|
|
self.host = host
|
|
self.port = port
|
|
self.ssid = ssid
|
|
self.security = security
|
|
self.password = password
|
|
self.sta_list = station_list
|
|
self.prefix = prefix
|
|
self.station_radio = station_radio
|
|
self.monitor_radio = monitor_radio
|
|
self.debug = _debug_on
|
|
self.upstream = upstream
|
|
self.min_rate = min_rate_multi_cast
|
|
self.max_rate = max_rate_multi_cast
|
|
self.output_file = output_file_for_cap
|
|
self.enable_multicast_testing = False
|
|
self.enable_unicast_testing = False
|
|
self.filter = ""
|
|
self.report_dir = report_path
|
|
self.captured_file_name = ""
|
|
self.sta_mac = ""
|
|
self.ap_mac = ""
|
|
self.upstream_mac = ""
|
|
self.band = band
|
|
if interface_to_capture is not None:
|
|
self.live_cap_timeout = interface_to_capture
|
|
else:
|
|
self.live_cap_timeout = station_radio
|
|
if remote_host_cap_ip and remote_host_cap_interface is not None:
|
|
self.remote_cap_host = remote_host_cap_ip
|
|
self.remote_cap_interface = remote_host_cap_interface
|
|
|
|
self.new_monitor = self.new_wifi_monitor_profile()
|
|
self.test_duration = test_duration
|
|
# upload
|
|
self.cx_prof_upload = l3_cxprofile.L3CXProfile(self.host, self.port, local_realm=self,
|
|
side_a_min_bps=side_a_unicast_min_rate, side_b_min_bps=0,
|
|
side_a_max_bps=side_a_unicast_max_rate, side_b_max_bps=0,
|
|
side_a_min_pdu=pdu_size, side_a_max_pdu=pdu_size,
|
|
side_b_min_pdu=0, side_b_max_pdu=0, debug_=self.debug)
|
|
|
|
# download
|
|
self.cx_prof_download = l3_cxprofile.L3CXProfile(self.host, self.port, local_realm=self,
|
|
side_a_min_bps=0, side_b_min_bps=side_b_unicast_min_rate,
|
|
side_a_max_bps=0, side_b_max_bps=side_b_unicast_max_rate,
|
|
side_a_min_pdu=0, side_a_max_pdu=0,
|
|
side_b_min_pdu=pdu_size, side_b_max_pdu=pdu_size,
|
|
debug_=self.debug)
|
|
self.multi_cast_profile = multicast_profile.MULTICASTProfile(self.host, self.port, local_realm=self)
|
|
|
|
self.station_profile = station_profile.StationProfile(self.lfclient_url, local_realm=self, ssid=self.ssid,
|
|
ssid_pass=self.password,
|
|
security=self.security, number_template_=self.prefix,
|
|
mode=0,
|
|
up=True,
|
|
dhcp=True,
|
|
debug_=self.debug)
|
|
self.new_monitor = wifi_monitor.WifiMonitor(self.lfclient_url, local_realm=self, debug_=self.debug)
|
|
|
|
def build_station_profile(self):
|
|
print("Station radio..in 108 lanfo-py-scri",self.station_radio)
|
|
self.station_profile.use_security(self.security, ssid=self.ssid, passwd=self.password)
|
|
self.station_profile.set_number_template(self.prefix)
|
|
self.station_profile.set_command_flag("add_sta", "create_admin_down", 1)
|
|
self.station_profile.set_command_param("set_port", "report_timer", 1500)
|
|
self.station_profile.set_command_flag("set_port", "rpt_timer", 1)
|
|
self.station_profile.set_command_flag("add_sta", "power_save_enable", 1)
|
|
self.station_profile.create(radio=self.station_radio, sta_names_=self.sta_list)
|
|
self._pass("PASS: Station builds finished")
|
|
|
|
def build_monitor(self, channel):
|
|
self.new_monitor.create(resource_=1, channel=channel, radio_=self.monitor_radio, name_="moni0")
|
|
|
|
def build_multi_cast_profile(self):
|
|
self.multi_cast_profile.create_mc_tx("mc_udp", self.upstream, min_rate=self.min_rate, max_rate=self.max_rate)
|
|
self.multi_cast_profile.create_mc_rx("mc_udp", self.sta_list)
|
|
|
|
def build_layer3_upload(self):
|
|
self.cx_prof_upload.name_prefix = "UDP_up"
|
|
print("Creating upload cx profile ")
|
|
self.cx_prof_upload.create(endp_type="lf_tcp", side_a=self.station_profile.station_names, side_b=self.upstream,
|
|
sleep_time=.05)
|
|
|
|
def build_layer3_download(self):
|
|
self.cx_prof_download.name_prefix = "TCP_down"
|
|
print("Creating download cx profile")
|
|
self.cx_prof_download.create(endp_type="lf_tcp", side_a=self.station_profile.station_names,
|
|
side_b=self.upstream,
|
|
sleep_time=.05)
|
|
|
|
# channel = self.json_get("/port/1/%s/%s/"%(1,"wiphy0"))
|
|
# rint("The channel name is...")
|
|
|
|
# station_channel = self.json_get("/port/1/%s/%s")
|
|
# pprint.pprint(station_channel)
|
|
|
|
def get_channel(self):
|
|
self.station_profile.admin_up()
|
|
# self.new_monitor.set_flag()
|
|
# print(self.station_profile.station_names)
|
|
print("Querying for channel from station created to create monitor interface on that particular channel")
|
|
if self.wait_for_ip(self.station_profile.station_names):
|
|
self._pass("All stations got IPs")
|
|
else:
|
|
self._fail("Stations failed to get IPs")
|
|
exit(1)
|
|
channel_info = self.json_get(f"port/1/1/{self.sta_list[0]}?fields=channel")
|
|
print(channel_info)
|
|
if channel_info is not None:
|
|
if 'interfaces' in channel_info:
|
|
for item in channel_info['interfaces']:
|
|
for k, v in item.items():
|
|
print("sta_name %s" % v['alias'])
|
|
print("mac %s" % v['mac'])
|
|
print("ap %s\n" % v['ap'])
|
|
elif 'interface' in channel_info:
|
|
print("channel %s" % channel_info['interface']['channel'])
|
|
self.station_profile.admin_down()
|
|
time.sleep(0.5)
|
|
return int(channel_info['interface']['channel'])
|
|
|
|
def get_captured_file_and_location(self):
|
|
return self.captured_file_name, self.report_dir, self.filter
|
|
|
|
def start(self,channel_info):
|
|
# start one test, measure
|
|
# start second test, measure
|
|
cur_time = datetime.datetime.now()
|
|
allure.attach(name="Test Start Time:", body=str(f"{cur_time}"))
|
|
end_time = self.parse_time(self.test_duration) + cur_time
|
|
allure.attach(name="Test End Time:", body=str(f"{end_time}"))
|
|
# admin up on new monitor
|
|
self.new_monitor.admin_up()
|
|
now = datetime.datetime.now()
|
|
date_time = now.strftime("%Y-%m-%d-%H%M%S")
|
|
curr_mon_name = self.new_monitor.monitor_name
|
|
# ("date and time: ",date_time)
|
|
duration_to_sec = 120
|
|
if self.test_duration[-1] == 'm':
|
|
duration_to_sec = int(self.test_duration.strip('m')) * 60
|
|
elif self.test_duration[-1] == 'h':
|
|
duration_to_sec = int(self.test_duration.strip('h')) * 60 * 60
|
|
elif self.test_duration[-1] == 's':
|
|
duration_to_sec = int(self.test_duration.strip('s'))
|
|
print(duration_to_sec)
|
|
|
|
self.captured_file_name = curr_mon_name + "-" + date_time + ".pcap"
|
|
if self.band == "sixg":
|
|
print("enterted into sixg")
|
|
lf_sniff_radio_obj = SniffRadio(lfclient_host=self.host,channel=channel_info,monitor_name="moni0")
|
|
print("frequency...",lf_sniff_radio_obj.freq)
|
|
lf_sniff_radio_obj.set_freq(self.host,"lanforge", freq=lf_sniff_radio_obj.freq)
|
|
print("Starting sniffing")
|
|
self.new_monitor.start_sniff(capname=curr_mon_name + "-" + date_time + ".pcap",
|
|
duration_sec=duration_to_sec)
|
|
print(f"Sniffer started will continue for {duration_to_sec} seconds")
|
|
# admin up on station
|
|
|
|
print("station profile got admin up")
|
|
self.station_profile.admin_up()
|
|
# self.new_monitor.set_flag()
|
|
# print(self.station_profile.station_names)
|
|
if self.wait_for_ip(self.station_profile.station_names):
|
|
self._pass("All stations got IPs")
|
|
else:
|
|
self._fail("Stations failed to get IPs")
|
|
print("Stations didn't received ip")
|
|
allure.attach(name="FAILED", body="Stations didn't connected to AP")
|
|
exit(1)
|
|
|
|
if self.enable_multicast_testing:
|
|
print("started multicast traffic")
|
|
self.multi_cast_profile.start_mc()
|
|
elif self.enable_unicast_testing:
|
|
print("started unicast traffic")
|
|
self.start_layer3()
|
|
|
|
# print station + MAC, AP
|
|
temp = []
|
|
for station in self.station_profile.station_names:
|
|
temp.append(self.name_to_eid(station)[2])
|
|
port_info = self.json_get("port/1/1/%s?fields=alias,ap,mac" % ','.join(temp))
|
|
print("port_info.........", port_info)
|
|
if port_info is not None:
|
|
if 'interfaces' in port_info:
|
|
for item in port_info['interfaces']:
|
|
for k, v in item.items():
|
|
print("sta_name %s" % v['alias'])
|
|
print("mac %s" % v['mac'])
|
|
print("ap %s\n" % v['ap'])
|
|
elif 'interface' in port_info:
|
|
print("sta_name %s" % port_info['interface']['alias'])
|
|
print("mac %s" % port_info['interface']['mac'])
|
|
print("ap %s\n" % port_info['interface']['ap'])
|
|
self.filter = "wlan.addr==" + port_info['interface']['mac'] + " || " + "wlan.addr==" + \
|
|
port_info['interface']['ap']
|
|
self.sta_mac = str(port_info['interface']['mac'])
|
|
self.ap_mac = str(port_info['interface']['ap'])
|
|
allure.attach(name="AP MAC", body=str(port_info['interface']['ap']))
|
|
allure.attach(name="Station MAC", body=str(port_info['interface']['mac']))
|
|
print("filter=", self.filter)
|
|
print("filter=", self.filter)
|
|
else:
|
|
print('interfaces and interface not in port_mgr_response')
|
|
allure.attach(name="FAILED", body=str("station data is not in response"))
|
|
exit(1)
|
|
temp_1 = []
|
|
temp_1.append(self.name_to_eid(self.upstream)[2])
|
|
upstream_info = self.json_get("port/1/1/%s?fields=alias,mac" % ','.join(temp_1))
|
|
print("upstream_info.........", upstream_info)
|
|
if upstream_info is not None:
|
|
if 'interfaces' in upstream_info:
|
|
for item in upstream_info['interfaces']:
|
|
for k, v in item.items():
|
|
print("upstream_name %s" % v['alias'])
|
|
print("mac %s" % v['mac'])
|
|
elif 'interface' in upstream_info:
|
|
print("sta_name %s" % upstream_info['interface']['alias'])
|
|
print("mac %s" % upstream_info['interface']['mac'])
|
|
self.upstream_mac = str(upstream_info['interface']['mac'])
|
|
allure.attach(name="Upstream Port MAC", body=str(upstream_info['interface']['mac']))
|
|
else:
|
|
print('interfaces and interface not in port_mgr_upstream_response')
|
|
exit(1)
|
|
exit(1)
|
|
|
|
while cur_time < end_time:
|
|
# DOUBLE CHECK
|
|
interval_time = cur_time + datetime.timedelta(minutes=1)
|
|
while cur_time < interval_time:
|
|
cur_time = datetime.datetime.now()
|
|
time.sleep(1)
|
|
|
|
# pulls report from lanforge to specified location in self.report_dir
|
|
lf_rpt.pull_reports(hostname=self.host, port=22, username="lanforge", password="lanforge",
|
|
report_location="/home/lanforge/" + curr_mon_name + "-" + date_time + ".pcap",
|
|
report_dir=self.report_dir)
|
|
# allure.attach.file(source=self.report_dir + "/" + self.captured_file_name, name="pcap_file",
|
|
# attachment_type=allure.attachment_type.PCAP)
|
|
|
|
'''
|
|
def capture_live_pcap(self):
|
|
try:
|
|
self.live_pcap = ps.LiveCapture(interface=self.live_pcap_interface, output_file=self.output_file)
|
|
self.live_pcap.sniff(timeout=300)
|
|
except ValueError:
|
|
raise "Capture Error"
|
|
return self.live_pcap
|
|
|
|
def capture_remote_pcap(self):
|
|
try:
|
|
self.remote_pcap = ps.RemoteCapture(remote_host=self.remote_cap_host,
|
|
remote_interface=self.remote_cap_interface)
|
|
except ValueError:
|
|
raise "Host error"
|
|
return self.remote_pcap
|
|
'''
|
|
|
|
def verify_unicast_pcap(self, pcap_file, apply_filter=None):
|
|
self.pcap_file = self.report_dir + "/" + pcap_file
|
|
# self.ap_mac = "10:f9:20:fd:e2:0b"
|
|
# self.sta_mac = "a8:93:4a:df:b0:3b"
|
|
|
|
if apply_filter is not None:
|
|
self.apply_filter = apply_filter
|
|
try:
|
|
pcap = ps.FileCapture(input_file=self.pcap_file, display_filter=self.apply_filter)
|
|
except Exception as error:
|
|
raise error
|
|
check_unicast = False
|
|
traffic_awaiting = False
|
|
block_ack = False
|
|
data_packet = False
|
|
result_data = False
|
|
QOS_data_pkt_count = False
|
|
for pkt in pcap: # traversing through all packets one by one
|
|
# print(str(pkt.wlan.fc_type_subtype))
|
|
# print(str(pkt.wlan.fc_pwrmgt))
|
|
if str(pkt.wlan.fc_type_subtype) == "0x002c" and str(pkt.wlan.fc_pwrmgt) == '1':
|
|
check_unicast = True
|
|
continue
|
|
|
|
if check_unicast:
|
|
if str(pkt.wlan.fc_type_subtype) == "0x0008":
|
|
if "wlan.mgt" in pkt and str(pkt["wlan.mgt"].wlan_tim_partial_virtual_bitmap) != '00':
|
|
traffic_awaiting = True
|
|
# print(f"PASSED:Beacon with Traffic buffered is seen {pkt.number}")
|
|
# allure.attach(name="Beacon",body=str(f"PASSED:Beacon with Traffic buffered is seen {pkt.number}"))
|
|
# print("found", pkt.number)
|
|
continue
|
|
|
|
if traffic_awaiting:
|
|
if str(pkt.wlan.fc_type_subtype) == "0x002c" and str(pkt.wlan.fc_pwrmgt) == '0':
|
|
# print(f"PASSED:QOS Null Function frame with pwr mgt 0 is seen {pkt.number}")
|
|
# allure.attach(name="QOS Null Function",body=str(f"PASSED:QOS Null Function frame with pwr mgt 0 is seen {pkt.number}"))
|
|
data_packet = True
|
|
continue
|
|
|
|
if str(pkt.wlan.fc_type_subtype) == "0x002c" and str(pkt.wlan.fc_pwrmgt) == '1':
|
|
# print("WARNING:QOS Null function with pwr mgt bit 1 without QOS data being transmitted ")
|
|
check_unicast = False
|
|
traffic_awaiting = False
|
|
data_packet = False
|
|
continue
|
|
|
|
if str(pkt.wlan.fc_type_subtype) == "0x002c" and str(pkt.wlan.fc_pwrmgt) == '0':
|
|
# print("WARNING:QOS Null function with pwr mgt bit 0 without QOS data being transmitted ")
|
|
check_unicast = False
|
|
traffic_awaiting = False
|
|
data_packet = False
|
|
continue
|
|
|
|
if data_packet:
|
|
if str(pkt.wlan.fc_type_subtype) == "0x0028":
|
|
if str(pkt.wlan.ta) == self.ap_mac and str(pkt.wlan.da) == self.sta_mac:
|
|
QOS_data_pkt_count=True
|
|
if str(pkt.wlan.fc_moredata) == "1":
|
|
data_packet=True
|
|
elif str(pkt.wlan.fc_moredata) == "0":
|
|
check_unicast = False
|
|
traffic_awaiting = False
|
|
data_packet = False
|
|
continue
|
|
# print(f"PASSED:Unicast Packet Transmitted after client came from powersave {pkt.number}")
|
|
# allure.attach(name="PASSED", body=str(
|
|
# f"Unicast Packet Transmitted after client came from powersave,packet Number {pkt.number}"))
|
|
# break
|
|
|
|
elif str(pkt.wlan.fc_type_subtype) == "0x0028":
|
|
if str(pkt.wlan.ta) == self.ap_mac and str(pkt.wlan.da) == self.sta_mac:
|
|
QOS_data_pkt_count=True
|
|
print(f"FAILED:Unexpected unicast data packet in {pkt.number}")
|
|
result_data = True
|
|
# pass
|
|
# print(f"FAILED:Unexpected unicast data packet in {pkt.number}")
|
|
allure.attach(name="FAILED",
|
|
body=str(f"Unexpected unicast data packet in capture, packet number {pkt.number}"))
|
|
|
|
if not QOS_data_pkt_count:
|
|
result_data=True
|
|
allure.attach(name="FAILED",body="QOS Data packets sent by the AP are not found in the capture for unicast traffic validation")
|
|
|
|
allure.attach.file(source=self.pcap_file, name="pcap_file", attachment_type=allure.attachment_type.PCAP)
|
|
return result_data
|
|
|
|
def start_station_profile(self):
|
|
self.station_profile.admin_up()
|
|
|
|
def stop_station_profile(self):
|
|
self.station_profile.admin_down()
|
|
|
|
def stop_monitor(self):
|
|
# switch off new monitor
|
|
self.new_monitor.admin_down()
|
|
|
|
def start_layer3(self):
|
|
# self.cx_prof_upload.start_cx()
|
|
self.cx_prof_download.start_cx()
|
|
|
|
def stop_layer3(self):
|
|
# self.cx_prof_upload.stop_cx()
|
|
self.cx_prof_download.stop_cx()
|
|
|
|
def cleanup_station_profile(self):
|
|
self.station_profile.cleanup(desired_stations=self.sta_list)
|
|
|
|
def cleanup_layer3(self):
|
|
self.cx_prof_download.cleanup()
|
|
self.cx_prof_upload.cleanup()
|
|
|
|
def cleanup_monitor(self):
|
|
self.new_monitor.cleanup()
|
|
|
|
def unicast_download_testing(self,cleanup=True):
|
|
self.enable_unicast_testing = True
|
|
self.station_profile.cleanup(desired_stations=self.sta_list)
|
|
self.station_profile.local_realm.remove_all_cxs(remove_all_endpoints=True)
|
|
self.build_station_profile() # function to build station profile
|
|
channel_info = self.get_channel()
|
|
self.build_monitor(channel_info) # function to create monitor
|
|
self.build_layer3_download() # function to build unicast download traffic
|
|
|
|
self.start(channel_info) # function to start sniff by admin up station followed by /
|
|
# starting the unicast traffic with desired pdu size for some duration
|
|
if cleanup:
|
|
self.unicast_testing_stop_and_cleanup()
|
|
|
|
def unicast_testing_stop_and_cleanup(self):
|
|
self.new_monitor.admin_down()
|
|
self.cx_prof_download.stop_cx()
|
|
self.station_profile.admin_down()
|
|
|
|
self.new_monitor.cleanup()
|
|
self.cx_prof_download.cleanup()
|
|
self.station_profile.cleanup(desired_stations=self.sta_list)
|
|
|
|
|
|
def main():
|
|
# Realm.create_basic_argparse defined in lanforge-scripts/py-json/LANforge/lfcli_base.py
|
|
parser = Realm.create_basic_argparse(
|
|
prog='lf_power_save_test_cases_cisco.py',
|
|
formatter_class=argparse.RawTextHelpFormatter,
|
|
epilog='''\
|
|
lf_power_save_test_cases_cisco.py
|
|
|
|
''',
|
|
description='''\
|
|
Example of creating traffic on an l3 connection
|
|
''')
|
|
|
|
parser.add_argument('--monitor_radio', help="--monitor_radio radio to be used in monitor creation",
|
|
default="wiphy1")
|
|
parser.add_argument('--report_path', help="desired path to save your pcap file fetched from lanforge through ssh",
|
|
default="/home/mahesh/Desktop/lanforge-scripts/lanforge-scripts/py-scripts")
|
|
parser.add_argument('--mcast_min_rate', help="value for multicast minimum rate in kbps", type=int, default=9000)
|
|
parser.add_argument('--mcast_max_rate', help="value for multicast maximum rate in kbps", type=int, default=128000)
|
|
|
|
parser.add_argument('--side_a_unicast_min_rate', help="value for multicast maximum rate in kbps", type=int,
|
|
default=9000)
|
|
parser.add_argument('--side_b_unicast_min_rate', help="value for multicast maximum rate in kbps", type=int,
|
|
default=9000)
|
|
parser.add_argument('--side_a_unicast_max_rate', help="value for multicast maximum rate in kbps", type=int,
|
|
default=128000)
|
|
parser.add_argument('--side_b_unicast_max_rate', help="value for multicast maximum rate in kbps", type=int,
|
|
default=128000)
|
|
parser.add_argument('--test_duration', help='duration of the test eg: 30s, 2m, 4h', default="5m")
|
|
|
|
parser.add_argument('--pdu_size', help="pdu size in bytes", type=int, default=1400)
|
|
|
|
args = parser.parse_args()
|
|
|
|
lfjson_host = args.mgr
|
|
lfjson_port = 8080
|
|
station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=0, padding_number_=10000)
|
|
ip_powersave_test = UnicastPowersaveTraffic(lfjson_host, lfjson_port, ssid=args.ssid,
|
|
security=args.security,
|
|
password=args.passwd, station_list=station_list,
|
|
min_rate_multi_cast=args.mcast_min_rate,
|
|
max_rate_multi_cast=args.mcast_max_rate,
|
|
side_a_unicast_min_rate=args.side_a_unicast_min_rate,
|
|
side_b_unicast_min_rate=args.side_b_unicast_min_rate,
|
|
side_a_unicast_max_rate=args.side_a_unicast_max_rate,
|
|
side_b_unicast_max_rate=args.side_b_unicast_max_rate,
|
|
pdu_size=args.pdu_size,
|
|
station_radio=args.radio,
|
|
upstream=args.upstream_port,
|
|
monitor_radio=args.monitor_radio,
|
|
test_duration=args.test_duration,
|
|
interface_to_capture=None,
|
|
_debug_on=args.debug, remote_host_cap_ip=None,
|
|
remote_host_cap_interface=None,
|
|
report_path=args.report_path,
|
|
output_file_for_cap="/home/lanforge",
|
|
_exit_on_error=True, _exit_on_fail=True)
|
|
|
|
#ip_powersave_test.unicast_download_testing() # function to run unicast test
|
|
|
|
#captured_file_name, local_dir, filter = ip_powersave_test.get_captured_file_and_location() # function returns captured_file_name along \
|
|
# with location of local directory where the \
|
|
# captured file is pulled
|
|
|
|
# below function should be executed along with multicast_testing or unicast_testing if we wish to disect captured file.
|
|
#ip_powersave_test.verify_unicast_pcap(local_dir + '/' + captured_file_name,
|
|
# apply_filter=filter)
|
|
|
|
# below function can be run solely without calling above functions to disect multicast packets \
|
|
# if we have ready packet captured file to test dtim_multicast
|
|
|
|
# ip_powersave_test.verify_dtim_multicast_pcap("/home/mahesh/Documents/moni0-2022-02-17-170018.pcap",
|
|
# apply_filter="wlan.addr==04:f0:21:94:1e:4d || wlan.addr==3C:37:86:13:81:60")
|
|
|
|
# ''' UNICAST TEST FUNCTIONS'''
|
|
# ip_powersave_test.verify_unicast_pcap(
|
|
# "/home/mahesh/Desktop/powersave_unicast_sta_wiphy1_sniff_wiphy3_AX210_tb1_36.pcapng",
|
|
# apply_filter="wlan.addr == 04:f0:21:ad:82:8f || wlan.addr == 68:7d:b4:60:04:be")
|
|
|
|
ip_powersave_test.verify_unicast_pcap(
|
|
"unicast_test_1.pcapng",
|
|
apply_filter="wlan.addr==10:f9:20:fd:e2:0b||wlan.addr==a8:93:4a:df:b0:3b")
|
|
|
|
# wlan.fc.type_subtype != 5 to avoid probe responses which is not having TIM which might gonna break script
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|