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
|
from _pydevd_bundle.pydevd_io import IORedirector
from _pydevd_bundle.pydevd_net_command_factory_xml import NetCommandFactory
import pytest
import sys
def test_io_redirector():
class MyRedirection1(object):
encoding = "foo"
class MyRedirection2(object):
pass
my_redirector = IORedirector(MyRedirection1(), MyRedirection2(), wrap_buffer=True)
none_redirector = IORedirector(None, None, wrap_buffer=True)
assert my_redirector.encoding == "foo"
with pytest.raises(AttributeError):
none_redirector.encoding
# Check that we don't fail creating the IORedirector if the original
# doesn't have a 'buffer'.
for redirector in (
my_redirector,
none_redirector,
):
redirector.write("test")
redirector.flush()
assert not redirector.isatty()
class _DummyWriter(object):
__slots__ = ["commands", "command_meanings"]
def __init__(self):
self.commands = []
self.command_meanings = []
def add_command(self, cmd):
from _pydevd_bundle.pydevd_comm import ID_TO_MEANING
meaning = ID_TO_MEANING[str(cmd.id)]
self.command_meanings.append(meaning)
self.commands.append(cmd)
class _DummyPyDb(object):
def __init__(self):
self.cmd_factory = NetCommandFactory()
self.writer = _DummyWriter()
def test_patch_stdin():
from pydevd import _internal_patch_stdin
py_db = _DummyPyDb()
class _Stub(object):
pass
actions = []
class OriginalStdin(object):
def readline(self):
# On a readline we keep the patched version.
assert sys_mod.stdin is not original_stdin
actions.append("readline")
return "read"
def getpass_stub(*args, **kwargs):
# On getpass we need to revert to the original version.
actions.append("getpass")
assert sys_mod.stdin is original_stdin
return "pass"
sys_mod = _Stub()
original_stdin = sys_mod.stdin = OriginalStdin()
getpass_mod = _Stub()
getpass_mod.getpass = getpass_stub
_internal_patch_stdin(py_db, sys_mod, getpass_mod)
assert sys_mod.stdin.readline() == "read"
assert py_db.writer.command_meanings == ["CMD_INPUT_REQUESTED", "CMD_INPUT_REQUESTED"]
del py_db.writer.command_meanings[:]
assert actions == ["readline"]
del actions[:]
assert getpass_mod.getpass() == "pass"
assert py_db.writer.command_meanings == ["CMD_INPUT_REQUESTED", "CMD_INPUT_REQUESTED"]
del py_db.writer.command_meanings[:]
def test_debug_console():
from _pydev_bundle.pydev_console_utils import DebugConsoleStdIn
class OriginalStdin(object):
def readline(self):
return "read"
original_stdin = OriginalStdin()
py_db = _DummyPyDb()
debug_console_std_in = DebugConsoleStdIn(py_db, original_stdin)
assert debug_console_std_in.readline() == "read"
assert py_db.writer.command_meanings == ["CMD_INPUT_REQUESTED", "CMD_INPUT_REQUESTED"]
del py_db.writer.command_meanings[:]
with debug_console_std_in.notify_input_requested():
with debug_console_std_in.notify_input_requested():
pass
assert py_db.writer.command_meanings == ["CMD_INPUT_REQUESTED", "CMD_INPUT_REQUESTED"]
@pytest.yield_fixture
def _redirect_context():
from _pydevd_bundle.pydevd_io import RedirectToPyDBIoMessages
from _pydevd_bundle.pydevd_io import _RedirectionsHolder
py_db = _DummyPyDb()
_original_get_pydb = RedirectToPyDBIoMessages.get_pydb
_original_stack_stdout = _RedirectionsHolder._stack_stdout
_original_stack_stderr = _RedirectionsHolder._stack_stderr
_original_stdout_redirect = _RedirectionsHolder._pydevd_stdout_redirect_
_original_stderr_redirect = _RedirectionsHolder._pydevd_stderr_redirect_
RedirectToPyDBIoMessages.get_pydb = lambda *args, **kwargs: py_db
_RedirectionsHolder._stack_stdout = []
_RedirectionsHolder._stack_stderr = []
_RedirectionsHolder._pydevd_stdout_redirect_ = None
_RedirectionsHolder._pydevd_stderr_redirect_ = None
original_stdout = sys.stdout
original_stderr = sys.stderr
yield {"py_db": py_db}
sys.stdout = original_stdout
sys.stderr = original_stderr
RedirectToPyDBIoMessages.get_pydb = _original_get_pydb
_RedirectionsHolder._stack_stdout = _original_stack_stdout
_RedirectionsHolder._stack_stderr = _original_stack_stderr
_RedirectionsHolder._pydevd_stdout_redirect_ = _original_stdout_redirect
_RedirectionsHolder._pydevd_stderr_redirect_ = _original_stderr_redirect
def test_redirect_to_pyd_io_messages_basic(_redirect_context):
from _pydevd_bundle.pydevd_io import redirect_stream_to_pydb_io_messages
from _pydevd_bundle.pydevd_io import redirect_stream_to_pydb_io_messages_context
from _pydevd_bundle.pydevd_io import stop_redirect_stream_to_pydb_io_messages
from _pydevd_bundle.pydevd_io import _RedirectionsHolder
py_db = _redirect_context["py_db"]
redirect_stream_to_pydb_io_messages(std="stdout")
assert len(_RedirectionsHolder._stack_stdout) == 1
assert _RedirectionsHolder._pydevd_stdout_redirect_ is not None
sys.stdout.write("aaa")
assert py_db.writer.command_meanings == ["CMD_WRITE_TO_CONSOLE"]
with redirect_stream_to_pydb_io_messages_context():
assert len(_RedirectionsHolder._stack_stdout) == 1
assert _RedirectionsHolder._pydevd_stdout_redirect_ is not None
sys.stdout.write("bbb")
assert py_db.writer.command_meanings == ["CMD_WRITE_TO_CONSOLE", "CMD_WRITE_TO_CONSOLE"]
assert len(_RedirectionsHolder._stack_stdout) == 1
assert _RedirectionsHolder._pydevd_stdout_redirect_ is not None
sys.stdout.write("ccc")
assert py_db.writer.command_meanings == ["CMD_WRITE_TO_CONSOLE", "CMD_WRITE_TO_CONSOLE", "CMD_WRITE_TO_CONSOLE"]
stop_redirect_stream_to_pydb_io_messages(std="stdout")
assert len(_RedirectionsHolder._stack_stdout) == 0
assert _RedirectionsHolder._pydevd_stdout_redirect_ is None
sys.stdout.write("ddd")
assert py_db.writer.command_meanings == ["CMD_WRITE_TO_CONSOLE", "CMD_WRITE_TO_CONSOLE", "CMD_WRITE_TO_CONSOLE"]
@pytest.mark.parametrize("std", ["stderr", "stdout"])
def test_redirect_to_pyd_io_messages_user_change_stdout(_redirect_context, std):
from _pydevd_bundle.pydevd_io import redirect_stream_to_pydb_io_messages
from _pydevd_bundle.pydevd_io import stop_redirect_stream_to_pydb_io_messages
from _pydevd_bundle.pydevd_io import _RedirectionsHolder
py_db = _redirect_context["py_db"]
stack = getattr(_RedirectionsHolder, "_stack_%s" % (std,))
def get_redirect():
return getattr(_RedirectionsHolder, "_pydevd_%s_redirect_" % (std,))
def write(s):
getattr(sys, std).write(s)
redirect_stream_to_pydb_io_messages(std=std)
assert len(stack) == 1
assert get_redirect() is not None
write("aaa")
assert py_db.writer.command_meanings == ["CMD_WRITE_TO_CONSOLE"]
from io import StringIO
stream = StringIO()
setattr(sys, std, stream)
write("bbb")
assert py_db.writer.command_meanings == ["CMD_WRITE_TO_CONSOLE"]
assert stream.getvalue() == "bbb"
# i.e.: because the user changed the sys.stdout, we cannot change it to our previous version.
stop_redirect_stream_to_pydb_io_messages(std=std)
assert len(stack) == 0
assert get_redirect() is None
write("ccc")
assert py_db.writer.command_meanings == ["CMD_WRITE_TO_CONSOLE"]
assert stream.getvalue() == "bbbccc"
|
