test: generate junit test report according to executed cases

tools/tiny-test-fw/IDF/__init__.py

@@ -81,7 +81,7 @@ def log_performance(item, value):
     Utility.console_log(performance_msg, "orange")
     # update to junit test report
     current_junit_case = TinyFW.JunitReport.get_current_test_case()
-    current_junit_case.stdout += performance_msg
+    current_junit_case.stdout += performance_msg + "\r\n"
 
 
 def check_performance(item, value):

tools/unit-test-app/unit_test.py

@@ -23,7 +23,6 @@ import os
 import sys
 import time
 import argparse
-
 import threading
 
 # if we want to run test case outside `tiny-test-fw` folder,
@@ -132,6 +131,7 @@ def format_test_case_config(test_case_data):
 
     return case_config
 
+
 def replace_app_bin(dut, name, new_app_bin):
     if new_app_bin is None:
         return
@@ -142,6 +142,7 @@ def replace_app_bin(dut, name, new_app_bin):
             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.
@@ -160,7 +161,83 @@ def reset_dut(dut):
         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
@@ -190,74 +267,17 @@ def run_unit_test_cases(env, extra_data):
         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)
+            # create junit report test case
-
+            junit_test_case = TinyFW.JunitReport.create_test_case("[{}] {}".format(ut_config, one_case["name"]))
-            # run test case
+            try:
-            dut.write("\"{}\"".format(one_case["name"]))
+                run_one_normal_case(dut, one_case, junit_test_case, failed_cases)
-            dut.expect("Running " + one_case["name"] + "...")
+                TinyFW.JunitReport.test_case_finish(junit_test_case)
-
+            except Exception as e:
-            exception_reset_list = []
+                junit_test_case.add_error_info("Unexpected exception: " + str(e))
-
+                TinyFW.JunitReport.test_case_finish(junit_test_case)
-            # 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
 
     # raise exception if any case fails
     if failed_cases:
@@ -283,6 +303,7 @@ class Handler(threading.Thread):
         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
@@ -292,6 +313,9 @@ class Handler(threading.Thread):
         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)
@@ -301,6 +325,8 @@ class Handler(threading.Thread):
             """ 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
 
@@ -370,7 +396,7 @@ def get_dut(duts, env, name, ut_config, app_bin=None):
     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 = []
@@ -384,9 +410,11 @@ def case_run(duts, ut_config, env, one_case, failed_cases, app_bin):
     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]
 
@@ -394,10 +422,11 @@ def case_run(duts, ut_config, env, one_case, failed_cases, app_bin):
         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
@@ -421,11 +450,17 @@ def run_multiple_devices_cases(env, extra_data):
     """
     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")
@@ -434,7 +469,109 @@ def run_multiple_devices_cases(env, extra_data):
         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
@@ -461,98 +598,13 @@ def run_multiple_stage_cases(env, extra_data):
         dut.start_app()
 
         for one_case in case_config[ut_config]:
-            reset_dut(dut)
+            junit_test_case = TinyFW.JunitReport.create_test_case("[{}] {}".format(ut_config, one_case["name"]))
-            exception_reset_list = []
+            try:
-
+                run_one_multiple_stage_case(dut, one_case, failed_cases, junit_test_case)
-            for test_stage in range(one_case["child case num"]):
+                TinyFW.JunitReport.test_case_finish(junit_test_case)
-                # select multi stage test case name
+            except Exception as e:
-                dut.write("\"{}\"".format(one_case["name"]))
+                junit_test_case.add_error_info("Unexpected exception: " + str(e))
-                dut.expect("Running " + one_case["name"] + "...")
+                TinyFW.JunitReport.test_case_finish(junit_test_case)
-                # 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
 
     # raise exception if any case fails
     if failed_cases:
@@ -561,6 +613,7 @@ def run_multiple_stage_cases(env, extra_data):
             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)
@@ -624,6 +677,7 @@ def detect_update_unit_test_info(env, extra_data, app_bin):
         # 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(