import sys
import time
import argparse
-
import threading
# if we want to run test case outside `tiny-test-fw` folder,
return case_config
+
def replace_app_bin(dut, name, new_app_bin):
if new_app_bin is None:
return
Utility.console_log("The replaced application binary is {}".format(new_app_bin), "O")
break
+
def reset_dut(dut):
dut.reset()
# esptool ``run`` cmd takes quite long time.
raise AssertationError("Reset {} ({}) failed!".format(dut.name, dut.port))
-@IDF.idf_unit_test(env_tag="UT_T1_1")
+def run_one_normal_case(dut, one_case, junit_test_case, failed_cases):
+
+ reset_dut(dut)
+
+ dut.start_capture_raw_data()
+ # run test case
+ dut.write("\"{}\"".format(one_case["name"]))
+ dut.expect("Running " + one_case["name"] + "...")
+
+ exception_reset_list = []
+
+ # we want to set this flag in callbacks (inner functions)
+ # use list here so we can use append to set this flag
+ test_finish = list()
+
+ # expect callbacks
+ def one_case_finish(result):
+ """ one test finished, let expect loop break and log result """
+ test_finish.append(True)
+ output = dut.stop_capture_raw_data()
+ if result:
+ Utility.console_log("Success: " + one_case["name"], color="green")
+ else:
+ failed_cases.append(one_case["name"])
+ Utility.console_log("Failed: " + one_case["name"], color="red")
+ junit_test_case.add_failure_info(output)
+
+ def handle_exception_reset(data):
+ """
+ just append data to exception list.
+ exception list will be checked in ``handle_reset_finish``, once reset finished.
+ """
+ exception_reset_list.append(data[0])
+
+ def handle_test_finish(data):
+ """ test finished without reset """
+ # in this scenario reset should not happen
+ assert not exception_reset_list
+ if int(data[1]):
+ # case ignored
+ Utility.console_log("Ignored: " + one_case["name"], color="orange")
+ junit_test_case.add_skipped_info("ignored")
+ one_case_finish(not int(data[0]))
+
+ def handle_reset_finish(data):
+ """ reset happened and reboot finished """
+ assert exception_reset_list # reboot but no exception/reset logged. should never happen
+ result = False
+ if len(one_case["reset"]) == len(exception_reset_list):
+ for i, exception in enumerate(exception_reset_list):
+ if one_case["reset"][i] not in exception:
+ break
+ else:
+ result = True
+ if not result:
+ err_msg = "Reset Check Failed: \r\n\tExpected: {}\r\n\tGet: {}".format(one_case["reset"],
+ exception_reset_list)
+ Utility.console_log(err_msg, color="orange")
+ junit_test_case.add_error_info(err_msg)
+ one_case_finish(result)
+
+ while not test_finish:
+ try:
+ dut.expect_any((RESET_PATTERN, handle_exception_reset),
+ (EXCEPTION_PATTERN, handle_exception_reset),
+ (ABORT_PATTERN, handle_exception_reset),
+ (FINISH_PATTERN, handle_test_finish),
+ (UT_APP_BOOT_UP_DONE, handle_reset_finish),
+ timeout=one_case["timeout"])
+ except ExpectTimeout:
+ Utility.console_log("Timeout in expect", color="orange")
+ junit_test_case.add_error_info("timeout")
+ one_case_finish(False)
+ break
+
+
+@IDF.idf_unit_test(env_tag="UT_T1_1", junit_report_by_case=True)
def run_unit_test_cases(env, extra_data):
"""
extra_data can be three types of value
if len(case_config[ut_config]) > 0:
replace_app_bin(dut, "unit-test-app", case_config[ut_config][0].get('app_bin'))
dut.start_app()
+ Utility.console_log("Download finished, start running test cases", "O")
for one_case in case_config[ut_config]:
- reset_dut(dut)
-
- # run test case
- dut.write("\"{}\"".format(one_case["name"]))
- dut.expect("Running " + one_case["name"] + "...")
-
- exception_reset_list = []
-
- # we want to set this flag in callbacks (inner functions)
- # use list here so we can use append to set this flag
- test_finish = list()
-
- # expect callbacks
- def one_case_finish(result):
- """ one test finished, let expect loop break and log result """
- test_finish.append(True)
- if result:
- Utility.console_log("Success: " + one_case["name"], color="green")
- else:
- failed_cases.append(one_case["name"])
- Utility.console_log("Failed: " + one_case["name"], color="red")
-
- def handle_exception_reset(data):
- """
- just append data to exception list.
- exception list will be checked in ``handle_reset_finish``, once reset finished.
- """
- exception_reset_list.append(data[0])
-
- def handle_test_finish(data):
- """ test finished without reset """
- # in this scenario reset should not happen
- assert not exception_reset_list
- if int(data[1]):
- # case ignored
- Utility.console_log("Ignored: " + one_case["name"], color="orange")
- one_case_finish(not int(data[0]))
-
- def handle_reset_finish(data):
- """ reset happened and reboot finished """
- assert exception_reset_list # reboot but no exception/reset logged. should never happen
- result = False
- if len(one_case["reset"]) == len(exception_reset_list):
- for i, exception in enumerate(exception_reset_list):
- if one_case["reset"][i] not in exception:
- break
- else:
- result = True
- if not result:
- Utility.console_log("""Reset Check Failed: \r\n\tExpected: {}\r\n\tGet: {}"""
- .format(one_case["reset"], exception_reset_list),
- color="orange")
- one_case_finish(result)
-
- while not test_finish:
- try:
- dut.expect_any((RESET_PATTERN, handle_exception_reset),
- (EXCEPTION_PATTERN, handle_exception_reset),
- (ABORT_PATTERN, handle_exception_reset),
- (FINISH_PATTERN, handle_test_finish),
- (UT_APP_BOOT_UP_DONE, handle_reset_finish),
- timeout=one_case["timeout"])
- except ExpectTimeout:
- Utility.console_log("Timeout in expect", color="orange")
- one_case_finish(False)
- break
+ # create junit report test case
+ junit_test_case = TinyFW.JunitReport.create_test_case("[{}] {}".format(ut_config, one_case["name"]))
+ try:
+ run_one_normal_case(dut, one_case, junit_test_case, failed_cases)
+ TinyFW.JunitReport.test_case_finish(junit_test_case)
+ except Exception as e:
+ junit_test_case.add_error_info("Unexpected exception: " + str(e))
+ TinyFW.JunitReport.test_case_finish(junit_test_case)
# raise exception if any case fails
if failed_cases:
self.child_case_index = child_case_index + 1
self.finish = False
self.result = False
+ self.output = ""
self.fail_name = None
self.timeout = timeout
self.force_stop = threading.Event() # it show the running status
threading.Thread.__init__(self, name="{} Handler".format(dut))
def run(self):
+
+ self.dut.start_capture_raw_data()
+
def get_child_case_name(data):
self.child_case_name = data[0]
time.sleep(1)
""" one test finished, let expect loop break and log result """
self.finish = True
self.result = result
+ self.output = "[{}]\n\n{}\n".format(self.child_case_name,
+ self.dut.stop_capture_raw_data())
if not result:
self.fail_name = self.child_case_name
return dut
-def case_run(duts, ut_config, env, one_case, failed_cases, app_bin):
+def run_one_multiple_devices_case(duts, ut_config, env, one_case, failed_cases, junit_test_case):
lock = threading.RLock()
threads = []
send_signal_list = []
for thread in threads:
thread.setDaemon(True)
thread.start()
+ output = "Multiple Device Failed\n"
for thread in threads:
thread.join()
result = result and thread.result
+ output += thread.output
if not thread.result:
[thd.stop() for thd in threads]
Utility.console_log("Success: " + one_case["name"], color="green")
else:
failed_cases.append(one_case["name"])
+ junit_test_case.add_failure_info(output)
Utility.console_log("Failed: " + one_case["name"], color="red")
-@IDF.idf_unit_test(env_tag="UT_T2_1")
+@IDF.idf_unit_test(env_tag="UT_T2_1", junit_report_by_case=True)
def run_multiple_devices_cases(env, extra_data):
"""
extra_data can be two types of value
"""
failed_cases = []
case_config = format_test_case_config(extra_data)
- DUTS = {}
+ duts = {}
for ut_config in case_config:
Utility.console_log("Running unit test for config: " + ut_config, "O")
for one_case in case_config[ut_config]:
- case_run(DUTS, ut_config, env, one_case, failed_cases, one_case.get('app_bin'))
+ junit_test_case = TinyFW.JunitReport.create_test_case("[{}] {}".format(ut_config, one_case["name"]))
+ try:
+ run_one_multiple_devices_case(duts, ut_config, env, one_case, failed_cases, junit_test_case)
+ TinyFW.JunitReport.test_case_finish(junit_test_case)
+ except Exception as e:
+ junit_test_case.add_error_info("Unexpected exception: " + str(e))
+ TinyFW.JunitReport.test_case_finish(junit_test_case)
if failed_cases:
Utility.console_log("Failed Cases:", color="red")
raise AssertionError("Unit Test Failed")
-@IDF.idf_unit_test(env_tag="UT_T1_1")
+def run_one_multiple_stage_case(dut, one_case, failed_cases, junit_test_case):
+ reset_dut(dut)
+
+ dut.start_capture_raw_data()
+
+ exception_reset_list = []
+
+ for test_stage in range(one_case["child case num"]):
+ # select multi stage test case name
+ dut.write("\"{}\"".format(one_case["name"]))
+ dut.expect("Running " + one_case["name"] + "...")
+ # select test function for current stage
+ dut.write(str(test_stage + 1))
+
+ # we want to set this flag in callbacks (inner functions)
+ # use list here so we can use append to set this flag
+ stage_finish = list()
+
+ def last_stage():
+ return test_stage == one_case["child case num"] - 1
+
+ def check_reset():
+ if one_case["reset"]:
+ assert exception_reset_list # reboot but no exception/reset logged. should never happen
+ result = False
+ if len(one_case["reset"]) == len(exception_reset_list):
+ for i, exception in enumerate(exception_reset_list):
+ if one_case["reset"][i] not in exception:
+ break
+ else:
+ result = True
+ if not result:
+ err_msg = "Reset Check Failed: \r\n\tExpected: {}\r\n\tGet: {}".format(one_case["reset"],
+ exception_reset_list)
+ Utility.console_log(err_msg, color="orange")
+ junit_test_case.add_error_info(err_msg)
+ else:
+ # we allow omit reset in multi stage cases
+ result = True
+ return result
+
+ # expect callbacks
+ def one_case_finish(result):
+ """ one test finished, let expect loop break and log result """
+ # handle test finish
+ result = result and check_reset()
+ output = dut.stop_capture_raw_data()
+ if result:
+ Utility.console_log("Success: " + one_case["name"], color="green")
+ else:
+ failed_cases.append(one_case["name"])
+ Utility.console_log("Failed: " + one_case["name"], color="red")
+ junit_test_case.add_failure_info(output)
+ stage_finish.append("break")
+
+ def handle_exception_reset(data):
+ """
+ just append data to exception list.
+ exception list will be checked in ``handle_reset_finish``, once reset finished.
+ """
+ exception_reset_list.append(data[0])
+
+ def handle_test_finish(data):
+ """ test finished without reset """
+ # in this scenario reset should not happen
+ if int(data[1]):
+ # case ignored
+ Utility.console_log("Ignored: " + one_case["name"], color="orange")
+ junit_test_case.add_skipped_info("ignored")
+ # only passed in last stage will be regarded as real pass
+ if last_stage():
+ one_case_finish(not int(data[0]))
+ else:
+ Utility.console_log("test finished before enter last stage", color="orange")
+ one_case_finish(False)
+
+ def handle_next_stage(data):
+ """ reboot finished. we goto next stage """
+ if last_stage():
+ # already last stage, should never goto next stage
+ Utility.console_log("didn't finish at last stage", color="orange")
+ one_case_finish(False)
+ else:
+ stage_finish.append("continue")
+
+ while not stage_finish:
+ try:
+ dut.expect_any((RESET_PATTERN, handle_exception_reset),
+ (EXCEPTION_PATTERN, handle_exception_reset),
+ (ABORT_PATTERN, handle_exception_reset),
+ (FINISH_PATTERN, handle_test_finish),
+ (UT_APP_BOOT_UP_DONE, handle_next_stage),
+ timeout=one_case["timeout"])
+ except ExpectTimeout:
+ Utility.console_log("Timeout in expect", color="orange")
+ one_case_finish(False)
+ break
+ if stage_finish[0] == "break":
+ # test breaks on current stage
+ break
+
+
+@IDF.idf_unit_test(env_tag="UT_T1_1", junit_report_by_case=True)
def run_multiple_stage_cases(env, extra_data):
"""
extra_data can be 2 types of value
dut.start_app()
for one_case in case_config[ut_config]:
- reset_dut(dut)
- exception_reset_list = []
-
- for test_stage in range(one_case["child case num"]):
- # select multi stage test case name
- dut.write("\"{}\"".format(one_case["name"]))
- dut.expect("Running " + one_case["name"] + "...")
- # select test function for current stage
- dut.write(str(test_stage + 1))
-
- # we want to set this flag in callbacks (inner functions)
- # use list here so we can use append to set this flag
- stage_finish = list()
-
- def last_stage():
- return test_stage == one_case["child case num"] - 1
-
- def check_reset():
- if one_case["reset"]:
- assert exception_reset_list # reboot but no exception/reset logged. should never happen
- result = False
- if len(one_case["reset"]) == len(exception_reset_list):
- for i, exception in enumerate(exception_reset_list):
- if one_case["reset"][i] not in exception:
- break
- else:
- result = True
- if not result:
- Utility.console_log("""Reset Check Failed: \r\n\tExpected: {}\r\n\tGet: {}"""
- .format(one_case["reset"], exception_reset_list),
- color="orange")
- else:
- # we allow omit reset in multi stage cases
- result = True
- return result
-
- # expect callbacks
- def one_case_finish(result):
- """ one test finished, let expect loop break and log result """
- # handle test finish
- result = result and check_reset()
- if result:
- Utility.console_log("Success: " + one_case["name"], color="green")
- else:
- failed_cases.append(one_case["name"])
- Utility.console_log("Failed: " + one_case["name"], color="red")
- stage_finish.append("break")
-
- def handle_exception_reset(data):
- """
- just append data to exception list.
- exception list will be checked in ``handle_reset_finish``, once reset finished.
- """
- exception_reset_list.append(data[0])
-
- def handle_test_finish(data):
- """ test finished without reset """
- # in this scenario reset should not happen
- if int(data[1]):
- # case ignored
- Utility.console_log("Ignored: " + one_case["name"], color="orange")
- # only passed in last stage will be regarded as real pass
- if last_stage():
- one_case_finish(not int(data[0]))
- else:
- Utility.console_log("test finished before enter last stage", color="orange")
- one_case_finish(False)
-
- def handle_next_stage(data):
- """ reboot finished. we goto next stage """
- if last_stage():
- # already last stage, should never goto next stage
- Utility.console_log("didn't finish at last stage", color="orange")
- one_case_finish(False)
- else:
- stage_finish.append("continue")
-
- while not stage_finish:
- try:
- dut.expect_any((RESET_PATTERN, handle_exception_reset),
- (EXCEPTION_PATTERN, handle_exception_reset),
- (ABORT_PATTERN, handle_exception_reset),
- (FINISH_PATTERN, handle_test_finish),
- (UT_APP_BOOT_UP_DONE, handle_next_stage),
- timeout=one_case["timeout"])
- except ExpectTimeout:
- Utility.console_log("Timeout in expect", color="orange")
- one_case_finish(False)
- break
- if stage_finish[0] == "break":
- # test breaks on current stage
- break
+ junit_test_case = TinyFW.JunitReport.create_test_case("[{}] {}".format(ut_config, one_case["name"]))
+ try:
+ run_one_multiple_stage_case(dut, one_case, failed_cases, junit_test_case)
+ TinyFW.JunitReport.test_case_finish(junit_test_case)
+ except Exception as e:
+ junit_test_case.add_error_info("Unexpected exception: " + str(e))
+ TinyFW.JunitReport.test_case_finish(junit_test_case)
# raise exception if any case fails
if failed_cases:
Utility.console_log("\t" + _case_name, color="red")
raise AssertionError("Unit Test Failed")
+
def detect_update_unit_test_info(env, extra_data, app_bin):
case_config = format_test_case_config(extra_data)
# These options are the same for all configs, therefore there is no need to continue
break
+
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument(
projects / esp-idf / commitdiff