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 parser.add_argument('--log-file', type=str,
  33         help="Write all test interaction the named file")
  34 parser.add_argument('--soft', action='store_true',
  35         help="Report but do not fail tests after ';>>> soft=True'")
  36
  37 parser.add_argument('test_file', type=argparse.FileType('r'),
  38         help="a test file formatted as with mal test data")
  39 parser.add_argument('mal_cmd', nargs="*",
  40         help="Mal implementation command line. Use '--' to "
  41              "specify a Mal command line with dashed options.")
  42
  43 class Runner():
  44     def __init__(self, args, no_pty=False, log_file=None):
  45         #print "args: %s" % repr(args)
  46         self.no_pty = no_pty
  47
  48         if log_file: self.logf = open(log_file, "a")
  49         else:        self.logf = None
  50
  51         # Cleanup child process on exit
  52         atexit.register(self.cleanup)
  53
  54         self.p = None
  55         env = os.environ
  56         env['TERM'] = 'dumb'
  57         env['INPUTRC'] = '/dev/null'
  58         env['PERL_RL'] = 'false'
  59         if no_pty:
  60             self.p = Popen(args, bufsize=0,
  61                            stdin=PIPE, stdout=PIPE, stderr=STDOUT,
  62                            preexec_fn=os.setsid,
  63                            env=env)
  64             self.stdin = self.p.stdin
  65             self.stdout = self.p.stdout
  66         else:
  67             # provide tty to get 'interactive' readline to work
  68             master, slave = pty.openpty()
  69
  70             # Set terminal size large so that readline will not send
  71             # ANSI/VT escape codes when the lines are long.
  72             buf = array.array('h', [100, 200, 0, 0])
  73             fcntl.ioctl(master, termios.TIOCSWINSZ, buf, True)
  74
  75             self.p = Popen(args, bufsize=0,
  76                            stdin=slave, stdout=slave, stderr=STDOUT,
  77                            preexec_fn=os.setsid,
  78                            env=env)
  79             # Now close slave so that we will get an exception from
  80             # read when the child exits early
  81             # http://stackoverflow.com/questions/11165521
  82             os.close(slave)
  83             self.stdin = os.fdopen(master, 'r+b', 0)
  84             self.stdout = self.stdin
  85
  86         #print "started"
  87         self.buf = ""
  88         self.last_prompt = ""
  89
  90     def read_to_prompt(self, prompts, timeout):
  91         end_time = time.time() + timeout
  92         while time.time() < end_time:
  93             [outs,_,_] = select([self.stdout], [], [], 1)
  94             if self.stdout in outs:
  95                 new_data = self.stdout.read(1)
  96                 new_data = new_data.decode("utf-8") if IS_PY_3 else new_data
  97                 #print "new_data: '%s'" % new_data
  98                 self.log(new_data)
  99                 if self.no_pty:
 100                     self.buf += new_data.replace("\n", "\r\n")
 101                 else:
 102                     self.buf += new_data
 103                 for prompt in prompts:
 104                     regexp = re.compile(prompt)
 105                     match = regexp.search(self.buf)
 106                     if match:
 107                         end = match.end()
 108                         buf = self.buf[0:end-len(prompt)]
 109                         self.buf = self.buf[end:]
 110                         self.last_prompt = prompt
 111                         return buf
 112         return None
 113
 114     def log(self, data):
 115         if self.logf:
 116             self.logf.write(data)
 117             self.logf.flush()
 118
 119
 120     def writeline(self, str):
 121         def _to_bytes(s):
 122             return bytes(s, "utf-8") if IS_PY_3 else s
 123
 124         self.stdin.write(_to_bytes(str + "\n"))
 125
 126     def cleanup(self):
 127         #print "cleaning up"
 128         if self.p:
 129             try:
 130                 os.killpg(self.p.pid, signal.SIGTERM)
 131             except OSError:
 132                 pass
 133             self.p = None
 134
 135 class TestReader:
 136     def __init__(self, test_file):
 137         self.line_num = 0
 138         self.data = test_file.read().split('\n')
 139         self.soft = False
 140
 141     def next(self):
 142         self.form = None
 143         self.out = ""
 144         self.ret = None
 145
 146         while self.data:
 147             self.line_num += 1
 148             line = self.data.pop(0)
 149             if re.match(r"^\s*$", line):   # blank line
 150                 continue
 151             elif line[0:3] == ";;;":       # ignore comment
 152                 continue
 153             elif line[0:2] == ";;":        # output comment
 154                 print(line[3:])
 155                 continue
 156             elif line[0:5] == ";>>> ":     # settings/commands
 157                 settings = {}
 158                 exec(line[5:], {}, settings)
 159                 if 'soft' in settings: self.soft = True
 160                 continue
 161             elif line[0:1] == ";":         # unexpected comment
 162                 print("Test data error at line %d:\n%s" % (self.line_num, line))
 163                 return None
 164             self.form = line   # the line is a form to send
 165
 166             # Now find the output and return value
 167             while self.data:
 168                 line = self.data[0]
 169                 if line[0:3] == ";=>":
 170                     self.ret = line[3:].replace('\\r', '\r').replace('\\n', '\n')
 171                     self.line_num += 1
 172                     self.data.pop(0)
 173                     break
 174                 elif line[0:2] == "; ":
 175                     self.out = self.out + line[2:] + sep
 176                     self.line_num += 1
 177                     self.data.pop(0)
 178                 else:
 179                     self.ret = "*"
 180                     break
 181             if self.ret: break
 182
 183         return self.form
 184
 185 args = parser.parse_args(sys.argv[1:])
 186 # Workaround argparse issue with two '--' on command line
 187 if sys.argv.count('--') > 0:
 188     args.mal_cmd = sys.argv[sys.argv.index('--')+1:]
 189
 190 if args.rundir: os.chdir(args.rundir)
 191
 192 r = Runner(args.mal_cmd, no_pty=args.no_pty, log_file=args.log_file)
 193 t = TestReader(args.test_file)
 194
 195
 196 def assert_prompt(runner, prompts, timeout):
 197     # Wait for the initial prompt
 198     header = runner.read_to_prompt(prompts, timeout=timeout)
 199     if not header == None:
 200         if header:
 201             print("Started with:\n%s" % header)
 202     else:
 203         print("Did not one of following prompt(s): %s" % repr(prompts))
 204         print("    Got      : %s" % repr(r.buf))
 205         sys.exit(1)
 206
 207
 208 # Wait for the initial prompt
 209 assert_prompt(r, ['user> ', 'mal-user> '], args.start_timeout)
 210
 211 # Send the pre-eval code if any
 212 if args.pre_eval:
 213     sys.stdout.write("RUNNING pre-eval: %s" % args.pre_eval)
 214     p.write(args.pre_eval)
 215     assert_prompt(args.test_timeout)
 216
 217 fail_cnt = 0
 218 soft_fail_cnt = 0
 219
 220 while t.next():
 221     sys.stdout.write("TEST: %s -> [%s,%s]" % (t.form, repr(t.out), t.ret))
 222     sys.stdout.flush()
 223
 224     # The repeated form is to get around an occasional OS X issue
 225     # where the form is repeated.
 226     # https://github.com/kanaka/mal/issues/30
 227     expected = ["%s%s%s%s" % (t.form, sep, t.out, t.ret),
 228                 "%s%s%s%s%s%s" % (t.form, sep, t.form, sep, t.out, t.ret)]
 229
 230     r.writeline(t.form)
 231     try:
 232         res = r.read_to_prompt(['\r\nuser> ', '\nuser> ',
 233                                 '\r\nmal-user> ', '\nmal-user> '],
 234                                 timeout=args.test_timeout)
 235         #print "%s,%s,%s" % (idx, repr(p.before), repr(p.after))
 236         if t.ret == "*" or res in expected:
 237             print(" -> SUCCESS")
 238         else:
 239             if args.soft and t.soft:
 240                 print(" -> SOFT FAIL (line %d):" % t.line_num)
 241                 soft_fail_cnt += 1
 242             else:
 243                 print(" -> FAIL (line %d):" % t.line_num)
 244                 fail_cnt += 1
 245             print("    Expected : %s" % repr(expected))
 246             print("    Got      : %s" % repr(res))
 247     except:
 248         _, exc, _ = sys.exc_info()
 249         print("\nException: %s" % repr(exc))
 250         print("Output before exception:\n%s" % r.buf)
 251         sys.exit(1)
 252
 253 if soft_fail_cnt > 0:
 254     print("SOFT FAILURES: %d" % soft_fail_cnt)
 255 if fail_cnt > 0:
 256     print("FAILURES: %d" % fail_cnt)
 257     sys.exit(2)
 258 sys.exit(0)