tiny-test-fw: move to tools/esp_python_packages:

make `tiny_test_fw` as a package and move to root path of idf python
packages

tools/ci/python_packages/ttfw_idf/CIAssignUnitTest.py

@@ -0,0 +1,198 @@
+"""
+Command line tool to assign unit tests to CI test jobs.
+"""
+
+import re
+import argparse
+
+import yaml
+
+try:
+    from yaml import CLoader as Loader
+except ImportError:
+    from yaml import Loader as Loader
+
+from tiny_test_fw.Utility import CIAssignTest
+
+
+class Group(CIAssignTest.Group):
+    SORT_KEYS = ["test environment", "tags", "chip_target"]
+    MAX_CASE = 50
+    ATTR_CONVERT_TABLE = {
+        "execution_time": "execution time"
+    }
+    CI_JOB_MATCH_KEYS = ["test environment"]
+    DUT_CLS_NAME = {
+        "esp32": "ESP32DUT",
+        "esp32s2beta": "ESP32S2DUT",
+        "esp8266": "ESP8266DUT",
+    }
+
+    def __init__(self, case):
+        super(Group, self).__init__(case)
+        for tag in self._get_case_attr(case, "tags"):
+            self.ci_job_match_keys.add(tag)
+
+    @staticmethod
+    def _get_case_attr(case, attr):
+        if attr in Group.ATTR_CONVERT_TABLE:
+            attr = Group.ATTR_CONVERT_TABLE[attr]
+        return case[attr]
+
+    def add_extra_case(self, case):
+        """ If current group contains all tags required by case, then add succeed """
+        added = False
+        if self.accept_new_case():
+            for key in self.filters:
+                if self._get_case_attr(case, key) != self.filters[key]:
+                    if key == "tags":
+                        if self._get_case_attr(case, key).issubset(self.filters[key]):
+                            continue
+                    break
+            else:
+                self.case_list.append(case)
+                added = True
+        return added
+
+    def _create_extra_data(self, test_cases, test_function):
+        """
+        For unit test case, we need to copy some attributes of test cases into config file.
+        So unit test function knows how to run the case.
+        """
+        case_data = []
+        for case in test_cases:
+            one_case_data = {
+                "config": self._get_case_attr(case, "config"),
+                "name": self._get_case_attr(case, "summary"),
+                "reset": self._get_case_attr(case, "reset"),
+                "timeout": self._get_case_attr(case, "timeout"),
+            }
+
+            if test_function in ["run_multiple_devices_cases", "run_multiple_stage_cases"]:
+                try:
+                    one_case_data["child case num"] = self._get_case_attr(case, "child case num")
+                except KeyError as e:
+                    print("multiple devices/stages cases must contains at least two test functions")
+                    print("case name: {}".format(one_case_data["name"]))
+                    raise e
+
+            case_data.append(one_case_data)
+        return case_data
+
+    def _divide_case_by_test_function(self):
+        """
+        divide cases of current test group by test function they need to use
+
+        :return: dict of list of cases for each test functions
+        """
+        case_by_test_function = {
+            "run_multiple_devices_cases": [],
+            "run_multiple_stage_cases": [],
+            "run_unit_test_cases": [],
+        }
+
+        for case in self.case_list:
+            if case["multi_device"] == "Yes":
+                case_by_test_function["run_multiple_devices_cases"].append(case)
+            elif case["multi_stage"] == "Yes":
+                case_by_test_function["run_multiple_stage_cases"].append(case)
+            else:
+                case_by_test_function["run_unit_test_cases"].append(case)
+        return case_by_test_function
+
+    def output(self):
+        """
+        output data for job configs
+
+        :return: {"Filter": case filter, "CaseConfig": list of case configs for cases in this group}
+        """
+
+        target = self._get_case_attr(self.case_list[0], "chip_target")
+        if target:
+            overwrite = {
+                "dut": {
+                    "package": "ttfw_idf",
+                    "class": self.DUT_CLS_NAME[target],
+                }
+            }
+        else:
+            overwrite = dict()
+
+        case_by_test_function = self._divide_case_by_test_function()
+
+        output_data = {
+            # we don't need filter for test function, as UT uses a few test functions for all cases
+            "CaseConfig": [
+                {
+                    "name": test_function,
+                    "extra_data": self._create_extra_data(test_cases, test_function),
+                    "overwrite": overwrite,
+                } for test_function, test_cases in case_by_test_function.iteritems() if test_cases
+            ],
+        }
+        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)
+
+    def _search_cases(self, 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.
+        """
+
+        try:
+            with open(test_case_path, "r") as f:
+                raw_data = yaml.load(f, Loader=Loader)
+            test_cases = raw_data["test cases"]
+            for case in test_cases:
+                case["tags"] = set(case["tags"])
+        except IOError:
+            print("Test case path is invalid. Should only happen when use @bot to skip unit test.")
+            test_cases = []
+        # filter keys are lower case. Do map lower case keys with original keys.
+        try:
+            key_mapping = {x.lower(): x for x in test_cases[0].keys()}
+        except IndexError:
+            key_mapping = dict()
+        if case_filter:
+            for key in case_filter:
+                filtered_cases = []
+                for case in test_cases:
+                    try:
+                        mapped_key = key_mapping[key]
+                        # bot converts string to lower case
+                        if isinstance(case[mapped_key], str):
+                            _value = case[mapped_key].lower()
+                        else:
+                            _value = case[mapped_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
+        # sort cases with configs and test functions
+        # in later stage cases with similar attributes are more likely to be assigned to the same job
+        # it will reduce the count of flash DUT operations
+        test_cases.sort(key=lambda x: x["config"] + x["multi_stage"] + x["multi_device"])
+        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)