202 lines · python
1#!/usr/bin/env python32 3"""Dispatch to update_*_test_checks.py scripts automatically in bulk4 5Given a list of test files, this script will invoke the correct6update_test_checks-style script, skipping any tests which have not previously7had assertions autogenerated. If test name starts with '@' it's treated as8a name of file containing test list.9"""10 11from __future__ import print_function12 13import argparse14import os15import re16import subprocess17import sys18from concurrent.futures import ThreadPoolExecutor19 20RE_ASSERTIONS = re.compile(21 r"NOTE: Assertions have been autogenerated by ([^\s]+)( UTC_ARGS:.*)?$"22)23 24 25def find_utc_tool(search_path, utc_name):26 """27 Return the path to the given UTC tool in the search path, or None if not28 found.29 """30 for path in search_path:31 candidate = os.path.join(path, utc_name)32 if os.path.isfile(candidate):33 return candidate34 return None35 36 37def run_utc_tool(utc_name, utc_tool, testname, environment):38 result = subprocess.run(39 [utc_tool, testname],40 stdout=subprocess.PIPE,41 stderr=subprocess.PIPE,42 env=environment,43 )44 return (result.returncode, result.stdout, result.stderr)45 46 47def read_arguments_from_file(filename):48 try:49 with open(filename, "r") as file:50 return [line.rstrip() for line in file.readlines()]51 except FileNotFoundError:52 print(f"Error: File '{filename}' not found.")53 sys.exit(1)54 55 56def expand_listfile_args(arg_list):57 exp_arg_list = []58 for arg in arg_list:59 if arg.startswith("@"):60 exp_arg_list += read_arguments_from_file(arg[1:])61 else:62 exp_arg_list.append(arg)63 return exp_arg_list64 65 66def utc_lit_plugin(result, test, commands):67 testname = test.getFilePath()68 if not testname:69 return None70 71 script_name = os.path.abspath(__file__)72 utc_search_path = os.path.join(os.path.dirname(script_name), os.path.pardir)73 74 with open(testname, "r") as f:75 header = f.readline().strip()76 77 m = RE_ASSERTIONS.search(header)78 if m is None:79 return None80 81 utc_name = m.group(1)82 utc_tool = find_utc_tool([utc_search_path], utc_name)83 if not utc_tool:84 return f"update-utc-tests: {utc_name} not found"85 86 return_code, stdout, stderr = run_utc_tool(87 utc_name, utc_tool, testname, test.config.environment88 )89 90 stderr = stderr.decode(errors="replace")91 if return_code != 0:92 if stderr:93 return f"update-utc-tests: {utc_name} exited with return code {return_code}\n{stderr.rstrip()}"94 return f"update-utc-tests: {utc_name} exited with return code {return_code}"95 96 stdout = stdout.decode(errors="replace")97 if stdout:98 return f"update-utc-tests: updated {testname}\n{stdout.rstrip()}"99 return f"update-utc-tests: updated {testname}"100 101 102def main():103 from argparse import RawTextHelpFormatter104 105 parser = argparse.ArgumentParser(106 description=__doc__, formatter_class=RawTextHelpFormatter107 )108 parser.add_argument(109 "--jobs",110 "-j",111 default=1,112 type=int,113 help="Run the given number of jobs in parallel",114 )115 parser.add_argument(116 "--utc-dir",117 nargs="*",118 help="Additional directories to scan for update_*_test_checks scripts",119 )120 parser.add_argument(121 "--path",122 help="""Additional directories to scan for executables invoked by the update_*_test_checks scripts,123separated by the platform path separator""",124 )125 parser.add_argument("tests", nargs="+")126 config = parser.parse_args()127 128 if config.utc_dir:129 utc_search_path = config.utc_dir[:]130 else:131 utc_search_path = []132 script_name = os.path.abspath(__file__)133 utc_search_path.append(os.path.join(os.path.dirname(script_name), os.path.pardir))134 135 local_env = os.environ.copy()136 if config.path:137 local_env["PATH"] = config.path + os.pathsep + local_env["PATH"]138 139 not_autogenerated = []140 utc_tools = {}141 have_error = False142 143 tests = expand_listfile_args(config.tests)144 145 with ThreadPoolExecutor(max_workers=config.jobs) as executor:146 jobs = []147 148 for testname in tests:149 with open(testname, "r") as f:150 header = f.readline().strip()151 m = RE_ASSERTIONS.search(header)152 if m is None:153 not_autogenerated.append(testname)154 continue155 156 utc_name = m.group(1)157 if utc_name not in utc_tools:158 utc_tools[utc_name] = find_utc_tool(utc_search_path, utc_name)159 if not utc_tools[utc_name]:160 print(161 f"{utc_name}: not found (used in {testname})",162 file=sys.stderr,163 )164 have_error = True165 continue166 167 future = executor.submit(168 run_utc_tool, utc_name, utc_tools[utc_name], testname, local_env169 )170 jobs.append((testname, future))171 172 for testname, future in jobs:173 return_code, stdout, stderr = future.result()174 175 print(f"Update {testname}")176 stdout = stdout.decode(errors="replace")177 if stdout:178 print(stdout, end="")179 if not stdout.endswith("\n"):180 print()181 182 stderr = stderr.decode(errors="replace")183 if stderr:184 print(stderr, end="")185 if not stderr.endswith("\n"):186 print()187 if return_code != 0:188 print(f"Return code: {return_code}")189 have_error = True190 191 if have_error:192 sys.exit(1)193 194 if not_autogenerated:195 print("Tests without autogenerated assertions:")196 for testname in not_autogenerated:197 print(f" {testname}")198 199 200if __name__ == "__main__":201 main()202