makereport

#!/usr/bin/env python3

from builtins import object
from copy                    import deepcopy
from datetime                import datetime
from glob                    import iglob as glob
from test.test               import ensure_dir
from urllib.request          import urlopen
from concurrent.futures import ProcessPoolExecutor
import fnmatch
import importlib.util
import multiprocessing
import optparse
import os
import sys
import traceback

_report_name    = "stats-" + datetime.now().strftime("%Y.%m.%d")
_configurations = dict()
_out            = sys.stdout

_opts = optparse.OptionParser(version="%prog 0.1", usage="%prog [options]")
_opts.set_defaults(
    debug=False,
    verbose=False,
    compile_timeout=60,
    execute_timeout=30,
    threads=multiprocessing.cpu_count() + 1,
    show_disappeared=False,
    reportdir="reports/",
    builddir="build/",
    faillog_out=None,
    arch_cflags="-march=native -m32",
    cflags="-O3",
    ldflags="-lm",
    arch_ldflags="-m32",
    expect_url="https://raw.githubusercontent.com/libfirm/testresults/master/fail_expectations-i686-linux-gnu",
    fetch = True,
    default_dirs=[
        "cparser/should_work",
        "cparser/should_warn",
        "cparser/should_fail",
        "cparser/lit",
        "preproctest",
        "preproctest/should_warn",
        "preproctest/should_fail",
        "backend",
        "opt",
        "C",
        "C/pragmatic",
        "C/should_fail",
        "C/should_warn",
        "C/should_warn_pedantic",
        "C/nowarn",
        "C/gnu99",
        "C/gnu99/should_warn",
        "C++/should_fail",
        "ack",
        "langshootout",
        "lit",
        "llvm",
        "gcc"
    ],
    arch_dirs=[
        "x86code"
    ],
    runexe="")
_opts.add_option("-d", "--debug", dest="debug", action="store_true",
                 help="Enable debug messages")
_opts.add_option("-v", "--verbose", dest="verbose", action="store_true",
                 help="More output")
_opts.add_option("-c", "--compile-times", dest="compile_times",
                 action="store_true",
                 help="Display compile time of each program")
_opts.add_option("-t", "--threads", dest="threads", type="int",
                 help="Number of threads to use")
_opts.add_option("--faillog", dest="faillog_out",
                 help="Write faillog into file specified here")
_opts.add_option("--expect", dest="expect_url",
                 help="file/url with fail expectations")
_opts.add_option("--no-fetch", dest="fetch", action="store_false")
_opts.add_option("--relax-expect", dest="relax_expect",
                 action="store_true", help="expect any error instead of specific ones")
_opts.add_option("--show-disappeared", dest="show_disappeared",
                 action="store_true", help="show disappeared tests")
_opts.add_option("--compile-timeout", dest="compile_timeout", type="int", help="Timeout for test compilation")
_opts.add_option("--execute-timeout", dest="execute_timeout", type="int", help="Timeout for test execution")

# List of (matchfunc, factory) tuples.
# if matchfunc(filename) returns True then factory(filename, environment) will
# be used to construct a Test for this file.
_test_factories = []
# List of (wildcard, default_factory) tuples.
# If a user specifies a directory, then we will construct tests for files
# matching "directory/wildcard" and use a factory from the test_factories list,
# if none is found default_factory(filename, environment).
_wildcard_factories = []

_console_red    = "\033[1;31m"
_console_green  = "\033[1;32m"
_console_yellow = "\033[1;33m"
_console_bold   = "\033[1m"
_console_normal = "\033[m"
_console_clear  = "\r\033[K"
_rc             = 0  # application return code
if not sys.stdout.isatty():
    _console_red    = ""
    _console_green  = ""
    _console_yellow = ""
    _console_bold   = ""
    _console_normal = ""
    _console_clear  = ""

def load_expectations(fetched, url, environment):
    if fetched:
        if os.path.isfile(url):
            cachefile = url
            fetched = False
        else:
            try:
                fetched = True
                input   = urlopen(url)
                if input.getcode() is not None and input.getcode() != 200:
                    fetched = False
            except:
                fetched = False
            if not fetched:
                _out.write("Warning: Couldn't download from '%s', using cache file\n" % url)
            cachefile = environment.builddir + "/" + os.path.basename(url)
    else:
        cachefile = environment.builddir + "/" + os.path.basename(url)

    if not fetched:
        input = open(cachefile, "rb")

    results = dict()
    for line in input:
        line = line.decode("utf-8")
        try:
            i = line.index(" ")
        except ValueError:
            continue
        test_id          = line[:i]
        result           = line[i:].strip()
        results[test_id] = result

    cache = open(cachefile, "w")
    for (test_id, result) in results.items():
        cache.write("%-40s %s\n" % (test_id, result))
    cache.close()

    return results


def expectation_match(error, expectation, config):
    if error.endswith(" compile errors") and \
       expectation.endswith(" compile errors"):
        # exact number of compile errors does not matter
        return True
    if error == "compile: SIGXCPU" and expectation == "compile: SIGKILL" or \
       error == "execute: SIGXCPU" and expectation == "execute: SIGKILL":
        # BSD on timeout signals SIGXCPU and Linux SIGKILL
        return True
    if error == "compile: SIGBUS" and expectation == "compile: SIGSEGV" or \
       error == "execute: SIGBUS" and expectation == "execute: SIGSEGV":
        # OS X often signals SIGBUS where others signal SIGSEGV
        return True
    if config.relax_expect:
       # consider any kind of error equivalent
       for er in ("execute: ", "compile: ", "preprocess: "):
          for ex in ("execute: ", "compile: ", "preprocess: "):
             if error.startswith(er) and expectation.startswith(ex):
                return True
    return error == expectation


def console_output(config, test, expected):
    timing = ""
    prefix = ""
    if config.compile_times:
        time = None
        if hasattr(test, "stepresults") and "compile" in test.stepresults:
            result = test.stepresults["compile"]
            if hasattr(result, "time"):
                time = result.time
        if time is not None:
            timing = " [%s%.2fs%s]" % (_console_yellow, time, _console_normal)
    result = test.result
    write_details = False
    if expected and expectation_match(result, expected, config):
        if test.success and not config.verbose:
            newline = "\n" if _console_clear == "" else ""
            _out.write("%s%s...%s" % (_console_clear, test.id, newline))
            _out.flush()
            return
    else:
        if test.success:
            prefix = _console_green
        elif expected and expected != "ok":
            prefix = _console_yellow
        else:
            prefix = _console_red

        if expected and not test.success:
            global _rc
            _rc = 1

        if not test.success and config.verbose:
            write_details = True

        if expected:
            result += " (expected %s)" % expected
        else:
            result += " (new)"
    _out.write("%s%s%-40s %s%s%s\n" % (_console_clear, prefix, test.id, result, _console_normal, timing))
    if write_details:
        for step in test.steps:
            result = test.stepresults.get(step.name)
            if result is None:
                continue
            _out.write("===> Step '%s': %s\n" % (step.name, result.error))
            _out.write("> %s\n" % (result.cmd,))
            _out.write("retcode %s\n" % (result.retcode,))
            _out.write("time %s\n" % (result.time,))
            _out.write("%s\n" % (result.stdout.decode("utf-8", "replace"),))
            _out.write("%s\n" % (result.stderr.decode("utf-8", "replace"),))

    _out.flush()


class Report(object):
    def __init__(self):
        self.tests   = list()
        self.summary = (0, 0)

    def add_test(self, test):
        self.tests.append(test)
        fail = not test.success
        self.summary = (self.summary[0]+1, self.summary[1]+int(fail))

    def print_summary(self):
        _out.write("%s---------------------------\n" % (_console_clear))
        _out.write("Ran %d tests, of which %s%d failed%s.\n" %
                   (self.summary[0], _console_bold, self.summary[1], _console_normal))
        _out.flush()

    def write_faillog(self, fh, config):
        for test in self.tests:
            fh.write("%-40s %s\n" % (test.id, test.result))


def make_test(environment, filename, default_factory=None):
    """Given a filename create a new Test object"""
    global _test_factories
    global _wildcard_factories
    # Each test gets its own environment
    environment = deepcopy(environment)

    # Determine Tester to be used
    factory = default_factory
    for (test, tfactory) in _test_factories:
        if test(filename):
            factory = tfactory
            break
    # try wildcard factories
    if factory is None:
        for (pattern, tfactory) in _wildcard_factories:
            if fnmatch.fnmatch(filename, pattern):
                factory = tfactory
                break
    if factory is None:
        sys.stderr.write("Couldn't determine Test factory for '%s'\n" % filename)
        return None

    try:
        test = factory(environment, filename)
    except Exception as e:
        sys.stderr.write("Couldn't create test '%s': %s\n%s\n" % (filename, e, traceback.format_exc()))
        return None
    if test is None:
        sys.stderr.write("Factory returned None for test '%s'\n" % filename)
        return None
    if not hasattr(test, 'run'):
        sys.stderr.write("Factory created invalid test for '%s': test object has no run method\n" % (filename))
        return None

    return test


def make_tests_from_dir(environment, directory):
    global _wildcard_factories
    names = []
    for (wildcard, default_factory) in _wildcard_factories:
        for name in glob("%s/%s" % (directory, wildcard)):
            names.append((name, default_factory))
    tests = []
    for (name, default_factory) in names:
        environment.filename = name
        test = make_test(environment, name, default_factory)
        if test:
            tests.append(test)
    return tests


def default_create_testset(config, args):
    tests = []
    if not args:
        args = config.default_dirs + config.arch_dirs

    for arg in args:
        if os.path.isdir(arg):
            tests += make_tests_from_dir(config, arg)
        else:
            test = make_test(config, arg)
            if test:
                tests.append(test)
    # Sort tests by their name
    tests.sort(key=lambda x: x.id)
    return tests


def run_test(test):
    try:
        test.run()
    except Exception as e:
        sys.stderr.write("Exception in %s: %s\n%s\n" % (test.id, e, traceback.format_exc()))
        raise e
    return test


def makereport(config, args):
    ensure_dir(config.reportdir)
    ensure_dir(config.builddir)

    expectations = {}
    if config.expect_url:
        try:
            expectations = load_expectations(config.fetch, config.expect_url, config)
        except Exception as e:
            _out.write("Couldn't load fail expectations: %s\n" % e)

    tests = config.create_testset(config, args)
    # Avoid problems when pickling this function in multiprocessing...
    config.create_testset = None

    # create test futures for parallel evaluation
    queue    = list()
    executor = ProcessPoolExecutor(max_workers=config.threads)
    for test in tests:
        queue.append(executor.submit(run_test, test))

    # collect report
    r = Report()
    try:
        found = {}
        for promise in queue:
            test           = promise.result()
            found[test.id] = True
            expected       = expectations.get(test.id)
            console_output(config, test, expected)
            r.add_test(test)
        if config.show_disappeared:
            for t in expectations:
                if t not in found:
                    _out.write("%s%-40s %s%s\n" % (_console_red, t, "test disappeared", _console_normal))
                    _out.flush()

    except KeyboardInterrupt:
        _out.write("Received interrupt signal, shutting down\n")
        for promise in queue:
            promise.cancel()
        executor.shutdown(wait=False)
    faillog_out = config.faillog_out
    if faillog_out is None:
        faillog_out = config.reportdir + "/" + _report_name + ".faillog"
    r.write_faillog(open(faillog_out, 'w'), config)
    r.print_summary()


def main():
    global _rc
    os.putenv("LC_ALL", "C")  # normalize environment

    # Add new commandline options for all configurations
    for (config, setupfunc) in sorted(_configurations.items()):
        _opts.add_option("--%s" % config, action="callback", callback=setupfunc,
                         help="activate %s configuration" % config)

    _opts.set_defaults(create_testset=default_create_testset)

    config, args = _opts.parse_args()
    makereport(config, args)
    return _rc


def try_load_plugin(filename):
    global _test_factories
    global _wildcard_factories
    global _configurations
    global _opts

    # do not clutter our plugin directories with .pyc files...
    sys.dont_write_bytecode = True

    try:
        dirname      = os.path.dirname(filename)
        basename     = os.path.basename(filename)
        parentmodule = dirname.replace("/", ".")

        modulename = "%s.%s" % (parentmodule, basename.replace(".py", ""))
        spec = importlib.util.spec_from_file_location(modulename, filename)
        plugin = importlib.util.module_from_spec(spec)
        sys.modules[modulename] = plugin
        spec.loader.exec_module(plugin)
        if hasattr(plugin, "test_factories"):
            _test_factories += plugin.test_factories
        if hasattr(plugin, "wildcard_factories"):
            for (wildcard, default_factory) in plugin.wildcard_factories:
                for (ewildcard, _) in _wildcard_factories:
                    if ewildcard == wildcard:
                        sys.stderr.write("Warning: plugin '%s' provides a second factory for wildcard '%s'\n" % (filename, ewildcard))
                        break
            _wildcard_factories += plugin.wildcard_factories
        if hasattr(plugin, "configurations"):
            for (config, setup) in plugin.configurations.items():
                if config in _configurations:
                    sys.stderr.write("Warning: plugin '%s' provides a second configuration for '%s'\n" % (filename, config))
                    continue
                _configurations[config] = setup
        if hasattr(plugin, "register_options"):
            plugin.register_options(_opts)
    except Exception as e:
        sys.stderr.write("Warning: loading plugin '%s' failed: %s\n%s\n" % (filename, e, traceback.format_exc()))
        return

if __name__ == "__main__":
    # Load plugins
    pluginlist  = []
    pluginlist += list(glob("plugins/*.py"))
    pluginlist += list(glob("*/reportplugin.py"))
    pluginlist += list(glob("configs/*.py"))
    for pluginname in pluginlist:
        try_load_plugin(pluginname)

    sys.exit(main())