Commit eebee103 authored by Tom Niget's avatar Tom Niget

Add fancy parallel test runner

parent 3c55caf6
ALT_RUNNER='clang++-16 -O3 -Wno-deprecated-declarations -Wno-return-type -Wno-unused-result -std=c++20 $(python3 __main__.py --cpp-flags) -o {name_bin} {name_cpp_posix} && {run_file} && {name_bin}'
\ No newline at end of file
ALT_RUNNER='(clang++-16 -O3 -Wno-deprecated-declarations -Wno-return-type -Wno-unused-result -std=c++20 $(python3 __main__.py --cpp-flags) -o {name_bin} {name_cpp_posix} || exit 2) && ({run_file} || exit 0) && ({name_bin} || exit 3)'
\ No newline at end of file
# coding: utf-8
import argparse
import concurrent.futures
import enum
import logging
import os
import subprocess
#logging.basicConfig(level=logging.DEBUG)
import sys
from datetime import datetime
from os import system, environ
from pathlib import Path
import colorama
import colorful as cf
import signal
from transpiler import transpile
from transpiler.format import format_code
......@@ -11,6 +20,7 @@ from transpiler.format import format_code
# load .env file
from dotenv import load_dotenv
colorama.init()
load_dotenv()
# todo: promise https://lab.nexedi.com/nexedi/slapos.toolbox/blob/master/slapos/promise/plugin/check_socket_listening.py
......@@ -21,43 +31,99 @@ parser = argparse.ArgumentParser()
parser.add_argument("-c", "--compile", action="store_true")
parser.add_argument("-g", "--generate", action="store_true")
parser.add_argument("-o", "--only", nargs="+")
parser.add_argument("-s", "--silent", action="store_true")
parser.add_argument("-w", "--workers", type=int, default=None)
args = parser.parse_args()
class TestStatus(enum.Enum):
SUCCESS = 0
TYPON_ERROR = 1
CLANG_ERROR = 2 # returned from ALT_RUNNER
RUNTIME_ERROR = 3 # returned from ALT_RUNNER
SKIPPED = 4
PYTHON_ERROR = 5
def run_tests():
for path in sorted(Path('tests').glob('*.py')):
if args.only and path.stem not in args.only:
continue
print(path.name)
if path.name.startswith('_'):
print("Skipping")
continue
with open(path, "r", encoding="utf-8") as f:
code = f.read()
execute = "# norun" not in code
compile = "# nocompile" not in code
def ascii(self):
color, msg = {
TestStatus.SUCCESS: (cf.green, "✓"),
TestStatus.TYPON_ERROR: (cf.red, "Typon"),
TestStatus.CLANG_ERROR: (cf.red, "C++"),
TestStatus.RUNTIME_ERROR: (cf.red, "Run"),
TestStatus.SKIPPED: (cf.yellow, "Skipped"),
TestStatus.PYTHON_ERROR: (cf.red, "Python"),
}[self]
return color("{:^7s}".format(msg))
def exec_cmd(cmd, quiet):
stdout = subprocess.DEVNULL if quiet else None
stderr = subprocess.DEVNULL if quiet else None
return subprocess.run(cmd, shell=True, stdout=stdout, stderr=stderr).returncode
def run_test(path, quiet=True):
if quiet:
sys.stdout = open(os.devnull, 'w')
sys.stderr = open(os.devnull, 'w')
if args.only and path.stem not in args.only:
return TestStatus.SKIPPED
print(path.name)
if path.name.startswith('_'):
print("Skipping")
return TestStatus.SKIPPED
with open(path, "r", encoding="utf-8") as f:
code = f.read()
execute = "# norun" not in code
compile = "# nocompile" not in code
try:
res = format_code(transpile(code, path.name, path))
#print(res)
name_cpp = path.with_suffix('.cpp')
with open(name_cpp, "w", encoding="utf-8") as fcpp:
fcpp.write(res)
print(".cpp generated")
if args.compile:
continue
execute_str = "true" if (execute and not args.generate) else "false"
name_bin = path.with_suffix('').as_posix()
commands = [
f'bash -c "export PYTHONPATH=stdlib; if {execute_str}; then python3 ./{path.as_posix()}; fi"',
]
if compile and (alt := environ.get("ALT_RUNNER")):
commands.append(alt.format(name_bin=name_bin, name_cpp_posix=name_cpp.as_posix(), run_file=execute_str))
else:
print("no ALT_RUNNER")
for cmd in commands:
print(cmd)
if system(cmd) != 0:
print(f"Error running command: {cmd}")
break
except:
if not quiet:
raise
return TestStatus.TYPON_ERROR
#print(res)
name_cpp = path.with_suffix('.cpp')
with open(name_cpp, "w", encoding="utf-8") as fcpp:
fcpp.write(res)
print(".cpp generated")
if args.compile:
return TestStatus.SUCCESS
execute_str = "true" if (execute and not args.generate) else "false"
name_bin = path.with_suffix('').as_posix()
if exec_cmd(f'bash -c "export PYTHONPATH=stdlib; if {execute_str}; then python3 ./{path.as_posix()}; fi"', quiet) != 0:
return TestStatus.PYTHON_ERROR
if compile and (alt := environ.get("ALT_RUNNER")):
if (code := exec_cmd(alt.format(name_bin=name_bin, name_cpp_posix=name_cpp.as_posix(), run_file=execute_str), quiet)) != 0:
return TestStatus(code)
else:
print("no ALT_RUNNER")
return TestStatus.SUCCESS
def runner(path):
start = datetime.now()
result = run_test(path)
duration = datetime.now() - start
return path, result, duration
def sigint_handler(signum, frame):
sys.exit(1)
signal.signal(signal.SIGINT, sigint_handler)
def run_tests():
tests = sorted(Path('tests').glob('*.py'))
if args.silent:
pool = concurrent.futures.ProcessPoolExecutor(args.workers)
print("Running", len(tests), "tests as", pool._max_workers, "workers")
start = datetime.now()
with pool as executor:
futures = [executor.submit(runner, path) for path in tests]
for future in concurrent.futures.as_completed(futures):
path, status, duration = future.result()
print(f"[{status.ascii()}] ({duration.total_seconds():2.2f}s) {path.name}")
duration = datetime.now() - start
print("Done in", duration.total_seconds(), "seconds")
else:
for path in tests:
run_test(path, False)
if __name__ == "__main__":
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment