runtests.py

import re
import os
import glob
import shutil
import signal
import sys
import difflib
import argparse
from io import StringIO
import subprocess
from multiprocessing import Pool
import time
import tempfile
import configparser

#
# test properties
#


class TestOptions:
    testBuildFlags = []
    testRoot = os.path.dirname(os.path.abspath(__file__))
    runTestRoot = os.path.dirname(os.path.abspath(__file__))

    ceramicCompiler = None
    ceramicPlatform = None
    cleanUpLater = []


#
# utils
#


def indented(txt):
    return " " + txt.rstrip("\n").replace("\n", "\n ") + "\n"


#
# getCeramicCompiler
#


def getCeramicCompiler(opt):
    buildPath = ["..", "build", "compiler"]
    ceramicexe = None
    if sys.platform == "win32":
        ceramicexe = "ceramic.exe"
    else:
        ceramicexe = "ceramic"

    compiler = os.path.join(opt.testRoot, *(buildPath + [ceramicexe]))
    if not os.path.exists(compiler):
        compiler = shutil.which(ceramicexe)
        if compiler is None:
            print("could not find the ceramic compiler")
            sys.exit(1)
    return compiler


#
# getCeramicPlatform, fileForPlatform
#


def getCeramicPlatform(opt):
    with tempfile.NamedTemporaryFile(
        suffix=".crm", delete=False, mode="w", encoding="utf-8"
    ) as platform_ceramic:
        platform_ceramic.write(
            "import core.platform.(OSString, OSFamilyString, CPUString, CPUBits);\n"
        )
        platform_ceramic.write("import printer.(println);\n")
        platform_ceramic.write("main() {\n")
        platform_ceramic.write('    println("[Target]");\n')
        platform_ceramic.write('    println("platform: ", OSString);\n')
        platform_ceramic.write('    println("platform-family: ", OSFamilyString);\n')
        platform_ceramic.write('    println("architecture: ", CPUString);\n')
        platform_ceramic.write('    println("bits: ", CPUBits);\n')
        platform_ceramic.write("    return 0;\n")
        platform_ceramic.write("}\n")
        src_path = platform_ceramic.name

    try:
        cmd = [opt.ceramicCompiler] + opt.testBuildFlags + ["-o", "test.exe", src_path]
        subprocess.run(cmd, check=True)
    except subprocess.CalledProcessError:
        raise RuntimeError("could not build an executable with: " + " ".join(cmd))

    try:
        r = subprocess.run(
            ["./test.exe"], check=True, stdout=subprocess.PIPE, text=True
        )
        out_text = r.stdout
    except subprocess.CalledProcessError as ex:
        raise RuntimeError(f"executable failed with exit code {ex.returncode}")

    cfg = configparser.ConfigParser()
    cfg.read_string(out_text)

    # cleanup
    try:
        os.unlink(src_path)
        os.unlink("test.exe")
    except OSError:
        pass

    return (
        cfg.get("Target", "platform"),
        cfg.get("Target", "platform-family"),
        cfg.get("Target", "architecture"),
        cfg.get("Target", "bits"),
    )


def fileForPlatform(opt, folder, name, ext):
    platform, family, arch, bits = opt.ceramicPlatform
    platformNames = [
        "%s.%s.%s.%s.%s" % (name, platform, arch, bits, ext),
        "%s.%s.%s.%s.%s" % (name, family, arch, bits, ext),
        "%s.%s.%s.%s" % (name, platform, arch, ext),
        "%s.%s.%s.%s" % (name, platform, bits, ext),
        "%s.%s.%s.%s" % (name, family, arch, ext),
        "%s.%s.%s.%s" % (name, family, bits, ext),
        "%s.%s.%s.%s" % (name, arch, bits, ext),
        "%s.%s.%s" % (name, platform, ext),
        "%s.%s.%s" % (name, family, ext),
        "%s.%s.%s" % (name, arch, ext),
        "%s.%s.%s" % (name, bits, ext),
    ]
    for platformName in platformNames:
        fullName = os.path.join(folder, platformName)
        if os.path.isfile(fullName):
            return fullName

    return os.path.join(folder, "%s.%s" % (name, ext))


#
# RefText
#


def normalize(text):
    text = text.replace("\r", "")
    if not text or text.endswith("\n"):
        return text
    else:
        return text + "\n"


class RefText(object):
    def __init__(self, name, text):
        self.name = name
        text = normalize(text)
        self.text = text
        self.re = None
        xx = text.split("\n", 1)
        if len(xx) == 2 and xx[0] == "(re)":
            restring = xx[1].replace("\r", "").strip()
            self.re = re.compile(restring, re.DOTALL)
        else:
            self.re = None

    @staticmethod
    def load(name, path):
        with open(path, encoding="utf-8") as f:
            text = f.read().replace("\r", "")
        return RefText(name=name, text=text)

    @staticmethod
    def empty(name):
        return RefText(name=name, text="")

    def compare(self, actual):
        actual = normalize(actual)
        if self.re is not None:
            if self.re.match(actual):
                return None
            r = ""
            r += "expecting to match regexp:\n"
            r += indented(self.re.pattern)
            r += self.name + ":\n"
            r += indented(actual)
            return r
        else:
            if actual == self.text:
                return None
            r = ""
            for line in difflib.unified_diff(
                self.text.splitlines(True),
                actual.splitlines(True),
                fromfile=self.name,
                tofile="stdout",
            ):
                r += indented(line)
            return r


#
# TestCase
#


class TestCase(object):
    allCases = []
    opt = None

    def __init__(self, opt, folder, testfile, base=None):
        self.opt = opt
        entries = os.listdir(folder)
        self.testfile = os.path.basename(testfile)
        self.path = folder
        if os.path.isfile(testfile):
            self.loadTest(testfile)

        runscript = fileForPlatform(self.opt, folder, "run", "py")
        if os.path.isfile(runscript):
            self.runscript = runscript
        else:
            self.runscript = None

        buildflags = fileForPlatform(self.opt, folder, "buildflags", "txt")
        if os.path.isfile(buildflags):
            with open(buildflags, encoding="utf-8") as f:
                self.buildflags = f.read().split()
        else:
            self.buildflags = []

        for entry in entries:
            fullpath = os.path.join(folder, entry)
            if not os.path.isdir(fullpath):
                continue
            findTestCase(self.opt, fullpath, self)

    def loadTest(self, testfile):
        TestCase.allCases.append(self)
        self.testfile = testfile

    def cmdline(self, ceramic):
        r = (
            [ceramic, "-I" + self.path, "-o", "test.exe", "-Dtest.minimal"]
            + self.opt.testBuildFlags
            + self.buildflags
            + [self.testfile]
        )
        return r

    def pre_build(self):
        pass

    def post_build(self):
        pass

    def post_run(self):
        for f in glob.glob("temp*") + glob.glob("*.data"):
            try:
                os.unlink(f)
            except OSError:
                self.opt.cleanUpLater.append(f)

    def match(self, resultout, resulterr, returncode):
        compilererrfile = fileForPlatform(self.opt, ".", "compilererr", "txt")
        if os.path.isfile(compilererrfile):
            with open(compilererrfile, encoding="utf-8") as f:
                errpattern = f.read()
            errpattern = errpattern.replace("\r", "").strip()
            resulterr = resulterr.replace("\r", "")

            if returncode != "compiler error":
                self.testLogBuffer.write("Failure: compiler did not fail\n")
                self.testLogBuffer.write("Expected:\n")
                self.testLogBuffer.write(indented(errpattern))
                return "compiler did not fail"
            if re.search(errpattern, resulterr):
                return "ok"
            else:
                self.testLogBuffer.write("Failure: unexpected compiler error\n")
                self.testLogBuffer.write("Expected:\n")
                self.testLogBuffer.write(indented(errpattern))
                self.testLogBuffer.write("Actual:\n")
                self.testLogBuffer.write(indented(resulterr))
                return "unexpected compiler error"

        else:
            if returncode == "compiler error":
                self.testLogBuffer.write("Failure: compiler error\n")
                self.testLogBuffer.write("Error:\n")
                self.testLogBuffer.write(indented(resulterr))
                return "compiler error"
            outfile = fileForPlatform(self.opt, ".", "out", "txt")
            errfile = fileForPlatform(self.opt, ".", "err", "txt")
            warnfile = fileForPlatform(self.opt, ".", "warn", "txt")
            if not os.path.isfile(outfile):
                self.testLogBuffer.write("Failure: out.txt missing\n")
                self.testLogBuffer.write("Stdout:\n")
                self.testLogBuffer.write(indented(resultout))
                self.testLogBuffer.write("Stderr:\n")
                self.testLogBuffer.write(indented(resulterr))
                return "out.txt missing"
            refout = RefText.load("out.txt", outfile)
            referr = RefText.empty("err.txt")
            if os.path.isfile(errfile):
                referr = RefText.load("err.txt", errfile)
            refwarn = RefText.empty("warn.txt")
            if os.path.isfile(warnfile):
                refwarn = RefText.load("warn.txt", warnfile)
            outcompare = refout.compare(resultout)
            errcompare = referr.compare(resulterr)
            warncompare = refwarn.compare(self.compilerWarnings)
            if not outcompare and not errcompare and not warncompare:
                return "ok"
            failures = []
            if outcompare:
                failures.append("out.txt")
            if errcompare:
                failures.append("err.txt")
            if warncompare:
                failures.append("warn.txt")
            label = " and ".join(failures) + " mismatch"
            self.testLogBuffer.write("Failure: %s\n" % label)
            if outcompare:
                self.testLogBuffer.write("Diff-Stdout:\n")
                self.testLogBuffer.write(outcompare)
            if errcompare:
                self.testLogBuffer.write("Diff-Stderr:\n")
                self.testLogBuffer.write(errcompare)
            if warncompare:
                self.testLogBuffer.write("Diff-Warnings:\n")
                self.testLogBuffer.write(warncompare)
            return label

    def run(self):
        self.testLogBuffer = StringIO()
        os.chdir(self.path)
        self.pre_build()
        self.post_build()
        resultout, resulterr, returncode = self.runtest()
        self.post_run()
        r = self.match(resultout, resulterr, returncode)
        log = self.testLogBuffer.getvalue()
        self.testLogBuffer.close()

        return (r, log)

    def name(self):
        return os.path.relpath(self.path, self.opt.testRoot)

    def runtest(self):
        outfilename = "test.exe"
        outfilename = os.path.join(".", outfilename)
        self.compilerWarnings = ""
        process = subprocess.Popen(
            self.cmdline(self.opt.ceramicCompiler),
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        )
        compilerout, compilererr = process.communicate()
        if process.returncode != 0:
            return "", "%s\n%s" % (compilerout, compilererr), "compiler error"
        # warnings land on stderr of a successful compile
        self.compilerWarnings = compilererr
        if self.runscript is None:
            commandline = [outfilename]
        else:
            commandline = [
                sys.executable,
                self.runscript,
                outfilename,
            ] + self.opt.testBuildFlags

        process = subprocess.Popen(
            commandline, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
        )
        resultout, resulterr = process.communicate()
        self.removefile(outfilename)
        return resultout, resulterr, process.returncode

    def removefile(self, filename):
        # on windows, sometimes, deleting a file
        # immediately after executing it
        # results in a 'access denied' error.
        # so we wait and try again a few times.
        attempts = 1
        while (attempts <= 3) and os.path.exists(filename):
            try:
                os.unlink(filename)
            except OSError:
                time.sleep(1)
            attempts += 1


class TestModuleCase(TestCase):
    def match(self, resultout, resulterr, returncode):
        if returncode == 0:
            return "ok"
        elif returncode == "compiler error":
            self.testLogBuffer.write("Failure: compiler error\n")
            self.testLogBuffer.write("Error:\n")
            self.testLogBuffer.write(indented(resulterr))
            return "compiler error"
        else:
            self.testLogBuffer.write("Failure: exit code %d\n" % returncode)
            self.testLogBuffer.write("Stdout:\n")
            self.testLogBuffer.write(indented(resultout))
            self.testLogBuffer.write("Stderr:\n")
            self.testLogBuffer.write(indented(resulterr))
            return "exit code %d" % returncode


class TestDisabledCase(TestCase):
    def runtest(self):
        return "disabled", "", None

    def match(self, resultout, resulterr, returncode):
        return "disabled"


#
# runtests
#


def findTestCase(opt, folder, base=None):
    testPath = fileForPlatform(opt, folder, "test", "crm")
    mainPath = fileForPlatform(opt, folder, "main", "crm")
    testDisabledPath = fileForPlatform(opt, folder, "test-disabled", "crm")
    mainDisabledPath = fileForPlatform(opt, folder, "main-disabled", "crm")
    if os.path.isfile(testPath):
        TestModuleCase(opt, folder, testPath, base)
    elif os.path.isfile(testDisabledPath):
        TestDisabledCase(opt, folder, testDisabledPath, base)
    elif os.path.isfile(mainDisabledPath):
        TestDisabledCase(opt, folder, mainDisabledPath, base)
    else:
        TestCase(opt, folder, mainPath, base)


def findTestCases(opt):
    findTestCase(opt, opt.runTestRoot)
    return TestCase.allCases


def initWorker():
    signal.signal(signal.SIGINT, signal.SIG_IGN)


def runTest(t):
    return t.run()


def runTests(opt):
    testcases = findTestCases(opt)
    try:
        pool = Pool(initializer=initWorker)
        results = pool.imap(runTest, testcases)
    except ImportError:
        pool = None
        results = (runTest(t) for t in testcases)
    succeeded = []
    failed = []
    disabled = []
    failure_logs = []
    try:
        print("[TargetPlatform]")
        platform, family, arch, bits = opt.ceramicPlatform
        print("platform:", platform)
        print("cpu:", arch)
        print("bits:", bits)
        print()
        print("[Compiler]")
        print("exe:", opt.ceramicCompiler)
        print()
        print("[Tests]")
        for test in testcases:
            startTime = time.time()
            res, log = next(results)
            endTime = time.time()
            print("%s: %s (%fs)" % (test.name(), res, (endTime - startTime)))
            if res == "disabled":
                disabled.append(test.name())
            elif res != "ok":
                failed.append("%s: %s" % (test.name(), res))
                failure_logs.append(log)
            else:
                succeeded.append(test.name())
    except KeyboardInterrupt:
        print("\nInterrupted!")
        if pool:
            pool.terminate()

    if failed:
        print()
        print("[Failed]")
        for name, log in zip(failed, failure_logs):
            print(name)
            print(log)

    print()
    print("[Summary]")
    print("Passed: %d" % len(succeeded))
    if disabled:
        print("Disabled: %d" % len(disabled))
    if failed:
        print("Failed: %d" % len(failed))
    return len(failed)


def main():
    opt = TestOptions()

    argp = argparse.ArgumentParser(
        description="Run the Ceramic test suite.",
        usage="%(prog)s [options] [test/path]",
    )
    argp.add_argument(
        "-I",
        metavar="path",
        dest="libCeramicPaths",
        action="append",
        help="Specifies the Ceramic module path(s) to use.",
    )
    argp.add_argument(
        "--arch",
        metavar="arch",
        dest="arch",
        help="Specifies a Darwin target architecture to build for. (Apple only)",
    )
    argp.add_argument(
        "--target",
        metavar="cpu-vendor-os",
        dest="target",
        help="Specifies the target triple to build for.",
    )
    argp.add_argument(
        "--ceramic",
        metavar="path",
        dest="ceramicCompiler",
        help="Use the specified ceramic compiler (defaults to ceramic on path, or in build dir).",
    )
    argp.add_argument(
        "--buildflags",
        nargs="+",
        metavar="flags",
        dest="buildFlags",
        help="Additional build flags to pass through to the Ceramic compiler.",
    )
    argp.add_argument(
        "--root",
        metavar="path",
        dest="testRoot",
        help="Specifies the root directory out of which to run tests.",
    )

    argp.add_argument(
        "testname",
        nargs="?",
        help="Only run tests under the specified subdirectory in the test root.",
    )

    args = argp.parse_args()

    if args.arch is not None:
        opt.testBuildFlags += ["-arch", args.arch]
    if args.libCeramicPaths is not None:
        opt.testBuildFlags += [
            "-I" + os.path.abspath(path) for path in args.libCeramicPaths
        ]
    if args.target is not None:
        opt.testBuildFlags += ["-target", args.target]
    if args.testRoot is not None:
        opt.testRoot = os.path.abspath(args.testRoot)

    if args.testname is not None:
        opt.runTestRoot = os.path.join(opt.testRoot, args.testname)
    else:
        opt.runTestRoot = opt.testRoot

    if args.ceramicCompiler is not None:
        opt.ceramicCompiler = os.path.abspath(args.ceramicCompiler)
    else:
        opt.ceramicCompiler = getCeramicCompiler(opt)

    startTime = time.time()
    opt.ceramicPlatform = getCeramicPlatform(opt)
    failures = runTests(opt)
    for f in opt.cleanUpLater:
        try:
            os.unlink(f)
        except OSError:
            print("warning: unable to clean up temporary file", f)
    endTime = time.time()
    print()
    print("time-taken-seconds: %f" % (endTime - startTime))
    sys.exit(failures > 0)


if __name__ == "__main__":
    main()