TestHelper.py

Go to the documentation of this file.

#
# Copyright 2007, 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
#
 
 
 
# -- system imports ----------------------------------------------------------------------------{{{-
import base64
import os
import re
import sys
 
import _thread
import traceback
 
# ----------------------------------------------------------------------------------------------}}}-
 
# -- local imports -----------------------------------------------------------------------------{{{-
import mevis
 
from TestSupport.ChangeSet import ChangeSet
from TestSupport.MessageFilters import MessageHandling, ErrorMessageFilter, WarningMessageFilter, InfoMessageFilter
 
# ----------------------------------------------------------------------------------------------}}}-
 
_mlab_do_not_reload = True  # exempt this module from Python module reloading, value is currently ignored
 
gEnablePrettyLogging = False
gLogInfoMessages = True
 
 
class SpecialMessage:
    """Stores an additional message (usually the unformatted version) with a log message."""
 
    specialMessageMarker = re.compile(r".*<MLAB TC=(\d+)\/>")
    currentSpecialMessageId = 0
    specialMessages = {}
 
    @classmethod
    def add(cls, message, specialMessage):
        cls.currentSpecialMessageId += 1
        message += f"<MLAB TC={cls.currentSpecialMessageId}/>"
        cls.specialMessages[cls.currentSpecialMessageId] = specialMessage
        return message
 
    @classmethod
    def get(cls, message):
        match = re.match(cls.specialMessageMarker, message)
        if match:
            messageId = int(match.group(1))
            specialMessage = cls.specialMessages.get(messageId, None)
            if specialMessage is not None:
                del cls.specialMessages[messageId]
                return specialMessage
        return None
 
 
def emitSpecialMessage(loggingMethod, file, line, type, message, escapeHtml=True, formattedStack=""):
    if escapeHtml:
        escapedMsg = mevis.MLAB.escapeString(message)
        escapedMsg = escapedMsg.replace("\n", "<br>")
    else:
        escapedMsg = message
 
    if sys.version_info.major >= 3:
        if isinstance(formattedStack, str):
            formattedStack = formattedStack.encode()
 
    encodedStackTrace = str(base64.b64encode(formattedStack))
    # store internal message to be used from LogHandler:
    specialMessage = "##M[%s|%s|%s|%s]: %s" % (file, line, type, encodedStackTrace, escapedMsg)
    if gEnablePrettyLogging:
        escapedMsg = (
            "<font style='font-weight:normal'><font color=black>"
            + mevis.MLAB.createHyperLinkWithLine(file, line)
            + "("
            + str(line)
            + "):</font></font> "
            + escapedMsg
        )
        loggingMethod(SpecialMessage.add(escapedMsg, specialMessage))
    else:
        loggingMethod(specialMessage)
 
 
def emitSpecialCommand(file, line, type, fileList, message, escapeHtml=True):
    if escapeHtml:
        escapedMsg = mevis.MLAB.escapeString(message)
    else:
        escapedMsg = message
 
    # store internal message to be used from LogHandler:
    specialMessage = "##C[%s|%s|%s:%s] %s" % (file, line, type, ",".join(fileList), str(escapedMsg))
    if gEnablePrettyLogging:
        escapedMsg = (
            "<font style='font-weight:normal'><font color=black>"
            + mevis.MLAB.createHyperLinkWithLine(file, line)
            + "("
            + str(line)
            + "):</font></font> "
            + escapedMsg
        )
        for fileInList in fileList:
            if "|" in fileInList:
                fileInList = fileInList[fileInList.index("|") + 1 :]
            escapedMsg += " " + mevis.MLAB.createHyperLink(fileInList)
        escapedMsg = SpecialMessage.add(escapedMsg, specialMessage)
        mevis.MLAB.logHTML(escapedMsg)
    else:
        mevis.MLAB.logHTML(specialMessage)
 
 
def UseStackFrameFromCallerForLogging(func):
    """This decorator is used to specify the stack frame that is used to detect the location while logging.
    It uses the stack frame that calls the given function. Functions with this decorator can call each other
    recursively. The first stack frame of such a recursion is then used."""
 
    def setLoggingCallerStackFrameWrapper(*args, **kwargs):
        with SetLoggingCallerStackFrame(1):
            return func(*args, **kwargs)
 
    return setLoggingCallerStackFrameWrapper
 
 
# Alias for backward compatibility of old TestCases
IncreaseLoggingStackDepthDecorator = UseStackFrameFromCallerForLogging
 
 
class SetLoggingCallerStackFrame:
    def __init__(self, stackDepth, optional=False):
        self.__testHelper = TestHelper.getInstance()
        self.__stackDepth = stackDepth
        self.__hasSetStackFrame = False
        self.__optional = optional
 
    def __enter__(self):
        if not self.__testHelper.hasCallerStackFrame() or not self.__optional:
            stackFrame = self.getCallingStackFrame(self.__stackDepth + 1)
            self.__testHelper.pushCallerStackFrame(stackFrame)
            self.__hasSetStackFrame = True
 
    def __exit__(self, *_):
        if self.__hasSetStackFrame:
            self.__testHelper.popCallerStackFrame()
 
    def getCallingStackFrame(self, stackDepth):
        return sys._getframe(stackDepth + 1)
 
 
# Returns pattern attribute or repr(exp) if the object does not have a pattern
def regexpToString(exp):
    try:
        result = exp.pattern
    except:
        result = repr(exp)
    return result
 
 
# Helper object to supress errors,
# make sure that you call handleResult() in a finally: clause
# If no logErrorFunc it given, matching error will just be ignored, not expected, so no error will be logged if no match was found.
class SuppressedErrors:
    def __init__(self, errorRegExp, logErrorFunc, preventReplacingMessageSeverity=False):
        self.__logErrorFunc = logErrorFunc
        self.__errorRegExp = errorRegExp
        handling = MessageHandling.EXPECT if logErrorFunc is not None else MessageHandling.IGNORE
        self.__messageFilter = ErrorMessageFilter(handling, errorRegExp)
        self.__preventReplacingMessageSeverity = preventReplacingMessageSeverity
        self.__previousKeepMessageSeverity = None
        self.__enableErrorHandling(True)
 
    def __enableErrorHandling(self, enable):
        logHandler = TestHelper.getInstance().getLogHandler()
        logHandler.pushOrPopMessageFilter(enable, self.__messageFilter)
        if self.__preventReplacingMessageSeverity:
            if enable:
                self.__previousKeepMessageSeverity = logHandler.setKeepMessageSeverity(True)
            else:
                if self.__previousKeepMessageSeverity != None:
                    logHandler.setKeepMessageSeverity(self.__previousKeepMessageSeverity)
                    self.__previousKeepMessageSeverity = None
 
    def handleResult(self):
        mevis.MLAB.priv().flushPendingMessages()
        hasCaughtError = self.__messageFilter.hadMatch
        self.__enableErrorHandling(False)
        if self.__logErrorFunc and not hasCaughtError:
            self.__logErrorFunc("Expected error did not occur: " + regexpToString(self.__errorRegExp))
        return hasCaughtError
 
 
class SuppressedWarnings:
    """Helper object to supress warnings.
    Make sure that you call handleResult() in a finally: clause.
    If no logErrorFunc it given, matching warnings will just be ignored, not expected,
    so no error will be logged if no match was found.
    """
 
    def __init__(self, warningRegExp, logErrorFunc=None, expectWarning=True):
        self.__logErrorFunc = logErrorFunc
        handling = MessageHandling.EXPECT if logErrorFunc is not None else MessageHandling.IGNORE
        self.__warningRegExp = warningRegExp
        self.__expectWarning = expectWarning
        self.__messageFilter = WarningMessageFilter(handling, warningRegExp)
        self.__enableWarningHandling(True)
 
    def __enableWarningHandling(self, enable):
        TestHelper.getInstance().getLogHandler().pushOrPopMessageFilter(enable, self.__messageFilter)
 
    def handleResult(self):
        mevis.MLAB.priv().flushPendingMessages()
        hasCaughtWarning = self.__messageFilter.hadMatch
        self.__enableWarningHandling(False)
        if self.__logErrorFunc and hasCaughtWarning != self.__expectWarning:
            if self.__expectWarning:
                self.__logErrorFunc("Expected warning did not occur: " + regexpToString(self.__warningRegExp))
            else:
                self.__logErrorFunc("Unexpected warning did occur: " + regexpToString(self.__warningRegExp))
 
        return hasCaughtWarning == self.__expectWarning
 
 
# Helper object to expect infos,
# make sure that you call handleResult() in a finally: clause
class ExpectInfos:
    def __init__(self, infoRegExp, allowUnexpectedMessages, logErrorFunc):
        self.__infoRegExp = infoRegExp
        self.__allowUnexpectedMessages = allowUnexpectedMessages
        self.__logErrorFunc = logErrorFunc
        self.__messageFilter = InfoMessageFilter(MessageHandling.EXPECT, infoRegExp, allowUnexpectedMessages)
        self.__enableInfoHandling(True)
 
    def __enableInfoHandling(self, enable):
        TestHelper.getInstance().getLogHandler().pushOrPopMessageFilter(enable, self.__messageFilter)
 
    def handleResult(self):
        mevis.MLAB.priv().flushPendingMessages()
        hadUnexpectedInfo = self.__messageFilter.hadNonMatch
        hadExpectedInfo = self.__messageFilter.hadMatch
        self.__enableInfoHandling(False)
        if hadUnexpectedInfo:
            if not self.__allowUnexpectedMessages:
                self.__logErrorFunc("Unexpected info message occurred")
        if not hadExpectedInfo:
            self.__logErrorFunc("Expected info message did not occur: " + regexpToString(self.__infoRegExp))
 
 
# -- class TC_ChangeSet ------------------------------------------------------------------------{{{-
class TC_ChangeSet(ChangeSet):
    # -- def __init__ ----------------------------------------------------------------------------{{{-
    
    def __init__(self, context):
        ChangeSet.__init__(self, context)
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def _logDecorator -----------------------------------------------------------------------{{{-
    def _logDecorator(func):
        def wrapper(*args, **kwds):
            with SetLoggingCallerStackFrame(3):
                loggingMethod, message = func(*args, **kwds)
                if gLogInfoMessages or (loggingMethod != mevis.MLAB.logHTML):
                    stackFrame = TestHelper.getInstance().getCallerStackFrame()
                    stackLine = traceback.extract_stack(stackFrame, 1)[0]
                    emitSpecialMessage(loggingMethod, stackLine[0], stackLine[1], "ChangeSet", message, escapeHtml=True)
 
        return wrapper
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def _logInfo ----------------------------------------------------------------------------{{{-
    @_logDecorator
    def _logInfo(self, message):
        return mevis.MLAB.logHTML, message
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def _logError ---------------------------------------------------------------------------{{{-
    @_logDecorator
    def _logError(self, message):
        return mevis.MLAB.logErrorHTML, message
 
    # --------------------------------------------------------------------------------------------}}}-
 
 
# ----------------------------------------------------------------------------------------------}}}-
 
 
# Exception that is raised when a test execution is cancelled.
# This may happen because of the stop-on-first-error feature
# or because the user cancels a test run.
class CancelTestException(Exception):
    pass
 
 
# -- class TestHelper --------------------------------------------------------------------------{{{-
 
class TestHelper:
    # -- member variables ------------------------------------------------------------------------{{{-
    
    _mCtx = None
    
    _mTestCaseContext = None
    
    _mHelperModule = None
    
    _mResultDirectory = None
    
    _mPackageList = None
 
    
    _mTestCaseName = None
    
    _mModuleName = None
 
    
    _mTestCaseDataDirectory = None
    
    _mTestCaseResultDirectory = None
 
    
    _mChangeSetStack = None
 
    
    _mSecureTesting = False
 
    
    _mLogHandler = None
 
    
    _mStackFrameStack = None
 
    
    _mEnvironmentStack = None
 
    
    _mExtraTestCaseResults = None
 
    
    _mMacrosLogOnSuccess = True
 
    
    __lockObj = _thread.allocate_lock()
    
    __instance = None
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def __new__ -----------------------------------------------------------------------------{{{-
    
    def __new__(self, *args, **kargs):
        return self.getInstance(*args, **kargs)
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def __init__ ----------------------------------------------------------------------------{{{-
    
    def __init__(self, *args, **kargs):
        pass
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def getInstance -------------------------------------------------------------------------{{{-
    
    @classmethod
    def getInstance(self, *args, **kargs):
        """Static method to have a reference to **THE UNIQUE** instance"""
        if len(kargs) not in (0, 2):
            raise Exception(
                "Singleton's getInstance must be called with context and result directory arguments or without any!"
            )
 
        initialize = len(kargs) == 2
        # Critical section start
        self.__lockObj.acquire()
        try:
            if self.__instance is None:
                # (Some exception may be thrown...)
                if not "context" in kargs or not "resultDir" in kargs:
                    raise Exception("Singleton must be initialized with context and result directory!")
                # Initialize **the unique** instance
                self.__instance = object.__new__(self)
 
            if initialize:
                """Initialize object **here**, as you would do in __init__()..."""
                self._mResultDirectory = kargs["resultDir"]
                self._mCtx = kargs["context"]
 
                # The FunctionalTesting module contains some helpful modules that are used in the methods.
                if self._mHelperModule:
                    self._mHelperModule.remove()
                self._mHelperModule = self._mCtx.addModule(mevis.MLAB.moduleLiteral("TestCenterHelperModule"))
 
                self._mChangeSetStack = []
                self._mEnvironmentStack = []
                self._mStackFrameStack = []
 
        finally:
            #  Exit from critical section whatever happens
            self.__lockObj.release()
        # Critical section end
        return self.__instance
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def hasInstance -------------------------------------------------------------------------{{{-
    
    @classmethod
    def hasInstance(self):
        return self.__instance is not None
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def setPackageList ----------------------------------------------------------------------{{{-
    
    def setPackageList(self, packageList):
        self._mPackageList = packageList
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def getPackageList ----------------------------------------------------------------------{{{-
    
    def getPackageList(self):
        return self._mPackageList
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def getGlobalContext --------------------------------------------------------------------{{{-
    def getGlobalContext(self):
        return self._mCtx
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def setTestCaseContext ------------------------------------------------------------------{{{-
    
    def setTestCaseContext(self, ctx):
        self._mTestCaseContext = ctx
        self._mChangeSet = TC_ChangeSet(ctx)
        self._mExtraTestCaseResults = []
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def setMacrosLogOnSuccess ---------------------------------------------------------------{{{-
    def setMacrosLogOnSuccess(self, macrosLogOnSuccess):
        self._mMacrosLogOnSuccess_mMacrosLogOnSuccess = macrosLogOnSuccess
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def getMacrosLogOnSuccess ---------------------------------------------------------------{{{-
    def getMacrosLogOnSuccess(self):
        return self._mMacrosLogOnSuccess_mMacrosLogOnSuccess
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def getTestCaseContext ------------------------------------------------------------------{{{-
    
    def getTestCaseContext(self):
        return self._mTestCaseContext
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def getHelperContext --------------------------------------------------------------------{{{-
    def getHelperContext(self):
        return self._mHelperModule
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def getResultDirectory ------------------------------------------------------------------{{{-
    def getResultDirectory(self):
        return self._mResultDirectory
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def setModuleName -----------------------------------------------------------------------{{{-
    def setModuleName(self, moduleName):
        self._mModuleName = moduleName
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def getModuleName -----------------------------------------------------------------------{{{-
    def getModuleName(self):
        return self._mModuleName
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def setTestCaseName ---------------------------------------------------------------------{{{-
    def setTestCaseName(self, testCaseName):
        self._mTestCaseName = testCaseName
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def getTestCaseName ---------------------------------------------------------------------{{{-
    def getTestCaseName(self):
        return self._mTestCaseName
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def setTestCaseDataDirectory ------------------------------------------------------------{{{-
    def setTestCaseDataDirectory(self, testCaseDataPath):
        self._mTestCaseDataDirectory = testCaseDataPath
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def getTestCaseDataDirectory ------------------------------------------------------------{{{-
    def getTestCaseDataDirectory(self):
        return self._mTestCaseDataDirectory
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def setTestCaseResultDirectory ------------------------------------------------------------{{{-
    def setTestCaseResultDirectory(self, testCaseName, functionName):
        self._mTestCaseResultDirectory = os.path.join(self.getResultDirectory(), "files", testCaseName)
        if functionName != None:
            self._mTestCaseResultDirectory = os.path.join(self._mTestCaseResultDirectory, functionName)
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def getTestCaseResultDirectory ------------------------------------------------------------{{{-
    def getTestCaseResultDirectory(self):
        if self._mTestCaseResultDirectory and not os.path.isdir(self._mTestCaseResultDirectory):
            os.makedirs(self._mTestCaseResultDirectory)
        return self._mTestCaseResultDirectory
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def pushChangeSet -----------------------------------------------------------------------{{{-
    
    def pushChangeSet(self):
        self._mChangeSetStack.append(TC_ChangeSet(self._mTestCaseContext))
        return len(self._mChangeSetStack) - 1
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def popChangeSet ------------------------------------------------------------------------{{{-
    
    def popChangeSet(self):
        retVal = True
        if len(self._mChangeSetStack) > 0:
            cs = self._mChangeSetStack.pop()
            cs.autoRevert()
        else:
            retVal = False
        return retVal
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def popChangeSet ------------------------------------------------------------------------{{{-
    
    def clearChangeSet(self):
        for changeSet in self._mChangeSetStack:
            changeSet.enableAutoRevert(False)
        while self.popChangeSet():
            pass
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def getChangeSet ------------------------------------------------------------------------{{{-
    
    def getChangeSet(self):
        retVal = None
        if len(self._mChangeSetStack) > 0:
            retVal = self._mChangeSetStack[-1]
        return retVal
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def getChangeSetStackLength -------------------------------------------------------------{{{-
    
    def getChangeSetStackLength(self):
        return len(self._mChangeSetStack)
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def runTestInCurrentMlabInstance -----------------------------------------------------------------------{{{-
    
    def runTestInCurrentMlabInstance(self):
        self._mSecureTesting_mSecureTesting = False
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def runTestInCurrentMlabInstance -----------------------------------------------------------------------{{{-
    
    def runTestInSecureMode(self):
        self._mSecureTesting_mSecureTesting = True
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def runTestInCurrentMlabInstance -----------------------------------------------------------------------{{{-
    
    def testIsRunInSecureMode(self):
        return self._mSecureTesting_mSecureTesting
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def setLogHandler -----------------------------------------------------------------------{{{-
    
    def setLogHandler(self, logHandler):
        self._mLogHandler = logHandler
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def unsetLogHandler ---------------------------------------------------------------------{{{-
    
    def unsetLogHandler(self):
        self._mLogHandler = None
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def getLogHandler -----------------------------------------------------------------------{{{-
    
    def getLogHandler(self):
        return self._mLogHandler
 
    # --------------------------------------------------------------------------------------------}}}-
 
    def pushCallerStackFrame(self, stackFrame):
        self._mStackFrameStack.append(stackFrame)
 
    def getCallerStackFrame(self):
        return self._mStackFrameStack[-1] if self._mStackFrameStack else None
 
    def hasCallerStackFrame(self):
        return len(self._mStackFrameStack) > 0
 
    def popCallerStackFrame(self):
        self._mStackFrameStack.pop()
 
    # -- def addExtraTestCaseResult --------------------------------------------------------------{{{-
    def addExtraTestCaseResult(self, extraTestCaseResult):
        self._mExtraTestCaseResults.append(extraTestCaseResult)
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def takeExtraTestCaseResults ------------------------------------------------------------{{{-
    def takeExtraTestCaseResults(self):
        l = self._mExtraTestCaseResults
        self._mExtraTestCaseResults = []
        return l
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def pushEnvironment ---------------------------------------------------------------------{{{-
    
    def pushEnvironment(self):
        self._mEnvironmentStack.append(os.environ)
        os.environ = dict(os.environ)
 
    # --------------------------------------------------------------------------------------------}}}-
 
    # -- def popEnvironment ----------------------------------------------------------------------{{{-
    
    def popEnvironment(self):
        if len(self._mEnvironmentStack) > 0:
            os.environ = self._mEnvironmentStack.pop()
            return True
        return False
 
    # --------------------------------------------------------------------------------------------}}}-
 
 
# ----------------------------------------------------------------------------------------------}}}-