test-harness

#!/usr/bin/env python3

links = './links'

import sys, re, time, os, signal
import concurrent.futures, threading, multiprocessing
from os import path
from subprocess import Popen, PIPE

successes = failures = ignored = 0
TIMEOUT = 15 # seconds before a test is timed out.

the_lock = threading.Lock()

def FAIL(name, msg):
    return ('  error: %s' % name,'      (%s)' % msg)

# must only be called while having the lock
def OK(name):
    global successes
    successes+=1
    print(' SUCCESS: %s' % name)

def parse(stream):
    """Read test information separated by blank lines.  The first line
    is the test name; the second is the test code; the remaining lines
    are auxiliary options"""
    def tests():
        tests = []
        for line in stream:
            if not str.strip(line) and tests[-1]:
                yield tests
                tests = []
            else:
                tests.append(line)
        yield tests
    for test in tests():
        if not test:
            print('WARNING: empty test', file=sys.stderr)
        else:
            name = str.strip(test.pop(0))
            code = str.strip(test.pop(0))
            opts = dict([str.split(str.rstrip(line), ' : ', 1) for line in test])
            yield name, code, opts

# Lock not required for calling
def check_expected(name, item, got, expected, errors):
    """Check some data against an expected value or predicate.  If it
    matches, return True.  Otherwise, display an error and return
    False.
    """
    got, expected = got.rstrip(), expected.rstrip()
    if expected.startswith('@'):
        if not re.match(expected[1:], got, re.DOTALL):
            errors.append(FAIL(name, "Unexpected %s: expected `%s' (regex); got `%s'" %
                               (item, expected[1:], got)))
            return False
        else:
            return True
    elif expected != got:
        errors.append(FAIL(name, "Unexpected %s:\nexpected `%s';\n     got `%s'" % (item, expected, got)))
        return False
    else:
        return True

# Ensures "foo bar" is interpreted as a single argument and not two.
def patch_arguments(args = []):
    result = []
    i = 0
    while i < len(args):
        if args[i].startswith("\""):
            components = [args[i][1:]]
            i += 1
            while i < len(args):
                if args[i].endswith("\""):
                    components.append(args[i][0:len(args[i])-1])
                    i +=1
                    result.append(" ".join(components))
                    break
                else:
                    components.append(args[i])
                    i += 1
        else:
            result.append(args[i])
            i += 1
    return result

def split_arguments(args = []):
    links_args = []
    rest_args  = []
    i = 0
    for i in range(0, len(args)):
        if args[i] == "--":
            rest_args = args[i:]
            break
        else:
            links_args.append(args[i])
    return (links_args, rest_args)

def evaluate(name, code, config_file, stdout='', stderr='', exit = '0', env = None, filemode='', args='', ignore = None):
    arg_array, rest_array = split_arguments(patch_arguments(str.split(args)))
    global failures

    if config_file != None:
        for arg in arg_array:
            if arg.startswith("--config"):
                print(
                    "FAILURE: Test \"%s\" comes with an args entry to specify the config file," % name +
                    " but a config file was also passed as a command-line argument to the test harness" )
                failures += 1
                return

        arg_array = ["--config=" + config_file] + arg_array

    if not filemode.startswith('true'):
        arg_array += ["-e"]

    #starts shell executing links in new session. This allows us to kill the
    #shell and its sub-processes (i.e., links) in case of a timeout
    proc = Popen([links] + arg_array + [code] + rest_array, stdout=PIPE, stderr=PIPE, env=env, start_new_session=True)

    passed = True
    errors = []
    for i in range(0, TIMEOUT*100):
        rc = proc.poll()
        if rc != None:
            passed &= check_expected(name, 'return code', str(rc), exit, errors)
            passed &= check_expected(name, 'stdout', proc.stdout.read().decode('ascii'), stdout, errors)
            passed &= check_expected(name, 'stderr', str(proc.stderr.read().decode('ascii')), stderr, errors)
            if passed:
                with the_lock:
                    OK(name)
            else:
                if ignore != None:
                    global ignored
                    with the_lock:
                        ignored += 1
                        print('?IGNORED: %s (%s)' % (name, ignore))
                else:
                    with the_lock:
                        failures += 1
                        print('!FAILURE: %s' % name)
                        for i, j in errors:
                            print(i)
                            print(j)
            return
        else:
            time.sleep(0.01)

    #Kill the whole process group we started earlier due to timeout
    pgid = os.getpgid(proc.pid)
    os.killpg(pgid, signal.SIGKILL)
    with the_lock:
        failures += 1
        print('!FAILURE: %s [TIMED OUT]' % name)

def peek_line(f):
    pos = f.tell()
    line = f.readline()
    f.seek(pos)
    return line

def parseTestScript(stream):
    header = dict()
    if peek_line(stream) == "---\n":
        stream.readline()
        line_num = 1
        line = peek_line(stream)
        while not (line == "---\n"):
            raw_line = stream.readline()
            line_num += 1
            line = str.strip(raw_line)
            keyvalue = list(map(str.strip, str.split(line, ':', 1)))
            if len(keyvalue) == 2:
                header[keyvalue[0]] = keyvalue[1]
            else:
                raise SystemExit("Syntax error on line " + str(line_num) + ": " + raw_line)
            line = peek_line(stream)
        stream.readline()
        stream.readline()
    yield header
    yield from parse(stream)


def main():
    config_file = None

    if len(sys.argv) == 3:
        filename = sys.argv[1]
        config_file =  sys.argv[2]
    elif len(sys.argv) == 2:
        filename = sys.argv[1]
    else:
        raise SystemExit('Usage: run <test file> [<links config file>]')

    cpus=multiprocessing.cpu_count()
    tp = concurrent.futures.ThreadPoolExecutor(max_workers=cpus)

    stream = parseTestScript(open(filename, 'r'))
    # Process the header, if any.
    header = next(stream)
    if config_file == None and "config" in header:
        config_file = header["config"]
        # Must remove the "config" key because we combine the header-provided
        # options with the per-file ones later on. However, the global config
        # file must be provided via the config_file parameter of evaluate only.
        del header["config"]

    jobs = []
    # Process the body.
    for name, code, opts in stream:
        # Combines the options set in the header with the ones specific to this
        # test case. The latter ones overwrite the ones from the header.
        opts = {**header, **opts}

        # enqueues execution of
        # evaluate(name, code, config_file, **opts)
        # in the thread pool
        jobs.append((name, tp.submit(evaluate, name, code, config_file, **opts)))

    # Wait for all jobs to finish
    tp.shutdown(wait=True)

    # Check that none of the jobs raised an exception:
    for (job_name, result) in jobs:
        if result.exception():
            raise SystemExit(("Error: Test \"%s\" failed to execute, producing "
                              "the following exception:\n%s")
                             % (job_name, result.exception()))

    print("%d failures (+%d ignored)\n%d successes\n" % (failures, ignored, successes))
    if failures > 0:
        sys.exit(1)
    else:
        sys.exit(0)

if __name__ == '__main__':
    main()