/
tests.py
219 lines (185 loc) · 6.54 KB
/
tests.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
import functools
import os
import StringIO
import time
import uuid
import signal
from nose.tools import istest, assert_equal, assert_raises, assert_true
import spur
from .testing import create_ssh_shell
def test(func):
@functools.wraps(func)
def run_test():
for shell in _create_shells():
with shell:
yield func, shell
def _create_shells():
return [
spur.LocalShell(),
create_ssh_shell()
]
return istest(run_test)
@test
def output_of_run_is_stored(shell):
result = shell.run(["echo", "hello"])
assert_equal("hello\n", result.output)
@test
def output_is_not_truncated_when_not_ending_in_a_newline(shell):
result = shell.run(["echo", "-n", "hello"])
assert_equal("hello", result.output)
@test
def trailing_newlines_are_not_stripped_from_run_output(shell):
result = shell.run(["echo", "\n\n"])
assert_equal("\n\n\n", result.output)
@test
def stderr_output_of_run_is_stored(shell):
result = shell.run(["sh", "-c", "echo hello 1>&2"])
assert_equal("hello\n", result.stderr_output)
@test
def cwd_of_run_can_be_set(shell):
result = shell.run(["pwd"], cwd="/")
assert_equal("/\n", result.output)
@test
def environment_variables_can_be_added_for_run(shell):
result = shell.run(["sh", "-c", "echo $NAME"], update_env={"NAME": "Bob"})
assert_equal("Bob\n", result.output)
@test
def environment_variables_can_be_added_for_run(shell):
result = shell.run(["sh", "-c", "echo $NAME"], update_env={"NAME": "Bob"})
assert_equal("Bob\n", result.output)
@test
def exception_is_raised_if_return_code_is_not_zero(shell):
assert_raises(spur.RunProcessError, lambda: shell.run(["false"]))
@test
def exception_has_output_from_command(shell):
try:
shell.run(["sh", "-c", "echo Hello world!; false"])
assert_true(False)
except spur.RunProcessError as error:
assert_equal("Hello world!\n", error.output)
@test
def exception_has_stderr_output_from_command(shell):
try:
shell.run(["sh", "-c", "echo Hello world! 1>&2; false"])
assert_true(False)
except spur.RunProcessError as error:
assert_equal("Hello world!\n", error.stderr_output)
@test
def exception_message_contains_return_code_and_all_output(shell):
try:
shell.run(["sh", "-c", "echo starting; echo failed! 1>&2; exit 1"])
assert_true(False)
except spur.RunProcessError as error:
assert_equal(
"return code: 1\noutput: starting\n\nstderr output: failed!\n",
error.message
)
@test
def return_code_stored_if_errors_allowed(shell):
result = shell.run(["sh", "-c", "exit 14"], allow_error=True)
assert_equal(14, result.return_code)
@test
def can_get_result_of_spawned_process(shell):
process = shell.spawn(["echo", "hello"])
result = process.wait_for_result()
assert_equal("hello\n", result.output)
@test
def calling_wait_for_result_is_idempotent(shell):
process = shell.spawn(["echo", "hello"])
process.wait_for_result()
result = process.wait_for_result()
assert_equal("hello\n", result.output)
@test
def wait_for_result_raises_error_if_return_code_is_not_zero(shell):
process = shell.spawn(["false"])
assert_raises(spur.RunProcessError, process.wait_for_result)
@test
def can_write_to_stdin_of_spawned_processes(shell):
process = shell.spawn(["sh", "-c", "read value; echo $value"])
process.stdin_write("hello\n")
result = process.wait_for_result()
assert_equal("hello\n", result.output)
@test
def can_tell_if_spawned_process_is_running(shell):
process = shell.spawn(["sh", "-c", "echo after; read dont_care; echo after"])
assert_equal(True, process.is_running())
process.stdin_write("\n")
_wait_for_assertion(lambda: assert_equal(False, process.is_running()))
@test
def can_write_stdout_to_file_object_while_process_is_executing(shell):
output_file = StringIO.StringIO()
process = shell.spawn(
["sh", "-c", "echo hello; read dont_care;"],
stdout=output_file
)
_wait_for_assertion(lambda: assert_equal("hello\n", output_file.getvalue()))
assert process.is_running()
process.stdin_write("\n")
assert_equal("hello\n", process.wait_for_result().output)
@test
def can_write_stderr_to_file_object_while_process_is_executing(shell):
output_file = StringIO.StringIO()
process = shell.spawn(
["sh", "-c", "echo hello 1>&2; read dont_care;"],
stderr=output_file
)
_wait_for_assertion(lambda: assert_equal("hello\n", output_file.getvalue()))
assert process.is_running()
process.stdin_write("\n")
assert_equal("hello\n", process.wait_for_result().stderr_output)
@test
def can_get_process_id_of_process_if_store_pid_is_true(shell):
# TODO: document store_pid
process = shell.spawn(["sh", "-c", "echo $$"], store_pid=True)
result = process.wait_for_result()
assert_equal(int(result.output.strip()), process.pid)
@test
def process_id_is_not_available_if_store_pid_is_not_set(shell):
process = shell.spawn(["sh", "-c", "echo $$"])
assert not hasattr(process, "pid")
@test
def can_send_signal_to_process_if_store_pid_is_set(shell):
# TODO: document send_signal
process = shell.spawn(["cat"], store_pid=True)
assert process.is_running()
process.send_signal(signal.SIGTERM)
_wait_for_assertion(lambda: assert_equal(False, process.is_running()))
@test
def can_write_to_files_opened_by_open(shell):
path = "/tmp/{0}".format(uuid.uuid4())
f = shell.open(path, "w")
try:
f.write("hello")
f.flush()
assert_equal("hello", shell.run(["cat", path]).output)
finally:
f.close()
shell.run(["rm", path])
@test
def can_read_files_opened_by_open(shell):
path = "/tmp/{0}".format(uuid.uuid4())
shell.run(["sh", "-c", "echo hello > '{0}'".format(path)])
f = shell.open(path)
try:
assert_equal("hello\n", f.read())
finally:
f.close()
shell.run(["rm", path])
@test
def open_can_be_used_as_context_manager(shell):
path = "/tmp/{0}".format(uuid.uuid4())
shell.run(["sh", "-c", "echo hello > '{0}'".format(path)])
with shell.open(path) as f:
assert_equal("hello\n", f.read())
# TODO: timeouts in wait_for_result
def _wait_for_assertion(assertion):
timeout = 1
pause = 0.01
start = time.time()
while True:
try:
assertion()
return
except AssertionError:
if time.time() - start > timeout:
raise