scottk/wlan-lanforge-scripts
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-lanforge-scripts.git synced 2025-10-30 02:12:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' of github.com:greearb/lanforge-scripts

Browse Source
This commit is contained in:
Matthew Stidham
2020-12-28 13:11:35 -08:00
parent 875942bd75 a2564b2dca
commit e4b8266a20
9 changed files with 491 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
py-json/LANforge/LFUtils.py
View File

@@ -25,6 +25,7 @@ ADD_STA_FLAGS_DOWN_WPA2 = 68719477760
REPORT_TIMER_MS_FAST = 1500
REPORT_TIMER_MS_SLOW = 3000
#Used for Speed
def parse_size_bps(size_val):
if isinstance(size_val, str):
size_val.upper()
@@ -46,6 +47,7 @@ def parse_size_bps(size_val):
else:
return size_val
#Used for Size of file
def parse_size(size_val):
if isinstance(size_val, str):
size_val.upper()
@@ -56,11 +58,11 @@ def parse_size(size_val):
unit = str(td.group(2)).lower()
# print(1, size, unit)
if unit == 'g':
size *= 10000000
size *= 1073741824
elif unit == 'm':
size *= 100000
size *= 1048576
elif unit == 'k':
size *= 1000
size *= 1024
# print(2, size, unit)
return size
else:

52
py-scripts/connection_test.py
View File

@@ -22,7 +22,8 @@ import datetime
import time
import os
from test_utility import CreateHTML
import matplotlib.pyplot as plt
from test_utility import RuntimeUpdates
import pdfkit
webconsole_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.getcwd())))
@@ -30,7 +31,7 @@ class ConnectionTest(LFCliBase):
def __init__(self, lfclient_host="localhost", lfclient_port=8080, radio="wiphy1", sta_prefix="sta", start_id=0,
num_sta=2,
dut_ssid="lexusdut", dut_security="open", dut_passwd="[BLANK]", upstream="eth1", name_prefix="L3Test",
dut_ssid="lexusdut", dut_security="open", dut_passwd="[BLANK]", upstream="eth1", _test_update=None, name_prefix="L3Test",
session_id="Layer3Test", test_name="Client/s Connectivity Test", pass_criteria=20, _debug_on=False,
_exit_on_error=False, _exit_on_fail=False):
super().__init__(lfclient_host, lfclient_port, _debug=_debug_on, _halt_on_error=_exit_on_error,
@@ -52,10 +53,10 @@ class ConnectionTest(LFCliBase):
self.session_id = session_id
self.test_name = test_name
self.test_duration = 1
self.test_update = _test_update
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.station_profile = self.local_realm.new_station_profile()
self.pass_fail = ""
station_list = []
for i in range(0, self.num_sta):
@@ -66,7 +67,7 @@ class ConnectionTest(LFCliBase):
self.station_data[i] = "None"
print(self.station_data)
self.test_update.send_update({"test_status": '1', "data": 'None', "data": [], "label": "Client Connectivity Time"})
self.reports_path = webconsole_dir+"/reports/" + self.test_name + "_" + self.session_id + '/'
@@ -78,16 +79,16 @@ class ConnectionTest(LFCliBase):
self.station_list = LFUtils.portNameSeries(prefix_=self.sta_prefix, start_id_=self.sta_start_id,
end_id_=self.num_sta - 1, padding_number_=10000, radio=self.radio)
print(self.station_profile.station_names)
self.test_update.send_update({"test_status": '2', "data": 'None', "data": [], "label": "Client Connectivity Time"})
def precleanup(self):
print("precleanup started")
sta_list = []
for i in self.local_realm.station_list():
if list(i.keys())[0] == '1.1.wlan0':
if (list(i.keys())[0] == '1.1.wlan0'):
pass
elif list(i.keys())[0] == '1.1.wlan1':
elif (list(i.keys())[0] == '1.1.wlan1'):
pass
else:
sta_list.append(list(i.keys())[0])
@@ -99,8 +100,10 @@ class ConnectionTest(LFCliBase):
LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=sta_list,
debug=self.debug)
print("precleanup done")
self.test_update.send_update({"test_status": '3', "data": 'None', "data": [], "label": "Client Connectivity Time"})
def build(self):
print("Building Test Configuration")
self.station_profile.use_security(self.security, self.ssid, self.password)
self.station_profile.set_number_template("00")
@@ -111,13 +114,14 @@ class ConnectionTest(LFCliBase):
self.local_realm.wait_until_ports_appear(sta_list=self.station_list)
self.update(status="build complete")
print("Test Build done")
self.test_update.send_update({"test_status": '4', "data": 'None', "data": [], "label": "Client Connectivity Time"})
def update(self, status="None"):
for i in self.station_list:
print(self.json_get("port/1/1/" + i + "/?fields=ip,ap,down"))
self.station_data[i.split(".")[2]] = \
self.json_get("port/1/1/" + i.split(".")[2] + "/?fields=ip,ap,down,phantom&cx%20time%20(us)")['interface']
self.test_update.send_update({"test_status": '5', "data": 'None', "data": [], "label": "Client Connectivity Time"})
def start(self, print_pass=False, print_fail=False):
print("Test is starting")
@@ -131,11 +135,11 @@ class ConnectionTest(LFCliBase):
sta_status = self.json_get("port/1/1/" + str(sta_name).split(".")[2] + "?fields=port,alias,ip,ap",
debug_=self.debug)
print(sta_status)
if sta_status is None or sta_status['interface'] is None or sta_status['interface']['ap'] is None:
if (sta_status is None or sta_status['interface'] is None) or (sta_status['interface']['ap'] is None):
continue
if len(sta_status['interface']['ap']) == 17 and sta_status['interface']['ap'][-3] == ':':
if (len(sta_status['interface']['ap']) == 17) and (sta_status['interface']['ap'][-3] == ':'):
associated_map[sta_name] = 1
if sta_status['interface']['ip'] != '0.0.0.0':
if (sta_status['interface']['ip'] != '0.0.0.0'):
self.ip_map[sta_name] = 1
if (len(self.station_profile.station_names) == len(self.ip_map)) and (
len(self.station_profile.station_names) == len(associated_map)):
@@ -186,13 +190,15 @@ class ConnectionTest(LFCliBase):
sta_status = self.json_get(
"port/1/1/" + str(sta_name).split(".")[2] + "?fields=alias,ap,channel,cx%20time%20(us),ip,mac,mode,dhcp%20(ms)",
debug_=self.debug)
print("ironman")
print(sta_status['interface'])
self.test_result_data.append(sta_status['interface'])
print(self.test_result_data)
offset = 0
self.chart_data = {}
for data in self.test_result_data:
if data["cx time (us)"]/1000 <= self.pass_criteria and data["cx time (us)"]/1000 > 0:
if (data["cx time (us)"]/1000 <= self.pass_criteria) and (data["cx time (us)"]/1000 > 0):
self.chart_data[data['alias']] = data["cx time (us)"]/1000
data['Result'] = "PASS"
else:
@@ -205,22 +211,26 @@ class ConnectionTest(LFCliBase):
if offset == 0:
summary_result = 'PASS ' + str(len(self.ip_map)) + "/" + str(self.num_sta) + ' Clients are Connected in less than ' + str(self.pass_criteria) + " ms"
self.pass_fail = "FAIL"
else:
summary_result = 'FAIL ' + str(len(self.ip_map)) + "/" + str(self.num_sta) + ' Clients are Connected, and/or Some might got connected in more than ' + str(self.pass_criteria) + " ms"
self.pass_fail = "FAIL"
self.html = open(self.reports_path + self.test_name + "_" + self.session_id + ".html", 'w')
self.html_data = CreateHTML(path=self.reports_path, test_name=self.test_name, time_snap=str(datetime.datetime.now()), dut_ssid=self.ssid, test_conf_data={"Number of Clients":str(self.num_sta)},
objective=objective, test_results={"summary": summary_result, "detail":{"keys": self.keys, "data":self.test_result_data}}, chart_data=self.chart_data,
objective=objective, test_results={"summary": summary_result, "detail": {"keys": self.keys, "data": self.test_result_data}}, chart_data=self.chart_data,
chart_params={"chart_head": "Client Connection Time", "xlabel": "Clients", "ylabel": "Connection Time"})
self.html.write(self.html_data.report)
self.html.close()
self.test_update.send_update({"test_status": '6', "data": 'None', "data": [], "label": "Client Connectivity Time"})
def stop(self):
print("Stopping Test")
self.station_profile.admin_down()
LFUtils.wait_until_ports_admin_down(port_list=self.station_profile.station_names)
self.test_update.send_update({"test_status": '7', "data": 'None', "data": [], "label": "Client Connectivity Time"})
def postcleanup(self):
self.station_profile.cleanup()
@@ -228,11 +238,12 @@ class ConnectionTest(LFCliBase):
port_list=self.station_profile.station_names,
debug=self.debug)
print("Test Completed")
self.test_update.send_update({"test_status": '8', "data": 'None', "data": [], "label": "Client Connectivity Time"})
def main():
# This has --mgr, --mgr_port and --debug
parser = LFCliBase.create_bare_argparse(prog="layer3_test.py", formatter_class=argparse.RawTextHelpFormatter,
parser = LFCliBase.create_bare_argparse(prog="connection_test.py", formatter_class=argparse.RawTextHelpFormatter,
epilog="About This Script")
# Adding More Arguments for custom use
@@ -247,19 +258,20 @@ def main():
args = parser.parse_args()
# args.session_id = "local";
print(args)
update = RuntimeUpdates(args.session_id, {"test_status": '0', "data": 'None', "data": [], "label": "Client Connectivity Time"})
# Start Test
obj = ConnectionTest(lfclient_host="192.168.200.15", lfclient_port=args.mgr_port,
obj = ConnectionTest(lfclient_host="192.168.200.12", lfclient_port=args.mgr_port,
session_id=args.session_id, test_name=args.test_name,
dut_ssid=args.ssid, dut_passwd=args.passwd, dut_security=args.security,
num_sta=args.num_clients, radio=args.radio, pass_criteria=args.pass_criteria)
num_sta=args.num_clients, radio=args.radio, pass_criteria=args.pass_criteria, _test_update=update)
obj.precleanup()
obj.build()
obj.start()
obj.stop()
obj.postcleanup()
print(obj.chart_data)
update.send_update({"test_status": '10', "data": obj.chart_data, "label": ["Client Names","Client Connectivity Time (ms)"], "result": obj.pass_fail})
if __name__ == '__main__':
main()

161
py-scripts/layer3_test.py Normal file
View File

@@ -0,0 +1,161 @@
import sys
if 'py-json' not in sys.path:
sys.path.append('../py-json')
from LANforge import LFUtils
from LANforge import lfcli_base
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
import realm
from realm import Realm
import argparse
import datetime
import time
class Layer3Test(LFCliBase):
def __init__(self, lfclient_host="localhost", lfclient_port=8080, radio="wiphy1", sta_prefix="sta", start_id=0, num_sta=2,
dut_ssid="lexusdut", dut_security="open", dut_passwd="[BLANK]", upstream="eth1", name_prefix="L3Test",
traffic_type="lf_udp",side_a_speed="0M", side_b_speed="10M", session_id="Layer3Test", duration="1m",_debug_on=False, _exit_on_error=False, _exit_on_fail=False):
super().__init__(lfclient_host, lfclient_port, _debug=_debug_on, _halt_on_error=_exit_on_error, _exit_on_fail=_exit_on_fail)
print("Test is about to start")
self.host = lfclient_host
self.port = lfclient_port
self.radio = radio
self.upstream = upstream
self.monitor_interval = 1
self.sta_prefix = sta_prefix
self.sta_start_id = start_id
self.test_duration = duration
self.num_sta = num_sta
self.name_prefix = name_prefix
self.ssid = dut_ssid
self.security = dut_security
self.password = dut_passwd
self.session_id = session_id
self.traffic_type = traffic_type
self.side_a_speed = side_a_speed
self.side_b_speed = side_b_speed
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.station_profile = self.local_realm.new_station_profile()
self.cx_profile = self.local_realm.new_l3_cx_profile()
self.cx_profile.host = self.host
self.cx_profile.port = self.port
self.cx_profile.name_prefix = self.name_prefix
self.cx_profile.side_a_min_bps = self.local_realm.parse_speed(self.side_a_speed)
self.cx_profile.side_a_max_bps = self.local_realm.parse_speed(self.side_a_speed)
self.cx_profile.side_b_min_bps = self.local_realm.parse_speed(self.side_b_speed)
self.cx_profile.side_b_max_bps = self.local_realm.parse_speed(self.side_b_speed)
print("Test is Initialized")
def precleanup(self):
print("precleanup started")
self.station_list = LFUtils.portNameSeries(prefix_=self.sta_prefix, start_id_=self.sta_start_id, end_id_=self.num_sta - 1, padding_number_=10000, radio=self.radio)
self.cx_profile.cleanup_prefix()
for sta in self.station_list:
self.local_realm.rm_port(sta, check_exists=True)
time.sleep(1)
self.cx_profile.cleanup()
LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=self.station_profile.station_names,
debug=self.debug)
print("precleanup done")
pass
def build(self):
print("Building Test Configuration")
self.station_profile.use_security(self.security, self.ssid, self.password)
self.station_profile.set_number_template("00")
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.create(radio=self.radio, sta_names_=self.station_list, debug=self.debug)
self.local_realm.wait_until_ports_appear(sta_list=self.station_list)
self.cx_profile.create(endp_type=self.traffic_type, side_a=self.station_profile.station_names, side_b=self.upstream, sleep_time=0)
print("Test Build done")
pass
def start(self, print_pass=False, print_fail=False):
print("Test is starting")
self.cx_names =[]
self.station_profile.admin_up()
temp_stas = self.station_profile.station_names.copy()
temp_stas.append(self.upstream)
if (self.local_realm.wait_for_ip(temp_stas)):
self._pass("All stations got IPs", print_pass)
else:
self._fail("Stations failed to get IPs", print_fail)
exit(1)
self.cx_profile.start_cx()
try:
for i in self.cx_profile.get_cx_names():
self.cx_names.append(i)
while self.local_realm.json_get("/cx/" + i).get(i).get('state') != 'Run':
continue
except Exception as e:
pass
print("Test Started")
self.cur_time = datetime.datetime.now()
self.end_time = self.local_realm.parse_time(self.test_duration) + self.cur_time
print(self.end_time-self.cur_time)
self.start_monitor()
pass
def my_monitor(self):
print("Monitoring Test")
print(self.end_time - datetime.datetime.now())
if (datetime.datetime.now() > self.end_time):
self.stop_monitor()
for i in self.cx_names:
self.add_event(message= self.cx_profile.get_cx_report()[i]['bps rx b'], name=self.session_id)
return self.cx_profile.get_cx_report()
def stop(self):
print("Stopping Test")
self.cx_profile.stop_cx()
self.station_profile.admin_down()
pass
def postcleanup(self):
self.cx_profile.cleanup()
self.station_profile.cleanup()
LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=self.station_profile.station_names,
debug=self.debug)
print("Test Completed")
pass
def main():
# This has --mgr, --mgr_port and --debug
parser = LFCliBase.create_bare_argparse(prog="layer3_test.py", formatter_class=argparse.RawTextHelpFormatter, epilog="About This Script")
# Adding More Arguments for custom use
parser.add_argument('--ssid', help='--ssid of DUT', default="lexusdut")
parser.add_argument('--passwd', help='--passwd of dut', default="[BLANK]")
parser.add_argument('--radio', help='--radio to use on LANforge', default="wiphy1")
parser.add_argument('--security', help='--security of dut', default="open")
parser.add_argument('--test_duration', help='--test_duration sets the duration of the test', default="1m")
parser.add_argument('--session_id', help='--session_id is for websocket', default="local")
parser.add_argument('--num_client', type=int, help='--num_sta is number of stations you want to create', default=2)
parser.add_argument('--side_a_min_speed', help='--speed you want to monitor traffic with (max is 10G)', default="0M")
parser.add_argument('--side_b_min_speed', help='--speed you want to monitor traffic with (max is 10G)', default="10M")
parser.add_argument('--traffic_type', help='--traffic_type is used for traffic type (lf_udp, lf_tcp)', default="lf_udp")
args = parser.parse_args()
print(args)
# Start Test
obj = Layer3Test(lfclient_host=args.mgr, lfclient_port=args.mgr_port,
duration=args.test_duration, session_id=args.session_id,
traffic_type=args.traffic_type,
dut_ssid=args.ssid, dut_passwd=args.passwd, dut_security=args.security, num_sta=args.num_client, side_a_speed=args.side_a_min_speed, side_b_speed=args.side_b_min_speed, radio=args.radio)
obj.precleanup()
obj.build()
obj.start()
obj.monitor()
obj.stop()
obj.postcleanup()
if __name__ == '__main__':
main()

287
py-scripts/layer4_test.py Normal file
View File

@@ -0,0 +1,287 @@
"""
Candela Technologies Inc.
Info : Standard Script for Layer 4 Testing
Date :
Author : Shivam Thakur
"""
import sys
if 'py-json' not in sys.path:
sys.path.append('../py-json')
from LANforge import LFUtils
from LANforge import lfcli_base
from LANforge.lfcli_base import LFCliBase
from LANforge.LFUtils import *
import realm
from realm import PortUtils
import argparse
import datetime
import time
from test_utility import CreateHTML
from test_utility import RuntimeUpdates
import pdfkit
import json
import re
import os
webconsole_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.getcwd())))
print(webconsole_dir)
class HTTPTest(LFCliBase):
def __init__(self, lfclient_host="localhost", lfclient_port=8080, radio="wiphy1", sta_prefix="sta", start_id=0, num_sta=2,
dut_ssid="lexusdut", dut_security="open", dut_passwd="[BLANK]", upstream="eth1", name_prefix="L3Test", _test_update="",
url="", url_ps=600, session_id="Layer3Test", duration="1m",_debug_on=False, _exit_on_error=False, _exit_on_fail=False):
super().__init__(lfclient_host, lfclient_port, _debug=_debug_on, _halt_on_error=_exit_on_error, _exit_on_fail=_exit_on_fail)
print("Test is about to start")
self.host = lfclient_host
self.port = lfclient_port
self.radio = radio
self.upstream = upstream
self.monitor_interval = 1
self.sta_prefix = sta_prefix
self.sta_start_id = start_id
self.num_sta = num_sta
self.name_prefix = name_prefix
self.ssid = dut_ssid
self.security = dut_security
self.password = dut_passwd
self.session_id = session_id
self.url = url
self.urls_ps = url_ps
self.test_update =_test_update
self.test_update.send_update({"test_status": '1', "duration_left": "initializing...", "data": 'None'})
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port)
self.test_duration = self.local_realm.parse_time(duration)
self.station_profile = self.local_realm.new_station_profile()
self.port_util = PortUtils(self.local_realm)
self.cx_profile = self.local_realm.new_http_profile()
self.cx_profile.requests_per_ten = self.urls_ps
self.cx_profile.url = self.url
self.cx_profile.direction = "dl"
self.cx_profile.dest = "/dev/null"
print("Test Setup Initialized")
self.test_update.send_update({"test_status": '2', "duration_left": "Initialized...", "data": 'None'})
def precleanup(self):
print("precleanup started")
self.station_list = LFUtils.portNameSeries(prefix_=self.sta_prefix, start_id_=self.sta_start_id, end_id_=self.num_sta - 1, padding_number_=10000, radio=self.radio)
self.cx_profile.cleanup()
for sta in self.station_list:
self.local_realm.rm_port(sta, check_exists=True)
self.cx_profile.cleanup()
LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=self.station_profile.station_names,
debug=self.debug)
self.test_update.send_update({"test_status": '3', "duration_left": "building test...", "data": 'None'})
print("precleanup done")
pass
def build(self):
print("Building Test Configuration")
self.station_list = LFUtils.portNameSeries(prefix_=self.sta_prefix, start_id_=self.sta_start_id,
end_id_=self.num_sta - 1, padding_number_=10000, radio=self.radio)
#self.port_util.set_http(port_name=name, resource=resource, on=True)
self.station_profile.use_security(self.security, self.ssid, self.password)
self.station_profile.set_number_template("00")
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.create(radio=self.radio, sta_names_=self.station_list, debug=self.debug)
self.local_realm.wait_until_ports_appear(sta_list=self.station_list)
print("Test Build done")
self.test_update.send_update({"test_status": '4', "duration_left": "starting test...", "data": 'None'})
pass
def start(self, print_pass=False, print_fail=False):
print("Test is starting")
self.station_profile.admin_up()
temp_stas = self.station_profile.station_names.copy()
if (self.local_realm.wait_for_ip(temp_stas)):
self._pass("All stations got IPs", print_pass)
else:
self._fail("Stations failed to get IPs", print_fail)
exit(1)
self.cx_profile.create(ports=self.station_profile.station_names, sleep_time=.5, debug_=self.debug,
suppress_related_commands_=None, http=True)
time.sleep(5)
self.cx_profile.start_cx()
try:
for i in self.cx_profile.get_cx_names():
while self.local_realm.json_get("/cx/" + i).get(i).get('state') != 'Run':
continue
except Exception as e:
pass
print("Test Started")
self.test_update.send_update({"test_status": '5', "duration_left": "Test started...", "data": 'None'})
self.cur_time = datetime.datetime.now()
self.end_time = self.test_duration + self.cur_time
print(self.end_time-self.cur_time)
self.start_monitor()
pass
def my_monitor(self):
print("Monitoring Test")
print(self.end_time - datetime.datetime.now())
self.test_update.send_update({"test_status": '6', "duration_left": str(self.end_time - datetime.datetime.now()), "data": 'None'})
if (datetime.datetime.now() > self.end_time):
self.stop_monitor()
return self.cx_profile.get_cx_report()
def stop(self):
print("Stopping Test")
self.test_update.send_update({"test_status": '7', "duration_left": "stopping...", "data": "None"})
self.cx_profile.stop_cx()
for sta_name in self.station_list:
data = LFUtils.port_down_request(1, self.local_realm.name_to_eid(sta_name)[2])
url = "cli-json/set_port"
self.json_post(url, data)
def postcleanup(self):
self.test_update.send_update({"test_status": '8', "duration_left": "finishing...", "data": "None"})
self.cx_profile.cleanup()
for sta in self.station_list:
self.local_realm.rm_port(sta, check_exists=True)
time.sleep(1)
LFUtils.wait_until_ports_disappear(base_url=self.lfclient_url, port_list=self.station_list,
debug=self.debug)
print("Test Completed")
pass
def main():
# This has --mgr, --mgr_port and --debug
parser = LFCliBase.create_bare_argparse(prog="layer3_test.py", formatter_class=argparse.RawTextHelpFormatter, epilog="About This Script")
# Adding More Arguments for custom use
parser.add_argument('--ssid', help='--ssid of DUT', default="WebAP")
parser.add_argument('--passwd', help='--passwd of dut', default="[BLANK]")
parser.add_argument('--radio', help='--radio to use on LANforge', default="wiphy1")
parser.add_argument('--security', help='--security of dut', default="open")
parser.add_argument('--test_duration', help='--test_duration sets the duration of the test', default="1m")
parser.add_argument('--session_id', help='--session_id is for websocket', default="local")
parser.add_argument('--num_client', type=int, help='--num_sta is number of stations you want to create', default=2)
parser.add_argument('--url_ps', help='--speed you want to monitor traffic with (max is 10G)', default=600)
parser.add_argument('--url', help='--url on which you want to test HTTP', default="www.google.com")
parser.add_argument('--test_name', help='--test_name is for webconsole reports', default="HTTP Traffic Test")
args = parser.parse_args()
# print(args)
update = RuntimeUpdates(args.session_id, {"test_status": '0', "data": 'None'})
# # Start Test
http_obj = HTTPTest(lfclient_host="192.168.200.12", lfclient_port=args.mgr_port,
duration=args.test_duration, session_id=args.session_id,
dut_ssid=args.ssid, dut_passwd=args.passwd, dut_security=args.security, num_sta=args.num_client, url_ps=args.url_ps, radio=args.radio, _test_update=update)
http_obj.precleanup()
http_obj.build()
http_obj.start()
http_obj.monitor()
http_obj.stop()
http_obj.postcleanup()
http_obj.test_update.send_update({"test_status": '8', "duration_left": "generating report...", "data": "None"})
result_file = open("../py-run/"+args.session_id + ".txt", "r")
result_data = result_file.readlines()
final_data = result_data[len(result_data)-1]
test_detail_data = []
client_names = []
timesnap=[]
obj = re.sub('[\']', '"', final_data)
data = json.loads(obj)
for i in data:
for j in data[i]['endpoint']:
for k in j:
client_names.append(k)
summary_data = dict.fromkeys(client_names)
count = 0
for i in summary_data:
summary_data[i] = 0
for i in result_data:
obj = re.sub('[\']', '"', i)
data = json.loads(obj)
for j in data:
timesnap.append(j)
for k in range(0, len(client_names)):
if (data[j]['endpoint'][k][client_names[k]]['uc-avg'] == 0):
count += 1
else:
summary_data[client_names[k]] = summary_data[client_names[k]] + \
data[j]['endpoint'][k][client_names[k]]['uc-avg']
count = len(result_data) - count / len(client_names)
# print(summary_data)
for i in summary_data:
summary_data[i] = summary_data[i]/count
print("iron ee")
print(summary_data)
# Detailed Table Data
obj = re.sub('[\']', '"', final_data)
data = json.loads(obj)
for i in data:
for j in data[i]['endpoint']:
for k in j:
client_names.append(k)
if (summary_data[k] != 0) and (summary_data[k] < 3000):
temp = {"Client Name": k, "Avg. Response Time": str(int(summary_data[k]))+ " ms",
"RX Rate": str(int(j[k]['rx rate']) / 8000) + " kbps",
"Total URL's": j[k]['total-urls'], "Total Timeouts": j[k]['timeout'],
"Total Errors": j[k]['total-err'], "HTTP Range Error": j[k]['http-r'],
"HTTP Port Error": j[k]['http-p'], "result": "PASS"}
else:
temp = {"Client Name": k, "Avg. Response Time": str(int(summary_data[k])) + " ms",
"RX Rate": str(int(j[k]['rx rate']) / 8000) + " kbps",
"Total URL's": j[k]['total-urls'], "Total Timeouts": j[k]['timeout'],
"Total Errors": j[k]['total-err'], "HTTP Range Error": j[k]['http-r'],
"HTTP Port Error": j[k]['http-p'], "result": "FAIL"}
test_detail_data.append(temp)
reports_path = webconsole_dir + "/reports/" + args.test_name + "_" + args.session_id + '/'
if not os.path.exists(reports_path):
os.makedirs(reports_path)
html = open(reports_path + args.test_name + "_" + args.session_id + ".html", 'w')
objective = "The HTTP Traffic Test is designed to test the HTTP Performance of the Access Point. In this Test, Stations are sending HTTP Requests and it is monitoring the response time as a metric"
count =0
for i in summary_data:
if (summary_data[i] < 3000) and (summary_data[i] !=0):
pass
else:
count+=1
if (count == 0):
pass_fail = "PASS"
summary_result = "PASS, All clients are succesfully connected and Passed HTTP Test"
else:
pass_fail = "FAIL"
summary_result = "FAIL, Not All clients are succesfully connected and Passed HTTP Test"
html_data = CreateHTML(path=reports_path, test_name=args.test_name,
time_snap=str(datetime.datetime.now()), dut_ssid=args.ssid,
test_conf_data={"Number of Clients": str(args.num_client), "URL": args.url},
objective=objective, test_results={"summary": summary_result,
"detail": {"keys": ["Client Name", "Average Response Time", "RX Rate", "Total URL's", "Total Timeouts", "Total Errors", "HTTP Range Error", "HTTP Port Error","Result"],
"data": test_detail_data}},
chart_data=summary_data,
chart_params={"chart_head": "HTTP Response Time", "xlabel": "Time",
"ylabel": "Average Response Time"})
html.write(html_data.report)
html.close()
http_obj.test_update.send_update({"test_status": '10', "duration_left": "done", "data": "None", "result": pass_fail})
if __name__ == '__main__':
main()

2
py-scripts/test_all_scripts.sh
View File

@@ -20,7 +20,7 @@ testCommands=("./example_security_connection.py --num_stations $NUM_STA --ssid j
"./example_security_connection.py --num_stations $NUM_STA --ssid jedway-wep-48 --passwd jedway-wep-48 --radio $RADIO_USED --security wep"
"./example_security_connection.py --num_stations $NUM_STA --ssid jedway-wpa3-1 --passwd jedway-wpa3-1 --radio $RADIO_USED --security wpa3"
"./test_ipv4_connection.py --radio wiphy2 --num_stations $NUM_STA --ssid $SSID_USED --passwd $PASSWD_USED --security $SECURITY --debug --upstream_port eth1"
"./test_generic.py --mgr localhost --mgr_port 4122 --radio $RADIO_USED --ssid SSID_USED --passwd $PASSWD_USED --num_stations $NUM_STA --type lfping --dest 10.40.0.1 --security $SECURITY"
"./test_generic.py --mgr localhost --mgr_port 4122 --radio $RADIO_USED --ssid $SSID_USED --passwd $PASSWD_USED --num_stations $NUM_STA --type lfping --dest 10.40.0.1 --security $SECURITY"
"./test_generic.py --mgr localhost --mgr_port 4122 --radio $RADIO_USED --ssid $SSID_USED --passwd $PASSWD_USED --num_stations $NUM_STA --type speedtest --speedtest_min_up 20 --speedtest_min_dl 20 --speedtest_max_ping 150 --security $SECURITY"
"./test_ipv4_l4_urls_per_ten.py --upstream_port eth1 --radio $RADIO_USED --num_stations $NUM_STA --security $SECURITY --ssid $SSID_USED --passwd $PASSWD_USED --num_tests 1 --requests_per_ten 600 --target_per_ten 600"
"./test_ipv4_l4_wifi.py --upstream_port eth1 --radio wiphy0 --num_stations $NUM_STA --security $SECURITY --ssid jedway-wpa2-x2048-4-4 --passwd jedway-wpa2-x2048-4-4 --test_duration 3m"

3
py-scripts/test_generic.py
View File

@@ -219,8 +219,7 @@ python3 ./test_generic.py
IPERF3 (under construction):
./test_generic.py --mgr localhost --mgr_port 4122 --radio wiphy1 --num_stations 3 --ssid jedway-wpa2-x2048-4-1 --passwd jedway-wpa2-x2048-4-1 --security wpa2 --type iperf3
''')
required = parser.add_argument_group('required arguments')
required.add_argument('--security', help='WiFi Security protocol: < open | wep | wpa | wpa2 | wpa3 >', required=True)
parser.add_argument('--type', help='type of command to run: generic, lfping, iperf3-client, iperf3-server, lfcurl', default="lfping")
parser.add_argument('--cmd', help='specifies command to be run by generic type endp', default='')
parser.add_argument('--dest', help='destination IP for command', default="10.40.0.1")

3
py-scripts/test_ipv4_l4.py
View File

@@ -178,8 +178,7 @@ def main():
--debug (optional)
''')
required = parser.add_argument_group('required arguments')
required.add_argument('--security', help='WiFi Security protocol: < open | wep | wpa | wpa2 | wpa3 >', required=True)
parser.add_argument('--test_duration', help='--test_duration sets the duration of the test', default="5m")
parser.add_argument('--url', help='--url specifies upload/download, address, and dest', default="dl http://10.40.0.1 /dev/null")

3
py-scripts/test_status_msg.py
View File

@@ -236,7 +236,6 @@ class TestStatusMessage(LFCliBase):
def main():
lfjson_port = 8080
parser = LFCliBase.create_bare_argparse(
prog=__file__,
# formatter_class=argparse.RawDescriptionHelpFormatter,
@@ -268,7 +267,7 @@ Actions can be:
args = parser.parse_args()
status_messages = TestStatusMessage(args.mgr,
lfjson_port,
args.mgr_port,
_debug_on=False,
_exit_on_error=False,
_exit_on_fail=False)

4
py-scripts/test_utility.py
View File

@@ -2,9 +2,9 @@
"""
Candela Technologies Inc.
Info : Standard Script for Layer 4 Testing
Info : Standard Script for Webconsole Test Utility
Date :
Author : Shivam Thakur
"""