runtest.py

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