integrations/pjrt/test/test_jax.py

# Licensed under the Apache License v2.0 with LLVM Exceptions.
# See https://llvm.org/LICENSE.txt for license information.
# SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception

import argparse
import multiprocessing
import os
import random
import re
import subprocess
import sys
import time

from collections import namedtuple
from enum import Enum

parser = argparse.ArgumentParser(
    prog="test_jax.py", description="Run jax testsuite hermetically"
)
parser.add_argument("testfiles", nargs="*")
parser.add_argument("-t", "--timeout", default=60)
parser.add_argument("-l", "--logdir", default="/tmp/jaxtest")
parser.add_argument("-p", "--passing", default=None)
parser.add_argument("-f", "--failing", default=None)
parser.add_argument("-e", "--expected", default=None)
parser.add_argument("-j", "--jobs", default=None)

args = parser.parse_args()

PYTEST_CMD = [
    "pytest",
    "-p",
    "openxla_pjrt_artifacts",
    f"--openxla-pjrt-artifact-dir={args.logdir}",
]


def get_test(test):
    print("Fetching from:", test)
    stdout = subprocess.run(PYTEST_CMD + ["--setup-only", test], capture_output=True)
    lst = re.findall("::[^ ]*::[^ ]*", str(stdout))
    return [test + func for func in lst]


def get_tests(tests):
    fulltestlist = []
    with multiprocessing.Pool(os.cpu_count()) as p:
        fulltestlist = p.map(get_test, tests)
    fulltestlist = sorted([i for lst in fulltestlist for i in lst])
    return fulltestlist


def generate_test_commands(tests):
    cmds = []
    for test in tests:
        test_cmd = PYTEST_CMD + [test]
        cmds.append(test_cmd)

    return cmds


TestCase = namedtuple("TestCase", ["test", "timeout"])
TestResult = Enum("TestResult", ["SUCCESS", "FAILURE", "TIMEOUT"])


def exec_test(testcase):
    command, timeout = testcase
    if float(timeout) > 0:
        command = ["timeout", f"{timeout}"] + command

    start = time.perf_counter()
    result = subprocess.run(command, capture_output=True)
    end = time.perf_counter()
    ellapsed = end - start
    timedout = (float(timeout) > 0) and (ellapsed > float(timeout))

    if result.returncode == 0:
        sys.stdout.write(".")
        sys.stdout.flush()
        return TestResult.SUCCESS

    if timedout:
        sys.stdout.write("t")
        sys.stdout.flush()
        return TestResult.TIMEOUT

    sys.stdout.write("f")
    sys.stdout.flush()
    return TestResult.FAILURE


def exec_testsuite(commands, jobs, timeout):
    random.shuffle(commands)
    withTimeout = list(map(lambda x: TestCase(x, timeout), commands))

    results = []
    with multiprocessing.Pool(int(jobs)) as p:
        results = p.map(exec_test, withTimeout)

    passing, timeout, failing = [], [], []
    for result, cmd in zip(results, commands):
        if result == TestResult.SUCCESS:
            passing.append(cmd)

        if result == TestResult.TIMEOUT:
            timeout.append(cmd)

        if result == TestResult.FAILURE:
            failing.append(cmd)
    print("")

    return passing, timeout, failing


def get_testnames(cmd):
    names = []
    for c in cmd:
        testname = " ".join(c)
        testname = re.search("[^ /]*::[^ ]*::[^ ]*", testname)[0]
        names.append(testname)
    return names


def write_results(filename, results):
    if filename is not None:
        with open(filename, "w") as f:
            for line in results:
                f.write(line + "\n")


def load_results(filename):
    if not filename or not os.path.isfile(filename):
        return []
    expected = []
    with open(filename, "r") as f:
        for line in f:
            expected.append(line.strip())
    return expected


def compare_results(expected, passing):
    passing = set(passing)
    expected = set(expected)
    new_failures = expected - passing
    new_passing = passing - expected
    return new_passing, new_failures


print("Querying All Tests")
tests = get_tests(args.testfiles)

print("Generating test suite")
commands = generate_test_commands(tests)

print(
    f"Executing {len(commands)} tests across {args.jobs} threads with timeout = {args.timeout}"
)
passing, timeout, failing = exec_testsuite(
    commands, jobs=args.jobs, timeout=args.timeout
)

expected = load_results(args.expected)

# Break into passing vs failing
failing = failing + timeout

# Get the testnames
passing = get_testnames(passing)
failing = get_testnames(failing)

write_results(args.passing, passing)
write_results(args.failing, failing)

print("Total:", len(commands))
print("Passing:", len(passing))
print("Failing:", len(failing))
print("Failing (timed out):", len(timeout))

if expected:
    new_passing, new_failures = compare_results(expected, passing)

    if new_passing:
        print("Newly Passing Tests:")
        for test in new_passing:
            print(" ", test)

    if new_failures:
        print("Newly Failing Tests:")
        for test in new_failures:
            print(" ", test)

    if len(expected) > len(passing):
        exit(1)