mirror of
https://github.com/kanaka/mal.git
synced 2024-08-16 17:20:23 +03:00
9d7fb0fd8d
If a test times out, that means it hasn't returned a prompt within reasonable time. Continuing to type at it is unlikely to be helpful, since the lack of a prompt indicates that the subprocess has hung, so subsequent tests are likely to time out as well. This is particularly annoying in step 5, where the line that usually times out, "(def! res2 (sum2 10000 0))" has its output ignored, so a timeout is treated as success, meaning that runtest proceeds to the next test, which of course also times out.
351 lines
12 KiB
Python
Executable File
351 lines
12 KiB
Python
Executable File
#!/usr/bin/env python
|
|
|
|
from __future__ import print_function
|
|
import os, sys, re
|
|
import argparse, time
|
|
import signal, atexit
|
|
|
|
from subprocess import Popen, STDOUT, PIPE
|
|
from select import select
|
|
|
|
# Pseudo-TTY and terminal manipulation
|
|
import pty, array, fcntl, termios
|
|
|
|
IS_PY_3 = sys.version_info[0] == 3
|
|
|
|
debug_file = None
|
|
log_file = None
|
|
|
|
def debug(data):
|
|
if debug_file:
|
|
debug_file.write(data)
|
|
debug_file.flush()
|
|
|
|
def log(data, end='\n'):
|
|
if log_file:
|
|
log_file.write(data + end)
|
|
log_file.flush()
|
|
print(data, end=end)
|
|
sys.stdout.flush()
|
|
|
|
sep = "\n"
|
|
rundir = None
|
|
|
|
parser = argparse.ArgumentParser(
|
|
description="Run a test file against a Mal implementation")
|
|
parser.add_argument('--rundir',
|
|
help="change to the directory before running tests")
|
|
parser.add_argument('--start-timeout', default=10, type=int,
|
|
help="default timeout for initial prompt")
|
|
parser.add_argument('--test-timeout', default=20, type=int,
|
|
help="default timeout for each individual test action")
|
|
parser.add_argument('--pre-eval', default=None, type=str,
|
|
help="Mal code to evaluate prior to running the test")
|
|
parser.add_argument('--no-pty', action='store_true',
|
|
help="Use direct pipes instead of pseudo-tty")
|
|
parser.add_argument('--log-file', type=str,
|
|
help="Write messages to the named file in addition the screen")
|
|
parser.add_argument('--debug-file', type=str,
|
|
help="Write all test interaction the named file")
|
|
parser.add_argument('--hard', action='store_true',
|
|
help="Turn soft tests (soft, deferrable, optional) into hard failures")
|
|
|
|
# Control whether deferrable and optional tests are executed
|
|
parser.add_argument('--deferrable', dest='deferrable', action='store_true',
|
|
help="Enable deferrable tests that follow a ';>>> deferrable=True'")
|
|
parser.add_argument('--no-deferrable', dest='deferrable', action='store_false',
|
|
help="Disable deferrable tests that follow a ';>>> deferrable=True'")
|
|
parser.set_defaults(deferrable=True)
|
|
parser.add_argument('--optional', dest='optional', action='store_true',
|
|
help="Enable optional tests that follow a ';>>> optional=True'")
|
|
parser.add_argument('--no-optional', dest='optional', action='store_false',
|
|
help="Disable optional tests that follow a ';>>> optional=True'")
|
|
parser.set_defaults(optional=True)
|
|
|
|
parser.add_argument('test_file', type=str,
|
|
help="a test file formatted as with mal test data")
|
|
parser.add_argument('mal_cmd', nargs="*",
|
|
help="Mal implementation command line. Use '--' to "
|
|
"specify a Mal command line with dashed options.")
|
|
parser.add_argument('--crlf', dest='crlf', action='store_true',
|
|
help="Write \\r\\n instead of \\n to the input")
|
|
|
|
class Runner():
|
|
def __init__(self, args, no_pty=False, line_break="\n"):
|
|
#print "args: %s" % repr(args)
|
|
self.no_pty = no_pty
|
|
|
|
# Cleanup child process on exit
|
|
atexit.register(self.cleanup)
|
|
|
|
self.p = None
|
|
env = os.environ
|
|
env['TERM'] = 'dumb'
|
|
env['INPUTRC'] = '/dev/null'
|
|
env['PERL_RL'] = 'false'
|
|
if no_pty:
|
|
self.p = Popen(args, bufsize=0,
|
|
stdin=PIPE, stdout=PIPE, stderr=STDOUT,
|
|
preexec_fn=os.setsid,
|
|
env=env)
|
|
self.stdin = self.p.stdin
|
|
self.stdout = self.p.stdout
|
|
else:
|
|
# provide tty to get 'interactive' readline to work
|
|
master, slave = pty.openpty()
|
|
|
|
# Set terminal size large so that readline will not send
|
|
# ANSI/VT escape codes when the lines are long.
|
|
buf = array.array('h', [100, 200, 0, 0])
|
|
fcntl.ioctl(master, termios.TIOCSWINSZ, buf, True)
|
|
|
|
self.p = Popen(args, bufsize=0,
|
|
stdin=slave, stdout=slave, stderr=STDOUT,
|
|
preexec_fn=os.setsid,
|
|
env=env)
|
|
# Now close slave so that we will get an exception from
|
|
# read when the child exits early
|
|
# http://stackoverflow.com/questions/11165521
|
|
os.close(slave)
|
|
self.stdin = os.fdopen(master, 'r+b', 0)
|
|
self.stdout = self.stdin
|
|
|
|
#print "started"
|
|
self.buf = ""
|
|
self.last_prompt = ""
|
|
|
|
self.line_break = line_break
|
|
|
|
def read_to_prompt(self, prompts, timeout):
|
|
end_time = time.time() + timeout
|
|
while time.time() < end_time:
|
|
[outs,_,_] = select([self.stdout], [], [], 1)
|
|
if self.stdout in outs:
|
|
new_data = self.stdout.read(1)
|
|
new_data = new_data.decode("utf-8") if IS_PY_3 else new_data
|
|
#print("new_data: '%s'" % new_data)
|
|
debug(new_data)
|
|
# Perform newline cleanup
|
|
self.buf += new_data.replace("\r", "")
|
|
for prompt in prompts:
|
|
regexp = re.compile(prompt)
|
|
match = regexp.search(self.buf)
|
|
if match:
|
|
end = match.end()
|
|
buf = self.buf[0:match.start()]
|
|
self.buf = self.buf[end:]
|
|
self.last_prompt = prompt
|
|
return buf
|
|
return None
|
|
|
|
def writeline(self, str):
|
|
def _to_bytes(s):
|
|
return bytes(s, "utf-8") if IS_PY_3 else s
|
|
|
|
self.stdin.write(_to_bytes(str.replace('\r', '\x16\r') + self.line_break))
|
|
|
|
def cleanup(self):
|
|
#print "cleaning up"
|
|
if self.p:
|
|
try:
|
|
os.killpg(self.p.pid, signal.SIGTERM)
|
|
except OSError:
|
|
pass
|
|
self.p = None
|
|
|
|
class TestReader:
|
|
def __init__(self, test_file):
|
|
self.line_num = 0
|
|
f = open(test_file, newline='') if IS_PY_3 else open(test_file)
|
|
self.data = f.read().split('\n')
|
|
self.soft = False
|
|
self.deferrable = False
|
|
self.optional = False
|
|
|
|
def next(self):
|
|
self.msg = None
|
|
self.form = None
|
|
self.out = ""
|
|
self.ret = None
|
|
|
|
while self.data:
|
|
self.line_num += 1
|
|
line = self.data.pop(0)
|
|
if re.match(r"^\s*$", line): # blank line
|
|
continue
|
|
elif line[0:3] == ";;;": # ignore comment
|
|
continue
|
|
elif line[0:2] == ";;": # output comment
|
|
self.msg = line[3:]
|
|
return True
|
|
elif line[0:5] == ";>>> ": # settings/commands
|
|
settings = {}
|
|
exec(line[5:], {}, settings)
|
|
if 'soft' in settings:
|
|
self.soft = settings['soft']
|
|
if 'deferrable' in settings and settings['deferrable']:
|
|
self.deferrable = "\nSkipping deferrable and optional tests"
|
|
return True
|
|
if 'optional' in settings and settings['optional']:
|
|
self.optional = "\nSkipping optional tests"
|
|
return True
|
|
continue
|
|
elif line[0:1] == ";": # unexpected comment
|
|
raise Exception("Test data error at line %d:\n%s" % (self.line_num, line))
|
|
self.form = line # the line is a form to send
|
|
|
|
# Now find the output and return value
|
|
while self.data:
|
|
line = self.data[0]
|
|
if line[0:3] == ";=>":
|
|
self.ret = line[3:]
|
|
self.line_num += 1
|
|
self.data.pop(0)
|
|
break
|
|
elif line[0:2] == ";/":
|
|
self.out = self.out + line[2:] + sep
|
|
self.line_num += 1
|
|
self.data.pop(0)
|
|
else:
|
|
self.ret = ""
|
|
break
|
|
if self.ret != None: break
|
|
|
|
if self.out[-1:] == sep and not self.ret:
|
|
# If there is no return value, output should not end in
|
|
# separator
|
|
self.out = self.out[0:-1]
|
|
return self.form
|
|
|
|
args = parser.parse_args(sys.argv[1:])
|
|
# Workaround argparse issue with two '--' on command line
|
|
if sys.argv.count('--') > 0:
|
|
args.mal_cmd = sys.argv[sys.argv.index('--')+1:]
|
|
|
|
if args.rundir: os.chdir(args.rundir)
|
|
|
|
if args.log_file: log_file = open(args.log_file, "a")
|
|
if args.debug_file: debug_file = open(args.debug_file, "a")
|
|
|
|
r = Runner(args.mal_cmd, no_pty=args.no_pty, line_break="\r\n" if args.crlf else "\n")
|
|
t = TestReader(args.test_file)
|
|
|
|
|
|
def assert_prompt(runner, prompts, timeout):
|
|
# Wait for the initial prompt
|
|
header = runner.read_to_prompt(prompts, timeout=timeout)
|
|
if not header == None:
|
|
if header:
|
|
log("Started with:\n%s" % header)
|
|
else:
|
|
log("Did not receive one of following prompt(s): %s" % repr(prompts))
|
|
log(" Got : %s" % repr(r.buf))
|
|
sys.exit(1)
|
|
|
|
|
|
# Wait for the initial prompt
|
|
try:
|
|
assert_prompt(r, ['[^\s()<>]+> '], args.start_timeout)
|
|
except:
|
|
_, exc, _ = sys.exc_info()
|
|
log("\nException: %s" % repr(exc))
|
|
log("Output before exception:\n%s" % r.buf)
|
|
sys.exit(1)
|
|
|
|
# Send the pre-eval code if any
|
|
if args.pre_eval:
|
|
sys.stdout.write("RUNNING pre-eval: %s" % args.pre_eval)
|
|
r.writeline(args.pre_eval)
|
|
assert_prompt(r, ['[^\s()<>]+> '], args.test_timeout)
|
|
|
|
test_cnt = 0
|
|
pass_cnt = 0
|
|
fail_cnt = 0
|
|
soft_fail_cnt = 0
|
|
failures = []
|
|
|
|
class TestTimeout(Exception):
|
|
pass
|
|
|
|
while t.next():
|
|
if args.deferrable == False and t.deferrable:
|
|
log(t.deferrable)
|
|
break
|
|
|
|
if args.optional == False and t.optional:
|
|
log(t.optional)
|
|
break
|
|
|
|
if t.msg != None:
|
|
log(t.msg)
|
|
continue
|
|
|
|
if t.form == None: continue
|
|
|
|
log("TEST: %s -> [%s,%s]" % (repr(t.form), repr(t.out), t.ret), end='')
|
|
|
|
# The repeated form is to get around an occasional OS X issue
|
|
# where the form is repeated.
|
|
# https://github.com/kanaka/mal/issues/30
|
|
expects = [".*%s%s%s" % (sep, t.out, re.escape(t.ret)),
|
|
".*%s.*%s%s%s" % (sep, sep, t.out, re.escape(t.ret))]
|
|
|
|
r.writeline(t.form)
|
|
try:
|
|
test_cnt += 1
|
|
res = r.read_to_prompt(['\r\n[^\s()<>]+> ', '\n[^\s()<>]+> '],
|
|
timeout=args.test_timeout)
|
|
#print "%s,%s,%s" % (idx, repr(p.before), repr(p.after))
|
|
if (res == None):
|
|
log(" -> TIMEOUT (line %d)" % t.line_num)
|
|
raise TestTimeout("TIMEOUT (line %d)" % t.line_num)
|
|
elif (t.ret == "" and t.out == ""):
|
|
log(" -> SUCCESS (result ignored)")
|
|
pass_cnt += 1
|
|
elif (re.search(expects[0], res, re.S) or
|
|
re.search(expects[1], res, re.S)):
|
|
log(" -> SUCCESS")
|
|
pass_cnt += 1
|
|
else:
|
|
if t.soft and not args.hard:
|
|
log(" -> SOFT FAIL (line %d):" % t.line_num)
|
|
soft_fail_cnt += 1
|
|
fail_type = "SOFT "
|
|
else:
|
|
log(" -> FAIL (line %d):" % t.line_num)
|
|
fail_cnt += 1
|
|
fail_type = ""
|
|
log(" Expected : %s" % repr(expects[0]))
|
|
log(" Got : %s" % repr(res))
|
|
failed_test = """%sFAILED TEST (line %d): %s -> [%s,%s]:
|
|
Expected : %s
|
|
Got : %s""" % (fail_type, t.line_num, t.form, repr(t.out),
|
|
t.ret, repr(expects[0]), repr(res))
|
|
failures.append(failed_test)
|
|
except:
|
|
_, exc, _ = sys.exc_info()
|
|
log("\nException: %s" % repr(exc))
|
|
log("Output before exception:\n%s" % r.buf)
|
|
sys.exit(1)
|
|
|
|
if len(failures) > 0:
|
|
log("\nFAILURES:")
|
|
for f in failures:
|
|
log(f)
|
|
|
|
results = """
|
|
TEST RESULTS (for %s):
|
|
%3d: soft failing tests
|
|
%3d: failing tests
|
|
%3d: passing tests
|
|
%3d: total tests
|
|
""" % (args.test_file, soft_fail_cnt, fail_cnt,
|
|
pass_cnt, test_cnt)
|
|
log(results)
|
|
|
|
debug("\n") # add some separate to debug log
|
|
|
|
if fail_cnt > 0:
|
|
sys.exit(1)
|
|
sys.exit(0)
|