-
-
Notifications
You must be signed in to change notification settings - Fork 146
/
harness.py
229 lines (167 loc) · 5.13 KB
/
harness.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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
#!/usr/bin/env python3
"""
State Machine style tests with pexpect, e.g. for interactive mode.
To invoke this file, run the shell wrapper:
test/stateful.sh all
"""
from __future__ import print_function
import os
import pexpect
import signal
import sys
from core import ansi
from test import spec_lib # Using this for a common interface
log = spec_lib.log
def expect_prompt(sh):
sh.expect(r'.*\$')
def get_pid_by_name(name):
"""Return the pid of the process matching `name`."""
# XXX: make sure this is restricted to subprocesses under us.
# This could be problematic on the continuous build if many tests are running
# in parallel.
output = pexpect.run('pgrep --exact --newest %s' % name)
#log('pgrep output %r' % output)
return int(output.split()[-1])
def send_signal(name, sig_num):
"""Kill the most recent process matching `name`."""
os.kill(get_pid_by_name(name), sig_num)
# XXX: osh.sendcontrol("z") does not suspend the foreground process :(
#
# why does osh.sendcontrol("c") generate SIGINT, while osh.sendcontrol("z")
# appears to do nothing?
def stop_process__hack(name):
"""Send sigstop to the most recent process matching `name`"""
send_signal(name, signal.SIGSTOP)
# Mutated by each test file.
CASES = []
def register(skip_shells=None):
if skip_shells is None:
skip_shells = []
def decorator(func):
CASES.append((func.__doc__, func, skip_shells))
return func
return decorator
class Result(object):
SKIP = 1
OK = 2
FAIL = 3
def RunCases(cases, case_predicate, shell_pairs, results):
for i, (desc, func, skip_shells) in enumerate(cases):
if not case_predicate(i, desc):
continue
result_row = [i]
for shell_label, shell_path in shell_pairs:
skip = shell_label in skip_shells
skip_str = 'SKIP' if skip else ''
print()
print('%s\t%d\t%s\t%s' % (skip_str, i, shell_label, desc))
print()
if skip:
result_row.append(Result.SKIP)
continue
env = None
sh_argv = []
if shell_label in ('bash', 'osh'):
sh_argv.extend(['--rcfile', '/dev/null'])
# Why the heck is --norc different from --rcfile /dev/null in bash??? This
# makes it so the prompt of the parent shell doesn't leak. Very annoying.
if shell_label == 'bash':
sh_argv.append('--norc')
#print(sh_argv)
# Python 3: encoding required
sh = pexpect.spawn(
shell_path, sh_argv, env=env, encoding='utf-8', timeout=1.0)
sh.shell_label = shell_label # for tests to use
# Generally don't want local echo, it gets confusing fast.
sh.setecho(False)
ok = True
try:
func(sh)
except Exception as e:
import traceback
traceback.print_exc(file=sys.stderr)
result_row.append(Result.FAIL)
ok = False
finally:
sh.close()
if ok:
result_row.append(Result.OK)
result_row.append(desc)
results.append(result_row)
def PrintResults(shell_pairs, results):
f = sys.stdout
if f.isatty():
fail_color = ansi.BOLD + ansi.RED
ok_color = ansi.BOLD + ansi.GREEN
bold = ansi.BOLD
reset = ansi.RESET
else:
fail_color = ''
ok_color = ''
bold = ''
reset = ''
f.write('\n')
# TODO: Might want an HTML version too
sh_labels = [shell_label for shell_label, _ in shell_pairs]
f.write(bold)
f.write('case\t') # case number
for sh_label in sh_labels:
f.write(sh_label)
f.write('\t')
f.write(reset)
f.write('\n')
num_failures = 0
for row in results:
case_num = row[0]
desc = row[-1]
f.write('%d\t' % case_num)
for cell in row[1:-1]:
if cell == Result.SKIP:
f.write('SKIP\t')
elif cell == Result.FAIL:
num_failures += 1
f.write('%sFAIL%s\t' % (fail_color, reset))
elif cell == Result.OK:
f.write('%sok%s\t' % (ok_color, reset))
else:
raise AssertionError(cell)
f.write(desc)
f.write('\n')
return num_failures
def main(argv):
# NOTE: Some options are ignored
o = spec_lib.Options()
opts, argv = o.parse_args(argv)
# List test cases and return
if opts.do_list:
for i, (desc, _, _) in enumerate(CASES):
print('%d\t%s' % (i, desc))
return
shells = argv[1:]
if not shells:
raise RuntimeError('Expected shells to run')
shell_pairs = spec_lib.MakeShellPairs(shells)
if opts.range:
begin, end = spec_lib.ParseRange(opts.range)
case_predicate = spec_lib.RangePredicate(begin, end)
elif opts.regex:
desc_re = re.compile(opts.regex, re.IGNORECASE)
case_predicate = spec_lib.RegexPredicate(desc_re)
else:
case_predicate = lambda i, case: True
if 0:
print(shell_pairs)
print(CASES)
results = [] # each row is a list
RunCases(CASES, case_predicate, shell_pairs, results)
num_failures = PrintResults(shell_pairs, results)
if opts.osh_failures_allowed != num_failures:
log('test/interactive: Expected %d failures, got %d', opts.osh_failures_allowed, num_failures)
return 1
return 0
if __name__ == '__main__':
try:
sys.exit(main(sys.argv))
except RuntimeError as e:
print('FATAL: %s' % e, file=sys.stderr)
sys.exit(1)