Source code for paraview.smtesting

# This is a module for Server Manager testing using Python.
# This provides several utility functions useful for testing
import os
import re
import sys
from paraview.modules.vtkRemotingServerManager import *
from paraview.modules.vtkRemotingApplication import *
from paraview.modules.vtkRemotingMisc import *

# we get different behavior based on how we import servermanager
# so we want to import servermanager the same way in this module
# as we do in any module that is importing this
SMModuleName = 'paraview.servermanager'
if 'paraview.simple' in sys.modules:
    SMModuleName = 'paraview.simple'

sm = __import__(SMModuleName)
servermanager = sm.servermanager

[docs]class TestError(Exception): pass
__ProcessedCommandLineArguments__ = False DataDir = "" TempDir = "" BaselineImage = "" Threshold = 10.0 SMStatesDir = "" StateXMLFileName = "" UseSavedStateForRegressionTests = False
[docs]def Error(message): print("ERROR: %s" % message) return False
[docs]def ProcessCommandLineArguments(): """Processes the command line areguments.""" global DataDir global TempDir global BaselineImage global Threshold global StateXMLFileName global UseSavedStateForRegressionTests global SMStatesDir global __ProcessedCommandLineArguments__ if __ProcessedCommandLineArguments__: return __ProcessedCommandLineArguments__ = True length = len(sys.argv) index = 1 while index < length: key = sys.argv[index - 1] value = sys.argv[index] index += 2 if key == "-D": DataDir = value elif key == "-V": BaselineImage = value elif key == "-T": TempDir = value elif key == "-S": SMStatesDir = value elif key == "--threshold": Threshold = float(value) elif key == "--state": StateXMLFileName = value elif key == "--use_saved_state": UseSavedStateForRegressionTests = True index -= 1 else: index -= 1 return
[docs]def LoadServerManagerState(filename): """This function loads the servermanager state xml/pvsm. Returns the status of the load.""" global DataDir ProcessCommandLineArguments() parser = servermanager.vtkPVXMLParser() try: fp = open(filename, "r") data = fp.close() except: return Error("Failed to open state file %s" % filename) regExp = re.compile("\${DataDir}") data = regExp.sub(DataDir, data) if not parser.Parse(data): return Error("Failed to parse") loader = servermanager.vtkSMStateLoader() loader.SetSession(servermanager.ActiveConnection.Session) root = parser.GetRootElement() if loader.LoadState(root): pxm = servermanager.vtkSMProxyManager.GetProxyManager().GetActiveSessionProxyManager() pxm.UpdateRegisteredProxiesInOrder(0); pxm.UpdateRegisteredProxies(0) return True return Error("Failed to load state file %s" % filename)
[docs]def DoRegressionTesting(rmProxy=None): """Perform regression testing.""" global TempDir global BaselineImage global Threshold ProcessCommandLineArguments() testing = vtkSMTesting() testing.AddArgument("-T") testing.AddArgument(TempDir) testing.AddArgument("-V") testing.AddArgument(BaselineImage) if not rmProxy: rmProxy = servermanager.GetRenderView() if rmProxy: rmProxy = rmProxy.SMProxy if not rmProxy: raise "Failed to locate view to perform regression testing." # pyProxy(rmProxy).SetRenderWindowSize(300, 300); # rmProxy.GetProperty("RenderWindowSize").SetElement(0, 300) # rmProxy.GetProperty("RenderWindowSize").SetElement(1, 300) # rmProxy.UpdateVTKObjects() rmProxy.StillRender() testing.SetRenderViewProxy(rmProxy) if testing.RegressionTest(Threshold) == 1: return True return Error("Regression Test Failed!")
[docs]def GetUniqueTempDirectory(prefix): """A convenience method to generate a temporary directory unique to the current ParaView process (or MPI group of processes)""" global TempDir pm = servermanager.vtkProcessModule.GetProcessModule() if pm.GetNumberOfLocalPartitions() > 1 and pm.GetSymmetricMPIMode(): if pm.GetPartitionId() == 0: import os pid = os.getpid() else: pid = None from mpi4py import MPI comm = MPI.COMM_WORLD pid = comm.bcast(pid, root=0) else: import os pid = os.getpid() import os.path tempdir = os.path.join(TempDir, "%s%d" % (prefix, pid)) if pm.GetPartitionId() == 0: try: os.makedirs(tempdir) except OSError: pass if pm.GetNumberOfLocalPartitions() > 1 and pm.GetSymmetricMPIMode(): from mpi4py import MPI comm = MPI.COMM_WORLD comm.Barrier() return tempdir
if __name__ == "__main__": # This script loads the state, saves out a temp state and loads the saved state. # This saved state is used for testing -- this will ensure load/save SM state # is working fine. servermanager.Connect() ProcessCommandLineArguments() ret = 1 if StateXMLFileName: if LoadServerManagerState(StateXMLFileName): pxm = servermanager.vtkSMProxyManager.GetProxyManager().GetActiveSessionProxyManager() if UseSavedStateForRegressionTests: saved_state = os.path.join(TempDir, "temp.pvsm") pxm.SaveState(saved_state) pxm.UnRegisterProxies(); LoadServerManagerState(saved_state) try: os.remove(saved_state) except: pass if DoRegressionTesting(): ret = 0 else: Error("No ServerManager state file specified") if ret: # This leads to vtkDebugLeaks reporting leaks, hence we do this # only when the tests failed. sys.exit(ret)