espressif/esp-idf
1
0
Fork 1
mirror of https://github.com/espressif/esp-idf.git synced 2025-08-02 20:24:32 +02:00
Code Issues Packages Projects Releases Wiki Activity

test: fix unit test script code style warnings

Browse Source
This commit is contained in:
He Yin Ling
2018-11-20 22:18:04 +08:00
parent 644571f78b
commit 5b3d09d5c8
1 changed files with 33 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
tools/unit-test-app/unit_test.py
View File

@@ -149,7 +149,8 @@ def reset_dut(dut):
# before reset finish, serial port is closed. therefore DUT could already bootup before serial port opened. # before reset finish, serial port is closed. therefore DUT could already bootup before serial port opened.
# this could cause checking bootup print failed. # this could cause checking bootup print failed.
# now use input cmd `-` and check test history to check if DUT is bootup. # now use input cmd `-` and check test history to check if DUT is bootup.
# we'll retry this step for a few times in case `dut.reset` returns during DUT bootup (when DUT can't process any command). # we'll retry this step for a few times,
# in case `dut.reset` returns during DUT bootup (when DUT can't process any command).
for _ in range(DUT_STARTUP_CHECK_RETRY_COUNT): for _ in range(DUT_STARTUP_CHECK_RETRY_COUNT):
dut.write("-") dut.write("-")
try: try:
@@ -158,7 +159,7 @@ def reset_dut(dut):
except ExpectTimeout: except ExpectTimeout:
pass pass
else: else:
raise AssertationError("Reset {} ({}) failed!".format(dut.name, dut.port)) raise AssertionError("Reset {} ({}) failed!".format(dut.name, dut.port))
def run_one_normal_case(dut, one_case, junit_test_case, failed_cases): def run_one_normal_case(dut, one_case, junit_test_case, failed_cases):
@@ -250,6 +251,7 @@ def run_unit_test_cases(env, extra_data):
3. as list of string or dict: 3. as list of string or dict:
[case1, case2, case3, {"name": "restart from PRO CPU", "reset": "SW_CPU_RESET"}, ...] [case1, case2, case3, {"name": "restart from PRO CPU", "reset": "SW_CPU_RESET"}, ...]
:param env: test env instance
:param extra_data: the case name or case list or case dictionary :param extra_data: the case name or case list or case dictionary
:return: None :return: None
""" """
@@ -287,7 +289,6 @@ def run_unit_test_cases(env, extra_data):
raise AssertionError("Unit Test Failed") raise AssertionError("Unit Test Failed")
class Handler(threading.Thread): class Handler(threading.Thread):
WAIT_SIGNAL_PATTERN = re.compile(r'Waiting for signal: \[(.+)\]!') WAIT_SIGNAL_PATTERN = re.compile(r'Waiting for signal: \[(.+)\]!')
@@ -356,7 +357,6 @@ class Handler(threading.Thread):
Utility.console_log("Ignored: " + self.child_case_name, color="orange") Utility.console_log("Ignored: " + self.child_case_name, color="orange")
one_device_case_finish(not int(data[0])) one_device_case_finish(not int(data[0]))
try: try:
time.sleep(1) time.sleep(1)
self.dut.write("\"{}\"".format(self.parent_case_name)) self.dut.write("\"{}\"".format(self.parent_case_name))
@@ -365,7 +365,8 @@ class Handler(threading.Thread):
Utility.console_log("No case detected!", color="orange") Utility.console_log("No case detected!", color="orange")
while not self.finish and not self.force_stop.isSet(): while not self.finish and not self.force_stop.isSet():
try: try:
self.dut.expect_any((re.compile('\(' + str(self.child_case_index) + '\)\s"(\w+)"'), get_child_case_name), self.dut.expect_any((re.compile('\(' + str(self.child_case_index) + '\)\s"(\w+)"'),
get_child_case_name),
(self.WAIT_SIGNAL_PATTERN, device_wait_action), # wait signal pattern (self.WAIT_SIGNAL_PATTERN, device_wait_action), # wait signal pattern
(self.SEND_SIGNAL_PATTERN, device_send_action), # send signal pattern (self.SEND_SIGNAL_PATTERN, device_send_action), # send signal pattern
(self.FINISH_PATTERN, handle_device_test_finish), # test finish pattern (self.FINISH_PATTERN, handle_device_test_finish), # test finish pattern
@@ -396,7 +397,7 @@ def get_dut(duts, env, name, ut_config, app_bin=None):
return dut return dut
def run_one_multiple_devices_case(duts, ut_config, env, one_case, failed_cases, junit_test_case): def run_one_multiple_devices_case(duts, ut_config, env, one_case, failed_cases, app_bin, junit_test_case):
lock = threading.RLock() lock = threading.RLock()
threads = [] threads = []
send_signal_list = [] send_signal_list = []
@@ -456,7 +457,8 @@ def run_multiple_devices_cases(env, extra_data):
for one_case in case_config[ut_config]: for one_case in case_config[ut_config]:
junit_test_case = TinyFW.JunitReport.create_test_case("[{}] {}".format(ut_config, one_case["name"])) junit_test_case = TinyFW.JunitReport.create_test_case("[{}] {}".format(ut_config, one_case["name"]))
try: try:
run_one_multiple_devices_case(duts, ut_config, env, one_case, failed_cases, junit_test_case) run_one_multiple_devices_case(duts, ut_config, env, one_case, failed_cases,
one_case.get('app_bin'), junit_test_case)
TinyFW.JunitReport.test_case_finish(junit_test_case) TinyFW.JunitReport.test_case_finish(junit_test_case)
except Exception as e: except Exception as e:
junit_test_case.add_error_info("Unexpected exception: " + str(e)) junit_test_case.add_error_info("Unexpected exception: " + str(e))
@@ -579,6 +581,7 @@ def run_multiple_stage_cases(env, extra_data):
3. as list of string or dict: 3. as list of string or dict:
[case1, case2, case3, {"name": "restart from PRO CPU", "child case num": 2}, ...] [case1, case2, case3, {"name": "restart from PRO CPU", "child case num": 2}, ...]
:param env: test env instance
:param extra_data: the case name or case list or case dictionary :param extra_data: the case name or case list or case dictionary
:return: None :return: None
""" """
@@ -629,14 +632,14 @@ def detect_update_unit_test_info(env, extra_data, app_bin):
dut.write("") dut.write("")
dut.expect("Here's the test menu, pick your combo:", timeout=DEFAULT_TIMEOUT) dut.expect("Here's the test menu, pick your combo:", timeout=DEFAULT_TIMEOUT)
def find_update_dic(name, t, timeout, child_case_num=None): def find_update_dic(name, _t, _timeout, child_case_num=None):
for dic in extra_data: for _case_data in extra_data:
if dic['name'] == name: if _case_data['name'] == name:
dic['type'] = t _case_data['type'] = _t
if 'timeout' not in dic: if 'timeout' not in _case_data:
dic['timeout'] = timeout _case_data['timeout'] = _timeout
if child_case_num: if child_case_num:
dic['child case num'] = child_case_num _case_data['child case num'] = child_case_num
try: try:
while True: while True:
@@ -666,9 +669,9 @@ def detect_update_unit_test_info(env, extra_data, app_bin):
if data[1] and re.search(END_LIST_STR, data[1]): if data[1] and re.search(END_LIST_STR, data[1]):
break break
# check if the unit test case names are correct, i.e. they could be found in the device # check if the unit test case names are correct, i.e. they could be found in the device
for dic in extra_data: for _dic in extra_data:
if 'type' not in dic: if 'type' not in _dic:
raise ValueError("Unit test \"{}\" doesn't exist in the flashed device!".format(dic.get('name'))) raise ValueError("Unit test \"{}\" doesn't exist in the flashed device!".format(_dic.get('name')))
except ExpectTimeout: except ExpectTimeout:
Utility.console_log("Timeout during getting the test list", color="red") Utility.console_log("Timeout during getting the test list", color="red")
finally: finally:
@@ -727,12 +730,12 @@ if __name__ == '__main__':
env_config['app'] = UT env_config['app'] = UT
env_config['dut'] = IDF.IDFDUT env_config['dut'] = IDF.IDFDUT
env_config['test_suite_name'] = 'unit_test_parsing' env_config['test_suite_name'] = 'unit_test_parsing'
env = Env.Env(**env_config) test_env = Env.Env(**env_config)
detect_update_unit_test_info(env, extra_data=list_of_dicts, app_bin=args.app_bin) detect_update_unit_test_info(test_env, extra_data=list_of_dicts, app_bin=args.app_bin)
for i in range(1, args.repeat+1): for index in range(1, args.repeat+1):
if args.repeat > 1: if args.repeat > 1:
Utility.console_log("Repetition {}".format(i), color="green") Utility.console_log("Repetition {}".format(index), color="green")
for dic in list_of_dicts: for dic in list_of_dicts:
t = dic.get('type', SIMPLE_TEST_ID) t = dic.get('type', SIMPLE_TEST_ID)
if t == SIMPLE_TEST_ID: if t == SIMPLE_TEST_ID: