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 # Control whether deferrable and optional tests are executed
  56 parser.add_argument('--deferrable', dest='deferrable', action='store_true',
  57         help="Enable deferrable tests that follow a ';>>> deferrable=True'")
  58 parser.add_argument('--no-deferrable', dest='deferrable', action='store_false',
  59         help="Disable deferrable tests that follow a ';>>> deferrable=True'")
  60 parser.set_defaults(deferrable=True)
  61 parser.add_argument('--optional', dest='optional', action='store_true',
  62         help="Enable optional tests that follow a ';>>> optional=True'")
  63 parser.add_argument('--no-optional', dest='optional', action='store_false',
  64         help="Disable optional tests that follow a ';>>> optional=True'")
  65 parser.set_defaults(optional=True)
  66
  67 parser.add_argument('test_file', type=argparse.FileType('r'),
  68         help="a test file formatted as with mal test data")
  69 parser.add_argument('mal_cmd', nargs="*",
  70         help="Mal implementation command line. Use '--' to "
  71              "specify a Mal command line with dashed options.")
  72
  73 class Runner():
  74     def __init__(self, args, no_pty=False):
  75         #print "args: %s" % repr(args)
  76         self.no_pty = no_pty
  77
  78         # Cleanup child process on exit
  79         atexit.register(self.cleanup)
  80
  81         self.p = None
  82         env = os.environ
  83         env['TERM'] = 'dumb'
  84         env['INPUTRC'] = '/dev/null'
  85         env['PERL_RL'] = 'false'
  86         if no_pty:
  87             self.p = Popen(args, bufsize=0,
  88                            stdin=PIPE, stdout=PIPE, stderr=STDOUT,
  89                            preexec_fn=os.setsid,
  90                            env=env)
  91             self.stdin = self.p.stdin
  92             self.stdout = self.p.stdout
  93         else:
  94             # provide tty to get 'interactive' readline to work
  95             master, slave = pty.openpty()
  96
  97             # Set terminal size large so that readline will not send
  98             # ANSI/VT escape codes when the lines are long.
  99             buf = array.array('h', [100, 200, 0, 0])
 100             fcntl.ioctl(master, termios.TIOCSWINSZ, buf, True)
 101
 102             self.p = Popen(args, bufsize=0,
 103                            stdin=slave, stdout=slave, stderr=STDOUT,
 104                            preexec_fn=os.setsid,
 105                            env=env)
 106             # Now close slave so that we will get an exception from
 107             # read when the child exits early
 108             # http://stackoverflow.com/questions/11165521
 109             os.close(slave)
 110             self.stdin = os.fdopen(master, 'r+b', 0)
 111             self.stdout = self.stdin
 112
 113         #print "started"
 114         self.buf = ""
 115         self.last_prompt = ""
 116
 117     def read_to_prompt(self, prompts, timeout):
 118         end_time = time.time() + timeout
 119         while time.time() < end_time:
 120             [outs,_,_] = select([self.stdout], [], [], 1)
 121             if self.stdout in outs:
 122                 new_data = self.stdout.read(1)
 123                 new_data = new_data.decode("utf-8") if IS_PY_3 else new_data
 124                 #print("new_data: '%s'" % new_data)
 125                 debug(new_data)
 126                 if self.no_pty:
 127                     self.buf += new_data.replace("\n", "\r\n")
 128                 else:
 129                     self.buf += new_data
 130                 for prompt in prompts:
 131                     regexp = re.compile(prompt)
 132                     match = regexp.search(self.buf)
 133                     if match:
 134                         end = match.end()
 135                         buf = self.buf[0:end-len(prompt)]
 136                         self.buf = self.buf[end:]
 137                         self.last_prompt = prompt
 138                         return buf
 139         return None
 140
 141     def writeline(self, str):
 142         def _to_bytes(s):
 143             return bytes(s, "utf-8") if IS_PY_3 else s
 144
 145         self.stdin.write(_to_bytes(str + "\n"))
 146
 147     def cleanup(self):
 148         #print "cleaning up"
 149         if self.p:
 150             try:
 151                 os.killpg(self.p.pid, signal.SIGTERM)
 152             except OSError:
 153                 pass
 154             self.p = None
 155
 156 class TestReader:
 157     def __init__(self, test_file):
 158         self.line_num = 0
 159         self.data = test_file.read().split('\n')
 160         self.soft = False
 161         self.deferrable = False
 162         self.optional = False
 163
 164     def next(self):
 165         self.msg = None
 166         self.form = None
 167         self.out = ""
 168         self.ret = None
 169
 170         while self.data:
 171             self.line_num += 1
 172             line = self.data.pop(0)
 173             if re.match(r"^\s*$", line):   # blank line
 174                 continue
 175             elif line[0:3] == ";;;":       # ignore comment
 176                 continue
 177             elif line[0:2] == ";;":        # output comment
 178                 self.msg = line[3:]
 179                 return True
 180             elif line[0:5] == ";>>> ":     # settings/commands
 181                 settings = {}
 182                 exec(line[5:], {}, settings)
 183                 if 'soft' in settings:
 184                     self.soft = settings['soft']
 185                 if 'deferrable' in settings and settings['deferrable']:
 186                     self.deferrable = "\nSkipping deferrable and optional tests"
 187                     return True
 188                 if 'optional' in settings and settings['optional']:
 189                     self.optional = "\nSkipping optional tests"
 190                     return True
 191                 continue
 192             elif line[0:1] == ";":         # unexpected comment
 193                 log("Test data error at line %d:\n%s" % (self.line_num, line))
 194                 return None
 195             self.form = line   # the line is a form to send
 196
 197             # Now find the output and return value
 198             while self.data:
 199                 line = self.data[0]
 200                 if line[0:3] == ";=>":
 201                     self.ret = line[3:]
 202                     self.line_num += 1
 203                     self.data.pop(0)
 204                     break
 205                 elif line[0:2] == "; ":
 206                     self.out = self.out + line[2:] + sep
 207                     self.line_num += 1
 208                     self.data.pop(0)
 209                 else:
 210                     self.ret = "*"
 211                     break
 212             if self.ret: break
 213
 214         return self.form
 215
 216 args = parser.parse_args(sys.argv[1:])
 217 # Workaround argparse issue with two '--' on command line
 218 if sys.argv.count('--') > 0:
 219     args.mal_cmd = sys.argv[sys.argv.index('--')+1:]
 220
 221 if args.rundir: os.chdir(args.rundir)
 222
 223 if args.log_file:   log_file   = open(args.log_file, "a")
 224 if args.debug_file: debug_file = open(args.debug_file, "a")
 225
 226 r = Runner(args.mal_cmd, no_pty=args.no_pty)
 227 t = TestReader(args.test_file)
 228
 229
 230 def assert_prompt(runner, prompts, timeout):
 231     # Wait for the initial prompt
 232     header = runner.read_to_prompt(prompts, timeout=timeout)
 233     if not header == None:
 234         if header:
 235             log("Started with:\n%s" % header)
 236     else:
 237         log("Did not one of following prompt(s): %s" % repr(prompts))
 238         log("    Got      : %s" % repr(r.buf))
 239         sys.exit(1)
 240
 241
 242 # Wait for the initial prompt
 243 assert_prompt(r, ['user> ', 'mal-user> '], args.start_timeout)
 244
 245 # Send the pre-eval code if any
 246 if args.pre_eval:
 247     sys.stdout.write("RUNNING pre-eval: %s" % args.pre_eval)
 248     p.write(args.pre_eval)
 249     assert_prompt(args.test_timeout)
 250
 251 test_cnt = 0
 252 pass_cnt = 0
 253 fail_cnt = 0
 254 soft_fail_cnt = 0
 255 failures = []
 256
 257 while t.next():
 258     if args.deferrable == False and t.deferrable:
 259         log(t.deferrable)
 260         break
 261
 262     if args.optional == False and t.optional:
 263         log(t.optional)
 264         break
 265
 266     if t.msg != None:
 267         log(t.msg)
 268         continue
 269
 270     if t.form == None: continue
 271
 272     log("TEST: %s -> [%s,%s]" % (t.form, repr(t.out), t.ret), end='')
 273
 274     # The repeated form is to get around an occasional OS X issue
 275     # where the form is repeated.
 276     # https://github.com/kanaka/mal/issues/30
 277     expected = ["%s%s%s%s" % (t.form, sep, t.out, t.ret),
 278                 "%s%s%s%s%s%s" % (t.form, sep, t.form, sep, t.out, t.ret)]
 279
 280     r.writeline(t.form)
 281     try:
 282         test_cnt += 1
 283         res = r.read_to_prompt(['\r\nuser> ', '\nuser> ',
 284                                 '\r\nmal-user> ', '\nmal-user> '],
 285                                 timeout=args.test_timeout)
 286         #print "%s,%s,%s" % (idx, repr(p.before), repr(p.after))
 287         if t.ret == "*" or res in expected:
 288             log(" -> SUCCESS")
 289             pass_cnt += 1
 290         else:
 291             if t.soft and not args.hard:
 292                 log(" -> SOFT FAIL (line %d):" % t.line_num)
 293                 soft_fail_cnt += 1
 294                 fail_type = "SOFT "
 295             else:
 296                 log(" -> FAIL (line %d):" % t.line_num)
 297                 fail_cnt += 1
 298                 fail_type = ""
 299             log("    Expected : %s" % repr(expected[0]))
 300             log("    Got      : %s" % repr(res))
 301             failed_test = """%sFAILED TEST (line %d): %s -> [%s,%s]:
 302     Expected : %s
 303     Got      : %s""" % (fail_type, t.line_num, t.form, repr(t.out), t.ret, repr(expected[0]), repr(res))
 304             failures.append(failed_test)
 305     except:
 306         _, exc, _ = sys.exc_info()
 307         log("\nException: %s" % repr(exc))
 308         log("Output before exception:\n%s" % r.buf)
 309         sys.exit(1)
 310
 311 if len(failures) > 0:
 312     log("\nFAILURES:")
 313     for f in failures:
 314         log(f)
 315
 316 results = """
 317 TEST RESULTS (for %s):
 318   %3d: soft failing tests
 319   %3d: failing tests
 320   %3d: passing tests
 321   %3d: total tests
 322 """ % (args.test_file.name, soft_fail_cnt, fail_cnt,
 323         pass_cnt, test_cnt)
 324 log(results)
 325
 326 debug("\n") # add some separate to debug log
 327
 328 if fail_cnt > 0:
 329     sys.exit(1)
 330 sys.exit(0)