llvm/utils/update_any_test_checks.py

   1 #!/usr/bin/env python3
   2
   3 """Dispatch to update_*_test_checks.py scripts automatically in bulk
   4
   5 Given a list of test files, this script will invoke the correct
   6 update_test_checks-style script, skipping any tests which have not previously
   7 had assertions autogenerated. If test name starts with '@' it's treated as
   8 a name of file containing test list.
   9 """
  10
  11 from __future__ import print_function
  12
  13 import argparse
  14 import os
  15 import re
  16 import subprocess
  17 import sys
  18 from concurrent.futures import ThreadPoolExecutor
  19
  20 RE_ASSERTIONS = re.compile(
  21     r"NOTE: Assertions have been autogenerated by ([^\s]+)( UTC_ARGS:.*)?$"
  22 )
  23
  24
  25 def find_utc_tool(search_path, utc_name):
  26     """
  27     Return the path to the given UTC tool in the search path, or None if not
  28     found.
  29     """
  30     for path in search_path:
  31         candidate = os.path.join(path, utc_name)
  32         if os.path.isfile(candidate):
  33             return candidate
  34     return None
  35
  36
  37 def run_utc_tool(utc_name, utc_tool, testname):
  38     result = subprocess.run(
  39         [utc_tool, testname], stdout=subprocess.PIPE, stderr=subprocess.PIPE
  40     )
  41     return (result.returncode, result.stdout, result.stderr)
  42
  43
  44 def read_arguments_from_file(filename):
  45     try:
  46         with open(filename, "r") as file:
  47             return [line.rstrip() for line in file.readlines()]
  48     except FileNotFoundError:
  49         print(f"Error: File '{filename}' not found.")
  50         sys.exit(1)
  51
  52
  53 def expand_listfile_args(arg_list):
  54     exp_arg_list = []
  55     for arg in arg_list:
  56         if arg.startswith("@"):
  57             exp_arg_list += read_arguments_from_file(arg[1:])
  58         else:
  59             exp_arg_list.append(arg)
  60     return exp_arg_list
  61
  62
  63 def main():
  64     from argparse import RawTextHelpFormatter
  65
  66     parser = argparse.ArgumentParser(
  67         description=__doc__, formatter_class=RawTextHelpFormatter
  68     )
  69     parser.add_argument(
  70         "--jobs",
  71         "-j",
  72         default=1,
  73         type=int,
  74         help="Run the given number of jobs in parallel",
  75     )
  76     parser.add_argument(
  77         "--utc-dir",
  78         nargs="*",
  79         help="Additional directories to scan for update_*_test_checks scripts",
  80     )
  81     parser.add_argument("tests", nargs="+")
  82     config = parser.parse_args()
  83
  84     if config.utc_dir:
  85         utc_search_path = config.utc_dir[:]
  86     else:
  87         utc_search_path = []
  88     script_name = os.path.abspath(__file__)
  89     utc_search_path.append(os.path.join(os.path.dirname(script_name), os.path.pardir))
  90
  91     not_autogenerated = []
  92     utc_tools = {}
  93     have_error = False
  94
  95     tests = expand_listfile_args(config.tests)
  96
  97     with ThreadPoolExecutor(max_workers=config.jobs) as executor:
  98         jobs = []
  99
 100         for testname in tests:
 101             with open(testname, "r") as f:
 102                 header = f.readline().strip()
 103                 m = RE_ASSERTIONS.search(header)
 104                 if m is None:
 105                     not_autogenerated.append(testname)
 106                     continue
 107
 108                 utc_name = m.group(1)
 109                 if utc_name not in utc_tools:
 110                     utc_tools[utc_name] = find_utc_tool(utc_search_path, utc_name)
 111                     if not utc_tools[utc_name]:
 112                         print(
 113                             f"{utc_name}: not found (used in {testname})",
 114                             file=sys.stderr,
 115                         )
 116                         have_error = True
 117                         continue
 118
 119                 future = executor.submit(
 120                     run_utc_tool, utc_name, utc_tools[utc_name], testname
 121                 )
 122                 jobs.append((testname, future))
 123
 124         for testname, future in jobs:
 125             return_code, stdout, stderr = future.result()
 126
 127             print(f"Update {testname}")
 128             stdout = stdout.decode(errors="replace")
 129             if stdout:
 130                 print(stdout, end="")
 131                 if not stdout.endswith("\n"):
 132                     print()
 133
 134             stderr = stderr.decode(errors="replace")
 135             if stderr:
 136                 print(stderr, end="")
 137                 if not stderr.endswith("\n"):
 138                     print()
 139             if return_code != 0:
 140                 print(f"Return code: {return_code}")
 141                 have_error = True
 142
 143     if have_error:
 144         sys.exit(1)
 145
 146     if not_autogenerated:
 147         print("Tests without autogenerated assertions:")
 148         for testname in not_autogenerated:
 149             print(f"  {testname}")
 150
 151
 152 if __name__ == "__main__":
 153     main()