dvander@195: # vim: set ts=2 sw=4 tw=99 ft=python et:
dvander@195: import os
dvander@195: import re
dvander@195: import sys
dvander@195: import time
dvander@195: import os.path
dvander@195: from test import Test
dvander@195: from subprocess import *
dvander@195: from threading import Thread
dvander@195: from display import Interactive
dvander@195: 
dvander@195: def FindTests(tests, root, folder):
dvander@195:   folders = []
dvander@195:   for name in os.listdir(os.path.join(root, folder)):
dvander@195:     path = os.path.join(root, folder, name)
dvander@195:     if os.path.isdir(path):
dvander@195:       folders.append(name)
dvander@195:       continue
dvander@195: 
dvander@195:     out = re.match("^([^\.]+)\.out$", name)
dvander@195:     err = re.match("^([^\.]+)\.err$", name)
dvander@195:     if out:
dvander@205:         name = os.path.join(folder, out.group(1))
dvander@205:         tests.append(Test(root, name, 'out'))
dvander@195:     elif err:
dvander@205:         name = os.path.join(folder, err.group(1))
dvander@205:         tests.append(Test(root, name, 'err'))
dvander@195: 
dvander@195:   for name in folders:
dvander@195:     path = os.path.join(folder, name)
dvander@195:     FindTests(tests, root, path)
dvander@195: 
dvander@195: def IsTestRoot(path, name):
dvander@195:   return os.path.isdir(path)
dvander@195: 
dvander@195: def RunTest_(argv, options, L):
dvander@195:   if sys.platform != 'win32':
dvander@195:     options['close_fds'] = True
dvander@195: 
dvander@195:   p = Popen(argv, stdin=PIPE, stdout=PIPE, stderr=PIPE, **options)
dvander@195: 
dvander@195:   L[0] = p
dvander@195:   out, err = p.communicate()
dvander@195:   L[1] = (out, err, p.returncode)
dvander@195: 
dvander@195: def RunTest(test, options, shell):
dvander@195:   argv = [shell, test.cmd()]
dvander@195: 
dvander@195:   L = [None, None]
dvander@195:   timeout = False
dvander@195:   th = Thread(target = RunTest_, args = (argv, options, L))
dvander@195:   th.start()
dvander@214:   th.join(10.0)
dvander@195:   while th.isAlive():
dvander@195:     if L[0] is not None:
dvander@195:       try:
dvander@195:         L[0].kill()
dvander@195:         time.sleep(.1)
dvander@195:         timeout = True
dvander@205:       except e:
dvander@195:         pass
dvander@195: 
dvander@195:   th.join()
dvander@195:   (out, err, code) = L[1]
dvander@195: 
dvander@195:   return test.check(out, err, code, timeout)
dvander@195: 
dvander@195: def RunTests(display, tests, shell):
dvander@195:   display.begin(tests)
dvander@195:   ok = True
dvander@195:   try:
dvander@195:     for test in tests:
dvander@195:       options = { }
dvander@195:       result = RunTest(test, options, shell)
dvander@195: 
dvander@195:       ok = ok and result.ok
dvander@195:       
dvander@195:       display.result(test, result)
dvander@195:   except KeyboardInterrupt:
dvander@195:     display.cancel()
dvander@195:     ok = False
dvander@195: 
dvander@195:   return ok
dvander@195: 
dvander@195:   
dvander@195: def main(argv):
dvander@195:   root = os.path.dirname(os.path.abspath(__file__))
dvander@195: 
dvander@195:   from optparse import OptionParser
dvander@195:   op = OptionParser(usage='%prog [options] <shell>')
dvander@195:   (options, args) = op.parse_args(argv)
dvander@195: 
dvander@195:   if len(args) != 1:
dvander@195:     op.error('missing <shell> argument')
dvander@195: 
dvander@195:   shell = os.path.normpath(args[0])
dvander@195: 
dvander@195:   tests = []
dvander@195:   for name in os.listdir(root):
dvander@195:     path = os.path.join(root, name)
dvander@195:     if IsTestRoot(path, name):
dvander@195:       FindTests(tests, root, name)
dvander@195: 
dvander@195:   if len(tests) == 0:
dvander@195:     op.error('no test cases found')
dvander@195: 
dvander@195:   display = Interactive()
dvander@195:   ok = RunTests(display, tests, shell)
dvander@195:   if ok:
dvander@195:     sys.exit(0)
dvander@195:   else:
dvander@195:     sys.exit(1)
dvander@195: 
dvander@195: if __name__ == '__main__':
dvander@195:   main(sys.argv[1:])
dvander@195: