bin/appliskel/salome_tester/python_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: python_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 runTestBegin(command):
  33   print("Running:", " ".join(command))
  34   p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  35   return p
  36
  37 def runTestEffective(process):
  38   out, err = process.communicate()
  39   res = process.returncode
  40   # About res value:
  41   # A negative value -N indicates that the child was terminated by signal N (Unix only).
  42   # On Unix, the value 11 generally corresponds to a segmentation fault.
  43   return res, out, err
  44 #
  45
  46 # Display output and errors
  47 def processResult(res, out, err):
  48   # Decode output
  49   out = out.decode('utf_8', errors='ignore') if out else ''
  50   err = err.decode('utf_8', errors='ignore') if err else ''
  51
  52   # Execute hook if it is installed
  53   if getattr(processResult, '__hook', None):
  54     res, out, err = getattr(processResult, '__hook')(res, out, err)
  55
  56   # Print standard output
  57   if out:
  58     print(out)
  59
  60   # Print standard error
  61   if err:
  62     print("    ** Detected error **")
  63     print("Error code: ", res)
  64     print(err, end=' \n')
  65     print("    ** end of message **")
  66
  67   return res
  68 #
  69
  70 def installHook(hook=None):
  71   """Install custome hook to process test result."""
  72   with suppress(AttributeError, IndexError):
  73     hook = hook.split(',')
  74     hook_path = hook.pop(0)
  75     hook_func = hook.pop(0)
  76     if os.path.exists(hook_path) and hook_func:
  77       with open(hook_path, 'rb') as hook_script:
  78         dic = {}
  79         exec(hook_script.read(), dic)
  80         processResult.__hook = dic.get(hook_func)
  81         print("Custom hook has been installed")
  82 #
  83
  84 subPIDToKill = None
  85
  86 # Timeout management
  87 class TimeoutException(Exception):
  88   """Exception raised when test timeout is reached."""
  89 #
  90 def timeoutHandler(signum, frame):
  91   if subPIDToKill:
  92     try:
  93       os.kill(subPIDToKill, signal.SIGTERM)
  94     except:
  95       pass
  96   raise TimeoutException()
  97 #
  98
  99 if __name__ == "__main__":
 100   timeout_delay = sys.argv[1]
 101   args = sys.argv[2:]
 102
 103   # Install hook if supplied via command line
 104   with suppress(IndexError, ValueError):
 105     index = args.index('--hook')
 106     args.pop(index)
 107     hook = args.pop(index)
 108     installHook(hook)
 109
 110   # Add explicit call to python executable if a Python script is passed as
 111   # first argument
 112   if not args:
 113     print("Invalid arguments for python_test_driver.py. No command defined.")
 114     sys.exit(1)
 115   _, ext = os.path.splitext(args[0])
 116   if ext == ".py":
 117     test_and_args = [sys.executable] + args
 118   else:
 119     test_and_args = args
 120
 121   # Set timeout handler
 122   print("Test timeout explicitly set to: %s seconds"%timeout_delay)
 123   timeout_sec = abs(int(timeout_delay)-10)
 124   if sys.platform == 'win32':
 125     from threading import Timer
 126     timer = Timer(timeout_sec, timeoutHandler)
 127     timer.start()
 128   else:
 129     signal.alarm(timeout_sec)
 130     signal.signal(signal.SIGALRM, timeoutHandler)
 131
 132   res = 1
 133   try:
 134     process = runTestBegin(test_and_args)
 135     subPIDToKill = process.pid
 136     res, out, err = runTestEffective(process)
 137     res = processResult(res, out, err)
 138   except TimeoutException:
 139     print("FAILED : timeout(%s) is reached"%timeout_delay)
 140   except:
 141     import traceback
 142     traceback.print_exc()
 143     pass
 144   if sys.platform == 'win32':
 145     timer.cancel()
 146   print("Exit test with status code:", res)
 147   sys.exit(res)
 148 #