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('--mono', action='store_true',
  28         help="Use workarounds Mono/.Net Console misbehaviors")
  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, mono=False):
  38         #print "args: %s" % repr(args)
  39         self.mono = mono
  40
  41         # Cleanup child process on exit
  42         atexit.register(self.cleanup)
  43
  44         if mono:
  45             self.p = Popen(args, bufsize=0,
  46                            stdin=PIPE, stdout=PIPE, stderr=STDOUT,
  47                            preexec_fn=os.setsid)
  48             self.stdin = self.p.stdin
  49             self.stdout = self.p.stdout
  50         else:
  51             # provide tty to get 'interactive' readline to work
  52             master, slave = pty.openpty()
  53             self.p = Popen(args, bufsize=0,
  54                            stdin=slave, stdout=slave, stderr=STDOUT,
  55                            preexec_fn=os.setsid)
  56             self.stdin = os.fdopen(master, 'r+b', 0)
  57             self.stdout = self.stdin
  58
  59         #print "started"
  60         self.buf = ""
  61         self.last_prompt = ""
  62
  63     def read_to_prompt(self, prompts, timeout):
  64         end_time = time.time() + timeout
  65         while time.time() < end_time:
  66             [outs,_,_] = select([self.stdout], [], [], 1)
  67             if self.stdout in outs:
  68                 new_data = self.stdout.read(1)
  69                 #print "new_data: '%s'" % new_data
  70                 new_data = new_data.decode("utf-8") if IS_PY_3 else new_data
  71                 if self.mono:
  72                     self.buf += new_data.replace("\n", "\r\n")
  73                 else:
  74                     self.buf += new_data
  75                 for prompt in prompts:
  76                     regexp = re.compile(prompt)
  77                     match = regexp.search(self.buf)
  78                     if match:
  79                         end = match.end()
  80                         buf = self.buf[0:end-len(prompt)]
  81                         self.buf = self.buf[end:]
  82                         self.last_prompt = prompt
  83                         return buf
  84         return None
  85
  86     def writeline(self, str):
  87         def _to_bytes(s):
  88             return bytes(s, "utf-8") if IS_PY_3 else s
  89
  90         self.stdin.write(_to_bytes(str + "\n"))
  91         if self.mono:
  92             # Simulate echo
  93             self.buf += _to_bytes(str + "\r\n")
  94
  95     def cleanup(self):
  96         #print "cleaning up"
  97         if self.p:
  98             os.killpg(self.p.pid, signal.SIGTERM)
  99             self.p = None
 100
 101
 102 args = parser.parse_args(sys.argv[1:])
 103 test_data = args.test_file.read().split('\n')
 104
 105 if args.rundir: os.chdir(args.rundir)
 106
 107 r = Runner(args.mal_cmd, mono=args.mono)
 108
 109
 110 test_idx = 0
 111 def read_test(data):
 112     global test_idx
 113     form, output, ret = None, "", None
 114     while data:
 115         test_idx += 1
 116         line = data.pop(0)
 117         if re.match(r"^\s*$", line):   # blank line
 118             continue
 119         elif line[0:3] == ";;;":       # ignore comment
 120             continue
 121         elif line[0:2] == ";;":        # output comment
 122             print(line[3:])
 123             continue
 124         elif line[0:2] == ";":         # unexpected comment
 125             print("Test data error at line %d:\n%s" % (test_idx, line))
 126             return None, None, None, test_idx
 127         form = line   # the line is a form to send
 128
 129         # Now find the output and return value
 130         while data:
 131             line = data[0]
 132             if line[0:3] == ";=>":
 133                 ret = line[3:].replace('\\r', '\r').replace('\\n', '\n')
 134                 test_idx += 1
 135                 data.pop(0)
 136                 break
 137             elif line[0:2] == "; ":
 138                 output = output + line[2:] + sep
 139                 test_idx += 1
 140                 data.pop(0)
 141             else:
 142                 ret = "*"
 143                 break
 144         if ret: break
 145
 146     return form, output, ret, test_idx
 147
 148 def assert_prompt(timeout):
 149     # Wait for the initial prompt
 150     header = r.read_to_prompt(['user> ', 'mal-user> '], timeout=timeout)
 151     if not header == None:
 152         if header:
 153             print("Started with:\n%s" % header)
 154     else:
 155         print("Did not get 'user> ' or 'mal-user> ' prompt")
 156         print("    Got      : %s" % repr(r.buf))
 157         sys.exit(1)
 158
 159
 160 # Wait for the initial prompt
 161 assert_prompt(args.start_timeout)
 162
 163 # Send the pre-eval code if any
 164 if args.pre_eval:
 165     sys.stdout.write("RUNNING pre-eval: %s" % args.pre_eval)
 166     p.write(args.pre_eval)
 167     assert_prompt(args.test_timeout)
 168
 169 fail_cnt = 0
 170
 171 while test_data:
 172     form, out, ret, line_num = read_test(test_data)
 173     if form == None:
 174         break
 175     sys.stdout.write("TEST: %s -> [%s,%s]" % (form, repr(out), repr(ret)))
 176     sys.stdout.flush()
 177     expected = "%s%s%s%s" % (form, sep, out, ret)
 178
 179     r.writeline(form)
 180     try:
 181         res = r.read_to_prompt(['\r\nuser> ', '\nuser> ',
 182                                 '\r\nmal-user> ', '\nmal-user> '],
 183                                 timeout=args.test_timeout)
 184         #print "%s,%s,%s" % (idx, repr(p.before), repr(p.after))
 185         if ret == "*" or res == expected:
 186             print(" -> SUCCESS")
 187         else:
 188             print(" -> FAIL (line %d):" % line_num)
 189             print("    Expected : %s" % repr(expected))
 190             print("    Got      : %s" % repr(res))
 191             fail_cnt += 1
 192     except:
 193         print("Got Exception")
 194         sys.exit(1)
 195
 196 if fail_cnt > 0:
 197     print("FAILURES: %d" % fail_cnt)
 198     sys.exit(2)
 199 sys.exit(0)