lldb/test/dosep.ty

#!/usr/bin/env python

"""
Run the test suite using a separate process for each test file.
"""

import os, sys, platform
import Queue, threading
import multiprocessing

from optparse import OptionParser

# Command template of the invocation of the test driver.
template = '%s %s/dotest.py %s -p %s %s'

def process_dir(root, files, test_root, dotest_options):
    """Examine a directory for tests, and invoke any found within it."""
    failed = []
    passed = []
    for name in files:
        path = os.path.join(root, name)

        # We're only interested in the test file with the "Test*.py" naming pattern.
        if not name.startswith("Test") or not name.endswith(".py"):
            continue

        # Neither a symbolically linked file.
        if os.path.islink(path):
            continue

        command = template % (sys.executable, test_root, dotest_options if dotest_options else "", name, root)
        if 0 != os.system(command):
            failed.append(name)
        else:
            passed.append(name)
    return (failed, passed)

in_q = None
out_q = None

def process_dir_worker():
    """Worker thread main loop when in multithreaded mode.
    Takes one directory specification at a time and works on it."""
    while True:
        (root, files, test_root, dotest_options) = in_q.get()
        (dir_failed, dir_passed) = process_dir(root, files, test_root, dotest_options)
        out_q.put((dir_failed, dir_passed))
        in_q.task_done()

def walk_and_invoke(test_root, dotest_options, num_threads):
    """Look for matched files and invoke test driver on each one.
    In single-threaded mode, each test driver is invoked directly.
    In multi-threaded mode, submit each test driver to a worker
    queue, and then wait for all to complete."""
    failed = []
    passed = []
    if (num_threads > 1):
        print("Running multithreaded with %d threads" % num_threads)
        global in_q
        global out_q
        in_q = Queue.Queue()
        out_q = Queue.Queue()
        for i in range(num_threads):
            t = threading.Thread(target=process_dir_worker)
            t.daemon = True
            t.start()
    else:
        print("Running single-threaded")
    for root, dirs, files in os.walk(test_root, topdown=False):
        if (num_threads > 1):
            in_q.put((root, files, test_root, dotest_options))
        else:
            (dir_failed, dir_passed) = process_dir(root, files, test_root, dotest_options)
            failed += dir_failed
            passed += dir_passed
    if (num_threads > 1):
        in_q.join()
        while not out_q.empty():
            (dir_failed, dir_passed) = out_q.get()
            failed += dir_failed
            passed += dir_passed
    return (failed, passed)

def main():
    test_root = sys.path[0]

    parser = OptionParser(usage="""\
Run lldb test suite using a separate process for each test file.
""")
    parser.add_option('-o', '--options',
                      type='string', action='store',
                      dest='dotest_options',
                      help="""The options passed to 'dotest.py' if specified.""")

    parser.add_option('-t', '--threads',
                      type='int',
                      dest='num_threads',
                      help="""The number of threads to use when running tests separately.""",
                      default=multiprocessing.cpu_count())

    opts, args = parser.parse_args()
    dotest_options = opts.dotest_options
    num_threads = opts.num_threads
    if num_threads < 1:
        num_threads_str = os.environ.get("LLDB_TEST_THREADS")
        if num_threads_str:
            num_threads = int(num_threads_str)
            if num_threads < 1:
                num_threads = 1
        else:
            num_threads = 1

    system_info = " ".join(platform.uname())
    (failed, passed) = walk_and_invoke(test_root, dotest_options, num_threads)
    num_tests = len(failed) + len(passed)

    print "Ran %d tests." % num_tests
    if len(failed) > 0:
        print "Failing Tests (%d)" % len(failed)
        for f in failed:
          print "FAIL: LLDB (suite) :: %s (%s)" % (f, system_info)
        sys.exit(1)
    sys.exit(0)

if __name__ == '__main__':
    main()