Skip to content

Instantly share code, notes, and snippets.

@perbu
Created March 9, 2015 10:11
Show Gist options
  • Save perbu/3a722df0b168ac8f73bb to your computer and use it in GitHub Desktop.
Save perbu/3a722df0b168ac8f73bb to your computer and use it in GitHub Desktop.
#!/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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment