-
Notifications
You must be signed in to change notification settings - Fork 13
/
doctests.py
177 lines (130 loc) · 5.64 KB
/
doctests.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
#!/usr/bin/env python3
"""
A shell-executable script to run all the commands contained in README.md or specified in EXTRA_TESTS and
assert that Wheatley does not crash. With decent test coverage, this prevents mistakes like silly crashes
and argument errors.
"""
from subprocess import STDOUT, TimeoutExpired, Popen, PIPE
import os
import shlex
import sys
# ID of a tower that I made called 'DO NOT ENTER'
ROOM_ID = "238915467"
EXAMPLE_METHOD = "Plain Bob Major"
IGNORE_STRING = "<!--- doctest-ignore -->"
# Tests that should be run on top of the examples from README.md
EXTRA_TESTS = []
def get_all_tests():
"""Get all commands that should be tested (these should start with 'wheatley [ID NUMBER]')."""
tests = []
with open("README.md") as readme:
is_in_code = False
ignore_next_examples = False
for i, l in enumerate(readme.read().split("\n")):
stripped_line = l.strip()
# If we see the line IGNORE_STRING, we should ignore the next block of examples
if not is_in_code and stripped_line == IGNORE_STRING:
ignore_next_examples = True
if stripped_line.startswith("```"):
# Alternate into or out of code blocks
is_in_code = not is_in_code
# Clear the ignore flag whenever we *finish* a code block
if not is_in_code:
ignore_next_examples = False
# If this is a code line that starts with `wheatley`, then it should be used as an example
if is_in_code and stripped_line.startswith("wheatley"):
if ignore_next_examples:
print(f"Ignoring README.md:{i}: {stripped_line}")
else:
tests.append((f"README.md:{i}", stripped_line))
return tests + [(f"EXTRA_TESTS:{i}", cmd) for i, cmd in EXTRA_TESTS]
def command_to_converted_args(location, command):
# Bit of a hack, but this will convert commands starting with 'wheatley' to start with './run-wheatley'
# and make sure that [ID NUMBER] is replaced with a valid room ID. This way, the tests will run on the
# version contained in the current commit, rather than a release version.
edited_command = "python ./run-" + command.replace("[ID NUMBER]", ROOM_ID).replace(
"[METHOD TITLE]", '"' + EXAMPLE_METHOD + '"'
)
insert_arg_index = 2
# Check if running inside venv, if so we need to activate in each process
if sys.prefix != sys.base_prefix and os.name == "nt":
# Windows path `\` gets dropped, but Windows is happy to accept `/` instead
activate_script = sys.prefix.replace("\\", "/") + "/Scripts/activate"
activate_script += ".bat"
edited_command = activate_script + " & " + edited_command
insert_arg_index = 4
args = shlex.split(edited_command)
args.insert(insert_arg_index, "integration-test")
return (args, location, edited_command)
# I'm not sure how to do this in python. I want `run_test` to return one of 3 types: an 'OK', a 'TIMEOUT'
# or an 'ERROR' with a return code and an output. These two classes and None are trying to represent the
# enum:
#
# enum TestResult {
# Ok,
# Timeout,
# Error(usize, String),
# }
class Timeout:
def __init__(self):
pass
def result_text(self):
return "TIMEOUT"
class CommandError:
def __init__(self, return_code, output):
self.return_code = return_code
self.output = output
def result_text(self):
return "ERROR"
def check_test(proc):
try:
out, err = proc.communicate(timeout=10)
if proc.returncode != 0:
return CommandError(proc.returncode, out.decode("utf-8"))
except TimeoutExpired:
proc.kill()
return Timeout()
def main():
"""Generate and run all the tests, asserting that Wheatley does not crash."""
errors = []
procs = []
max_processes = 16
# Generate all the edited commands upfront, so that we can line up all the errors
converted_commands = [command_to_converted_args(location, cmd) for (location, cmd) in get_all_tests()]
max_command_length = max([len(cmd) for (_, _, cmd) in converted_commands])
converted_commands_count = len(converted_commands)
process_index = min(converted_commands_count, max_processes)
# Start the original set of processes
for i in range(process_index):
(args, location, edited_command) = converted_commands[i]
procs.append(Popen(args, stderr=STDOUT, stdout=PIPE))
print("Jobs started")
for i in range(converted_commands_count):
(args, location, edited_command) = converted_commands[i]
error = check_test(procs[i])
# Start next process if needed
if process_index < converted_commands_count:
(args, location, edited_command) = converted_commands[process_index]
procs.append(Popen(args, stderr=STDOUT, stdout=PIPE))
process_index += 1
result_text = "ok"
if error is not None:
errors.append((location, edited_command, error))
result_text = error.result_text()
print(edited_command + " " + "." * (max_command_length + 3 - len(edited_command)) + " " + result_text)
# Iterate over the errors
if len(errors) == 0:
print("ALL OK")
return
print("ERRORS FOUND:")
for location, command, e in errors:
print("\n")
print(f" >>> {location}: {command}")
if type(e) == CommandError:
print(f"RETURN CODE: {e.return_code}")
print(f"OUTPUT:\n{e.output}")
if type(e) == Timeout:
print("TIMED OUT")
exit(1)
if __name__ == "__main__":
main()