-
Notifications
You must be signed in to change notification settings - Fork 9
/
run.py
executable file
·99 lines (81 loc) · 3.26 KB
/
run.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# "THE JUICE-WARE LICENSE" (Revision 42)
#
# <[email protected]> wrote this file. As long as you
# keep this notice, you can do whatever you want with this stuff.
# If we meet someday, and you think this stuff is worth it, you
# can buy me a glass of juice in return. Yegor Derevenets
import argparse, subprocess, sys, threading
class ExecutionTimeout(Exception):
def __str__(self):
return "Execution timeout."
class UnexpectedExitCode(Exception):
def __init__(self, returned, expected):
self.returned = returned
self.expected = expected
def __str__(self):
return "Exit code is %d, expected %d." % (self.returned, self.expected)
def run(cmdline, stdout=None, stderr=None, **kwargs):
try:
out = None
err = None
if (stdout is not None):
out = open(stdout, "w")
if (stderr is not None):
err = open(stderr, "w")
return execute(cmdline, stdout=out, stderr=err, **kwargs)
finally:
if out is not None:
out.close()
if err is not None:
err.close()
def execute(cmdline, timeout=None, **kwargs):
class Launcher(object):
def __init__(self, cmdline, **kwargs):
self.cmdline = cmdline
self.kwargs = kwargs
self.process = None
self.exception = None
def __call__(self):
try:
self.process = subprocess.Popen(self.cmdline, **self.kwargs)
self.process.communicate()
except OSError as e:
self.exception = e
launcher = Launcher(cmdline, **kwargs)
thread = threading.Thread(target=launcher)
thread.start()
thread.join(timeout)
if thread.is_alive():
launcher.process.terminate()
thread.join()
raise ExecutionTimeout()
if launcher.exception is not None:
raise launcher.exception
return launcher.process.returncode
def main():
parser = argparse.ArgumentParser(description='Runs an executable.')
parser.add_argument('--stdout', metavar='FILE', help='Redirect standard output of the program to this file.')
parser.add_argument('--stderr', metavar='FILE', help='Redirect standard error of the program to this file.')
parser.add_argument('--timeout', metavar='SEC', type=float, default=None, help='Kill the process after the given timeout.')
parser.add_argument('--exit-code', type=int, default=None, help='Fail if the program does not exit with the given code.')
parser.add_argument('program', help='Command to execute.')
parser.add_argument('argument', nargs='*', help='Arguments to the program.')
args = parser.parse_args()
try:
exit_code = run([args.program] + args.argument, stdout=args.stdout, stderr=args.stderr, timeout=args.timeout)
if args.exit_code is not None and args.exit_code != exit_code:
raise UnexpectedExitCode(exit_code, args.exit_code)
except ExecutionTimeout as e:
sys.stderr.write(str(e) + '\n')
sys.exit(100)
except UnexpectedExitCode as e:
sys.stderr.write(str(e) + '\n')
sys.exit(101)
if args.exit_code == None:
sys.exit(exit_code)
else:
sys.exit(0)
if __name__ == '__main__':
main()