gistfile1.py

#!/usr/local/bin/python

import getopt
import threading
import Queue
import time
import glob
import subprocess
import sys
import os
import logging
import textwrap

# Workqueue for tests.
q = Queue.Queue()

# Queue up fail tests for retry here:
failq    = [ ]
running  = 1 # set to 0 to tell the threads to bail
failblog = [ ]
finalfail = [ ]
varnishtestopts = [ ]

retry_tests = True         # retry failed tests in single threaded mode.
logfile  = "testrunner.log"
nthreads = 8
verbose  = False 
varnishtest = "varnishtest"
testdir  = "tests"
loglevel = "WARNING"
progress = False

printlock = threading.Lock()

def lprint(*stuff):
    printlock.acquire()
    print " ".join(stuff)
    printlock.release()

def run_test(test, finalrun):
    global finalfail
    global varnishtestopts
    cmd = []
    try:
        cmd.append(varnishtest)
        if (varnishtestopts): 
            cmd.append(varnishtestopts)
        cmd.append(test)
        logging.info("varnishtest cmd line: %s", str(cmd))
        output = subprocess.check_output( cmd )
        if progress:
            sys.stdout.write('.')
    except subprocess.CalledProcessError as e:
        failblog.append(e.output)
        if finalrun:
            # This is the final run. Log all errors.
            finalfail.append(test)
            if progress:
                sys.stdout.write('!')
        else:
            # Append the output to the faillog
            # Reschedule test for single threaded retry
            if progress:
                sys.stdout.write('w')
            failq.append(test)

def worker(tn,finalrun):
    logging.info("starting worker %i (final: %s)",tn,str(finalrun))
    while (running):
        try:
            t = q.get(True, 0.1)
            logging.debug("Running %s", t)
            run_test(t, finalrun)
            q.task_done()
        except Queue.Empty:
            pass


def halt_threads():
    global running # silly scoping
    running = 0
    for thread in threading.enumerate():
        if thread is not threading.currentThread():
            thread.join()
    running = 1
    # lprint("Threads halted")


def usage():
        print textwrap.dedent("""\

            Usage: testrunner <option>
             [-h]               - Show help and exit.
             [-j <threads>]     - paralellize testing (default: 8)
             [-v]               - verbose mode  (default is off)
             [-o <file>]        - log to this file (default testrunner.log)
             [-e <varnishtest>] - set varnishtest path 
             [-l <level>]       - set log level (DEBUG,INFO,WARNING,ERROR)
             [-t <dir>]         - look for tests in this directory (default: tests)
             [-p ]              - enable progress
             [-a <arg>]         - pass this argument to varnishtest. Can be used multiple times.
            """)

# ==============
# main block
# ==============


try:
    optlist, rem = getopt.getopt(sys.argv[1:], 'dVvj:o:e:l:t:p')

    for opt, arg in optlist:
        if opt == '-j':
            nthreads = int(arg)
        elif opt == '-v':
            verbose = True
        elif opt == '-o':
            logfile = arg
        elif opt == '-e':
            varnishtest = arg
        elif opt == '-l':
            loglevel = arg
        elif opt == '-t':
            testdir = arg
        elif opt == '-p':
            progress = True
        elif opt == '-a':
            varnishtestopts.append(arg)
        elif opt == '-h':
            usage()
            sys.exit(0)
except getopt.GetoptError as err:
    print str(err) # will print something like "option -a not recognized"
    usage()
    sys.exit(2)

numeric_level = getattr(logging, loglevel.upper(), None)
if not isinstance(numeric_level, int):
    raise ValueError('Invalid log level: %s' % loglevel)
logging.basicConfig(level=numeric_level)

logging.info("Options parsed. Reading tests.")


for test in glob.glob(testdir + "/*.vtc"):
    q.put(test)

logging.info("%i tests scheduled. Starting %i threads", q.qsize(), nthreads )

for tn in range(nthreads):
    t = threading.Thread(target=worker, args = (tn, False))
    t.start()

q.join()     # block the queue until empty

halt_threads()
logging.info("Tests done. Threads halted. Scehduling failed tests for single threaded execution.")

# re-scehdule all the failed tests
for test in failq:
    q.put(test)

# spin up a single thread
if retry_tests:
    failblog.append("Retrying any failed threads in single threaded mode.\n")
    t = threading.Thread(target=worker, args = (tn, True))
    t.start()
    q.join()
    halt_threads()

# progress indicator ends here. cr.
sys.stdout.write(os.linesep)

f = open(logfile, 'w')
for line in failblog:
    f.write(line)
    f.write(os.linesep)
if (finalfail):
    print "Tests failed: ", os.linesep.join(finalfail)
print "Log written to", logfile