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
|
import os
import platform
import time
from mock import Mock
from pytest import skip, raises
from invoke import (
run,
Local,
Context,
ThreadException,
Responder,
FailingResponder,
WatcherError,
Failure,
CommandTimedOut,
)
from _util import assert_cpu_usage
PYPY = platform.python_implementation() == "PyPy"
class Runner_:
def setup(self):
os.chdir(os.path.join(os.path.dirname(__file__), "_support"))
class responding:
def base_case(self):
# Basic "doesn't explode" test: respond.py will exit nonzero unless
# this works, causing a Failure.
watcher = Responder(r"What's the password\?", "Rosebud\n")
# Gotta give -u or Python will line-buffer its stdout, so we'll
# never actually see the prompt.
run(
"python -u respond_base.py",
watchers=[watcher],
hide=True,
timeout=5,
)
def both_streams(self):
watchers = [
Responder("standard out", "with it\n"),
Responder("standard error", "between chair and keyboard\n"),
]
run(
"python -u respond_both.py",
watchers=watchers,
hide=True,
timeout=5,
)
def watcher_errors_become_Failures(self):
watcher = FailingResponder(
pattern=r"What's the password\?",
response="Rosebud\n",
sentinel="You're not Citizen Kane!",
)
try:
run(
"python -u respond_fail.py",
watchers=[watcher],
hide=True,
timeout=5,
)
except Failure as e:
assert isinstance(e.reason, WatcherError)
assert e.result.exited is None
else:
assert False, "Did not raise Failure!"
class stdin_mirroring:
def piped_stdin_is_not_conflated_with_mocked_stdin(self):
# Re: GH issue #308
# Will die on broken-pipe OSError if bug is present.
run("echo 'lollerskates' | inv -c nested_or_piped foo", hide=True)
def nested_invoke_sessions_not_conflated_with_mocked_stdin(self):
# Also re: GH issue #308. This one will just hang forever. Woo!
run("inv -c nested_or_piped calls-foo", hide=True)
def isnt_cpu_heavy(self):
"stdin mirroring isn't CPU-heavy"
# CPU measurement under PyPy is...rather different. NBD.
if PYPY:
skip()
# Python 3.5 has been seen using up to ~6.0s CPU time under Travis
with assert_cpu_usage(lt=7.0):
run("python -u busywork.py 10", pty=True, hide=True)
def doesnt_break_when_stdin_exists_but_null(self):
# Re: #425 - IOError occurs when bug present
run("inv -c nested_or_piped foo < /dev/null", hide=True)
class IO_hangs:
"IO hangs"
def _hang_on_full_pipe(self, pty):
class Whoops(Exception):
pass
runner = Local(Context())
# Force runner IO thread-body method to raise an exception to mimic
# real world encoding explosions/etc. When bug is present, this
# will make the test hang until forcibly terminated.
runner.handle_stdout = Mock(side_effect=Whoops, __name__="sigh")
# NOTE: both Darwin (10.10) and Linux (Travis' docker image) have
# this file. It's plenty large enough to fill most pipe buffers,
# which is the triggering behavior.
try:
runner.run("cat /usr/share/dict/words", pty=pty)
except ThreadException as e:
assert len(e.exceptions) == 1
assert e.exceptions[0].type is Whoops
else:
assert False, "Did not receive expected ThreadException!"
def pty_subproc_should_not_hang_if_IO_thread_has_an_exception(self):
self._hang_on_full_pipe(pty=True)
def nonpty_subproc_should_not_hang_if_IO_thread_has_an_exception(self):
self._hang_on_full_pipe(pty=False)
class timeouts:
def does_not_fire_when_command_quick(self):
assert run("sleep 1", timeout=5)
def triggers_exception_when_command_slow(self):
before = time.time()
with raises(CommandTimedOut) as info:
run("sleep 5", timeout=0.5)
after = time.time()
# Fudge real time check a bit, <=0.5 typically fails due to
# overhead etc. May need raising further to avoid races? Meh.
assert (after - before) <= 0.75
# Sanity checks of the exception obj
assert info.value.timeout == 0.5
assert info.value.result.command == "sleep 5"
|
