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

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