3 """Dispatch to update_*_test_checks.py scripts automatically in bulk
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.
10 from __future__
import print_function
17 from concurrent
.futures
import ThreadPoolExecutor
19 RE_ASSERTIONS
= re
.compile(
20 r
"NOTE: Assertions have been autogenerated by ([^\s]+)( UTC_ARGS:.*)?$"
24 def find_utc_tool(search_path
, utc_name
):
26 Return the path to the given UTC tool in the search path, or None if not
29 for path
in search_path
:
30 candidate
= os
.path
.join(path
, utc_name
)
31 if os
.path
.isfile(candidate
):
36 def run_utc_tool(utc_name
, utc_tool
, testname
):
37 result
= subprocess
.run(
38 [utc_tool
, testname
], stdout
=subprocess
.PIPE
, stderr
=subprocess
.PIPE
40 return (result
.returncode
, result
.stdout
, result
.stderr
)
44 from argparse
import RawTextHelpFormatter
46 parser
= argparse
.ArgumentParser(
47 description
=__doc__
, formatter_class
=RawTextHelpFormatter
54 help="Run the given number of jobs in parallel",
59 help="Additional directories to scan for update_*_test_checks scripts",
61 parser
.add_argument("tests", nargs
="+")
62 config
= parser
.parse_args()
65 utc_search_path
= config
.utc_dir
[:]
68 script_name
= os
.path
.abspath(__file__
)
69 utc_search_path
.append(os
.path
.join(os
.path
.dirname(script_name
), os
.path
.pardir
))
71 not_autogenerated
= []
75 with
ThreadPoolExecutor(max_workers
=config
.jobs
) as executor
:
78 for testname
in config
.tests
:
79 with
open(testname
, "r") as f
:
80 header
= f
.readline().strip()
81 m
= RE_ASSERTIONS
.search(header
)
83 not_autogenerated
.append(testname
)
87 if utc_name
not in utc_tools
:
88 utc_tools
[utc_name
] = find_utc_tool(utc_search_path
, utc_name
)
89 if not utc_tools
[utc_name
]:
91 f
"{utc_name}: not found (used in {testname})",
97 future
= executor
.submit(
98 run_utc_tool
, utc_name
, utc_tools
[utc_name
], testname
100 jobs
.append((testname
, future
))
102 for testname
, future
in jobs
:
103 return_code
, stdout
, stderr
= future
.result()
105 print(f
"Update {testname}")
106 stdout
= stdout
.decode(errors
="replace")
108 print(stdout
, end
="")
109 if not stdout
.endswith("\n"):
112 stderr
= stderr
.decode(errors
="replace")
114 print(stderr
, end
="")
115 if not stderr
.endswith("\n"):
118 print(f
"Return code: {return_code}")
124 if not_autogenerated
:
125 print("Tests without autogenerated assertions:")
126 for testname
in not_autogenerated
:
127 print(f
" {testname}")
130 if __name__
== "__main__":