runtest.py

   1 #!/usr/bin/env python
   2
   3 import os, sys, re
   4 import argparse, time
   5
   6 import pty
   7 from subprocess import Popen, STDOUT, PIPE
   8 from select import select
   9
  10 # TODO: do we need to support '\n' too
  11 sep = "\r\n"
  12 #sep = "\n"
  13 rundir = None
  14
  15 parser = argparse.ArgumentParser(
  16         description="Run a test file against a Mal implementation")
  17 parser.add_argument('--rundir',
  18         help="change to the directory before running tests")
  19 parser.add_argument('--start-timeout', default=10, type=int,
  20         help="default timeout for initial prompt")
  21 parser.add_argument('--test-timeout', default=20, type=int,
  22         help="default timeout for each individual test action")
  23 parser.add_argument('--pre-eval', default=None, type=str,
  24         help="Mal code to evaluate prior to running the test")
  25 parser.add_argument('--redirect', action='store_true',
  26         help="Run implementation in bash and redirect output to /dev/null")
  27
  28 parser.add_argument('test_file', type=argparse.FileType('r'),
  29         help="a test file formatted as with mal test data")
  30 parser.add_argument('mal_cmd', nargs="*",
  31         help="Mal implementation command line. Use '--' to "
  32              "specify a Mal command line with dashed options.")
  33
  34 class Runner():
  35     def __init__(self, args, redirect=False):
  36         self.redirect = redirect
  37         print "args: %s" % repr(args)
  38         if redirect:
  39             #cmd = '/bin/bash -c "' + " ".join(args) + ' |tee /dev/null"'
  40             #cmd = " ".join(args) + ' |tee /dev/null'
  41             #cmd = " ".join(args) + ' |tee /tmp/joel'
  42             #print "using redirect cmd: '%s'" % cmd
  43             self.p = Popen(args, bufsize=0, stdin=PIPE, stdout=PIPE, stderr=STDOUT)
  44             #self.p = Popen(cmd, bufsize=0, stdin=PIPE, stdout=PIPE, stderr=STDOUT, shell=True)
  45             #self.p = Popen(["/bin/bash", "-c", cmd], bufsize=0, stdin=PIPE, stdout=PIPE, stderr=STDOUT)
  46             self.stdin = self.p.stdin
  47             self.stdout = self.p.stdout
  48         else:
  49             # provide tty to get 'interactive' readline to work
  50             master, slave = pty.openpty()
  51             self.p = Popen(args, bufsize=0, stdin=slave, stdout=slave, stderr=STDOUT)
  52             self.stdin = os.fdopen(master, 'r+b', 0)
  53             self.stdout = self.stdin
  54
  55         #print "started"
  56         self.buf = ""
  57         self.last_prompt = ""
  58
  59     def read_to_prompt(self, prompts, timeout):
  60         end_time = time.time() + timeout
  61         while time.time() < end_time:
  62             [outs,_,_] = select([self.stdout], [], [], 1)
  63             if self.stdout in outs:
  64                 new_data = self.stdout.read(1)
  65                 #print "new_data: '%s'" % new_data
  66                 self.buf += new_data
  67                 for prompt in prompts:
  68                     regexp = re.compile(prompt)
  69                     match = regexp.search(self.buf)
  70                     if match:
  71                         end = match.end()
  72                         buf = self.buf[0:end-len(prompt)]
  73                         self.buf = self.buf[end:]
  74                         self.last_prompt = prompt
  75                         return buf
  76         return None
  77
  78     def writeline(self, str):
  79         self.stdin.write(str + "\n")
  80         if self.redirect:
  81             # Simulate echo
  82             self.buf += str + "\r\n"
  83
  84     def cleanup(self):
  85         if self.p:
  86             self.p.kill()
  87             self.p = None
  88
  89
  90 args = parser.parse_args(sys.argv[1:])
  91 test_data = args.test_file.read().split('\n')
  92
  93 if args.rundir: os.chdir(args.rundir)
  94
  95 r = Runner(args.mal_cmd, redirect=args.redirect)
  96
  97
  98 test_idx = 0
  99 def read_test(data):
 100     global test_idx
 101     form, output, ret = None, "", None
 102     while data:
 103         test_idx += 1
 104         line = data.pop(0)
 105         if re.match(r"^\s*$", line):   # blank line
 106             continue
 107         elif line[0:3] == ";;;":       # ignore comment
 108             continue
 109         elif line[0:2] == ";;":        # output comment
 110             print line[3:]
 111             continue
 112         elif line[0:2] == ";":         # unexpected comment
 113             print "Test data error at line %d:\n%s" % (test_idx, line)
 114             return None, None, None, test_idx
 115         form = line   # the line is a form to send
 116
 117         # Now find the output and return value
 118         while data:
 119             line = data[0]
 120             if line[0:3] == ";=>":
 121                 ret = line[3:].replace('\\r', '\r').replace('\\n', '\n')
 122                 test_idx += 1
 123                 data.pop(0)
 124                 break
 125             elif line[0:2] == "; ":
 126                 output = output + line[2:] + sep
 127                 test_idx += 1
 128                 data.pop(0)
 129             else:
 130                 ret = "*"
 131                 break
 132         if ret: break
 133
 134     return form, output, ret, test_idx
 135
 136 def assert_prompt(timeout):
 137     # Wait for the initial prompt
 138     header = r.read_to_prompt(['user> ', 'mal-user> '], timeout=timeout)
 139     if not header == None:
 140         if header:
 141             print "Started with:\n%s" % header
 142     else:
 143         print "Did not get 'user> ' or 'mal-user> ' prompt"
 144         print "    Got      : %s" % repr(r.buf)
 145         r.cleanup()
 146         sys.exit(1)
 147
 148
 149 # Wait for the initial prompt
 150 assert_prompt(args.start_timeout)
 151
 152 # Send the pre-eval code if any
 153 if args.pre_eval:
 154     sys.stdout.write("RUNNING pre-eval: %s" % args.pre_eval)
 155     p.write(args.pre_eval)
 156     assert_prompt(args.test_timeout)
 157
 158 fail_cnt = 0
 159
 160 while test_data:
 161     form, out, ret, line_num = read_test(test_data)
 162     if form == None:
 163         break
 164     sys.stdout.write("TEST: %s -> [%s,%s]" % (form, repr(out), repr(ret)))
 165     sys.stdout.flush()
 166     expected = "%s%s%s%s" % (form, sep, out, ret)
 167
 168     r.writeline(form)
 169     try:
 170         res = r.read_to_prompt(['\r\nuser> ', '\nuser> ',
 171                                 '\r\nmal-user> ', '\nmal-user> '],
 172                                 timeout=args.test_timeout)
 173         #print "%s,%s,%s" % (idx, repr(p.before), repr(p.after))
 174         if ret == "*" or res == expected:
 175             print " -> SUCCESS"
 176         else:
 177             print " -> FAIL (line %d):" % line_num
 178             print "    Expected : %s" % repr(expected)
 179             print "    Got      : %s" % repr(res)
 180             fail_cnt += 1
 181     except:
 182         print "Got Exception"
 183         r.cleanup()
 184         sys.exit(1)
 185 r.cleanup()
 186
 187 if fail_cnt > 0:
 188     print "FAILURES: %d" % fail_cnt
 189     sys.exit(2)
 190 sys.exit(0)