hphp/hack/test/verify.py

   1 #!/usr/bin/env python3
   2 # pyre-strict
   3
   4 import argparse
   5 import difflib
   6 import os
   7 import os.path
   8 import re
   9 import shlex
  10 import subprocess
  11 import sys
  12 from concurrent.futures import ThreadPoolExecutor
  13 from dataclasses import dataclass
  14 from typing import Callable, Dict, List, Optional, Tuple
  15
  16 from hphp.hack.test.parse_errors import Error, parse_errors, sprint_errors
  17
  18
  19 max_workers = 48
  20 verbose = False
  21 dump_on_failure = False
  22 batch_size = 500
  23
  24
  25 @dataclass
  26 class TestCase:
  27     file_path: str
  28     input: Optional[str]
  29     expected: str
  30
  31
  32 @dataclass
  33 class Result:
  34     test_case: TestCase
  35     output: str
  36     is_failure: bool
  37
  38
  39 """
  40 Per-test flags passed to test executable. Expected to be in a file with
  41 same name as test, but with .flags extension.
  42 """
  43
  44
  45 def compare_errors_by_line_no(
  46     errors_exp: List[Error], errors_out: List[Error]
  47 ) -> Tuple[List[Error], List[Error]]:
  48     i_out = 0
  49     i_exp = 0
  50     len_out = len(errors_out)
  51     len_exp = len(errors_exp)
  52     errors_in_out_not_in_exp = []
  53     errors_in_exp_not_in_out = []
  54     while i_out < len_out and i_exp < len_exp:
  55         err_out = errors_out[i_out]
  56         err_exp = errors_exp[i_exp]
  57         l_out = err_out.message.position.line
  58         l_exp = err_exp.message.position.line
  59         if l_out < l_exp:
  60             errors_in_out_not_in_exp.append(err_out)
  61             i_out += 1
  62         elif l_exp < l_out:
  63             errors_in_exp_not_in_out.append(err_exp)
  64             i_exp += 1
  65         else:
  66             i_out += 1
  67             i_exp += 1
  68     if i_out >= len_out:
  69         for i in range(i_exp, len_exp):
  70             errors_in_exp_not_in_out.append(errors_exp[i])
  71     elif i_exp >= len_exp:
  72         for i in range(i_out, len_out):
  73             errors_in_out_not_in_exp.append(errors_out[i])
  74     return (errors_in_exp_not_in_out, errors_in_out_not_in_exp)
  75
  76
  77 def compare_output_files_error_lines_only(
  78     file_out: str, file_exp: str
  79 ) -> Tuple[bool, str]:
  80     out = ""
  81     failed = False
  82     try:
  83         errors_out = parse_errors(file_out)
  84         errors_exp = parse_errors(file_exp)
  85         (
  86             errors_in_exp_not_in_out,
  87             errors_in_out_not_in_exp,
  88         ) = compare_errors_by_line_no(errors_out=errors_out, errors_exp=errors_exp)
  89
  90         failed = bool(errors_in_exp_not_in_out) or bool(errors_in_out_not_in_exp)
  91         if errors_in_exp_not_in_out:
  92             out += f"""\033[93mExpected errors which were not produced:\033[0m
  93 {sprint_errors(errors_in_exp_not_in_out)}
  94 """
  95         if errors_in_out_not_in_exp:
  96             out += f"""\033[93mProduced errors which were not expected:\033[0m
  97 {sprint_errors(errors_in_out_not_in_exp)}
  98 """
  99     except IOError as e:
 100         out = f"Warning: {e}"
 101     return (failed, out)
 102
 103
 104 def check_output_error_lines_only(
 105     test: str, out_ext: str = ".out", exp_ext: str = ".exp"
 106 ) -> Tuple[bool, str]:
 107     file_out = test + out_ext
 108     file_exp = test + exp_ext
 109     return compare_output_files_error_lines_only(file_out=file_out, file_exp=file_exp)
 110
 111
 112 def get_test_flags(path: str) -> List[str]:
 113     prefix, _ext = os.path.splitext(path)
 114     path = prefix + ".flags"
 115
 116     if not os.path.isfile(path):
 117         return []
 118     with open(path) as file:
 119         return shlex.split(file.read().strip())
 120
 121
 122 def check_output(
 123     case: TestCase,
 124     out_extension: str,
 125     default_expect_regex: Optional[str],
 126     ignore_error_text: bool,
 127     only_compare_error_lines: bool,
 128 ) -> Result:
 129     if only_compare_error_lines:
 130         (failed, out) = check_output_error_lines_only(case.file_path)
 131         return Result(test_case=case, output=out, is_failure=failed)
 132     else:
 133         out_path = case.file_path + out_extension
 134         try:
 135             with open(out_path, "r") as f:
 136                 output: str = f.read()
 137         except FileNotFoundError:
 138             out_path = os.path.realpath(out_path)
 139             output = "Output file " + out_path + " was not found!"
 140         return check_result(case, default_expect_regex, ignore_error_text, output)
 141
 142
 143 def debug_cmd(cwd: str, cmd: List[str]) -> None:
 144     if verbose:
 145         print("From directory", os.path.realpath(cwd))
 146         print("Executing", " ".join(cmd))
 147         print()
 148
 149
 150 def run_batch_tests(
 151     test_cases: List[TestCase],
 152     program: str,
 153     default_expect_regex: Optional[str],
 154     ignore_error_text: bool,
 155     no_stderr: bool,
 156     force_color: bool,
 157     mode_flag: List[str],
 158     get_flags: Callable[[str], List[str]],
 159     out_extension: str,
 160     only_compare_error_lines: bool = False,
 161 ) -> List[Result]:
 162     """
 163     Run the program with batches of files and return a list of results.
 164     """
 165     # Each directory needs to be in a separate batch because flags are different
 166     # for each directory.
 167     # Compile a list of directories to test cases, and then
 168     dirs_to_files: Dict[str, List[TestCase]] = {}
 169     for case in test_cases:
 170         test_dir = os.path.dirname(case.file_path)
 171         dirs_to_files.setdefault(test_dir, []).append(case)
 172
 173     # run a list of test cases.
 174     # The contract here is that the program will write to
 175     # filename.out_extension for each file, and we read that
 176     # for the output.
 177     def run(test_cases: List[TestCase]) -> List[Result]:
 178         if not test_cases:
 179             raise AssertionError()
 180         first_test = test_cases[0]
 181         test_dir = os.path.dirname(first_test.file_path)
 182         flags = get_flags(test_dir)
 183         test_flags = get_test_flags(first_test.file_path)
 184         cmd = [program]
 185         cmd += mode_flag
 186         cmd += ["--batch-files", "--out-extension", out_extension]
 187         cmd += flags + test_flags
 188         cmd += [os.path.basename(case.file_path) for case in test_cases]
 189         debug_cmd(test_dir, cmd)
 190         env = os.environ.copy()
 191         env["FORCE_ERROR_COLOR"] = "true" if force_color else "false"
 192         try:
 193             return_code = subprocess.call(
 194                 cmd,
 195                 stderr=None if no_stderr else subprocess.STDOUT,
 196                 cwd=test_dir,
 197                 universal_newlines=True,
 198                 env=env,
 199             )
 200         except subprocess.CalledProcessError:
 201             # we don't care about nonzero exit codes... for instance, type
 202             # errors cause hh_single_type_check to produce them
 203             return_code = None
 204         if return_code == -11:
 205             print(
 206                 "Segmentation fault while running the following command "
 207                 + "from directory "
 208                 + os.path.realpath(test_dir)
 209             )
 210             print(" ".join(cmd))
 211             print()
 212         results = []
 213         for case in test_cases:
 214             result = check_output(
 215                 case,
 216                 out_extension=out_extension,
 217                 default_expect_regex=default_expect_regex,
 218                 ignore_error_text=ignore_error_text,
 219                 only_compare_error_lines=only_compare_error_lines,
 220             )
 221             results.append(result)
 222         return results
 223
 224     # Create a list of batched cases.
 225     all_batched_cases: List[List[TestCase]] = []
 226
 227     # For each directory, we split all the test cases
 228     # into chunks of batch_size. Then each of these lists
 229     # is a separate job for each thread in the threadpool.
 230     for cases in dirs_to_files.values():
 231         batched_cases: List[List[TestCase]] = [
 232             cases[i : i + batch_size] for i in range(0, len(cases), batch_size)
 233         ]
 234         all_batched_cases += batched_cases
 235
 236     executor = ThreadPoolExecutor(max_workers=max_workers)
 237     futures = [executor.submit(run, test_batch) for test_batch in all_batched_cases]
 238
 239     results = [future.result() for future in futures]
 240     # Flatten the list
 241     return [item for sublist in results for item in sublist]
 242
 243
 244 def run_test_program(
 245     test_cases: List[TestCase],
 246     program: str,
 247     default_expect_regex: Optional[str],
 248     ignore_error_text: bool,
 249     no_stderr: bool,
 250     force_color: bool,
 251     mode_flag: List[str],
 252     get_flags: Callable[[str], List[str]],
 253     timeout: Optional[float] = None,
 254 ) -> List[Result]:
 255
 256     """
 257     Run the program and return a list of results.
 258     """
 259
 260     def run(test_case: TestCase) -> Result:
 261         test_dir, test_name = os.path.split(test_case.file_path)
 262         flags = get_flags(test_dir)
 263         test_flags = get_test_flags(test_case.file_path)
 264         cmd = [program]
 265         cmd += mode_flag
 266         if test_case.input is None:
 267             cmd.append(test_name)
 268         cmd += flags + test_flags
 269         debug_cmd(test_dir, cmd)
 270         env = os.environ.copy()
 271         env["FORCE_ERROR_COLOR"] = "true" if force_color else "false"
 272         try:
 273             output = subprocess.check_output(
 274                 cmd,
 275                 stderr=None if no_stderr else subprocess.STDOUT,
 276                 cwd=test_dir,
 277                 universal_newlines=True,
 278                 # pyre-ignore
 279                 input=test_case.input,
 280                 timeout=timeout,
 281                 errors="replace",
 282                 env=env,
 283             )
 284         except subprocess.TimeoutExpired as e:
 285             output = "Timed out. " + str(e.output)
 286         except subprocess.CalledProcessError as e:
 287             # we don't care about nonzero exit codes... for instance, type
 288             # errors cause hh_single_type_check to produce them
 289             output = str(e.output)
 290         return check_result(test_case, default_expect_regex, ignore_error_text, output)
 291
 292     executor = ThreadPoolExecutor(max_workers=max_workers)
 293     futures = [executor.submit(run, test_case) for test_case in test_cases]
 294
 295     return [future.result() for future in futures]
 296
 297
 298 def filter_ocaml_stacktrace(text: str) -> str:
 299     """take a string and remove all the lines that look like
 300     they're part of an OCaml stacktrace"""
 301     assert isinstance(text, str)
 302     it = text.splitlines()
 303     out = []
 304     for x in it:
 305         drop_line = x.lstrip().startswith("Called") or x.lstrip().startswith("Raised")
 306         if drop_line:
 307             pass
 308         else:
 309             out.append(x)
 310     return "\n".join(out)
 311
 312
 313 def filter_version_field(text: str) -> str:
 314     """given a string, remove the part that looks like the schema version"""
 315     assert isinstance(text, str)
 316     return re.sub(
 317         r'"version":"\d{4}-\d{2}-\d{2}-\d{4}"', r'"version":"sanitised"', text, count=1
 318     )
 319
 320
 321 def compare_expected(expected: str, out: str) -> bool:
 322     if expected == "No errors\n" or out == "No errors\n":
 323         return expected == out
 324     else:
 325         return True
 326
 327
 328 # Strip leading and trailing whitespace from every line
 329 def strip_lines(text: str) -> str:
 330     return "\n".join(line.strip() for line in text.splitlines())
 331
 332
 333 def check_result(
 334     test_case: TestCase,
 335     default_expect_regex: Optional[str],
 336     ignore_error_messages: bool,
 337     out: str,
 338 ) -> Result:
 339     """
 340     Check that the output of the test in :out corresponds to the expected
 341     output, or if a :default_expect_regex is provided,
 342     check that the output in :out contains the provided regex.
 343     """
 344     expected = filter_version_field(strip_lines(test_case.expected))
 345     normalized_out = filter_version_field(strip_lines(out))
 346     is_ok = (
 347         expected == normalized_out
 348         or (ignore_error_messages and compare_expected(expected, normalized_out))
 349         or expected == filter_ocaml_stacktrace(normalized_out)
 350         or (
 351             default_expect_regex is not None
 352             and re.search(default_expect_regex, normalized_out) is not None
 353         )
 354     )
 355
 356     return Result(test_case=test_case, output=out, is_failure=not is_ok)
 357
 358
 359 def record_results(results: List[Result], out_ext: str) -> None:
 360     for result in results:
 361         outfile = result.test_case.file_path + out_ext
 362         with open(outfile, "wb") as f:
 363             f.write(bytes(result.output, "UTF-8"))
 364
 365
 366 def find_in_ancestors_rec(dir: str, path: str) -> str:
 367     if path == "" or os.path.dirname(path) == path:
 368         raise Exception("Could not find directory %s in ancestors." % dir)
 369     if os.path.basename(path) == dir:
 370         return path
 371     return find_in_ancestors_rec(dir, os.path.dirname(path))
 372
 373
 374 def find_in_ancestors(dir: str, path: str) -> str:
 375     try:
 376         return find_in_ancestors_rec(dir, path)
 377     except Exception:
 378         raise Exception("Could not find directory %s in ancestors of %s." % (dir, path))
 379
 380
 381 def get_exp_out_dirs(test_file: str) -> Tuple[str, str]:
 382     if (
 383         os.environ.get("HACK_BUILD_ROOT") is not None
 384         and os.environ.get("HACK_SOURCE_ROOT") is not None
 385     ):
 386         exp_dir = os.environ["HACK_SOURCE_ROOT"]
 387         out_dir = os.environ["HACK_BUILD_ROOT"]
 388     else:
 389         fbcode = find_in_ancestors("fbcode", test_file)
 390         exp_dir = os.path.join(fbcode, "hphp", "hack")
 391         out_dir = os.path.dirname(find_in_ancestors("test", test_file))
 392     return exp_dir, out_dir
 393
 394
 395 def report_failures(
 396     total: int,
 397     failures: List[Result],
 398     out_extension: str,
 399     expect_extension: str,
 400     fallback_expect_extension: Optional[str],
 401     no_copy: bool = False,
 402     only_compare_error_lines: bool = False,
 403 ) -> None:
 404     if only_compare_error_lines:
 405         for failure in failures:
 406             print(f"\033[95m{failure.test_case.file_path}\033[0m")
 407             print(failure.output)
 408             print()
 409     elif failures != []:
 410         record_results(failures, out_extension)
 411         if dump_on_failure:
 412             dump_failures(failures)
 413         fnames = [failure.test_case.file_path for failure in failures]
 414         print("To review the failures, use the following command: ")
 415         fallback_expect_ext_var = ""
 416         if fallback_expect_extension is not None:
 417             fallback_expect_ext_var = "FALLBACK_EXP_EXT=%s " % fallback_expect_extension
 418         first_test_file = os.path.realpath(failures[0].test_case.file_path)
 419         out_dir: str  # for Pyre
 420         (exp_dir, out_dir) = get_exp_out_dirs(first_test_file)
 421         output_dir_var = "SOURCE_ROOT=%s OUTPUT_ROOT=%s " % (exp_dir, out_dir)
 422
 423         # Get a full path to 'review.sh' so this command be run
 424         # regardless of your current directory.
 425         review_script = os.path.join(
 426             os.path.dirname(os.path.realpath(__file__)), "review.sh"
 427         )
 428         if not os.path.isfile(review_script):
 429             review_script = "./hphp/hack/test/review.sh"
 430
 431         def fname_map_var(f: str) -> str:
 432             return "hphp/hack/" + os.path.relpath(f, out_dir)
 433
 434         print(
 435             "OUT_EXT=%s EXP_EXT=%s %s%sNO_COPY=%s %s %s"
 436             % (
 437                 out_extension,
 438                 expect_extension,
 439                 fallback_expect_ext_var,
 440                 output_dir_var,
 441                 "true" if no_copy else "false",
 442                 review_script,
 443                 " ".join(map(fname_map_var, fnames)),
 444             )
 445         )
 446
 447
 448 def dump_failures(failures: List[Result]) -> None:
 449     for f in failures:
 450         expected = f.test_case.expected
 451         actual = f.output
 452         diff = difflib.ndiff(expected.splitlines(True), actual.splitlines(True))
 453         print("Details for the failed test %s:" % f.test_case.file_path)
 454         print("\n>>>>>  Expected output  >>>>>>\n")
 455         print(expected)
 456         print("\n=====   Actual output   ======\n")
 457         print(actual)
 458         print("\n<<<<< End Actual output <<<<<<<\n")
 459         print("\n>>>>>       Diff        >>>>>>>\n")
 460         print("".join(diff))
 461         print("\n<<<<<     End Diff      <<<<<<<\n")
 462
 463
 464 def get_hh_flags(test_dir: str) -> List[str]:
 465     path = os.path.join(test_dir, "HH_FLAGS")
 466     if not os.path.isfile(path):
 467         if verbose:
 468             print("No HH_FLAGS file found")
 469         return []
 470     with open(path) as f:
 471         return shlex.split(f.read())
 472
 473
 474 def files_with_ext(files: List[str], ext: str) -> List[str]:
 475     """
 476     Returns the set of filenames in :files that end in :ext
 477     """
 478     filtered_files: List[str] = []
 479     for file in files:
 480         prefix, suffix = os.path.splitext(file)
 481         if suffix == ext:
 482             filtered_files.append(prefix)
 483     return filtered_files
 484
 485
 486 def list_test_files(root: str, disabled_ext: str, test_ext: str) -> List[str]:
 487     if os.path.isfile(root):
 488         if root.endswith(test_ext):
 489             return [root]
 490         else:
 491             return []
 492     elif os.path.isdir(root):
 493         result: List[str] = []
 494         children = os.listdir(root)
 495         disabled = files_with_ext(children, disabled_ext)
 496         for child in children:
 497             if child != "disabled" and child not in disabled:
 498                 result.extend(
 499                     list_test_files(os.path.join(root, child), disabled_ext, test_ext)
 500                 )
 501         return result
 502     elif os.path.islink(root):
 503         # Some editors create broken symlinks as part of their locking scheme,
 504         # so ignore those.
 505         return []
 506     else:
 507         raise Exception("Could not find test file or directory at %s" % root)
 508
 509
 510 def get_content_(file_path: str, ext: str) -> str:
 511     with open(file_path + ext, "r") as fexp:
 512         return fexp.read()
 513
 514
 515 def get_content(
 516     file_path: str, ext: str = "", fallback_ext: Optional[str] = None
 517 ) -> str:
 518     try:
 519         return get_content_(file_path, ext)
 520     except FileNotFoundError:
 521         if fallback_ext is not None:
 522             try:
 523                 return get_content_(file_path, fallback_ext)
 524             except FileNotFoundError:
 525                 return ""
 526         else:
 527             return ""
 528
 529
 530 def run_tests(
 531     files: List[str],
 532     expected_extension: str,
 533     fallback_expect_extension: Optional[str],
 534     out_extension: str,
 535     use_stdin: str,
 536     program: str,
 537     default_expect_regex: Optional[str],
 538     batch_mode: str,
 539     ignore_error_text: bool,
 540     no_stderr: bool,
 541     force_color: bool,
 542     mode_flag: List[str],
 543     get_flags: Callable[[str], List[str]],
 544     timeout: Optional[float] = None,
 545     only_compare_error_lines: bool = False,
 546 ) -> List[Result]:
 547
 548     # for each file, create a test case
 549     test_cases = [
 550         TestCase(
 551             file_path=file,
 552             expected=get_content(file, expected_extension, fallback_expect_extension),
 553             input=get_content(file) if use_stdin else None,
 554         )
 555         for file in files
 556     ]
 557     if batch_mode:
 558         results = run_batch_tests(
 559             test_cases,
 560             program,
 561             default_expect_regex,
 562             ignore_error_text,
 563             no_stderr,
 564             force_color,
 565             mode_flag,
 566             get_flags,
 567             out_extension,
 568             only_compare_error_lines,
 569         )
 570     else:
 571         results = run_test_program(
 572             test_cases,
 573             program,
 574             default_expect_regex,
 575             ignore_error_text,
 576             no_stderr,
 577             force_color,
 578             mode_flag,
 579             get_flags,
 580             timeout=timeout,
 581         )
 582
 583     failures = [result for result in results if result.is_failure]
 584
 585     num_results = len(results)
 586     if failures == []:
 587         print(
 588             "All tests in the suite passed! "
 589             "The number of tests that ran: %d\n" % num_results
 590         )
 591     else:
 592         print("The number of tests that failed: %d/%d\n" % (len(failures), num_results))
 593         report_failures(
 594             num_results,
 595             failures,
 596             args.out_extension,
 597             args.expect_extension,
 598             args.fallback_expect_extension,
 599             only_compare_error_lines=only_compare_error_lines,
 600         )
 601         sys.exit(1)  # this exit code fails the suite and lets Buck know
 602
 603     return results
 604
 605
 606 def run_idempotence_tests(
 607     results: List[Result],
 608     expected_extension: str,
 609     fallback_expect_extension: Optional[str],
 610     out_extension: str,
 611     program: str,
 612     default_expect_regex: Optional[str],
 613     mode_flag: List[str],
 614     get_flags: Callable[[str], List[str]],
 615 ) -> None:
 616     idempotence_test_cases = [
 617         TestCase(
 618             file_path=result.test_case.file_path,
 619             expected=result.test_case.expected,
 620             input=result.output,
 621         )
 622         for result in results
 623     ]
 624
 625     idempotence_results = run_test_program(
 626         idempotence_test_cases,
 627         program,
 628         default_expect_regex,
 629         False,
 630         False,
 631         False,
 632         mode_flag,
 633         get_flags,
 634     )
 635
 636     num_idempotence_results = len(idempotence_results)
 637
 638     idempotence_failures = [
 639         result for result in idempotence_results if result.is_failure
 640     ]
 641
 642     if idempotence_failures == []:
 643         print(
 644             "All idempotence tests in the suite passed! The number of "
 645             "idempotence tests that ran: %d\n" % num_idempotence_results
 646         )
 647     else:
 648         print(
 649             "The number of idempotence tests that failed: %d/%d\n"
 650             % (len(idempotence_failures), num_idempotence_results)
 651         )
 652         report_failures(
 653             num_idempotence_results,
 654             idempotence_failures,
 655             out_extension + out_extension,  # e.g., *.out.out
 656             expected_extension,
 657             fallback_expect_extension,
 658             no_copy=True,
 659         )
 660         sys.exit(1)  # this exit code fails the suite and lets Buck know
 661
 662
 663 def get_flags_cache(args_flags: List[str]) -> Callable[[str], List[str]]:
 664     flags_cache: Dict[str, List[str]] = {}
 665
 666     def get_flags(test_dir: str) -> List[str]:
 667         if test_dir not in flags_cache:
 668             flags_cache[test_dir] = get_hh_flags(test_dir)
 669         flags = flags_cache[test_dir]
 670         if args_flags is not None:
 671             flags = flags + args_flags
 672         return flags
 673
 674     return get_flags
 675
 676
 677 if __name__ == "__main__":
 678     parser = argparse.ArgumentParser()
 679     parser.add_argument("test_path", help="A file or a directory. ")
 680     parser.add_argument("--program", type=os.path.abspath)
 681     parser.add_argument("--out-extension", type=str, default=".out")
 682     parser.add_argument("--expect-extension", type=str, default=".exp")
 683     parser.add_argument("--fallback-expect-extension", type=str)
 684     parser.add_argument("--default-expect-regex", type=str)
 685     parser.add_argument("--in-extension", type=str, default=".php")
 686     parser.add_argument("--disabled-extension", type=str, default=".no_typecheck")
 687     parser.add_argument("--verbose", action="store_true")
 688     parser.add_argument(
 689         "--idempotence",
 690         action="store_true",
 691         help="Verify that the output passed to the program "
 692         "as input results in the same output.",
 693     )
 694     parser.add_argument("--max-workers", type=int, default="48")
 695     parser.add_argument(
 696         "--diff",
 697         action="store_true",
 698         help="On test failure, show the content of " "the files and a diff",
 699     )
 700     parser.add_argument("--mode-flag", type=str)
 701     parser.add_argument("--flags", nargs=argparse.REMAINDER)
 702     parser.add_argument(
 703         "--stdin", action="store_true", help="Pass test input file via stdin"
 704     )
 705     parser.add_argument(
 706         "--no-stderr",
 707         action="store_true",
 708         help="Do not include stderr output in the output file",
 709     )
 710     parser.add_argument(
 711         "--batch", action="store_true", help="Run tests in batches to the test program"
 712     )
 713     parser.add_argument(
 714         "--ignore-error-text",
 715         action="store_true",
 716         help="Do not compare error text when verifying output",
 717     )
 718     parser.add_argument(
 719         "--only-compare-error-lines",
 720         action="store_true",
 721         help="Does not care about exact expected error message, "
 722         "but only compare the error line numbers.",
 723     )
 724     parser.add_argument(
 725         "--timeout",
 726         type=int,
 727         help="Timeout in seconds for each test, in non-batch mode.",
 728     )
 729     parser.add_argument(
 730         "--force-color",
 731         action="store_true",
 732         help="Set the FORCE_ERROR_COLOR environment variable, "
 733         "which causes the test output to retain terminal escape codes.",
 734     )
 735     parser.epilog = (
 736         "%s looks for a file named HH_FLAGS in the same directory"
 737         " as the test files it is executing. If found, the "
 738         "contents will be passed as arguments to "
 739         "<program> in addition to any arguments "
 740         "specified by --flags" % parser.prog
 741     )
 742     args: argparse.Namespace = parser.parse_args()
 743
 744     max_workers = args.max_workers
 745     verbose = args.verbose
 746
 747     dump_on_failure = args.diff
 748     if os.getenv("SANDCASTLE") is not None:
 749         dump_on_failure = True
 750
 751     if not os.path.isfile(args.program):
 752         raise Exception("Could not find program at %s" % args.program)
 753
 754     files: List[str] = list_test_files(
 755         args.test_path, args.disabled_extension, args.in_extension
 756     )
 757
 758     if len(files) == 0:
 759         raise Exception("Could not find any files to test in " + args.test_path)
 760
 761     mode_flag: List[str] = [] if args.mode_flag is None else [args.mode_flag]
 762     get_flags: Callable[[str], List[str]] = get_flags_cache(args.flags)
 763
 764     results: List[Result] = run_tests(
 765         files,
 766         args.expect_extension,
 767         args.fallback_expect_extension,
 768         args.out_extension,
 769         args.stdin,
 770         args.program,
 771         args.default_expect_regex,
 772         args.batch,
 773         args.ignore_error_text,
 774         args.no_stderr,
 775         args.force_color,
 776         mode_flag,
 777         get_flags,
 778         timeout=args.timeout,
 779         only_compare_error_lines=args.only_compare_error_lines,
 780     )
 781
 782     # Doesn't make sense to check failures for idempotence
 783     successes: List[Result] = [result for result in results if not result.is_failure]
 784
 785     if args.idempotence and successes:
 786         run_idempotence_tests(
 787             successes,
 788             args.expect_extension,
 789             args.fallback_expect_extension,
 790             args.out_extension,
 791             args.program,
 792             args.default_expect_regex,
 793             mode_flag,
 794             get_flags,
 795         )