CodeTest.py

Go to the documentation of this file.

#
# Copyright 2016, MeVis Medical Solutions AG
#
# The user may use this file in accordance with the license agreement provided with
# the Software or, alternatively, in accordance with the terms contained in a
# written agreement between the user and MeVis Medical Solutions AG.
#
# For further information use the contact form at https://www.mevislab.de/contact
#
 
 
 
from TestSupport.Macros import EXPECT_EQ, EXPECT_TRUE, ASSERT_TRUE
from TestSupport.Logging import infoHTML, errorHTML, info
from mevis import MLAB, MLABPackageManager, MLABTestCaseDatabase
import os
import os.path
import sys
import re
import json
 
 
# Allows to inject test functions for a compiled GoogleTest executable into a FunctionalTestCase
# Example (to be put into a TestCase's python file:
#   from TestSupport import CodeTest
#   CodeTest.inject("$(MLAB_MeVisLab_Standard)/CodeTests/bin/MLBaseTest", ctx)
# The test will fail if the unit test fails, OR if the test logs either
# "error" or "failed" (case-insensitive) to stderr. By optionally
# providing allowedStdErrRegExp, one can exclude certain stderr lines from
# this additional error detection.
def inject(executable, context, allowedStdErrRegExp=None):
    # replace package variable with the path to the binary directory in this case:
    # (the default would be the source directory, which might not be the same for some setups)
    match = re.search(r"\$\‍(MLAB_([^_)]+)_([^_)]+)\‍)", executable)
    if match:
        package = MLABPackageManager.packageByIdentifier(match.group(1) + "/" + match.group(2))
        if package:
            executable = executable.replace(match.group(0), package.binariesPath())
    test = CodeTest(executable=context.expandFilename(executable))
    test.injectTestFunctions(context, allowedStdErrRegExp=allowedStdErrRegExp)
 
 
pathToBuildSystemScripts = MLABPackageManager.getPathInPackage("MeVis/BuildSystem", "BuildTools/Scripts")
if os.path.isdir(pathToBuildSystemScripts):
    if not pathToBuildSystemScripts in sys.path:
        sys.path.insert(0, pathToBuildSystemScripts)
    from MeVisPython import MeVisPython
del pathToBuildSystemScripts
 
 
def _parseGoogleTestOutput(result, allowRestart=False, allowedStdErrRegExp=None):
    colorOk = "color: rgb(78,154,6);"
    colorFailed = "color: rgb(204,0,0);"
    commonStyle = "font-family: monospace;"
 
    statusFormat = '<div style="%s"><span style="%s">%%s</span>%%s</div>'
    errorFormat = statusFormat % (commonStyle, colorFailed)
    infoFormat = statusFormat % (commonStyle, colorOk)
 
    statusRegExp = re.compile("(?P<status>\\[\\s*\\S+\\s*\\])(?P<message>.*)")
 
    hasNotifiedStdErrOutput = False
 
    # Check for concrete error or failure information in stderr message. If yes then
    # really take it as an error. Often we have normal stderr prints of suppressed
    # leaks from sanitizer or stupidly other informational but non-failure-messages,
    # which should not be handled as error. If they do not actually print
    # something like "error:" or "failed:" they should be tolerated.
    lowerStderrMsg = result["stderr"].lower()
    if lowerStderrMsg and allowedStdErrRegExp is not None:
        info(f"NOTE: Test is configured to ignore stderr output that matches the regex '{allowedStdErrRegExp}'")
        lowerStderrMsg = re.sub(allowedStdErrRegExp, "", lowerStderrMsg, flags=re.IGNORECASE)
    stdErrContainsError = (lowerStderrMsg.find("error") >= 0) or (lowerStderrMsg.find("failed") >= 0)
 
    needsRestartWithoutSanitizer = False
    if stdErrContainsError:
        # Check whether test failed with a sanitizer failure caused by a sanitizer bug. It makes a
        # kind of unit test incompatible.
        if (
            lowerStderrMsg.find("sanitizer check failed:") >= 0
            and lowerStderrMsg.find("src/libsanitizer/asan/asan_malloc_linux.cc") >= 0
            and lowerStderrMsg.find("allocated_for_dlsym") >= 0
        ):
            needsRestartWithoutSanitizer = allowRestart
 
    if not needsRestartWithoutSanitizer:
 
        EXPECT_EQ(result["status"], 0, msg="The test status is not 0.", logOnSuccess=False)
        EXPECT_TRUE(result["ok"], msg="The test run failed.", logOnSuccess=False)
        ASSERT_TRUE("stdout" in result, msg="The test output does not exists.", logOnSuccess=False)
 
        for outChannel in ["stdout", "stderr"]:
            for line in result[outChannel].splitlines():
                # Before first std::err message print a notice that there are also std::cerr messages.
                if outChannel == "stderr" and not hasNotifiedStdErrOutput:
                    if stdErrContainsError:
                        errorHTML(
                            'Note: The stderr channel output contained "error" or "failed" which is handled as test failure.'
                        )
                    infoHTML("The test printed the following messages to stderr channel:")
                    hasNotifiedStdErrOutput = True
 
                m = statusRegExp.match(line)
                if m:
                    status = m.group("status").replace(" ", "&nbsp;").replace("\t", "&nbsp;" * 4)
                    message = m.group("message").replace(" ", "&nbsp;").replace("\t", "&nbsp;" * 4)
                    if "FAILED" in status:
                        errorHTML(errorFormat % (status, message))
                    else:
                        infoHTML(infoFormat % (status, message))
                else:
                    line = line.replace(" ", "&nbsp;").replace("\t", "&nbsp;" * 4)
                    infoHTML(('<div style="%s">' + line + "</div>") % commonStyle)
 
    return needsRestartWithoutSanitizer
 
 
class TestFunctionGroup:
    def __init__(self, name, sourceName):
        self.name = name
        self.sourceName = sourceName
        self.__testFunctions = []
 
    def addFunction(self, function):
        self.__testFunctions.append(function)
 
    def getFunctions(self):
        return self.__testFunctions
 
 
class TestFunction:
    def __init__(self, name, sourceName):
        self.name = name
        self.sourceName = sourceName
 
 
class CodeTest:
    # Either cmakeFile or executable need to be passed in. Executable is without debug and "
    def __init__(self, cmakeFile=None, executable=None):
        self.__testFunctionGroups = {}
        self.__infoDict = None
        self.__uniqueFunctionNames = None
        self.cmakeFile = cmakeFile
        self.executable = executable
        self.__searchedExecutable = False
 
        if cmakeFile is not None:
            self.name = self._parseCMakeFile(cmakeFile)
        elif executable is not None:
            self.name = os.path.basename(executable)
 
    def _getUniqueTestCaseName(self, name):
        uniqueName = name
        i = 0
        while MLABTestCaseDatabase.testCaseInfo(uniqueName):
            if i == 0:
                # first append _CodeTest, and later also a number
                name += "_CodeTest"
                uniqueName = name
            else:
                uniqueName = name + str(i)
            i += 1
        return uniqueName
 
    def getInfoDict(self):
        if self.__infoDict == None:
            package = MLABPackageManager.findPackageContainingPath(self.cmakeFile)
            self.__infoDict = {
                "package": package.packageIdentifier(),
                "author": "",
                "file": self.cmakeFile,
                "lineno": 1,
                "type": "CodeTestTestCase",
                "scriptFile": "ScriptFileNotAvailable",
            }
        return self.__infoDict
 
    def getTestFunctionGroups(self):
        return self.__testFunctionGroups
 
    def getExecutable(self):
        if self.executable is None:
            if not self.__searchedExecutable:
                self.__searchedExecutable = True
                self.executable = self._findExecutable()
                if not self.__executableExists(self.executable):
                    MLAB.logError(f"Code test executable for {self.name} not found.")
        elif not self.__executableExists(self.executable):
            orginalExecutable = self.executable
            self.executable = self.__appendSuffix(self.executable)
            self.__searchedExecutable = True
            if not self.__executableExists(self.executable):
                self.executable = orginalExecutable
                MLAB.logError(f"Code test executable {self.executable} not found.")
        return self.executable
 
    @staticmethod
    def __appendSuffix(executable):
        # turn base filename into something probably existing
        exeSuffix = ".exe" if MLAB.isWindows() else ""
        if MLAB.isDebug():
            executable += "_d"
        if os.path.exists(executable + exeSuffix):
            executable += exeSuffix
        elif not MLAB.isDebug() and os.path.exists(executable + "_d" + exeSuffix):
            executable += "_d" + exeSuffix
        elif MLAB.isDebug():
            executable += exeSuffix
            executable = executable.replace("_d" + exeSuffix, exeSuffix)
        return executable
 
    @staticmethod
    def __executableExists(executable):
        return executable and os.path.exists(executable)
 
    def existsExecutable(self):
        return self.__executableExists(self.getExecutable())
 
    def buildProject(self):
        if self.existsExecutable():
            os.remove(self.executable)
        executable = "cmake"
        arguments = ["--build", ".", "--config", ("Debug" if MLAB.isDebug() else "Release"), "--target", self.name]
        result = MLAB.runCommandStdInOut(executable, arguments, self._getBuildRoot())
        # force to search for the executable again:
        self.executable = None
        self.__searchedExecutable = False
        if not self.existsExecutable():
            MLAB.showCritical(self._getError(result))
 
        if result["ok"] != True:
            MLAB.logError(self._getError(result))
        return result["ok"] == True
 
    def _getError(self, result):
        message = []
        stderr = result.get("stderr")
        stdout = result.get("stdout")
        if stdout:
            message.append(stdout.encode("latin1").decode("UTF-8"))
        if stderr:
            message.append(stderr.encode("latin1").decode("UTF-8"))
        if not stdout and not stderr:
            message.append("An unknown error occurred")
        return "\n".join(message)
 
    def _parseCMakeFile(self, cmakeFile):
        testRegex = re.compile(r"^(?!#)\s*mlab_add_test\‍(\s*(\w*)")
        target = None
        with open(cmakeFile, "r") as f:
            for line in f:
                match = testRegex.match(line)
                if match and match.group(1):
                    target = match.group(1)
                    return target
        return target
 
    def _getBuildRoot(self):
        if "MLAB_BUILD" in os.environ:
            buildRoot = os.environ["MLAB_BUILD"]
        else:
            buildRoot = os.path.normpath(
                os.path.join(MLABPackageManager.packageByIdentifier("MeVisLab/IDE").binariesPath(), "..", "..")
            )
        return buildRoot
 
    def _findExecutable(self):
        executable = None
        command = "ctest"
        testArguments = ["-V", "-R", self.name, "-C", ("Debug" if MLAB.isDebug() else "Release"), "--show-only=json-v1"]
        result = MLAB.runCommandStdInOut(command, testArguments, self._getBuildRoot())
        if result["ok"] == True:
            testDetails = json.loads(result["stdout"])
            if (
                "tests" in testDetails
                and len(testDetails["tests"]) > 0
                and "command" in testDetails["tests"][0]
                and len(testDetails["tests"][0]["command"]) > 0
            ):
                executable = testDetails["tests"][0]["command"][0]
        # workaround for code tests not in MeVisLab/IDE
        if executable is None and self.cmakeFile:
            package = MLABPackageManager.findPackageContainingPath(self.cmakeFile)
            executable = self.__appendSuffix(os.path.join(package.binariesPath(), "CodeTests", "bin", self.name))
        return executable
 
    def loadTestFunctions(self):
        if not self.existsExecutable():
            if self.cmakeFile:
                self.buildProject()
        if self.existsExecutable():
            executable = self.executable
            result = MLAB.runCommandStdInOut(executable, ["--gtest_list_tests"], os.path.dirname(executable))
            if result["ok"] == True:
                self._parseTestFunctionsFromOutput(result["stdout"])
 
    def __getUniqueFunctionName(self, name):
        i = 0
        uniqueName = name
        while uniqueName in self.__uniqueFunctionNames:
            i += 1
            uniqueName = name + str(i)
        self.__uniqueFunctionNames.add(uniqueName)
        return uniqueName
 
    def _parseTestFunctionsFromOutput(self, output):
        identifierRegExp = re.compile(r"^[\S]+$")
        self.__testFunctionGroups = []
        self.__uniqueFunctionNames = set()
        group = None
        groupIdx = 1
        functionIdx = 1
        for line in output.splitlines():
            commentIdx = line.find("#")
            if commentIdx >= 0:
                line = line[:commentIdx].rstrip(" ")
            if not line.startswith(" "):
                name = line.rstrip(".")
                if identifierRegExp.match(name):
                    sourceName = "GROUP%03d_%s" % (groupIdx, name.replace("/", "_"))
                    group = TestFunctionGroup(name, sourceName)
                    self.__testFunctionGroups.append(group)
                    groupIdx += 1
            else:
                name = line.strip()
                if identifierRegExp.match(name):
                    isDisabled = False
                    uniqueName = name
                    if uniqueName.startswith("DISABLED_"):
                        uniqueName = uniqueName[len("DISABLED_") :]
                        isDisabled = True
                    uniqueName = self.__getUniqueFunctionName(uniqueName)
                    sourceName = "TEST%03d_%s" % (functionIdx, uniqueName.replace("/", "_"))
                    if isDisabled:
                        sourceName = "DISABLED_" + sourceName
                    function = TestFunction(name, sourceName)
                    group.addFunction(function)
                    functionIdx += 1
 
    def injectTestFunctions(self, context, allowedStdErrRegExp=None):
        self.loadTestFunctions()
 
        testCode = ""
        ctx = context
 
        globalSymbols = globals()
        localSymbols = locals()
 
        for group in self.getTestFunctionGroups():
            for function in group.getFunctions():
                testCode += self._functionCode(group, function, allowedStdErrRegExp=allowedStdErrRegExp)
            testCode += self._groupCode(group)
 
        eval(compile(testCode, "<string>", "exec"), globalSymbols, localSymbols)
 
    def _groupCode(self, group):
        functions = []
        for f in group.getFunctions():
            functions.append("'" + f.sourceName + "'")
        return """def %(sourceName)s(): 
  return %(functions)s,
ctx.setScriptVariable(%(sourceName)s.__name__, %(sourceName)s)
""" % dict(
            sourceName=group.sourceName, functions=",".join(functions)
        )
 
    def _hasSanitizerSuppression(self, sanitizerPrefix, envVar, compilerVersion):
        mlabRootPath = MLAB.variable("MLAB_MeVis_BuildSystem")
        suppressionFile = (
            mlabRootPath
            + "/BuildTools/linux/"
            + sanitizerPrefix
            + "SanitizerSuppressions_gcc"
            + compilerVersion
            + ".supp"
        )
        # print(suppressionFile)
        return os.path.isfile(suppressionFile) and os.path.exists(suppressionFile)
 
    def _determineSanitizerSetUpCode(self):
        if MLAB.isLinux():
            compilerInfo = MLAB.compilerInfo()
            compilerVersion = "4"
            if compilerInfo.find("GCC_64_7") >= 0:
                compilerVersion = "7"
            elif compilerInfo.find("GCC_64_9") >= 0:
                compilerVersion = "9"
            elif compilerInfo.find("") >= 0:
                compilerVersion = "5"
 
            hasSuppressions = (
                self._hasSanitizerSuppression("Leak", "LSAN_OPTIONS", compilerVersion)
                and self._hasSanitizerSuppression("Address", "ASAN_OPTIONS", compilerVersion)
                and self._hasSanitizerSuppression("UB", "UBSAN_OPTIONS", compilerVersion)
                and self._hasSanitizerSuppression("Thread", "TSAN_OPTIONS", compilerVersion)
            )
            if hasSuppressions:
                return """
  MLAB.log(u"Running in sanitizer environment of gcc%(compilerVer)s ('asan' and 'ubsan' libs with LD_PRELOAD as well as suppression files).")
  mlabRootPath = MLAB.variable("MLAB_MeVis_BuildSystem")
  sanitizerEnvVarIntro = "suppressions=" + mlabRootPath + os.sep
  os.environ["LSAN_OPTIONS"] = sanitizerEnvVarIntro + u"BuildTools/linux/LeakSanitizerSuppressions_gcc%(compilerVer)s.supp"
  os.environ["ASAN_OPTIONS"] = sanitizerEnvVarIntro + u"BuildTools/linux/AddressSanitizerSuppressions_gcc%(compilerVer)s.supp"
  os.environ["UBSAN_OPTIONS"] = sanitizerEnvVarIntro + u"BuildTools/linux/UBSanitizerSuppressions_gcc%(compilerVer)s.supp"
  os.environ["TSAN_OPTIONS"] = sanitizerEnvVarIntro + u"linux/ThreadSanitizerSuppressions_gcc%(compilerVer)s.supp"
  os.environ["LD_PRELOAD"] = u"/usr/lib/gcc/x86_64-linux-gnu/%(compilerVer)s/libasan.so:/usr/lib/gcc/x86_64-linux-gnu/%(compilerVer)s/libubsan.so"
  """ % dict(
                    compilerVer=compilerVersion
                )
            else:
                return """
  MLAB.log(u"Could not find (all) sanitizer suppression files: not running test in gcc sanitizer environment.")
  """
        else:
            return ""
 
    def _functionCode(self, group, function, allowedStdErrRegExp=None):
        exePath = self.getExecutable()
        exePath = exePath.replace("\\", "\\\\")
        saniCode = self._determineSanitizerSetUpCode()
        return """def %(sourceName)s():
  executable = "%(executable)s"
  for package in MLABPackageManager.packages():
    os.environ[package.packageEnvVariableName()] = package.scriptsPath()
%(sanitizerSetUpCode)s
  result = MLAB.runCommandStdInOut(executable, ['--gtest_also_run_disabled_tests', '--gtest_filter=%(groupName)s.%(functionName)s'], os.path.dirname(executable))
  from TestSupport.CodeTest import _parseGoogleTestOutput
  needsRestartWithoutSanitizer = _parseGoogleTestOutput(result, allowRestart=True, allowedStdErrRegExp=r"%(allowedStdErrRegExp)s")
  if needsRestartWithoutSanitizer:
    os.environ["LD_PRELOAD"] = ""
    MLAB.log(u"Sanitizer is incompatible with unittest. Restarting test without sanitzer preload.")
    result = MLAB.runCommandStdInOut(executable, ['--gtest_also_run_disabled_tests', '--gtest_filter=%(groupName)s.%(functionName)s'], os.path.dirname(executable))
    _parseGoogleTestOutput(result, allowedStdErrRegExp=r"%(allowedStdErrRegExp)s")
ctx.setScriptVariable(%(sourceName)s.__name__, %(sourceName)s)
""" % dict(
            sourceName=function.sourceName,
            executable=exePath,
            sanitizerSetUpCode=saniCode,
            groupName=group.name,
            functionName=function.name,
            allowedStdErrRegExp=allowedStdErrRegExp,
        )