Changes in tests/test.py [103c292:dcfedca]
- File:
-
- 1 edited
-
tests/test.py (modified) (5 diffs)
Legend:
- Unmodified
- Added
- Removed
-
tests/test.py
r103c292 rdcfedca 10 10 import tempfile 11 11 import time 12 13 import os14 import psutil15 import signal16 12 17 13 ################################################################################ … … 147 143 # build, skipping to next test on error 148 144 with Timed() as comp_dur: 149 make_ret, _ = make( test.target(), output _file=subprocess.DEVNULL, error=out_file, error_file = err_file )145 make_ret, _ = make( test.target(), output=subprocess.DEVNULL, error=out_file, error_file = err_file ) 150 146 151 147 run_dur = None 152 148 # run everything in a temp directory to make sure core file are handled properly 153 149 with tempdir(): 154 # if the make command succe eds continue otherwise skip to diff150 # if the make command succeds continue otherwise skip to diff 155 151 if success(make_ret): 156 152 with Timed() as run_dur: 157 153 if settings.dry_run or is_exe(exe_file): 158 154 # run test 159 retcode, _ = sh(exe_file, output _file=out_file, input_file=in_file, timeout=True)155 retcode, _ = sh(exe_file, output=out_file, input=in_file, timeout=True) 160 156 else : 161 157 # simply cat the result into the output … … 214 210 except KeyboardInterrupt: 215 211 return False, "" 216 except Exception as ex:217 print("Unexpected error in worker thread : %s" % ex, file=sys.stderr)212 except: 213 print("Unexpected error in worker thread", file=sys.stderr) 218 214 sys.stderr.flush() 219 215 return False, "" … … 223 219 def run_tests(tests, jobs) : 224 220 # clean the sandbox from previous commands 225 make('clean', output_file=subprocess.DEVNULL, error=subprocess.DEVNULL) 226 227 # since python prints stacks by default on a interrupt, redo the interrupt handling to be silent 228 def worker_init(): 229 def sig_int(signal_num, frame): 230 pass 231 232 signal.signal(signal.SIGINT, sig_int) 221 make('clean', output=subprocess.DEVNULL, error=subprocess.DEVNULL) 233 222 234 223 # create the executor for our jobs and handle the signal properly 235 pool = multiprocessing.Pool(jobs , worker_init)224 pool = multiprocessing.Pool(jobs) 236 225 237 226 failed = False 238 239 def stop(x, y):240 print("Tests interrupted by user", file=sys.stderr)241 sys.exit(1)242 signal.signal(signal.SIGINT, stop)243 227 244 228 # for each test to run … … 276 260 277 261 # clean the workspace 278 make('clean', output _file=subprocess.DEVNULL, error=subprocess.DEVNULL)262 make('clean', output=subprocess.DEVNULL, error=subprocess.DEVNULL) 279 263 280 264 return 1 if failed else 0
Note:
See TracChangeset
for help on using the changeset viewer.