gistfile1.py

# Django tests with multiprocessing (via Brian Rosner)

def run_tests_faster(test_labels, verbosity=1, interactive=True, extra_tests=[]):
    from Queue import Empty
    from processing import Queue, cpuCount, enableLogging
    from processing import currentProcess, Process, freezeSupport

    #enableLogging()

    # the number of processes to run in parallel.
    num_processes = cpuCount()

    def process_print(msg):
        print "%s: %s" % (currentProcess().getName(), msg)

    def worker(old_name, verbosity, interactive, ready_queue, app_queue, done_queue):
        process_print("setting up environment")
        setup_test_environment()
        create_test_db(verbosity, autoclobber=not interactive)
        ready_queue.put(1)
        for app_label in iter(app_queue.get, "STOP"):
            process_print(app_label)
            suite = build_suite(get_app(app_label))
            result = unittest.TestResult()
            suite.run(result)
            done_queue.put((app_label, len(result.failures), len(result.errors)))
            # done_queue.put((app_label, 0, 0)) # FOR TESTING
        process_print("tearing down environment")
        destroy_test_db(old_name, verbosity)
        teardown_test_environment()

    ready_queue = Queue()
    app_queue = Queue()
    done_queue = Queue()
    apps = []

    for installed_app in settings.INSTALLED_APPS:
        apps.append(installed_app.split(".")[-1])

    for i in range(num_processes):
        Process(target=worker, args=[
            settings.DATABASE_NAME, verbosity, interactive, ready_queue,
            app_queue, done_queue
        ]).start()

    ready_count = 0
    done_apps = []
    total_failures = 0
    total_errors = 0

    while ready_count != num_processes:
        ready_queue.get()
        ready_count += 1

    print "start timer"
    start_time = time.time()
    app_queue.putmany(apps)

    while len(apps) != len(done_apps):
        app_label, failures, errors = done_queue.get()
        total_failures += failures
        total_errors += errors
        done_apps.append(app_label)

    stop_time = time.time()
    print "timer ended"
    time_taken = stop_time - start_time

    for i in range(num_processes):
        app_queue.put("STOP")

    print "time: %.3fs, failures: %d, errors: %d" % (time_taken, total_failures, total_errors)

    # give the workers some time to finish
    time.sleep(2)

    return 0