bin/appliskel/salome_tester/salome_test_driver.py

   1 #!/usr/bin/env python3
   2 # Copyright (C) 2015-2023  CEA, EDF, OPEN CASCADE
   3 #
   4 # This library is free software; you can redistribute it and/or
   5 # modify it under the terms of the GNU Lesser General Public
   6 # License as published by the Free Software Foundation; either
   7 # version 2.1 of the License, or (at your option) any later version.
   8 #
   9 # This library is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12 # Lesser General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Lesser General Public
  15 # License along with this library; if not, write to the Free Software
  16 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  17 #
  18 # See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
  19 #
  20
  21 """
  22 Usage: salome_test_driver.py <timeout_delay> <test command> [test command arguments]
  23 """
  24
  25 import sys
  26 import os
  27 import subprocess
  28 import signal
  29 from contextlib import suppress
  30
  31 # Run test
  32 def runTest(command):
  33   print("Running:", " ".join(command))
  34   p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  35   out, err = p.communicate()
  36   res = p.returncode
  37   # About res value:
  38   # A negative value -N indicates that the child was terminated by signal N (Unix only).
  39   # On Unix, the value 11 generally corresponds to a segmentation fault.
  40   return res, out, err
  41 #
  42
  43 # Display output and errors
  44 def processResult(res, out, err):
  45   # Decode output
  46   out = out.decode('utf_8', errors='ignore') if out else ''
  47   err = err.decode('utf_8', errors='ignore') if err else ''
  48
  49   # Execute hook if it is installed
  50   if getattr(processResult, '__hook', None):
  51     res, out, err = getattr(processResult, '__hook')(res, out, err)
  52
  53   # Print standard output
  54   if out:
  55     print(out)
  56
  57   # Print standard error
  58   if err:
  59     print("    ** Detected error **")
  60     print("Error code: ", res)
  61     print(err, end=' \n')
  62     print("    ** end of message **")
  63
  64   return res
  65 #
  66
  67 def installHook(hook=None):
  68   """Install custome hook to process test result."""
  69   with suppress(AttributeError, IndexError):
  70     hook = hook.split(',')
  71     hook_path = hook.pop(0)
  72     hook_func = hook.pop(0)
  73     if os.path.exists(hook_path) and hook_func:
  74       with open(hook_path, 'rb') as hook_script:
  75         dic = {}
  76         exec(hook_script.read(), dic)
  77         processResult.__hook = dic.get(hook_func)
  78         print("Custom hook has been installed")
  79 #
  80
  81 # Timeout management
  82 class TimeoutException(Exception):
  83   """Exception raised when test timeout is reached."""
  84 #
  85 def timeoutHandler(signum, frame):
  86   raise TimeoutException()
  87 #
  88
  89 if __name__ == "__main__":
  90   timeout_delay = sys.argv[1]
  91   args = sys.argv[2:]
  92
  93   # Install hook if supplied via command line
  94   with suppress(IndexError, ValueError):
  95     index = args.index('--hook')
  96     args.pop(index)
  97     hook = args.pop(index)
  98     installHook(hook)
  99
 100   # Add explicit call to python executable if a Python script is passed as
 101   # first argument
 102   if not args:
 103     print("Invalid arguments for salome_test_driver.py. No command defined.")
 104     sys.exit(1)
 105   _, ext = os.path.splitext(args[0])
 106   if ext == ".py":
 107     test_and_args = [sys.executable] + args
 108   else:
 109     test_and_args = args
 110
 111   # Ensure OMNIORB_USER_PATH is set
 112   from salomeContextUtils import setOmniOrbUserPath
 113   setOmniOrbUserPath()
 114
 115   # Set timeout handler
 116   print("Test timeout explicitly set to: %s seconds"%timeout_delay)
 117   timeout_sec = abs(int(timeout_delay)-10)
 118   if sys.platform == 'win32':
 119     from threading import Timer
 120     timer = Timer(timeout_sec, timeoutHandler)
 121     timer.start()
 122   else:
 123     signal.alarm(timeout_sec)
 124     signal.signal(signal.SIGALRM, timeoutHandler)
 125
 126   # Run test in a new SALOME instance
 127   from salome_instance import SalomeInstance
 128   res = 1
 129   try:
 130     salome_instance = SalomeInstance.start(shutdown_servers=True)
 131     port = salome_instance.get_port()
 132     res, out, err = runTest(test_and_args)
 133     res = processResult(res, out, err)
 134   except TimeoutException:
 135     print("FAILED : timeout(%s) is reached"%timeout_delay)
 136   except Exception:
 137     import traceback
 138     traceback.print_exc()
 139     pass
 140   try:
 141     salome_instance.stop()
 142     os.kill(pid, signal.SIGTERM)
 143   except Exception:
 144     pass
 145   if sys.platform == 'win32':
 146     timer.cancel()
 147   print("Exit test with status code:", res)
 148   sys.exit(res)
 149 #