Script to run tests of preCICE one by one

precice-run-tests

#! python

import subprocess
import concurrent.futures
from colorama import Style, Fore
import sys
import pathlib
import tempfile
import os
import collections
import argparse

TEST_TIMEOUT = 20


def get_all_tests():
    path = pathlib.Path("./testprecice")
    assert path.exists() and path.is_file(), "testprecice doesn't exist"
    result = subprocess.run(["./testprecice", "--list_units"],
                            stderr=subprocess.PIPE, stdout=subprocess.PIPE, check=True)
    return result.stdout.decode().splitlines(False)


def filtered_tests(tests, filters):
    if not filters:
        return tests

    return [t
            for t in tests
            if all((filter in t
                    for filter in filters))
            ]


def build_test():
    ret = subprocess.run(["cmake", "--build", ".", "--target", "testprecice", "--parallel", str(os.cpu_count())]).returncode
    if ret != 0:
        sys.exit(ret)


def launch_test(test: str, dir: pathlib.Path, verbose: bool):
    exec = pathlib.Path("./testprecice").absolute()
    level = "all" if verbose else "message"
    return subprocess.run(["mpirun", "-n", "4", "--bind-to", "L3CACHE", exec, f"--run_test={test}", "-l", level], cwd=dir, timeout=TEST_TIMEOUT, stderr=subprocess.PIPE, stdout=subprocess.PIPE)


Result = collections.namedtuple("Result", "state test code out err")
Index = collections.namedtuple("Index", "current total")


def format_index(i: Index) -> str:
    totalLength = str(len(str(i.total)))
    return ("[{:" + totalLength + "}/{}]").format(i.current+1, i.total)


def run_test(test: str, index: Index, temp_dir: pathlib.Path, verbose: bool):
    temp_dir.mkdir(parents=False, exist_ok=True)
    timeout = False
    pre = format_index(index) + " "
    try:
        run = launch_test(test, temp_dir, verbose)
        stdout = run.stdout.decode()
        stderr = run.stderr.decode()
    except subprocess.TimeoutExpired as e:
        print(f"{pre}{Fore.YELLOW + Style.BRIGHT}TIMEOUT{Style.RESET_ALL} {test} ")
        stdout = e.stdout.decode() if e.stdout else "Nothing captured"
        stderr = e.stderr.decode() if e.stderr else "Nothing captured"
        return Result("T", test, -1, stdout, stderr)

    if run.returncode == 0:
        print(f"{pre}{Fore.GREEN + Style.BRIGHT}SUCCESS{Style.RESET_ALL} {test}")
        return Result("S", test, run.returncode, stdout, stderr)
    else:
        print(
            f"{pre}{Fore.RED + Style.BRIGHT}ERROR {run.returncode}{Style.RESET_ALL} {test}")
        return Result("E", test, run.returncode, stdout, stderr)


def print_failed_test(result: Result):
    if result.state == "S":
        return

    print("-"*60)
    print(result.test)
    print(f"Exit code {result.code}")
    print(f"STDOUT")
    print(result.out)
    print()
    print(f"STDERR")
    print(result.err)
    return False


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("-b", "--build", action="store_true")
    parser.add_argument("-s", "--serial", action="store_true")
    parser.add_argument("-v", "--verbose", action="store_true")
    parser.add_argument("-j", "--jobs", type=int, default=((os.cpu_count() or 2) / 2))
    parser.add_argument("filters", nargs="*")
    args = parser.parse_args()

    if args.build:
        print(f"Building tests")
        build_test()

    tests = get_all_tests()
    print(f"Total tests: {len(tests)}")

    tests = filtered_tests(tests, args.filters)
    print(f"Filtered tests: {len(tests)}")

    if len(tests) == 0:
        print("Nothing to do")
        return

    total = len(tests)
    results = []

    with tempfile.TemporaryDirectory(prefix="precice-tests-") as tdn:
        dir = pathlib.Path(tdn)
        print(f"Temp folder: {dir}")
        print(f"Running")

        workers = 1 if args.serial else args.jobs
        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
            futures = []
            for i, test in enumerate(tests):
                testdir = dir / str(i)
                futures.append(pool.submit(
                    run_test, test, Index(i, total), testdir, args.verbose))

            for f in concurrent.futures.as_completed(futures):
                result = f.result()
                if result.state != "S":
                    results.append(result)
                    print_failed_test(result)

    failed = [r.test for r in results if r.state == "E"]
    if failed:
        print()
        print("Failed:")
        for t in failed:
            print(t)
    timedout = [r.test for r in results if r.state == "T"]
    if timedout:
        print()
        print("Timed out:")
        for t in timedout:
            print(t)


if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print("Interrupted by keyboard")
        sys.exit(1)