Merge branch 'test/ut_script' into 'master'

test: use the tiny-test-fw to run the unit test in CI

See merge request idf/esp-idf!1558

Author: heyinling

.gitlab-ci.yml

@@ -446,22 +446,21 @@ assign_test:
       - components/idf_test/*/CIConfigs
       - components/idf_test/*/TC.sqlite
       - $EXAMPLE_CONFIG_OUTPUT_PATH
+      - tools/unit-test-app/output
     expire_in: 1 mos
   before_script: *add_gitlab_key_before
   script:
     # first move test bins together: test_bins/CHIP_SDK/TestApp/bin_files
     - mkdir -p $OUTPUT_BIN_PATH
-    # copy and rename folder name to "UT_config"
-    - for CONFIG in $(ls $UT_BIN_PATH); do cp -r "$UT_BIN_PATH/$CONFIG" "$OUTPUT_BIN_PATH/UT_$CONFIG"; done
     - cp -r SSC/ssc_bin/* $OUTPUT_BIN_PATH
     # assign example tests
     - python $TEST_FW_PATH/CIAssignExampleTest.py $IDF_PATH/examples $IDF_PATH/.gitlab-ci.yml $EXAMPLE_CONFIG_OUTPUT_PATH
+    # assign unit test cases
+    - python $TEST_FW_PATH/CIAssignUnitTest.py $IDF_PATH/components/idf_test/unit_test/TestCaseAll.yml $IDF_PATH/.gitlab-ci.yml $IDF_PATH/components/idf_test/unit_test/CIConfigs
     # clone test script to assign tests
     - git clone $TEST_SCRIPT_REPOSITORY
     - cd auto_test_script
     - python $CHECKOUT_REF_SCRIPT auto_test_script
-    # assign unit test cases
-    - python CIAssignTestCases.py -t $IDF_PATH/components/idf_test/unit_test -c $IDF_PATH/.gitlab-ci.yml -b $IDF_PATH/test_bins
     # assgin integration test cases
     - python CIAssignTestCases.py -t $IDF_PATH/components/idf_test/integration_test -c $IDF_PATH/.gitlab-ci.yml -b $IDF_PATH/test_bins
 
@@ -493,6 +492,17 @@ assign_test:
     # run test
     - python Runner.py $TEST_CASE_PATH -c $CONFIG_FILE
 
+.unit_test_template: &unit_test_template
+  <<: *example_test_template
+  stage: unit_test
+  dependencies:
+    - assign_test
+  variables:
+    TEST_FW_PATH: "$CI_PROJECT_DIR/tools/tiny-test-fw"
+    TEST_CASE_PATH: "$CI_PROJECT_DIR/tools/unit-test-app"
+    CONFIG_FILE: "$CI_PROJECT_DIR/components/idf_test/unit_test/CIConfigs/$CI_JOB_NAME.yml"
+    LOG_PATH: "$CI_PROJECT_DIR/TEST_LOGS"
+
 .test_template: &test_template
   stage: test
   when: on_success
@@ -530,18 +540,6 @@ assign_test:
     # run test
     - python CIRunner.py -l "$LOG_PATH/$CI_JOB_NAME" -c $CONFIG_FILE -e $LOCAL_ENV_CONFIG_PATH -t $TEST_CASE_FILE_PATH -m $MODULE_UPDATE_FILE
 
-# template for unit test jobs
-.unit_test_template: &unit_test_template
-  <<: *test_template
-  allow_failure: false
-  stage: unit_test
-  variables:
-    LOCAL_ENV_CONFIG_PATH: "$CI_PROJECT_DIR/ci-test-runner-configs/$CI_RUNNER_DESCRIPTION/ESP32_IDF"
-    LOG_PATH: "$CI_PROJECT_DIR/$CI_COMMIT_SHA"
-    TEST_CASE_FILE_PATH: "$CI_PROJECT_DIR/components/idf_test/unit_test"
-    MODULE_UPDATE_FILE: "$CI_PROJECT_DIR/components/idf_test/ModuleDefinition.yml"
-    CONFIG_FILE: "$CI_PROJECT_DIR/components/idf_test/unit_test/CIConfigs/$CI_JOB_NAME.yml"
-
 nvs_compatible_test:
   <<: *test_template
   artifacts:

components/driver/test/test_adc2.c

@@ -40,7 +40,7 @@ static esp_err_t event_handler(void *ctx, system_event_t *event)
     return ESP_OK;
 }
 
-TEST_CASE("adc2 work with wifi","[adc]")
+TEST_CASE("adc2 work with wifi","[adc][ignore]")
 {
     int     read_raw;
     int     target_value;

tools/tiny-test-fw/CIAssignExampleTest.py

@@ -22,147 +22,16 @@
 import re
 import argparse
 
-import yaml
-
 test_fw_path = os.getenv("TEST_FW_PATH")
 if test_fw_path:
     sys.path.insert(0, test_fw_path)
 
-from Utility import CaseConfig, SearchCases, GitlabCIJob
-
-
-class Group(object):
-
-    MAX_EXECUTION_TIME = 30
-    MAX_CASE = 15
-    SORT_KEYS = ["env_tag"]
-
-    def __init__(self, case):
-        self.execution_time = 0
-        self.case_list = [case]
-        self.filters = dict(zip(self.SORT_KEYS, [case.case_info[x] for x in self.SORT_KEYS]))
-
-    def accept_new_case(self):
-        """
-        check if allowed to add any case to this group
-
-        :return: True or False
-        """
-        max_time = (sum([x.case_info["execution_time"] for x in self.case_list]) < self.MAX_EXECUTION_TIME)
-        max_case = (len(self.case_list) < self.MAX_CASE)
-        return max_time and max_case
-
-    def add_case(self, case):
-        """
-        add case to current group
-
-        :param case: test case
-        :return: True if add succeed, else False
-        """
-        added = False
-        if self.accept_new_case():
-            for key in self.filters:
-                if case.case_info[key] != self.filters[key]:
-                    break
-            else:
-                self.case_list.append(case)
-                added = True
-        return added
-
-    def output(self):
-        """
-        output data for job configs
-
-        :return: {"Filter": case filter, "CaseConfig": list of case configs for cases in this group}
-        """
-        output_data = {
-            "Filter": self.filters,
-            "CaseConfig": [{"name": x.case_info["name"]} for x in self.case_list],
-        }
-        return output_data
-
+from Utility.CIAssignTest import AssignTest
 
-class AssignTest(object):
-    """
-    Auto assign tests to CI jobs.
-
-    :param test_case: path of test case file(s)
-    :param ci_config_file: path of ``.gitlab-ci.yml``
-    """
 
+class CIExampleAssignTest(AssignTest):
     CI_TEST_JOB_PATTERN = re.compile(r"^example_test_.+")
 
-    def __init__(self, test_case, ci_config_file):
-        self.test_cases = self._search_cases(test_case)
-        self.jobs = self._parse_gitlab_ci_config(ci_config_file)
-
-    def _parse_gitlab_ci_config(self, ci_config_file):
-
-        with open(ci_config_file, "r") as f:
-            ci_config = yaml.load(f)
-
-        job_list = list()
-        for job_name in ci_config:
-            if self.CI_TEST_JOB_PATTERN.search(job_name) is not None:
-                job_list.append(GitlabCIJob.Job(ci_config[job_name], job_name))
-        return job_list
-
-    @staticmethod
-    def _search_cases(test_case, case_filter=None):
-        """
-        :param test_case: path contains test case folder
-        :param case_filter: filter for test cases
-        :return: filtered test case list
-        """
-        test_methods = SearchCases.Search.search_test_cases(test_case)
-        return CaseConfig.filter_test_cases(test_methods, case_filter if case_filter else dict())
-
-    def _group_cases(self):
-        """
-        separate all cases into groups according group rules. each group will be executed by one CI job.
-
-        :return: test case groups.
-        """
-        groups = []
-        for case in self.test_cases:
-            for group in groups:
-                # add to current group
-                if group.add_case(case):
-                    break
-            else:
-                # create new group
-                groups.append(Group(case))
-        return groups
-
-    def assign_cases(self):
-        """
-        separate test cases to groups and assign test cases to CI jobs.
-
-        :raise AssertError: if failed to assign any case to CI job.
-        :return: None
-        """
-        failed_to_assign = []
-        test_groups = self._group_cases()
-        for group in test_groups:
-            for job in self.jobs:
-                if job.match_group(group):
-                    job.assign_group(group)
-                    break
-            else:
-                failed_to_assign.append(group)
-        assert not failed_to_assign
-
-    def output_configs(self, output_path):
-        """
-
-        :param output_path: path to output config files for each CI job
-        :return: None
-        """
-        if not os.path.exists(output_path):
-            os.makedirs(output_path)
-        for job in self.jobs:
-            job.output_config(output_path)
-
 
 if __name__ == '__main__':
     parser = argparse.ArgumentParser()
@@ -174,6 +43,6 @@ def output_configs(self, output_path):
                         help="output path of config files")
     args = parser.parse_args()
 
-    assign_test = AssignTest(args.test_case, args.ci_config_file)
+    assign_test = CIExampleAssignTest(args.test_case, args.ci_config_file)
     assign_test.assign_cases()
     assign_test.output_configs(args.output_path)

tools/tiny-test-fw/CIAssignUnitTest.py

@@ -0,0 +1,121 @@
+"""
+Command line tool to assign unit tests to CI test jobs.
+"""
+
+import re
+import os
+import sys
+import argparse
+
+import yaml
+
+test_fw_path = os.getenv("TEST_FW_PATH")
+if test_fw_path:
+    sys.path.insert(0, test_fw_path)
+
+from Utility import CIAssignTest
+
+
+class Group(CIAssignTest.Group):
+    SORT_KEYS = ["Test App", "SDK", "test environment"]
+    MAX_CASE = 30
+    ATTR_CONVERT_TABLE = {
+        "execution_time": "execution time"
+    }
+
+    @staticmethod
+    def _get_case_attr(case, attr):
+        if attr in Group.ATTR_CONVERT_TABLE:
+            attr = Group.ATTR_CONVERT_TABLE[attr]
+        return case[attr]
+
+    @staticmethod
+    def _get_ut_config(test_app):
+        # we format test app "UT_ + config" when parsing test cases
+        # now we need to extract config
+        assert test_app[:3] == "UT_"
+        return test_app[3:]
+
+    def _create_extra_data(self):
+        case_data = []
+        for case in self.case_list:
+            if self._get_case_attr(case, "cmd set") == "multiple_devices_case":
+                case_data.append({
+                    "config": self._get_ut_config(self._get_case_attr(case, "Test App")),
+                    "name": self._get_case_attr(case, "summary"),
+                    "child case num": self._get_case_attr(case, "child case num")
+                })
+            else:
+                case_data.append({
+                    "config": self._get_ut_config(self._get_case_attr(case, "Test App")),
+                    "name": self._get_case_attr(case, "summary"),
+                    "reset": self._get_case_attr(case, "reset") ,
+                })
+        return case_data
+
+    def output(self):
+        """
+        output data for job configs
+
+        :return: {"Filter": case filter, "CaseConfig": list of case configs for cases in this group}
+        """
+        output_data = {
+            # we don't need filter for test function, as UT uses a few test functions for all cases
+            "CaseConfig": [
+                {
+                    "name": self.case_list[0]["cmd set"] if isinstance(self.case_list[0]["cmd set"], str) else self.case_list[0]["cmd set"][0],
+                    "extra_data": self._create_extra_data(),
+                }
+            ]
+        }
+        return output_data
+
+
+class UnitTestAssignTest(CIAssignTest.AssignTest):
+    CI_TEST_JOB_PATTERN = re.compile(r"^UT_.+")
+
+    def __init__(self, test_case_path, ci_config_file):
+        CIAssignTest.AssignTest.__init__(self, test_case_path, ci_config_file, case_group=Group)
+
+    @staticmethod
+    def _search_cases(test_case_path, case_filter=None):
+        """
+        For unit test case, we don't search for test functions.
+        The unit test cases is stored in a yaml file which is created in job build-idf-test.
+        """
+
+        with open(test_case_path, "r") as f:
+            raw_data = yaml.load(f)
+        test_cases = raw_data["test cases"]
+        if case_filter:
+            for key in case_filter:
+                filtered_cases = []
+                for case in test_cases:
+                    try:
+                        # bot converts string to lower case
+                        if isinstance(case[key], str):
+                            _value = case[key].lower()
+                        else:
+                            _value = case[key]
+                        if _value in case_filter[key]:
+                            filtered_cases.append(case)
+                    except KeyError:
+                        # case don't have this key, regard as filter success
+                        filtered_cases.append(case)
+                test_cases = filtered_cases
+        return test_cases
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser()
+    parser.add_argument("test_case",
+                        help="test case folder or file")
+    parser.add_argument("ci_config_file",
+                        help="gitlab ci config file")
+    parser.add_argument("output_path",
+                        help="output path of config files")
+    args = parser.parse_args()
+
+    assign_test = UnitTestAssignTest(args.test_case, args.ci_config_file)
+    assign_test.assign_cases()
+    assign_test.output_configs(args.output_path)

tools/tiny-test-fw/DUT.py

@@ -316,7 +316,7 @@ def write(self, data, eol="\r\n", flush=True):
         if flush:
             self.data_cache.flush()
         # do write if cache
-        if data:
+        if data is not None:
             self._port_write(data + eol if eol else data)
 
     @_expect_lock
@@ -557,9 +557,9 @@ def _format_data(data):
         :return: formatted data (str)
         """
         timestamp = time.time()
-        timestamp = "{}:{}".format(time.strftime("%m-%d %H:%M:%S", time.localtime(timestamp)),
-                                   str(timestamp % 1)[2:5])
-        formatted_data = "[{}]:\r\n{}\r\n".format(timestamp, _decode_data(data))
+        timestamp = "[{}:{}]".format(time.strftime("%m-%d %H:%M:%S", time.localtime(timestamp)),
+                                     str(timestamp % 1)[2:5])
+        formatted_data = timestamp.encode() + b"\r\n" + data + b"\r\n"
         return formatted_data
 
     def _port_open(self):
@@ -571,11 +571,13 @@ def _port_close(self):
     def _port_read(self, size=1):
         data = self.port_inst.read(size)
         if data:
-            with open(self.log_file, "a+") as _log_file:
+            with open(self.log_file, "ab+") as _log_file:
                 _log_file.write(self._format_data(data))
         return data
 
     def _port_write(self, data):
+        if isinstance(data, str):
+            data = data.encode()
         self.port_inst.write(data)
 
     @classmethod