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
|
from contextlib import contextmanager
import pytest
from _pydev_bundle.pydev_override import overrides
from tests_python.debugger_fixtures import DebuggerRunnerSimple, debugger_runner_simple
from tests_python.debugger_unittest import (
AbstractWriterThread,
SHOW_OTHER_DEBUG_INFO,
start_in_daemon_thread,
wait_for_condition,
IS_JYTHON,
)
from _pydev_bundle.pydev_localhost import get_socket_names, get_socket_name
from _pydev_bundle.pydev_imports import xmlrpclib
from _pydev_bundle.pydev_imports import _queue as queue
from _pydev_bundle.pydev_imports import SimpleXMLRPCServer
import time
import socket
builtin_qualifier = "builtins"
@pytest.fixture
def console_setup(tmpdir):
server_queue = queue.Queue()
def notify_finished(more):
server_queue.put(("notify_finished", more))
return ""
class ConsoleRunner(DebuggerRunnerSimple):
@overrides(DebuggerRunnerSimple.add_command_line_args)
def add_command_line_args(self, args, dap=False):
port, client_port = get_socket_names(2, close=True)
args.extend((writer.get_pydevconsole_file(), str(port[1]), str(client_port[1])))
self.port = port
self.client_port = client_port
server = SimpleXMLRPCServer(client_port)
server.register_function(notify_finished, "NotifyFinished")
start_in_daemon_thread(server.serve_forever, [])
self.proxy = xmlrpclib.ServerProxy("http://%s:%s/" % port)
return args
class WriterThread(AbstractWriterThread):
if IS_JYTHON:
EXPECTED_RETURNCODE = "any"
@overrides(AbstractWriterThread.additional_output_checks)
def additional_output_checks(self, stdout, stderr):
print("output found: %s - %s" % (stdout, stderr))
@overrides(AbstractWriterThread.write_dump_threads)
def write_dump_threads(self):
pass # no-op (may be called on timeout).
def execute_line(self, command, more=False):
runner.proxy.execLine(command)
assert server_queue.get(timeout=5.0) == ("notify_finished", more)
def hello(self):
def _hello():
try:
msg = runner.proxy.hello("ignored")
if msg is not None:
if isinstance(msg, (list, tuple)):
msg = next(iter(msg))
if msg.lower().startswith("hello"):
return True
except:
# That's ok, communication still not ready.
pass
return False
wait_for_condition(_hello)
def close(self):
try:
runner.proxy.close()
except:
# Ignore any errors on close.
pass
def connect_to_debugger(self, debugger_port):
runner.proxy.connectToDebugger(debugger_port)
runner = ConsoleRunner(tmpdir)
writer = WriterThread()
class CaseSetup(object):
@contextmanager
def check_console(self, **kwargs):
for key, value in kwargs.items():
assert hasattr(WriterThread, key)
setattr(WriterThread, key, value)
self.writer = writer
args = runner.get_command_line()
args = runner.add_command_line_args(args)
if SHOW_OTHER_DEBUG_INFO:
print("executing: %s" % (" ".join(args),))
try:
with runner.run_process(args, writer) as dct_with_stdout_stder:
writer.get_stdout = lambda: "".join(dct_with_stdout_stder["stdout"])
writer.get_stderr = lambda: "".join(dct_with_stdout_stder["stderr"])
# Make sure communication is setup.
writer.hello()
yield writer
finally:
writer.log = []
stdout = dct_with_stdout_stder["stdout"]
stderr = dct_with_stdout_stder["stderr"]
writer.additional_output_checks("".join(stdout), "".join(stderr))
return CaseSetup()
def test_console_simple(console_setup):
with console_setup.check_console() as writer:
writer.execute_line("a = 10")
writer.execute_line('print("TEST SUCEEDED")')
writer.close()
writer.finished_ok = True
def test_console_debugger_connected(console_setup):
class _DebuggerWriterThread(AbstractWriterThread):
FORCE_KILL_PROCESS_WHEN_FINISHED_OK = True
def __init__(self):
AbstractWriterThread.__init__(self)
socket_name = get_socket_name(close=True)
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.bind(socket_name)
server_socket.listen(1)
self.port = socket_name[1]
self.__server_socket = server_socket
def run(self):
print("waiting for second process")
self.sock, addr = self.__server_socket.accept()
print("accepted second process")
from tests_python.debugger_unittest import ReaderThread
self.reader_thread = ReaderThread(self.sock)
self.reader_thread.start()
self._sequence = -1
# initial command is always the version
self.write_version()
self.log.append("start_socket")
self.write_make_initial_run()
time.sleep(1)
seq = self.write_list_threads()
msg = self.wait_for_list_threads(seq)
assert msg.thread["name"] == "MainThread"
assert msg.thread["id"] == "console_main"
self.write_get_frame("console_main", "1")
self.wait_for_vars(
[
[
'<var name="a" type="int" qualifier="%s" value="int: 10"' % (builtin_qualifier,),
'<var name="a" type="int" value="int', # jython
],
]
)
self.finished_ok = True
with console_setup.check_console() as writer:
writer.execute_line("a = 10")
debugger_writer_thread = _DebuggerWriterThread()
debugger_writer_thread.start()
writer.connect_to_debugger(debugger_writer_thread.port)
wait_for_condition(lambda: debugger_writer_thread.finished_ok)
writer.execute_line('print("TEST SUCEEDED")')
writer.close()
writer.finished_ok = True
|
