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.
   8 """
   9
  10 from __future__ import print_function
  11
  12 import argparse
  13 import os
  14 import re
  15 import subprocess
  16 import sys
  17 from concurrent.futures import ThreadPoolExecutor
  18
  19 RE_ASSERTIONS = re.compile(
  20     r'NOTE: Assertions have been autogenerated by ([^\s]+)( UTC_ARGS:.*)?$')
  21
  22 def find_utc_tool(search_path, utc_name):
  23   """
  24   Return the path to the given UTC tool in the search path, or None if not
  25   found.
  26   """
  27   for path in search_path:
  28     candidate = os.path.join(path, utc_name)
  29     if os.path.isfile(candidate):
  30       return candidate
  31   return None
  32
  33 def run_utc_tool(utc_name, utc_tool, testname):
  34   result = subprocess.run([utc_tool, testname], stdout=subprocess.PIPE,
  35                           stderr=subprocess.PIPE)
  36   return (result.returncode, result.stdout, result.stderr)
  37
  38 def main():
  39   from argparse import RawTextHelpFormatter
  40   parser = argparse.ArgumentParser(description=__doc__,
  41                                    formatter_class=RawTextHelpFormatter)
  42   parser.add_argument(
  43       '--jobs', '-j', default=1, type=int,
  44       help='Run the given number of jobs in parallel')
  45   parser.add_argument(
  46       '--utc-dir', nargs='*',
  47       help='Additional directories to scan for update_*_test_checks scripts')
  48   parser.add_argument('tests', nargs='+')
  49   config = parser.parse_args()
  50
  51   if config.utc_dir:
  52     utc_search_path = config.utc_dir[:]
  53   else:
  54     utc_search_path = []
  55   script_name = os.path.abspath(__file__)
  56   utc_search_path.append(os.path.join(os.path.dirname(script_name),
  57                                       os.path.pardir))
  58
  59   not_autogenerated = []
  60   utc_tools = {}
  61   have_error = False
  62
  63   with ThreadPoolExecutor(max_workers=config.jobs) as executor:
  64     jobs = []
  65
  66     for testname in config.tests:
  67       with open(testname, 'r') as f:
  68         header = f.readline().strip()
  69         m = RE_ASSERTIONS.search(header)
  70         if m is None:
  71           not_autogenerated.append(testname)
  72           continue
  73
  74         utc_name = m.group(1)
  75         if utc_name not in utc_tools:
  76           utc_tools[utc_name] = find_utc_tool(utc_search_path, utc_name)
  77           if not utc_tools[utc_name]:
  78             print(f"{utc_name}: not found (used in {testname})",
  79                   file=sys.stderr)
  80             have_error = True
  81             continue
  82
  83         future = executor.submit(run_utc_tool, utc_name, utc_tools[utc_name],
  84                                  testname)
  85         jobs.append((testname, future))
  86
  87     for testname, future in jobs:
  88       return_code, stdout, stderr = future.result()
  89
  90       print(f"Update {testname}")
  91       stdout = stdout.decode(errors='replace')
  92       if stdout:
  93         print(stdout, end='')
  94         if not stdout.endswith('\n'):
  95           print()
  96
  97       stderr = stderr.decode(errors='replace')
  98       if stderr:
  99         print(stderr, end='')
 100         if not stderr.endswith('\n'):
 101           print()
 102       if return_code != 0:
 103         print(f"Return code: {return_code}")
 104         have_error = True
 105
 106   if have_error:
 107     sys.exit(1)
 108
 109   if not_autogenerated:
 110     print("Tests without autogenerated assertions:")
 111     for testname in not_autogenerated:
 112       print(f"  {testname}")
 113
 114 if __name__ == '__main__':
 115   main()