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
#
 
from future import standard_library
standard_library.install_aliases()
from builtins import str
from builtins import object
 
# Already ported to Python 3
 
 
# -- system imports ----------------------------------------------------------------------------{{{-
import base64
import os
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
 
gCurrentSpecialMessage = ""
gSpecialMessageMarker = "<MLAB TC=1/>"
gEnablePrettyLogging = False
gLogInfoMessages = True
 
def takeCurrentSpecialMessage():
  global gCurrentSpecialMessage
  msg = gCurrentSpecialMessage
  gCurrentSpecialMessage = ""
  return msg
 
def emitSpecialMessage(loggingMethod, file, line, type, message, escapeHtml=True, formattedStack=''):
  global gCurrentSpecialMessage
  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:
    gCurrentSpecialMessage = specialMessage
    escapedMsg = "<font style='font-weight:normal'><font color=black>" + mevis.MLAB.createHyperLinkWithLine(file, line) + "(" + str(line) + "):</font></font> " + escapedMsg
    escapedMsg += gSpecialMessageMarker
    loggingMethod(escapedMsg)
  else:
    loggingMethod(specialMessage)
 
def emitSpecialCommand(file, line, type, fileList, message, escapeHtml=True):
  global gCurrentSpecialMessage
  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:
    gCurrentSpecialMessage = specialMessage
    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 += gSpecialMessageMarker
    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(object):
  def __init__(self, stackDepth):
    self.__testHelper__testHelper = TestHelper.getInstance()
    self.__stackDepth__stackDepth = stackDepth
    self.__hasSetStackFrame__hasSetStackFrame = False
    
  def __enter__(self):
    if not self.__testHelper__testHelper.hasCallerStackFrame():
      stackFrame = self.getCallingStackFramegetCallingStackFrame(self.__stackDepth__stackDepth+1)
      self.__testHelper__testHelper.setCallerStackFrame(stackFrame)
      self.__hasSetStackFrame__hasSetStackFrame = True
      
  def __exit__(self, *_):
    if self.__hasSetStackFrame__hasSetStackFrame:
      self.__testHelper__testHelper.resetCallerStackFrame()
    
  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(object):
  def __init__(self, errorRegExp, logErrorFunc, preventReplacingMessageSeverity=False):
    self.__logErrorFunc__logErrorFunc = logErrorFunc
    self.__errorRegExp__errorRegExp = errorRegExp
    handling = MessageHandling.EXPECT if logErrorFunc is not None else MessageHandling.IGNORE
    self.__messageFilter__messageFilter = ErrorMessageFilter( handling, errorRegExp )
    self.__preventReplacingMessageSeverity__preventReplacingMessageSeverity = preventReplacingMessageSeverity
    self.__previousKeepMessageSeverity__previousKeepMessageSeverity = None
    self.__enableErrorHandling__enableErrorHandling( True )
      
  def __enableErrorHandling( self, enable ):
    logHandler = TestHelper.getInstance().getLogHandler()
    logHandler.pushOrPopMessageFilter(enable, self.__messageFilter__messageFilter)
    if self.__preventReplacingMessageSeverity__preventReplacingMessageSeverity:
      if enable:
        self.__previousKeepMessageSeverity__previousKeepMessageSeverity = logHandler.setKeepMessageSeverity(True)
      else:
        if self.__previousKeepMessageSeverity__previousKeepMessageSeverity != None:
          logHandler.setKeepMessageSeverity(self.__previousKeepMessageSeverity__previousKeepMessageSeverity)
          self.__previousKeepMessageSeverity__previousKeepMessageSeverity = None
 
  def handleResult(self):
    mevis.MLAB.priv().flushPendingMessages()
    hasCaughtError = self.__messageFilter__messageFilter.hadMatch
    self.__enableErrorHandling__enableErrorHandling( False )    
    if self.__logErrorFunc__logErrorFunc and not hasCaughtError:
      self.__logErrorFunc__logErrorFunc("Expected error did not occur: " + regexpToString(self.__errorRegExp__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 = logErrorFunc
    handling = MessageHandling.EXPECT if logErrorFunc is not None else MessageHandling.IGNORE
    self.__warningRegExp__warningRegExp = warningRegExp
    self.__expectWarning__expectWarning = expectWarning
    self.__messageFilter__messageFilter = WarningMessageFilter( handling, warningRegExp )
    self.__enableWarningHandling__enableWarningHandling( True )
      
  def __enableWarningHandling( self, enable ):
    TestHelper.getInstance().getLogHandler().pushOrPopMessageFilter(enable, self.__messageFilter__messageFilter)
 
  def handleResult(self):
    mevis.MLAB.priv().flushPendingMessages()
    hasCaughtWarning = self.__messageFilter__messageFilter.hadMatch
    self.__enableWarningHandling__enableWarningHandling(False)
    if self.__logErrorFunc__logErrorFunc and hasCaughtWarning != self.__expectWarning__expectWarning:
      if self.__expectWarning__expectWarning:
        self.__logErrorFunc__logErrorFunc("Expected warning did not occur: " + regexpToString(self.__warningRegExp__warningRegExp))
      else:
        self.__logErrorFunc__logErrorFunc("Unexpected warning did occur: " + regexpToString(self.__warningRegExp__warningRegExp))
 
    return hasCaughtWarning == self.__expectWarning__expectWarning
 
 
# Helper object to expect infos,
# make sure that you call handleResult() in a finally: clause
class ExpectInfos(object):
  def __init__(self, infoRegExp, allowUnexpectedMessages, logErrorFunc):
    self.__infoRegExp__infoRegExp = infoRegExp
    self.__allowUnexpectedMessages__allowUnexpectedMessages = allowUnexpectedMessages
    self.__logErrorFunc__logErrorFunc = logErrorFunc
    self.__messageFilter__messageFilter = InfoMessageFilter( MessageHandling.EXPECT, infoRegExp )
    self.__enableInfoHandling__enableInfoHandling( True )
 
  def __enableInfoHandling( self, enable ):
    TestHelper.getInstance().getLogHandler().pushOrPopMessageFilter(enable, self.__messageFilter__messageFilter)
    
  def handleResult(self):
    mevis.MLAB.priv().flushPendingMessages()
    hadUnexpectedInfo = self.__messageFilter__messageFilter.hadNonMatch
    hadExpectedInfo = self.__messageFilter__messageFilter.hadMatch
    self.__enableInfoHandling__enableInfoHandling( False )
    if hadUnexpectedInfo:
      if not self.__allowUnexpectedMessages__allowUnexpectedMessages:
        self.__logErrorFunc__logErrorFunc("Unexpected info message occurred")
    if not hadExpectedInfo:
      self.__logErrorFunc__logErrorFunc("Expected info message did not occur: " + regexpToString(self.__infoRegExp__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 (object):
  # -- member variables ------------------------------------------------------------------------{{{-
  
  _mCtx = None
  
  _mTestCaseContext = None
  
  _mHelperModule = None
  
  _mResultDirectory = None
  
  _mPackageList = None
 
  
  _mTestCaseName = None
  
  _mModuleName = None
 
  
  _mTestCaseDataDirectory = None
  
  _mTestCaseResultDirectory = None
 
  
  _mChangeSetStack = None
 
  
  _mLogHandler = None
  
  
  _mStackFrame = None
 
  
  _mEnvironmentStack = None
 
  
  _mExtraTestCaseResults = None
  
  
  _mMacrosLogOnSuccess = True
 
  
  __lockObj = _thread.allocate_lock()
  
  __instance = None
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def __new__ -----------------------------------------------------------------------------{{{-
  
  def __new__ (self, *args, **kargs):
    return self.getInstancegetInstance(*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__lockObj.acquire()
    try:
      if self.__instance__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__instance = object.__new__(self)
 
      if initialize:
        '''Initialize object **here**, as you would do in __init__()...'''
        self._mResultDirectory_mResultDirectory = kargs["resultDir"]
        self._mCtx_mCtx = kargs["context"]
 
        # The FunctionalTesting module contains some helpful modules that are used in the methods.
        if self._mHelperModule_mHelperModule:
          self._mHelperModule_mHelperModule.remove()
        self._mHelperModule_mHelperModule = self._mCtx_mCtx.addModule(mevis.MLAB.moduleLiteral("TestCenterHelperModule"))
 
        self._mChangeSetStack_mChangeSetStack = []
        self._mEnvironmentStack_mEnvironmentStack = []
 
    finally:
      #  Exit from critical section whatever happens
      self.__lockObj__lockObj.release()
    # Critical section end
    return self.__instance__instance
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def hasInstance -------------------------------------------------------------------------{{{-
  
  @classmethod
  def hasInstance(self):
    return self.__instance__instance is not None
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def setPackageList ----------------------------------------------------------------------{{{-
  
  def setPackageList (self, packageList):
    self._mPackageList_mPackageList = packageList
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def getPackageList ----------------------------------------------------------------------{{{-
  
  def getPackageList (self):
    return self._mPackageList_mPackageList
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def getGlobalContext --------------------------------------------------------------------{{{-
  def getGlobalContext (self):
    return self._mCtx_mCtx
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def setTestCaseContext ------------------------------------------------------------------{{{-
  
  def setTestCaseContext (self, ctx):
    self._mTestCaseContext_mTestCaseContext = ctx
    self._mChangeSet_mChangeSet = TC_ChangeSet(ctx)
    self._mExtraTestCaseResults_mExtraTestCaseResults = []
  # --------------------------------------------------------------------------------------------}}}-
  
  # -- def setMacrosLogOnSuccess ---------------------------------------------------------------{{{-
  def setMacrosLogOnSuccess (self, macrosLogOnSuccess):
    self._mMacrosLogOnSuccess_mMacrosLogOnSuccess_mMacrosLogOnSuccess = macrosLogOnSuccess
  # --------------------------------------------------------------------------------------------}}}-
  
  # -- def getMacrosLogOnSuccess ---------------------------------------------------------------{{{-
  def getMacrosLogOnSuccess (self):
    return self._mMacrosLogOnSuccess_mMacrosLogOnSuccess_mMacrosLogOnSuccess
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def getTestCaseContext ------------------------------------------------------------------{{{-
  
  def getTestCaseContext (self):
    return self._mTestCaseContext_mTestCaseContext
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def getHelperContext --------------------------------------------------------------------{{{-
  def getHelperContext (self):
    return self._mHelperModule_mHelperModule
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def getResultDirectory ------------------------------------------------------------------{{{-
  def getResultDirectory (self):
    return self._mResultDirectory_mResultDirectory
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def setModuleName -----------------------------------------------------------------------{{{-
  def setModuleName (self, moduleName):
    self._mModuleName_mModuleName = moduleName
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def getModuleName -----------------------------------------------------------------------{{{-
  def getModuleName (self):
    return self._mModuleName_mModuleName
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def setTestCaseName ---------------------------------------------------------------------{{{-
  def setTestCaseName (self, testCaseName):
    self._mTestCaseName_mTestCaseName = testCaseName
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def getTestCaseName ---------------------------------------------------------------------{{{-
  def getTestCaseName (self):
    return self._mTestCaseName_mTestCaseName
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def setTestCaseDataDirectory ------------------------------------------------------------{{{-
  def setTestCaseDataDirectory (self, testCaseDataPath):
    self._mTestCaseDataDirectory_mTestCaseDataDirectory = testCaseDataPath
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def getTestCaseDataDirectory ------------------------------------------------------------{{{-
  def getTestCaseDataDirectory (self):
    return self._mTestCaseDataDirectory_mTestCaseDataDirectory
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def setTestCaseResultDirectory ------------------------------------------------------------{{{-
  def setTestCaseResultDirectory (self, testCaseName, functionName):
    self._mTestCaseResultDirectory_mTestCaseResultDirectory = os.path.join(self.getResultDirectorygetResultDirectory(), "files", testCaseName)
    if functionName != None:
      self._mTestCaseResultDirectory_mTestCaseResultDirectory = os.path.join(self._mTestCaseResultDirectory_mTestCaseResultDirectory, functionName)
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def getTestCaseResultDirectory ------------------------------------------------------------{{{-
  def getTestCaseResultDirectory (self):
    if self._mTestCaseResultDirectory_mTestCaseResultDirectory and not os.path.isdir(self._mTestCaseResultDirectory_mTestCaseResultDirectory):
      os.makedirs(self._mTestCaseResultDirectory_mTestCaseResultDirectory)
    return self._mTestCaseResultDirectory_mTestCaseResultDirectory
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def pushChangeSet -----------------------------------------------------------------------{{{-
  
  def pushChangeSet (self):
    self._mChangeSetStack_mChangeSetStack.append(TC_ChangeSet(self._mTestCaseContext_mTestCaseContext))
    return len(self._mChangeSetStack_mChangeSetStack)-1
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def popChangeSet ------------------------------------------------------------------------{{{-
  
  def popChangeSet (self):
    retVal = True
    if len(self._mChangeSetStack_mChangeSetStack) > 0:
      cs = self._mChangeSetStack_mChangeSetStack.pop()
      cs.autoRevert()
    else:
      retVal = False
    return retVal
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def popChangeSet ------------------------------------------------------------------------{{{-
  
  def clearChangeSet (self):
    for changeSet in self._mChangeSetStack_mChangeSetStack:
      changeSet.enableAutoRevert(False)
    while self.popChangeSetpopChangeSet():
      pass
  # --------------------------------------------------------------------------------------------}}}-
 
# -- def getChangeSet ------------------------------------------------------------------------{{{-
  
  def getChangeSet (self):
    retVal = None
    if len(self._mChangeSetStack_mChangeSetStack) > 0:
      retVal = self._mChangeSetStack_mChangeSetStack[-1]
    return retVal
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def getChangeSetStackLength -------------------------------------------------------------{{{-
  
  def getChangeSetStackLength (self):
    return len(self._mChangeSetStack_mChangeSetStack)
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def setLogHandler -----------------------------------------------------------------------{{{-
  
  def setLogHandler (self, logHandler):
    self._mLogHandler_mLogHandler = logHandler
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def unsetLogHandler ---------------------------------------------------------------------{{{-
  
  def unsetLogHandler (self):
    self._mLogHandler_mLogHandler = None
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def getLogHandler -----------------------------------------------------------------------{{{-
  
  def getLogHandler (self):
    return self._mLogHandler_mLogHandler
  # --------------------------------------------------------------------------------------------}}}-
  
  def setCallerStackFrame(self, stackFrame):
    self._mStackFrame_mStackFrame = stackFrame
    
  def getCallerStackFrame(self):
    return self._mStackFrame_mStackFrame
    
  def hasCallerStackFrame(self):
    return self._mStackFrame_mStackFrame != None
    
  def resetCallerStackFrame(self):
    self._mStackFrame_mStackFrame = None
 
  # -- def addExtraTestCaseResult --------------------------------------------------------------{{{-
  def addExtraTestCaseResult (self, extraTestCaseResult):
    self._mExtraTestCaseResults_mExtraTestCaseResults.append(extraTestCaseResult)
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def takeExtraTestCaseResults ------------------------------------------------------------{{{-
  def takeExtraTestCaseResults (self):
    l = self._mExtraTestCaseResults_mExtraTestCaseResults
    self._mExtraTestCaseResults_mExtraTestCaseResults = []
    return l
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def pushEnvironment ---------------------------------------------------------------------{{{-
  
  def pushEnvironment (self):
    self._mEnvironmentStack_mEnvironmentStack.append(os.environ)
    os.environ = dict(os.environ)
  # --------------------------------------------------------------------------------------------}}}-
 
  # -- def popEnvironment ----------------------------------------------------------------------{{{-
  
  def popEnvironment (self):
    if len(self._mEnvironmentStack_mEnvironmentStack) > 0:
      os.environ = self._mEnvironmentStack_mEnvironmentStack.pop()
      return True
    return False
  # --------------------------------------------------------------------------------------------}}}-
 
# ----------------------------------------------------------------------------------------------}}}-