2 # Copyright (C) 2015-2023 CEA, EDF, OPEN CASCADE
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.
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.
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
18 # See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
22 Usage: python_test_driver.py <timeout_delay> <test command> [test command arguments]
29 from contextlib import suppress
32 def runTestBegin(command):
33 print("Running:", " ".join(command))
34 p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
37 def runTestEffective(process):
38 out, err = process.communicate()
39 res = process.returncode
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.
46 # Display output and errors
47 def processResult(res, out, err):
49 out = out.decode('utf_8', errors='ignore') if out else ''
50 err = err.decode('utf_8', errors='ignore') if err else ''
52 # Execute hook if it is installed
53 if getattr(processResult, '__hook', None):
54 res, out, err = getattr(processResult, '__hook')(res, out, err)
56 # Print standard output
60 # Print standard error
62 print(" ** Detected error **")
63 print("Error code: ", res)
65 print(" ** end of message **")
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:
79 exec(hook_script.read(), dic)
80 processResult.__hook = dic.get(hook_func)
81 print("Custom hook has been installed")
87 class TimeoutException(Exception):
88 """Exception raised when test timeout is reached."""
90 def timeoutHandler(signum, frame):
93 os.kill(subPIDToKill, signal.SIGTERM)
96 raise TimeoutException()
99 if __name__ == "__main__":
100 timeout_delay = sys.argv[1]
103 # Install hook if supplied via command line
104 with suppress(IndexError, ValueError):
105 index = args.index('--hook')
107 hook = args.pop(index)
110 # Add explicit call to python executable if a Python script is passed as
113 print("Invalid arguments for python_test_driver.py. No command defined.")
115 _, ext = os.path.splitext(args[0])
117 test_and_args = [sys.executable] + args
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)
129 signal.alarm(timeout_sec)
130 signal.signal(signal.SIGALRM, timeoutHandler)
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)
142 traceback.print_exc()
144 if sys.platform == 'win32':
146 print("Exit test with status code:", res)