Witch of Git - ivy/blob - tools/lit.py

   1 #!/usr/bin/env python3
   2 import argparse
   3 import os
   4 import shlex
   5 import subprocess
   6 import sys
   7 from pathlib import Path
   8
   9 from fmt import green, red, yellow
  10
  11 ROOT = Path(__file__).parent.parent
  12 TOOLS = ROOT / "tools"
  13 TARGET_RELEASE = ROOT / "target" / "release"
  14
  15
  16 def relative(path):
  17     return str(Path(path).relative_to(Path.cwd()))
  18
  19
  20 def discover(roots):
  21     def inspect(path):
  22         with open(path, "r") as f:
  23             run_lines = [line for line in f if "RUN:" in line]
  24         if not run_lines:
  25             return
  26         run_lines = [line.split("RUN:")[1] for line in run_lines]
  27         yield (path, {"RUN": run_lines})
  28
  29     for root in roots:
  30         root = Path(root).resolve()
  31         if root.is_file():
  32             yield from inspect(root)
  33             continue
  34
  35         for root, dirs, files in os.walk(root):
  36             root = Path(root)
  37             for fname in files:
  38                 path = root / fname
  39                 yield from inspect(path)
  40
  41
  42 def split_seq(seq, split):
  43     chunk = []
  44     for item in seq:
  45         if item == split:
  46             yield chunk
  47             chunk = []
  48         else:
  49             chunk.append(item)
  50     yield chunk
  51
  52
  53 def run_test(source, pipeline, verbose=False, PATH=None):
  54     s = shlex.shlex(pipeline, posix=True, punctuation_chars=True)
  55     s.whitespace_split = True
  56     pipeline = list(split_seq(s, "|"))
  57     env = {}
  58     if PATH:
  59         env["PATH"] = PATH
  60
  61     print(f"{relative(source):<72}", end="", flush=True)
  62     next_stdin = subprocess.PIPE
  63     first_process = None
  64     processes = []
  65     stderr_r, stderr = os.pipe()
  66     for command in reversed(pipeline):
  67         command = [word.replace("%s", str(source.resolve())) for word in command]
  68         process = subprocess.Popen(
  69             command, stdin=subprocess.PIPE, stderr=stderr, stdout=next_stdin, env=env
  70         )
  71         next_stdin = process.stdin
  72         processes.append(process)
  73     processes.reverse()
  74     processes[0].communicate()
  75     any_failed = False
  76     for process in processes:
  77         process.stdin.close()
  78         process.wait()
  79         any_failed |= process.returncode != 0
  80     os.close(stderr)
  81     if any_failed:
  82         print(red("FAIL"))
  83         if verbose:
  84             print(yellow("stdout:"))
  85             for line in processes[-1].stdout:
  86                 sys.stdout.buffer.write(line)
  87             print()
  88             print(yellow("stderr:"))
  89             while chunk := os.read(stderr_r, 1024):
  90                 sys.stdout.buffer.write(chunk)
  91             print()
  92         return False
  93     else:
  94         print(green("PASS"))
  95         return True
  96
  97
  98 def report(tests):
  99     passes = [path for path, status in tests.items() if status == "PASS"]
 100     failures = [path for path, status in tests.items() if status == "FAIL"]
 101     print(
 102         f"""
 103 PASS: {len(passes)}
 104 FAIL: {len(failures)}
 105 """
 106     )
 107     if len(failures):
 108         print(yellow("Failures:"))
 109         print()
 110         for failure in failures:
 111             print(f"    {relative(failure)}")
 112         print(red("\nFAILED\n"))
 113         return 1
 114     else:
 115         print(green("PASSED\n"))
 116         return 0
 117
 118
 119 def main():
 120     parser = argparse.ArgumentParser()
 121     parser.add_argument("tests", nargs="+")
 122     parser.add_argument("-v", "--verbose", action="store_true")
 123     args = parser.parse_args()
 124     tests = {}
 125     PATH = ":".join([os.getenv("PATH"), str(TOOLS), str(TARGET_RELEASE)])
 126     for path, directives in discover(args.tests):
 127         tests[path] = "FAIL"
 128         for run_line in directives["RUN"]:
 129             if run_test(path, run_line, args.verbose, PATH):
 130                 tests[path] = "PASS"
 131     status = report(tests)
 132     sys.exit(status)
 133
 134
 135 if __name__ == "__main__":
 136     main()