runtest.py

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