1 | import os
|
---|
2 | import sys
|
---|
3 | import subprocess
|
---|
4 | import threading
|
---|
5 | import signal
|
---|
6 |
|
---|
7 |
|
---|
8 | def get_environment():
|
---|
9 | """
|
---|
10 | Get the environment parameter, depending on OS (Win/Unix).
|
---|
11 | """
|
---|
12 | if os.name == 'nt': # not tested!
|
---|
13 | environment = '--environment=WIN'
|
---|
14 | else:
|
---|
15 | environment = '--environment=UNIX'
|
---|
16 | return environment
|
---|
17 |
|
---|
18 |
|
---|
19 | def print_no_newline(string):
|
---|
20 | sys.stdout.write(str(string))
|
---|
21 | sys.stdout.flush()
|
---|
22 |
|
---|
23 |
|
---|
24 | def run_popen_with_timeout(
|
---|
25 | command, timeout, input_data, stdin, stdout, stderr, env=None):
|
---|
26 | """
|
---|
27 | Run a sub-program in subprocess.Popen, pass it the input_data,
|
---|
28 | kill it if the specified timeout has passed.
|
---|
29 | returns a tuple of success, stdout, stderr
|
---|
30 |
|
---|
31 | sample usage:
|
---|
32 |
|
---|
33 | timeout = 60 # seconds
|
---|
34 | path = '/path/to/event.log'
|
---|
35 | command = ['/usr/bin/tail', '-30', path]
|
---|
36 | input_data = ''
|
---|
37 | success, stdout, stderr = run_popen_with_timeout(command, timeout,
|
---|
38 | input_data)
|
---|
39 | if not success:
|
---|
40 | print('timeout on tail event.log output')
|
---|
41 | tail_output = stdout
|
---|
42 | """
|
---|
43 | kill_check = threading.Event()
|
---|
44 |
|
---|
45 | def _kill_process_after_a_timeout(pid):
|
---|
46 | try:
|
---|
47 | os.kill(pid, signal.SIGTERM)
|
---|
48 | except OSError:
|
---|
49 | # catch a possible race condition, the process terminated normally
|
---|
50 | # between the timer firing and our kill
|
---|
51 | return
|
---|
52 | kill_check.set() # tell the main routine that we had to kill
|
---|
53 | # use SIGKILL if hard to kill...
|
---|
54 | return
|
---|
55 |
|
---|
56 | stdout_l = []
|
---|
57 |
|
---|
58 | # don't use shell if command/options come in as list
|
---|
59 | use_shell = not isinstance(command, list)
|
---|
60 | try:
|
---|
61 | p = subprocess.Popen(command, bufsize=1, shell=use_shell,
|
---|
62 | stdin=stdin, stdout=stdout,
|
---|
63 | stderr=stderr, env=env)
|
---|
64 | except OSError as error_message:
|
---|
65 | stderr = 'OSError: ' + str(error_message)
|
---|
66 | return (False, '', stderr)
|
---|
67 | pid = p.pid
|
---|
68 |
|
---|
69 | watchdog = threading.Timer(timeout, _kill_process_after_a_timeout,
|
---|
70 | args=(pid, ))
|
---|
71 | watchdog.start()
|
---|
72 |
|
---|
73 | while True:
|
---|
74 | output = p.stdout.readline(1).decode('utf-8')
|
---|
75 | if output in ['', b''] and p.poll() is not None:
|
---|
76 | break
|
---|
77 | if output == '\n':
|
---|
78 | print(output)
|
---|
79 | else:
|
---|
80 | print_no_newline(output)
|
---|
81 | stdout_l.append(output)
|
---|
82 |
|
---|
83 | try:
|
---|
84 | (stdout, stderr) = p.communicate(input_data)
|
---|
85 | except OSError as error_message:
|
---|
86 | stdout = ''
|
---|
87 | stderr = 'OSError: ' + str(error_message)
|
---|
88 | p.returncode = -666
|
---|
89 |
|
---|
90 | watchdog.cancel() # if it's still waiting to run
|
---|
91 |
|
---|
92 | # if it timed out, success is False
|
---|
93 | success = (not kill_check.isSet()) and p.returncode >= 0
|
---|
94 | kill_check.clear()
|
---|
95 | return (success, ''.join(stdout_l), stderr)
|
---|
96 |
|
---|
97 |
|
---|
98 | class Runner(object):
|
---|
99 | """
|
---|
100 | Setup to run envjasmine "specs" (tests).
|
---|
101 |
|
---|
102 | To use it, probably best to put it inside a normal python
|
---|
103 | unit test suite, then just print out the output.
|
---|
104 | """
|
---|
105 |
|
---|
106 | def __init__(self, rootdir=None, testdir=None, configfile=None,
|
---|
107 | browser_configfile=None):
|
---|
108 | """
|
---|
109 | Set up paths, by default everything is
|
---|
110 | inside the "envjasmine" folder right here.
|
---|
111 | Giving no paths, the sample specs from envjasmine will be run.
|
---|
112 | XXX: it would be more practical if this raised an exception
|
---|
113 | and you know you're not running the tests you want.
|
---|
114 |
|
---|
115 | parameters:
|
---|
116 | testdir - the directory that holds the "mocks", "specs"
|
---|
117 | and "include" directories for the actual tests.
|
---|
118 | rootdir - the directory where the envjasmine code lives in.
|
---|
119 | configfile - path to an extra js config file that is run for the tests.
|
---|
120 | browser_configfile - path to an extra js config file for running
|
---|
121 | the tests in browser.
|
---|
122 | """
|
---|
123 | here = os.path.dirname(__file__)
|
---|
124 | self.libdir = here
|
---|
125 | self.rootdir = rootdir or os.path.join(here, 'envjasmine')
|
---|
126 | self.testdir = testdir or self.rootdir
|
---|
127 | self.configfile = configfile
|
---|
128 | self.browser_configfile = browser_configfile
|
---|
129 | self.runner_html = os.path.join(here, 'runner.html')
|
---|
130 |
|
---|
131 | def run(self, spec=None, timeout=None):
|
---|
132 | """
|
---|
133 | Run the js tests with envjasmine, return success (true/false) and
|
---|
134 | the captured stdout data
|
---|
135 |
|
---|
136 | spec: (relative) path to a spec file (run only that spec)
|
---|
137 | timeout: Set it to a given number of seconds and the process running
|
---|
138 | the js tests will be killed passed that time
|
---|
139 | """
|
---|
140 | environment = get_environment()
|
---|
141 | rhino_path = os.path.join(self.rootdir, 'lib', 'rhino', 'js.jar')
|
---|
142 | envjasmine_js_path = os.path.join(self.rootdir, 'lib', 'envjasmine.js')
|
---|
143 | rootdir_param = '--rootDir=%s' % self.rootdir
|
---|
144 | testdir_param = '--testDir=%s' % self.testdir
|
---|
145 | if self.browser_configfile and os.path.exists(self.browser_configfile):
|
---|
146 | self.write_browser_htmlfile()
|
---|
147 |
|
---|
148 | command = [
|
---|
149 | 'java',
|
---|
150 | '-Duser.timezone=US/Eastern',
|
---|
151 | '-Dfile.encoding=utf-8',
|
---|
152 | '-jar',
|
---|
153 | rhino_path,
|
---|
154 | envjasmine_js_path,
|
---|
155 | '--disableColor',
|
---|
156 | environment,
|
---|
157 | rootdir_param,
|
---|
158 | testdir_param
|
---|
159 | ]
|
---|
160 |
|
---|
161 | if self.configfile and os.path.exists(self.configfile):
|
---|
162 | command.append('--configFile=%s' % self.configfile)
|
---|
163 |
|
---|
164 | # if we were asked to test only some of the spec files,
|
---|
165 | # addd them to the command line:
|
---|
166 | if spec is not None:
|
---|
167 | if not isinstance(spec, list):
|
---|
168 | spec = [spec]
|
---|
169 | command.extend(spec)
|
---|
170 |
|
---|
171 | shell = False
|
---|
172 | stdin = None
|
---|
173 | stdout = subprocess.PIPE
|
---|
174 | stderr = subprocess.PIPE
|
---|
175 | input_data = ''
|
---|
176 |
|
---|
177 | success, stdout, stderr = run_popen_with_timeout(
|
---|
178 | command, timeout, input_data, stdin, stdout, stderr
|
---|
179 | )
|
---|
180 |
|
---|
181 | # success will be true if the subprocess did not timeout, now look
|
---|
182 | # for actual failures if there was not a timeout
|
---|
183 | if success:
|
---|
184 | success = self.did_test_pass(stdout)
|
---|
185 | return success, stdout
|
---|
186 |
|
---|
187 | def did_test_pass(self, stdout):
|
---|
188 | for line in stdout.splitlines():
|
---|
189 | if 'Failed' in line:
|
---|
190 | failed = line.split(':')[1].strip()
|
---|
191 | return failed == '0'
|
---|
192 | return False
|
---|
193 |
|
---|
194 | def write_browser_htmlfile(self):
|
---|
195 | markup = self.create_testRunnerHtml()
|
---|
196 | with open("browser.runner.html", 'w') as file:
|
---|
197 | file.write(markup)
|
---|
198 |
|
---|
199 | def create_testRunnerHtml(self):
|
---|
200 | with open(self.runner_html, 'r') as runner_html:
|
---|
201 | html = runner_html.read()
|
---|
202 | return html % {"libDir": os.path.normpath(self.libdir),
|
---|
203 | "testDir": os.path.normpath(self.testdir),
|
---|
204 | "browser_configfile": self.browser_configfile}
|
---|