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

import random
import time
import string
import threading

from TCAction import TCActionBase
from NativeLog import NativeLog
from TCAction import PerformanceTCBase
from Utility import Encoding


class STAJAPThread(threading.Thread):
    def __init__(self, test_action, port_name, ssid, password, delay1, delay2, change_mac):
        threading.Thread.__init__(self)
        self.setDaemon(True)
        self.test_action = test_action
        self.port_name = port_name
        self.ssid = ssid
        self.password = password
        self.delay1 = delay1
        self.delay2 = delay2
        self.change_mac = change_mac
        self.exit_flag = threading.Event()
        pass

    def exit(self):
        self.exit_flag.set()
        pass

    def run(self):
        total_test_count = 0
        fail_count = 0
        while self.exit_flag.isSet() is False:
            # change mac
            if self.change_mac is True:
                mac = Encoding.generate_random_mac()
                self.test_action.serial_write_line(self.port_name, "mac -S -o 1 -m %s" % mac)
                self.test_action.check_response(self.port_name, "+MAC:STA,OK")

            time.sleep(1)

            # JAP
            total_test_count += 1
            # flush current port data
            self.test_action.flush_data(self.port_name)
            self.test_action.serial_write_line(self.port_name, "sta -C -s %s -p %s" % (self.ssid, self.password))
            if self.test_action.check_response(self.port_name, "+JAP:CONNECTED", 45) is False:
                fail_count += 1
                NativeLog.add_trace_critical("[%s] Failed to JAP, Failed/Total : %d/%d"
                                             % (self.port_name, fail_count, total_test_count))
                continue
            time.sleep(random.randint(self.delay1[0], self.delay1[1]))

            # QAP
            self.test_action.serial_write_line(self.port_name, "sta -D")
            if self.test_action.check_response(self.port_name, "+QAP:OK", 5) is False:
                NativeLog.add_trace_critical("[%s] Failed to QAP" % self.port_name)
            time.sleep(random.randint(self.delay2[0], self.delay2[1]))

        # make sure quit AP
        self.test_action.serial_write_line(self.port_name, "sta -D")
        pass
    pass


class TestCase(PerformanceTCBase.PerformanceTCBase):

    def __init__(self, test_case, test_env, timeout=30, log_path=TCActionBase.LOG_PATH):
        PerformanceTCBase.PerformanceTCBase.__init__(self, test_case, test_env, timeout=timeout, log_path=log_path)
        self.sta_num = 0
        self.max_sta = 4
        self.test_time = 60
        self.delay1 = [5, 30]
        self.delay2 = [5, 10]
        self.change_mac = True
        self.channel = 11
        # 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
        pass

    def process(self):
        # configurable parameters
        try:
            sta_num = self.sta_num
            max_sta = self.max_sta
            test_time = self.test_time
            # delay between JAP succeed and QAP
            delay1 = self.delay1
            # delay between QAP and JAP
            delay2 = self.delay2
            # if change mac each time before JAP
            change_mac = self.change_mac
            # channel
            channel = self.channel
        except StandardError, e:
            raise StandardError("miss mandatory parameters")

        # step 0, set op mode and enable dhcp
        self.serial_write_line("SSC1", "op -S -o 2")
        if self.check_response("SSC1", "+MODE:OK", 2) is False:
            NativeLog.add_trace_critical("Failed to set ap mode")
            return
        self.serial_write_line("SSC1", "dhcp -E -o 2")
        if self.check_response("SSC1", "+DHCP:AP,OK", 2) is False:
            NativeLog.add_trace_critical("Failed to enable ap dhcp")
            return
        self.serial_write_line("SSC1", "dhcp -L -s 192.168.4.2 -e 192.168.4.100 -t 1")
        if self.check_response("SSC1", "+DHCP:LEASE,OK", 2) is False:
            NativeLog.add_trace_critical("Failed to enable ap dhcp")
            return
        self.serial_write_line("SSC1", "dhcp -S -o 2")
        if self.check_response("SSC1", "+DHCP:AP,OK", 2) is False:
            NativeLog.add_trace_critical("Failed to enable ap dhcp")
            return

        for i in range(sta_num):
            self.serial_write_line("SSC%d" % (i+2), "op -S -o 1")
            if self.check_response("SSC%d" % (i+2), "+MODE:OK", 2) is False:
                NativeLog.add_trace_critical("Failed to set sta mode")
                return
            self.serial_write_line("SSC%d" % (i+2), "dhcp -S -o 1")
            if self.check_response("SSC%d" % (i+2), "+DHCP:STA,OK", 2) is False:
                NativeLog.add_trace_critical("Failed to enable sta dhcp")

        # step 1, set ap config and load
        ap_ssid = "".join([random.choice(string.uppercase) for m in range(15)])
        ap_password = "".join([random.choice(string.lowercase) for m in range(15)])

        self.serial_write_line("SSC1", "ap -S -s %s -p %s -t 3 -m %s -n %s"
                               % (ap_ssid, ap_password, max_sta, channel))
        if self.check_response("SSC1", "+SAP:OK", 2) is False:
            NativeLog.add_trace_critical("Failed to set AP")
            return

        # step 2, start thread to let STA JAP
        sta_thread_list = []
        for i in range(sta_num):
            sta_thread_list.append(STAJAPThread(self, "SSC%d" % (i+2),
                                                ap_ssid, ap_password, delay1, delay2, change_mac))
        for sta_thread in sta_thread_list:
            sta_thread.start()

        # step 3, sleep for test time
        for i in range(test_time):
            self.flush_data("SSC1")
            time.sleep(60)

        # step 4, close all thread, will disconnect when exit thread
        for sta_thread in sta_thread_list:
            sta_thread.exit()
        for sta_thread in sta_thread_list:
            sta_thread.join()
        # wait and make sure disconnect done
        time.sleep(1)

        # step 5, join AP and check
        sta_num_temp = max_sta if sta_num > max_sta else sta_num

        for i in range(sta_num_temp):
            self.serial_write_line("SSC%d" % (i+2), "sta -C -s %s -p %s" % (ap_ssid, ap_password))
            if self.check_response("SSC%d" % (i+2), "+JAP:CONNECTED", 45) is False:
                self.set_result("Fail")
                break
            pass
        else:
            self.set_result("Succeed")


def main():
    pass

if __name__ == '__main__':
    main()