1  from __future__ import print_function
  2  
  3  import argparse
  4  
  5  import yaml
  6  import os
  7  import re
  8  import shutil
  9  import subprocess
 10  
 11  from copy import deepcopy
 12  import CreateSectionTable
 13  
 14  try:
 15      from yaml import CLoader as Loader
 16  except ImportError:
 17      from yaml import Loader as Loader
 18  
 19  TEST_CASE_PATTERN = {
 20      "initial condition": "UTINIT1",
 21      "chip_target": "esp32",
 22      "level": "Unit",
 23      "execution time": 0,
 24      "auto test": "Yes",
 25      "category": "Function",
 26      "test point 1": "basic function",
 27      "version": "v1 (2016-12-06)",
 28      "test environment": "UT_T1_1",
 29      "reset": "",
 30      "expected result": "1. set succeed",
 31      "cmd set": "test_unit_test_case",
 32      "Test App": "UT",
 33  }
 34  
 35  
 36  class Parser(object):
 37      """ parse unit test cases from build files and create files for test bench """
 38  
 39      TAG_PATTERN = re.compile(r"([^=]+)(=)?(.+)?")
 40      DESCRIPTION_PATTERN = re.compile(r"\[([^]\[]+)\]")
 41      CONFIG_PATTERN = re.compile(r"{([^}]+)}")
 42      TEST_GROUPS_PATTERN = re.compile(r"TEST_GROUPS=(.*)$")
 43  
 44      # file path (relative to idf path)
 45      TAG_DEF_FILE = os.path.join("tools", "unit-test-app", "tools", "TagDefinition.yml")
 46      MODULE_DEF_FILE = os.path.join("tools", "unit-test-app", "tools", "ModuleDefinition.yml")
 47      CONFIG_DEPENDENCY_FILE = os.path.join("tools", "unit-test-app", "tools", "ConfigDependency.yml")
 48      MODULE_ARTIFACT_FILE = os.path.join("components", "idf_test", "ModuleDefinition.yml")
 49      TEST_CASE_FILE_DIR = os.path.join("components", "idf_test", "unit_test")
 50      UT_CONFIG_FOLDER = os.path.join("tools", "unit-test-app", "configs")
 51      ELF_FILE = "unit-test-app.elf"
 52      SDKCONFIG_FILE = "sdkconfig"
 53      STRIP_CONFIG_PATTERN = re.compile(r"(.+?)(_\d+)?$")
 54      TOOLCHAIN_FOR_TARGET = {
 55          "esp32": "xtensa-esp32-elf-",
 56          "esp32s2": "xtensa-esp32s2-elf-",
 57      }
 58  
 59      def __init__(self, binary_folder):
 60          idf_path = os.getenv('IDF_PATH')
 61          idf_target = os.getenv('IDF_TARGET')
 62          self.test_env_tags = {}
 63          self.unit_jobs = {}
 64          self.file_name_cache = {}
 65          self.idf_path = idf_path
 66          self.idf_target = idf_target
 67          self.ut_bin_folder = binary_folder
 68          self.objdump = Parser.TOOLCHAIN_FOR_TARGET.get(idf_target, "") + "objdump"
 69          self.tag_def = yaml.load(open(os.path.join(idf_path, self.TAG_DEF_FILE), "r"), Loader=Loader)
 70          self.module_map = yaml.load(open(os.path.join(idf_path, self.MODULE_DEF_FILE), "r"), Loader=Loader)
 71          self.config_dependencies = yaml.load(open(os.path.join(idf_path, self.CONFIG_DEPENDENCY_FILE), "r"),
 72                                               Loader=Loader)
 73          # used to check if duplicated test case names
 74          self.test_case_names = set()
 75          self.parsing_errors = []
 76  
 77      def parse_test_cases_for_one_config(self, configs_folder, config_output_folder, config_name):
 78          """
 79          parse test cases from elf and save test cases need to be executed to unit test folder
 80          :param configs_folder: folder where per-config sdkconfig fragments are located (i.e. tools/unit-test-app/configs)
 81          :param config_output_folder: build folder of this config
 82          :param config_name: built unit test config name
 83          """
 84          tags = self.parse_tags(os.path.join(config_output_folder, self.SDKCONFIG_FILE))
 85          print("Tags of config %s: %s" % (config_name, tags))
 86  
 87          test_groups = self.get_test_groups(os.path.join(configs_folder, config_name))
 88  
 89          elf_file = os.path.join(config_output_folder, self.ELF_FILE)
 90          subprocess.check_output('{} -t {} | grep test_desc > case_address.tmp'.format(self.objdump, elf_file),
 91                                  shell=True)
 92          subprocess.check_output('{} -s {} > section_table.tmp'.format(self.objdump, elf_file), shell=True)
 93  
 94          table = CreateSectionTable.SectionTable("section_table.tmp")
 95          test_cases = []
 96  
 97          # we could split cases of same config into multiple binaries as we have limited rom space
 98          # we should regard those configs like `default` and `default_2` as the same config
 99          match = self.STRIP_CONFIG_PATTERN.match(config_name)
100          stripped_config_name = match.group(1)
101  
102          with open("case_address.tmp", "rb") as f:
103              for line in f:
104                  # process symbol table like: "3ffb4310 l     O .dram0.data	00000018 test_desc_33$5010"
105                  line = line.split()
106                  test_addr = int(line[0], 16)
107                  section = line[3]
108  
109                  name_addr = table.get_unsigned_int(section, test_addr, 4)
110                  desc_addr = table.get_unsigned_int(section, test_addr + 4, 4)
111                  function_count = table.get_unsigned_int(section, test_addr + 20, 4)
112                  name = table.get_string("any", name_addr)
113                  desc = table.get_string("any", desc_addr)
114  
115                  tc = self.parse_one_test_case(name, desc, config_name, stripped_config_name, tags)
116  
117                  # check if duplicated case names
118                  # we need to use it to select case,
119                  # if duplicated IDs, Unity could select incorrect case to run
120                  # and we need to check all cases no matter if it's going te be executed by CI
121                  # also add app_name here, we allow same case for different apps
122                  if (tc["summary"] + stripped_config_name) in self.test_case_names:
123                      self.parsing_errors.append("duplicated test case ID: " + tc["summary"])
124                  else:
125                      self.test_case_names.add(tc["summary"] + stripped_config_name)
126  
127                  test_group_included = True
128                  if test_groups is not None and tc["group"] not in test_groups:
129                      test_group_included = False
130  
131                  if tc["CI ready"] == "Yes" and test_group_included:
132                      # update test env list and the cases of same env list
133                      if tc["test environment"] in self.test_env_tags:
134                          self.test_env_tags[tc["test environment"]].append(tc["ID"])
135                      else:
136                          self.test_env_tags.update({tc["test environment"]: [tc["ID"]]})
137  
138                      if function_count > 1:
139                          tc.update({"child case num": function_count})
140  
141                      # only add  cases need to be executed
142                      test_cases.append(tc)
143  
144          os.remove("section_table.tmp")
145          os.remove("case_address.tmp")
146  
147          return test_cases
148  
149      def parse_case_properties(self, tags_raw):
150          """
151          parse test case tags (properties) with the following rules:
152              * first tag is always group of test cases, it's mandatory
153              * the rest tags should be [type=value].
154                  * if the type have default value, then [type] equal to [type=default_value].
155                  * if the type don't don't exist, then equal to [type=omitted_value]
156              default_value and omitted_value are defined in TagDefinition.yml
157          :param tags_raw: raw tag string
158          :return: tag dict
159          """
160          tags = self.DESCRIPTION_PATTERN.findall(tags_raw)
161          assert len(tags) > 0
162          p = dict([(k, self.tag_def[k]["omitted"]) for k in self.tag_def])
163          p["module"] = tags[0]
164  
165          # Use the original value of the first tag as test group name
166          p["group"] = p["module"]
167  
168          if p["module"] not in self.module_map:
169              p["module"] = "misc"
170  
171          # parsing rest tags, [type=value], =value is optional
172          for tag in tags[1:]:
173              match = self.TAG_PATTERN.search(tag)
174              assert match is not None
175              tag_type = match.group(1)
176              tag_value = match.group(3)
177              if match.group(2) == "=" and tag_value is None:
178                  # [tag_type=] means tag_value is empty string
179                  tag_value = ""
180              if tag_type in p:
181                  if tag_value is None:
182                      p[tag_type] = self.tag_def[tag_type]["default"]
183                  else:
184                      p[tag_type] = tag_value
185              else:
186                  # ignore not defined tag type
187                  pass
188          return p
189  
190      @staticmethod
191      def parse_tags_internal(sdkconfig, config_dependencies, config_pattern):
192          required_tags = []
193  
194          def compare_config(config):
195              return config in sdkconfig
196  
197          def process_condition(condition):
198              matches = config_pattern.findall(condition)
199              if matches:
200                  for config in matches:
201                      compare_result = compare_config(config)
202                      # replace all configs in condition with True or False according to compare result
203                      condition = re.sub(config_pattern, str(compare_result), condition, count=1)
204                  # Now the condition is a python condition, we can use eval to compute its value
205                  ret = eval(condition)
206              else:
207                  # didn't use complex condition. only defined one condition for the tag
208                  ret = compare_config(condition)
209              return ret
210  
211          for tag in config_dependencies:
212              if process_condition(config_dependencies[tag]):
213                  required_tags.append(tag)
214  
215          return required_tags
216  
217      def parse_tags(self, sdkconfig_file):
218          """
219          Some test configs could requires different DUTs.
220          For example, if CONFIG_ESP32_SPIRAM_SUPPORT is enabled, we need WROVER-Kit to run test.
221          This method will get tags for runners according to ConfigDependency.yml(maps tags to sdkconfig).
222  
223          We support to the following syntax::
224  
225              # define the config which requires the tag
226              'tag_a': 'config_a="value_a"'
227              # define the condition for the tag
228              'tag_b': '{config A} and (not {config B} or (not {config C} and {config D}))'
229  
230          :param sdkconfig_file: sdk config file of the unit test config
231          :return: required tags for runners
232          """
233          with open(sdkconfig_file, "r") as f:
234              configs_raw_data = f.read()
235  
236          configs = configs_raw_data.splitlines(False)
237  
238          return self.parse_tags_internal(configs, self.config_dependencies, self.CONFIG_PATTERN)
239  
240      def get_test_groups(self, config_file):
241          """
242          If the config file includes TEST_GROUPS variable, return its value as a list of strings.
243          :param config_file file under configs/ directory for given configuration
244          :return: list of test groups, or None if TEST_GROUPS wasn't set
245          """
246          with open(config_file, "r") as f:
247              for line in f:
248                  match = self.TEST_GROUPS_PATTERN.match(line)
249                  if match is not None:
250                      return match.group(1).split(' ')
251          return None
252  
253      def parse_one_test_case(self, name, description, config_name, stripped_config_name, tags):
254          """
255          parse one test case
256          :param name: test case name (summary)
257          :param description: test case description (tag string)
258          :param config_name: built unit test app name
259          :param stripped_config_name: strip suffix from config name because they're the same except test components
260          :param tags: tags to select runners
261          :return: parsed test case
262          """
263          prop = self.parse_case_properties(description)
264  
265          test_case = deepcopy(TEST_CASE_PATTERN)
266          test_case.update({"config": config_name,
267                            "module": self.module_map[prop["module"]]['module'],
268                            "group": prop["group"],
269                            "CI ready": "No" if prop["ignore"] == "Yes" else "Yes",
270                            "ID": "[{}] {}".format(stripped_config_name, name),
271                            "test point 2": prop["module"],
272                            "steps": name,
273                            "test environment": prop["test_env"],
274                            "reset": prop["reset"],
275                            "sub module": self.module_map[prop["module"]]['sub module'],
276                            "summary": name,
277                            "multi_device": prop["multi_device"],
278                            "multi_stage": prop["multi_stage"],
279                            "timeout": int(prop["timeout"]),
280                            "tags": tags,
281                            "chip_target": self.idf_target})
282          return test_case
283  
284      def dump_test_cases(self, test_cases):
285          """
286          dump parsed test cases to YAML file for test bench input
287          :param test_cases: parsed test cases
288          """
289          filename = os.path.join(self.idf_path, self.TEST_CASE_FILE_DIR, self.idf_target + ".yml")
290          try:
291              os.mkdir(os.path.dirname(filename))
292          except OSError:
293              pass
294          with open(os.path.join(filename), "w+") as f:
295              yaml.dump({"test cases": test_cases}, f, allow_unicode=True, default_flow_style=False)
296  
297      def copy_module_def_file(self):
298          """ copy module def file to artifact path """
299          src = os.path.join(self.idf_path, self.MODULE_DEF_FILE)
300          dst = os.path.join(self.idf_path, self.MODULE_ARTIFACT_FILE)
301          shutil.copy(src, dst)
302  
303      def parse_test_cases(self):
304          """ parse test cases from multiple built unit test apps """
305          test_cases = []
306  
307          output_folder = os.path.join(self.idf_path, self.ut_bin_folder, self.idf_target)
308          configs_folder = os.path.join(self.idf_path, self.UT_CONFIG_FOLDER)
309          test_configs = [item for item in os.listdir(output_folder)
310                          if os.path.isdir(os.path.join(output_folder, item))]
311          for config in test_configs:
312              config_output_folder = os.path.join(output_folder, config)
313              if os.path.exists(config_output_folder):
314                  test_cases.extend(self.parse_test_cases_for_one_config(configs_folder, config_output_folder, config))
315          test_cases.sort(key=lambda x: x["config"] + x["summary"])
316          self.dump_test_cases(test_cases)
317  
318  
319  def test_parser(binary_folder):
320      parser = Parser(binary_folder)
321      # test parsing tags
322      # parsing module only and module in module list
323      prop = parser.parse_case_properties("[esp32]")
324      assert prop["module"] == "esp32"
325      # module not in module list
326      prop = parser.parse_case_properties("[not_in_list]")
327      assert prop["module"] == "misc"
328      # parsing a default tag, a tag with assigned value
329      prop = parser.parse_case_properties("[esp32][ignore][test_env=ABCD][not_support1][not_support2=ABCD]")
330      assert prop["ignore"] == "Yes" and prop["test_env"] == "ABCD" \
331             and "not_support1" not in prop and "not_supported2" not in prop
332      # parsing omitted value
333      prop = parser.parse_case_properties("[esp32]")
334      assert prop["ignore"] == "No" and prop["test_env"] == "UT_T1_1"
335      # parsing with incorrect format
336      try:
337          parser.parse_case_properties("abcd")
338          assert False
339      except AssertionError:
340          pass
341      # skip invalid data parse, [type=] assigns empty string to type
342      prop = parser.parse_case_properties("[esp32]abdc aaaa [ignore=]")
343      assert prop["module"] == "esp32" and prop["ignore"] == ""
344      # skip mis-paired []
345      prop = parser.parse_case_properties("[esp32][[ignore=b]][]][test_env=AAA]]")
346      assert prop["module"] == "esp32" and prop["ignore"] == "b" and prop["test_env"] == "AAA"
347  
348      config_dependency = {
349          'a': '123',
350          'b': '456',
351          'c': 'not {123}',
352          'd': '{123} and not {456}',
353          'e': '{123} and not {789}',
354          'f': '({123} and {456}) or ({123} and {789})'
355      }
356      sdkconfig = ["123", "789"]
357      tags = parser.parse_tags_internal(sdkconfig, config_dependency, parser.CONFIG_PATTERN)
358      assert sorted(tags) == ['a', 'd', 'f']  # sorted is required for older Python3, e.g. 3.4.8
359  
360  
361  def main(binary_folder):
362      assert os.getenv('IDF_PATH'), 'IDF_PATH must be set to use this script'
363      assert os.getenv('IDF_TARGET'), 'IDF_TARGET must be set to use this script'
364      test_parser(binary_folder)
365  
366      parser = Parser(binary_folder)
367      parser.parse_test_cases()
368      parser.copy_module_def_file()
369      if len(parser.parsing_errors) > 0:
370          for error in parser.parsing_errors:
371              print(error)
372          exit(1)
373  
374  
375  if __name__ == '__main__':
376      parser = argparse.ArgumentParser()
377      parser.add_argument('bin_dir', help='Binary Folder')
378      args = parser.parse_args()
379      main(args.bin_dir)