runtest.py

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