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