esp-idf/components/idf_test/integration_test/TestCaseScript/WiFiStress/WifiJAP.py

220 lines
9.4 KiB
Python
Raw Normal View History

2016-12-06 02:41:02 -05:00
import os
import random
import time
import WifiConnUtility
from NativeLog import NativeLog
from TCAction import TCActionBase
from Utility import Encoding
from Utility import MakeFolder
STEPS = {"SCAN1": 0x01, "JAP": 0x02, "SCAN2": 0x04, "RECONNECT": 0x08}
AP_PROP = ("ssid", "ssid_len", "pwd",
"pwd_len", "channel", "enc", "apc")
JAP_TEST_METHOD = ("Normal", "OFF_ON", "OFF", "WRONG_PROP")
RECONNECT_TEST_METHOD = ("OFF_ON", "OFF")
LOG_FOLDER = os.path.join("Performance", "JAP")
SSID_LEN_RANGE = (1, 32) # in bytes
ENC_TYPE = (0, 2, 3, 4) # do not support WEP for 8266 soft AP
PWD_RANGE = {0: [0, 0],
1: [5, 5],
2: [8, 63],
3: [8, 63],
4: [8, 63],
}
class TestCase(TCActionBase.CommonTCActionBase):
def __init__(self, test_case, test_env, timeout=30, log_path=TCActionBase.LOG_PATH):
TCActionBase.CommonTCActionBase.__init__(self, test_case, test_env, timeout=timeout, log_path=log_path)
# default value for optional configurable params
self.performance_folder_path = log_path
self.pwd_len = [8, 64]
self.step_config = [0x03, 0x01, 0x02, 0x0B, 0x0F]
self.join_test_method = ["Normal"]
self.join_delay = [[1.5, 5], [1.5, 5]]
self.reconnect_test_method = ["OFF_ON"]
self.reconnect_delay = [[1.5, 5], [1.5, 6]]
# load param from excel
cmd_set = test_case["cmd set"]
for i in range(1, len(cmd_set)):
if cmd_set[i][0] != "dummy" and cmd_set[i][0] != "":
cmd_string = "self." + cmd_set[i][0]
exec cmd_string
# read AP list
self.ap_list = []
for i in range(1, len(cmd_set)):
for j in range(len(cmd_set[i][1])):
if cmd_set[i][1][j] != "":
cmd_string = "self.ap_list.append(dict(zip(AP_PROP, " + cmd_set[i][1][j] + ")))"
exec cmd_string
folder_path = MakeFolder.make_folder(self.performance_folder_path + "\\" + LOG_FOLDER)
file_name = "JAP_log_%s.log" % (time.strftime("%m%d%H%M%S", time.localtime()))
self._performance_log_file = os.path.join(folder_path, file_name)
# test statistics
self._succeed_count = self._fail_count = self._time_cost_count = 0
self._total_time = self._longest_time = 0
self.result_cntx = TCActionBase.ResultCheckContext(self, test_env, self.tc_name)
# get target type "SSC" or "AT"
self.target_type = ["SSC" if test_env.get_port_by_name("AT1") is None else "AT"]
self.target_type.append("SSC" if test_env.get_port_by_name("AT2") is None else "AT")
self._utility = WifiConnUtility.WifiConnUtility(self)
pass
def _generate_random_ap_prop(self):
ap_prop = dict.fromkeys(AP_PROP)
# generate target ap_value
ap_prop["ssid_len"] = random.randint(SSID_LEN_RANGE[0], SSID_LEN_RANGE[1])
ap_prop["channel"] = random.choice(range(1, 14))
ap_prop["enc"] = random.choice(ENC_TYPE)
ap_prop["pwd_len"] = random.randint(PWD_RANGE[ap_prop["enc"]][0], PWD_RANGE[ap_prop["enc"]][1])
# generate string
if self.target_type[0] == self.target_type[1] == "AT":
ap_prop["ssid"] = Encoding.generate_random_utf8_str(ap_prop["ssid_len"])
ap_prop["pwd"] = Encoding.generate_random_utf8_str(ap_prop["pwd_len"])
# NativeLog.add_trace_info("ssid hex is : %x" % ap_prop["ssid"])
# NativeLog.add_trace_info("pwd hex is : %x" % ap_prop["pwd"])
else:
ap_prop["ssid"] = Encoding.generate_random_printable_str(ap_prop["ssid_len"])
ap_prop["pwd"] = Encoding.generate_random_printable_str(ap_prop["pwd_len"])
return ap_prop
def _logging_performance(self, ssid, join_method="Normal", time_cost=0):
# log performance to performance log file
with open(self._performance_log_file, "ab+") as f:
# log time and ssid
f.write("\r\n[%s]:\r\n[AP name] %s\r\n" %
(time.strftime("%m-%d %H:%M:%S", time.localtime()), ssid))
if join_method == "Normal" or join_method == "OFF_ON":
if time_cost is not False:
self._succeed_count += 1
if join_method == "Normal":
f.write("[Succeed][%f]\r\n" % time_cost)
self._longest_time = (time_cost > self._longest_time and
[time_cost] or [self._longest_time])[0]
self._time_cost_count += 1
self._total_time += time_cost
else:
f.write("[Succeed][%s]\r\n" % join_method)
else:
self._fail_count += 1
f.write("[Fail][%s]\r\n" % join_method)
pass
def _logging_fail_step(self, ssid, step):
with open(self._performance_log_file, "ab+") as f:
f.write("\r\n[%s]:\r\n[AP name] %s\r\n" %
(time.strftime("%m-%d %H:%M:%S", time.localtime()), ssid))
f.write("[Fail][%s]\r\n" % step)
pass
def _generate_performance_report(self):
with open(self._performance_log_file, "ab+") as f:
f.write("[Test report] Succeed: %d\r\n" % self._succeed_count)
f.write("[Test report] Failed: %d\r\n" % self._fail_count)
if self._succeed_count > 0 or self._fail_count > 0:
f.write("[Test report] Pass Rate: %f\r\n" %
(self._succeed_count/(self._fail_count+self._succeed_count)))
if self._time_cost_count > 0:
f.write("[Test report] Average time: %f\r\n" % (self._total_time/self._time_cost_count))
f.write("[Test report] Longest time: %f\r\n" % self._longest_time)
def execute(self):
TCActionBase.TCActionBase.execute(self)
self.result_cntx.start()
# mandatory configurable params
try:
target_ap_num = self.target_ap_num
test_count = self.test_count
except StandardError, e:
NativeLog.add_trace_critical("Error configuration for WifiJAP script, error is %s" % e)
raise StandardError("Error configuration")
# prepare ap list
_ap_list = [["target", None]] * target_ap_num
for _ap_prop in self.ap_list:
_ap_list.append(["AP", _ap_prop])
# set to correct mode first
self._utility.set_mode([1, 2])
for i in xrange(test_count):
_ap = random.choice(_ap_list)
# arrange ap
_ap_type = _ap[0]
_ap_prop = _ap[1]
if _ap_type == "target":
_ap_prop = self._generate_random_ap_prop()
pass
# step 1 : mandatory step, set up AP
if self._utility.setup_ap(_ap_type, _ap_prop) is False:
self._logging_fail_step(_ap_prop["ssid"], "Set AP")
NativeLog.add_prompt_trace("[Step1] setup AP Fail")
continue
step_config = random.choice(self.step_config)
NativeLog.add_prompt_trace("[Step1] setup AP succeed")
# step 2 : optional step, do scan before connect
if step_config & STEPS["SCAN1"] != 0: # check option
if self._utility.do_scan(_ap_prop) is False:
self._logging_fail_step(_ap_prop["ssid"], "Scan before JAP")
NativeLog.add_prompt_trace("[Step2] Scan Done")
# step 3 : mandatory step, join AP
if step_config & STEPS["JAP"] != 0: # check option
_join_test_method = random.choice(self.join_test_method)
time_cost = self._utility.join_ap(_join_test_method, _ap_type, _ap_prop, self.join_delay)
# log performance to performance log file
self._logging_performance(_ap_prop["ssid"], _join_test_method, time_cost)
if time_cost is False:
# do scan once to check if AP exist
self._utility.do_scan(_ap_prop)
continue
NativeLog.add_prompt_trace("[Step3] Join AP done")
# step 4 : optional step, scan after join AP
if step_config & STEPS["SCAN2"] != 0: # check option
if self._utility.do_scan(_ap_prop) is False:
self._logging_fail_step(_ap_prop["ssid"], "Scan after JAP")
NativeLog.add_prompt_trace("[Step4] Scan done")
# step 5 : optional step, reconnect test
if step_config & STEPS["RECONNECT"] != 0: # check option
_reconnect_test_method = random.choice(self.reconnect_test_method)
if self._utility.do_reconnect(_reconnect_test_method,
_ap_type, _ap_prop, self.reconnect_delay) is False:
self._logging_fail_step(_ap_prop["ssid"], "Reconnect")
NativeLog.add_prompt_trace("[Step5] Reconnect done")
# continue to next loop
NativeLog.add_prompt_trace("[WifiJAP] Test count %d done" % i)
# generate report and cleanup
self._generate_performance_report()
self.result_cntx.set_result("Succeed")
def result_check(self, port_name, data):
TCActionBase.CommonTCActionBase.result_check(self, port_name, data)
self.result_cntx.append_data(port_name, data)
def main():
pass
if __name__ == '__main__':
main()