mirrors/debugpy
1
0
Fork 0
mirror of https://github.com/microsoft/debugpy.git synced 2025-12-23 08:48:12 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Support top-level async. Fixes #951

Browse source
This commit is contained in:
Fabio Zadrozny 2022-08-11 09:47:45 -03:00
parent 6e247fb17b
commit 71d42ed63f
9 changed files with 512 additions and 142 deletions
Download patch file Download diff file Expand all files Collapse all files

178
src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevconsole_code_for_ironpython.py → src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevconsole_code.py
View file

@ -1,11 +1,15 @@
"""Utilities needed to emulate Python's interactive interpreter.
"""
A copy of the code module in the standard library with some changes to work with
async evaluation.
Utilities needed to emulate Python's interactive interpreter.
"""
# Inspired by similar code by Jeff Epler and Fredrik Lundh.
import sys
import traceback
import inspect
# START --------------------------- from codeop import CommandCompiler, compile_command
# START --------------------------- from codeop import CommandCompiler, compile_command
@ -100,18 +104,21 @@ def _maybe_compile(compiler, source, filename, symbol):
try:
code1 = compiler(source + "\n", filename, symbol)
except SyntaxError as err1:
pass
except SyntaxError as e:
err1 = e
try:
code2 = compiler(source + "\n\n", filename, symbol)
except SyntaxError as err2:
pass
except SyntaxError as e:
err2 = e
if code:
return code
if not code1 and repr(err1) == repr(err2):
raise SyntaxError(err1)
try:
if code:
return code
if not code1 and repr(err1) == repr(err2):
raise err1
finally:
err1 = err2 = None
def _compile(source, filename, symbol):
@ -148,6 +155,12 @@ class Compile:
def __init__(self):
self.flags = PyCF_DONT_IMPLY_DEDENT
try:
from ast import PyCF_ALLOW_TOP_LEVEL_AWAIT
self.flags |= PyCF_ALLOW_TOP_LEVEL_AWAIT
except:
pass
def __call__(self, source, filename, symbol):
codeob = compile(source, filename, symbol, self.flags, 1)
for feature in _features:
@ -197,19 +210,33 @@ class CommandCompiler:
__all__ = ["InteractiveInterpreter", "InteractiveConsole", "interact",
"compile_command"]
from _pydev_bundle._pydev_saved_modules import threading
def softspace(file, newvalue):
oldvalue = 0
try:
oldvalue = file.softspace
except AttributeError:
pass
try:
file.softspace = newvalue
except (AttributeError, TypeError):
# "attribute-less object" or "read-only attributes"
pass
return oldvalue
class _EvalAwaitInNewEventLoop(threading.Thread):
def __init__(self, compiled, updated_globals, updated_locals):
threading.Thread.__init__(self)
self.daemon = True
self._compiled = compiled
self._updated_globals = updated_globals
self._updated_locals = updated_locals
# Output
self.evaluated_value = None
self.exc = None
async def _async_func(self):
return await eval(self._compiled, self._updated_locals, self._updated_globals)
def run(self):
try:
import asyncio
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
self.evaluated_value = asyncio.run(self._async_func())
except:
self.exc = sys.exc_info()
class InteractiveInterpreter:
@ -240,7 +267,7 @@ class InteractiveInterpreter:
Arguments are as for compile_command().
One several things can happen:
One of several things can happen:
1) The input is incorrect; compile_command() raised an
exception (SyntaxError or OverflowError). A syntax traceback
@ -287,14 +314,24 @@ class InteractiveInterpreter:
"""
try:
exec(code, self.locals)
is_async = False
if hasattr(inspect, 'CO_COROUTINE'):
is_async = inspect.CO_COROUTINE & code.co_flags == inspect.CO_COROUTINE
if is_async:
t = _EvalAwaitInNewEventLoop(code, self.locals, None)
t.start()
t.join()
if t.exc:
raise t.exc[1].with_traceback(t.exc[2])
else:
exec(code, self.locals)
except SystemExit:
raise
except:
self.showtraceback()
else:
if softspace(sys.stdout, 0):
sys.stdout.write('\n')
def showsyntaxerror(self, filename=None):
"""Display the syntax error that just occurred.
@ -308,24 +345,30 @@ class InteractiveInterpreter:
The output is written by self.write(), below.
"""
type, value, sys.last_traceback = sys.exc_info()
type, value, tb = sys.exc_info()
sys.last_type = type
sys.last_value = value
sys.last_traceback = tb
if filename and type is SyntaxError:
# Work hard to stuff the correct filename in the exception
try:
msg, (dummy_filename, lineno, offset, line) = value
except:
msg, (dummy_filename, lineno, offset, line) = value.args
except ValueError:
# Not the format we expect; leave it alone
pass
else:
# Stuff in the right filename
value = SyntaxError(msg, (filename, lineno, offset, line))
sys.last_value = value
list = traceback.format_exception_only(type, value)
map(self.write, list)
if sys.excepthook is sys.__excepthook__:
lines = traceback.format_exception_only(type, value)
self.write(''.join(lines))
else:
# If someone has set sys.excepthook, we let that take precedence
# over self.write
sys.excepthook(type, value, tb)
def showtraceback(self, *args, **kwargs):
def showtraceback(self):
"""Display the exception that just occurred.
We remove the first stack item because it is our own code.
@ -333,20 +376,18 @@ class InteractiveInterpreter:
The output is written by self.write(), below.
"""
sys.last_type, sys.last_value, last_tb = ei = sys.exc_info()
sys.last_traceback = last_tb
try:
type, value, tb = sys.exc_info()
sys.last_type = type
sys.last_value = value
sys.last_traceback = tb
tblist = traceback.extract_tb(tb)
del tblist[:1]
list = traceback.format_list(tblist)
if list:
list.insert(0, "Traceback (most recent call last):\n")
list[len(list):] = traceback.format_exception_only(type, value)
lines = traceback.format_exception(ei[0], ei[1], last_tb.tb_next)
if sys.excepthook is sys.__excepthook__:
self.write(''.join(lines))
else:
# If someone has set sys.excepthook, we let that take precedence
# over self.write
sys.excepthook(ei[0], ei[1], last_tb)
finally:
tblist = tb = None
map(self.write, list)
last_tb = ei = None
def write(self, data):
"""Write a string.
@ -384,23 +425,28 @@ class InteractiveConsole(InteractiveInterpreter):
"""Reset the input buffer."""
self.buffer = []
def interact(self, banner=None):
def interact(self, banner=None, exitmsg=None):
"""Closely emulate the interactive Python console.
The optional banner argument specify the banner to print
The optional banner argument specifies the banner to print
before the first interaction; by default it prints a banner
similar to the one printed by the real Python interpreter,
followed by the current class name in parentheses (so as not
to confuse this with the real interpreter -- since it's so
close!).
The optional exitmsg argument specifies the exit message
printed when exiting. Pass the empty string to suppress
printing an exit message. If exitmsg is not given or None,
a default message is printed.
"""
try:
sys.ps1 # @UndefinedVariable
sys.ps1
except AttributeError:
sys.ps1 = ">>> "
try:
sys.ps2 # @UndefinedVariable
sys.ps2
except AttributeError:
sys.ps2 = "... "
cprt = 'Type "help", "copyright", "credits" or "license" for more information.'
@ -408,21 +454,17 @@ class InteractiveConsole(InteractiveInterpreter):
self.write("Python %s on %s\n%s\n(%s)\n" %
(sys.version, sys.platform, cprt,
self.__class__.__name__))
else:
elif banner:
self.write("%s\n" % str(banner))
more = 0
while 1:
try:
if more:
prompt = sys.ps2 # @UndefinedVariable
prompt = sys.ps2
else:
prompt = sys.ps1 # @UndefinedVariable
prompt = sys.ps1
try:
line = self.raw_input(prompt)
# Can be None if sys.stdin was redefined
encoding = getattr(sys.stdin, "encoding", None)
if encoding and not isinstance(line, str):
line = line.decode(encoding)
except EOFError:
self.write("\n")
break
@ -432,6 +474,10 @@ class InteractiveConsole(InteractiveInterpreter):
self.write("\nKeyboardInterrupt\n")
self.resetbuffer()
more = 0
if exitmsg is None:
self.write('now exiting %s...\n' % self.__class__.__name__)
elif exitmsg != '':
self.write('%s\n' % exitmsg)
def push(self, line):
"""Push a line to the interpreter.
@ -461,14 +507,14 @@ class InteractiveConsole(InteractiveInterpreter):
When the user enters the EOF key sequence, EOFError is raised.
The base implementation uses the built-in function
raw_input(); a subclass may replace this with a different
input(); a subclass may replace this with a different
implementation.
"""
return input(prompt)
def interact(banner=None, readfunc=None, local=None):
def interact(banner=None, readfunc=None, local=None, exitmsg=None):
"""Closely emulate the interactive Python interpreter.
This is a backwards compatible interface to the InteractiveConsole
@ -480,6 +526,7 @@ def interact(banner=None, readfunc=None, local=None):
banner -- passed to InteractiveConsole.interact()
readfunc -- if not None, replaces InteractiveConsole.raw_input()
local -- passed to InteractiveInterpreter.__init__()
exitmsg -- passed to InteractiveConsole.interact()
"""
console = InteractiveConsole(local)
@ -490,9 +537,18 @@ def interact(banner=None, readfunc=None, local=None):
import readline
except ImportError:
pass
console.interact(banner)
console.interact(banner, exitmsg)
if __name__ == '__main__':
import pdb
pdb.run("interact()\n")
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument('-q', action='store_true',
help="don't print version and copyright messages")
args = parser.parse_args()
if args.q or sys.flags.quiet:
banner = ''
else:
banner = None
interact(banner)

2
src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_comm.py
View file

@ -833,7 +833,7 @@ class InternalGetArray(InternalThreadCommand):
def do_it(self, dbg):
try:
frame = dbg.find_frame(self.thread_id, self.frame_id)
var = pydevd_vars.eval_in_context(self.name, frame.f_globals, frame.f_locals)
var = pydevd_vars.eval_in_context(self.name, frame.f_globals, frame.f_locals, py_db=dbg)
xml = pydevd_vars.table_like_struct_to_xml(var, self.name, self.roffset, self.coffset, self.rows, self.cols, self.format)
cmd = dbg.cmd_factory.make_get_array_message(self.sequence, xml)
dbg.writer.add_command(cmd)

30
src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_console.py
View file

@ -2,8 +2,7 @@
'''
import sys
import traceback
from code import InteractiveConsole
from _pydevd_bundle.pydevconsole_code import InteractiveConsole, _EvalAwaitInNewEventLoop
from _pydev_bundle import _pydev_completer
from _pydev_bundle.pydev_console_utils import BaseInterpreterInterface, BaseStdIn
from _pydev_bundle.pydev_imports import Exec
@ -12,6 +11,8 @@ from _pydevd_bundle import pydevd_save_locals
from _pydevd_bundle.pydevd_io import IOBuf
from pydevd_tracing import get_exception_traceback_str
from _pydevd_bundle.pydevd_xml import make_valid_xml_value
import inspect
from _pydevd_bundle.pydevd_save_locals import update_globals_and_locals
CONSOLE_OUTPUT = "output"
CONSOLE_ERROR = "error"
@ -152,8 +153,29 @@ class DebugConsole(InteractiveConsole, BaseInterpreterInterface):
"""
try:
Exec(code, self.frame.f_globals, self.frame.f_locals)
pydevd_save_locals.save_locals(self.frame)
updated_globals = self.get_namespace()
initial_globals = updated_globals.copy()
updated_locals = None
is_async = False
if hasattr(inspect, 'CO_COROUTINE'):
is_async = inspect.CO_COROUTINE & code.co_flags == inspect.CO_COROUTINE
if is_async:
t = _EvalAwaitInNewEventLoop(code, updated_globals, updated_locals)
t.start()
t.join()
update_globals_and_locals(updated_globals, initial_globals, self.frame)
if t.exc:
raise t.exc[1].with_traceback(t.exc[2])
else:
try:
exec(code, updated_globals, updated_locals)
finally:
update_globals_and_locals(updated_globals, initial_globals, self.frame)
except SystemExit:
raise
except:

2
src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_dont_trace_files.py
View file

@ -70,7 +70,7 @@ DONT_TRACE = {
'pydev_umd.py': PYDEV_FILE,
'pydev_versioncheck.py': PYDEV_FILE,
'pydevconsole.py': PYDEV_FILE,
'pydevconsole_code_for_ironpython.py': PYDEV_FILE,
'pydevconsole_code.py': PYDEV_FILE,
'pydevd.py': PYDEV_FILE,
'pydevd_additional_thread_info.py': PYDEV_FILE,
'pydevd_additional_thread_info_regular.py': PYDEV_FILE,

27
src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_save_locals.py
View file

@ -47,6 +47,7 @@ def make_save_locals_impl():
pass
else:
if '__pypy__' in sys.builtin_module_names:
def save_locals_pypy_impl(frame):
save_locals(frame)
@ -58,6 +59,7 @@ def make_save_locals_impl():
except:
pass
else:
def save_locals_ctypes_impl(frame):
locals_to_fast(ctypes.py_object(frame), ctypes.c_int(0))
@ -67,3 +69,28 @@ def make_save_locals_impl():
save_locals_impl = make_save_locals_impl()
def update_globals_and_locals(updated_globals, initial_globals, frame):
# We don't have the locals and passed all in globals, so, we have to
# manually choose how to update the variables.
#
# Note that the current implementation is a bit tricky: it does work in general
# but if we do something as 'some_var = 10' and 'some_var' is already defined to have
# the value '10' in the globals, we won't actually put that value in the locals
# (which means that the frame locals won't be updated).
# Still, the approach to have a single namespace was chosen because it was the only
# one that enabled creating and using variables during the same evaluation.
assert updated_globals is not None
f_locals = None
for key, val in updated_globals.items():
if initial_globals.get(key) is not val:
if f_locals is None:
# Note: we call f_locals only once because each time
# we call it the values may be reset.
f_locals = frame.f_locals
f_locals[key] = val
if f_locals is not None:
save_locals(frame)

189
src/debugpy/_vendored/pydevd/_pydevd_bundle/pydevd_vars.py
View file

@ -3,7 +3,7 @@
"""
import pickle
from _pydevd_bundle.pydevd_constants import get_frame, get_current_thread_id, \
iter_chars, silence_warnings_decorator
iter_chars, silence_warnings_decorator, get_global_debugger
from _pydevd_bundle.pydevd_xml import ExceptionOnEvaluate, get_type, var_to_xml
from _pydev_bundle import pydev_log
@ -17,6 +17,10 @@ from _pydev_bundle._pydev_saved_modules import threading
from _pydevd_bundle import pydevd_save_locals, pydevd_timeout, pydevd_constants
from _pydev_bundle.pydev_imports import Exec, execfile
from _pydevd_bundle.pydevd_utils import to_string
import inspect
from _pydevd_bundle.pydevd_daemon_thread import PyDBDaemonThread
from _pydevd_bundle.pydevd_save_locals import update_globals_and_locals
from functools import lru_cache
SENTINEL_VALUE = []
@ -203,6 +207,7 @@ def custom_operation(dbg, thread_id, frame_id, scope, attrs, style, code_or_file
pydev_log.exception()
@lru_cache(3)
def _expression_to_evaluate(expression):
keepends = True
lines = expression.splitlines(keepends)
@ -240,13 +245,27 @@ def _expression_to_evaluate(expression):
return expression
def eval_in_context(expression, globals, locals=None):
def eval_in_context(expression, global_vars, local_vars, py_db=None):
result = None
try:
if locals is None:
result = eval(_expression_to_evaluate(expression), globals)
compiled = compile_as_eval(expression)
is_async = inspect.CO_COROUTINE & compiled.co_flags == inspect.CO_COROUTINE
if is_async:
if py_db is None:
py_db = get_global_debugger()
if py_db is None:
raise RuntimeError('Cannot evaluate async without py_db.')
t = _EvalAwaitInNewEventLoop(py_db, compiled, global_vars, local_vars)
t.start()
t.join()
if t.exc:
raise t.exc[1].with_traceback(t.exc[2])
else:
result = t.evaluated_value
else:
result = eval(_expression_to_evaluate(expression), globals, locals)
result = eval(compiled, global_vars, local_vars)
except (Exception, KeyboardInterrupt):
etype, result, tb = sys.exc_info()
result = ExceptionOnEvaluate(result, etype, tb)
@ -258,9 +277,9 @@ def eval_in_context(expression, globals, locals=None):
split = expression.split('.')
entry = split[0]
if locals is None:
locals = globals
curr = locals[entry] # Note: we want the KeyError if it's not there.
if local_vars is None:
local_vars = global_vars
curr = local_vars[entry] # Note: we want the KeyError if it's not there.
for entry in split[1:]:
if entry.startswith('__') and not hasattr(curr, entry):
entry = '_%s%s' % (curr.__class__.__name__, entry)
@ -353,60 +372,117 @@ def _evaluate_with_timeouts(original_func):
return new_func
_ASYNC_COMPILE_FLAGS = None
try:
from ast import PyCF_ALLOW_TOP_LEVEL_AWAIT
_ASYNC_COMPILE_FLAGS = PyCF_ALLOW_TOP_LEVEL_AWAIT
except:
pass
def compile_as_eval(expression):
'''
:param expression:
The expression to be compiled.
The expression to be _compiled.
:return: code object
:raises Exception if the expression cannot be evaluated.
'''
return compile(_expression_to_evaluate(expression), '<string>', 'eval')
expression_to_evaluate = _expression_to_evaluate(expression)
if _ASYNC_COMPILE_FLAGS is not None:
return compile(expression_to_evaluate, '<string>', 'eval', _ASYNC_COMPILE_FLAGS)
else:
return compile(expression_to_evaluate, '<string>', 'eval')
def _update_globals_and_locals(updated_globals, initial_globals, frame):
# We don't have the locals and passed all in globals, so, we have to
# manually choose how to update the variables.
#
# Note that the current implementation is a bit tricky: it does work in general
# but if we do something as 'some_var = 10' and 'some_var' is already defined to have
# the value '10' in the globals, we won't actually put that value in the locals
# (which means that the frame locals won't be updated).
# Still, the approach to have a single namespace was chosen because it was the only
# one that enabled creating and using variables during the same evaluation.
assert updated_globals is not None
f_locals = None
for key, val in updated_globals.items():
if initial_globals.get(key) is not val:
if f_locals is None:
# Note: we call f_locals only once because each time
# we call it the values may be reset.
f_locals = frame.f_locals
def _compile_as_exec(expression):
'''
f_locals[key] = val
:param expression:
The expression to be _compiled.
if f_locals is not None:
pydevd_save_locals.save_locals(frame)
:return: code object
:raises Exception if the expression cannot be evaluated.
'''
expression_to_evaluate = _expression_to_evaluate(expression)
if _ASYNC_COMPILE_FLAGS is not None:
return compile(expression_to_evaluate, '<string>', 'exec', _ASYNC_COMPILE_FLAGS)
else:
return compile(expression_to_evaluate, '<string>', 'exec')
class _EvalAwaitInNewEventLoop(PyDBDaemonThread):
def __init__(self, py_db, compiled, updated_globals, updated_locals):
PyDBDaemonThread.__init__(self, py_db)
self._compiled = compiled
self._updated_globals = updated_globals
self._updated_locals = updated_locals
# Output
self.evaluated_value = None
self.exc = None
async def _async_func(self):
return await eval(self._compiled, self._updated_locals, self._updated_globals)
def _on_run(self):
try:
import asyncio
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
self.evaluated_value = asyncio.run(self._async_func())
except:
self.exc = sys.exc_info()
@_evaluate_with_timeouts
def evaluate_expression(py_db, frame, expression, is_exec):
'''
There are some changes in this function depending on whether it's an exec or an eval.
:param str expression:
The expression to be evaluated.
When it's an exec (i.e.: is_exec==True):
This function returns None.
Any exception that happens during the evaluation is reraised.
If the expression could actually be evaluated, the variable is printed to the console if not None.
Note that if the expression is indented it's automatically dedented (based on the indentation
found on the first non-empty line).
When it's an eval (i.e.: is_exec==False):
This function returns the result from the evaluation.
If some exception happens in this case, the exception is caught and a ExceptionOnEvaluate is returned.
Also, in this case we try to resolve name-mangling (i.e.: to be able to add a self.__my_var watch).
i.e.: something as:
`
def method():
a = 1
`
becomes:
`
def method():
a = 1
`
Also, it's possible to evaluate calls with a top-level await (currently this is done by
creating a new event loop in a new thread and making the evaluate at that thread -- note
that this is still done synchronously so the evaluation has to finish before this
function returns).
:param is_exec: determines if we should do an exec or an eval.
There are some changes in this function depending on whether it's an exec or an eval.
When it's an exec (i.e.: is_exec==True):
This function returns None.
Any exception that happens during the evaluation is reraised.
If the expression could actually be evaluated, the variable is printed to the console if not None.
When it's an eval (i.e.: is_exec==False):
This function returns the result from the evaluation.
If some exception happens in this case, the exception is caught and a ExceptionOnEvaluate is returned.
Also, in this case we try to resolve name-mangling (i.e.: to be able to add a self.__my_var watch).
:param py_db:
The debugger. Only needed if some top-level await is detected (for creating a
PyDBDaemonThread).
'''
if frame is None:
return
@ -457,7 +533,7 @@ def evaluate_expression(py_db, frame, expression, is_exec):
if is_exec:
try:
# try to make it an eval (if it is an eval we can print it, otherwise we'll exec it and
# Try to make it an eval (if it is an eval we can print it, otherwise we'll exec it and
# it will have whatever the user actually did)
compiled = compile_as_eval(expression)
except Exception:
@ -465,18 +541,39 @@ def evaluate_expression(py_db, frame, expression, is_exec):
if compiled is None:
try:
Exec(_expression_to_evaluate(expression), updated_globals, updated_locals)
compiled = _compile_as_exec(expression)
is_async = inspect.CO_COROUTINE & compiled.co_flags == inspect.CO_COROUTINE
if is_async:
t = _EvalAwaitInNewEventLoop(py_db, compiled, updated_globals, updated_locals)
t.start()
t.join()
if t.exc:
raise t.exc[1].with_traceback(t.exc[2])
else:
Exec(compiled, updated_globals, updated_locals)
finally:
# Update the globals even if it errored as it may have partially worked.
_update_globals_and_locals(updated_globals, initial_globals, frame)
update_globals_and_locals(updated_globals, initial_globals, frame)
else:
result = eval(compiled, updated_globals, updated_locals)
is_async = inspect.CO_COROUTINE & compiled.co_flags == inspect.CO_COROUTINE
if is_async:
t = _EvalAwaitInNewEventLoop(py_db, compiled, updated_globals, updated_locals)
t.start()
t.join()
if t.exc:
raise t.exc[1].with_traceback(t.exc[2])
else:
result = t.evaluated_value
else:
result = eval(compiled, updated_globals, updated_locals)
if result is not None: # Only print if it's not None (as python does)
sys.stdout.write('%s\n' % (result,))
return
else:
ret = eval_in_context(expression, updated_globals, updated_locals)
ret = eval_in_context(expression, updated_globals, updated_locals, py_db)
try:
is_exception_returned = ret.__class__ == ExceptionOnEvaluate
except:
@ -485,7 +582,7 @@ def evaluate_expression(py_db, frame, expression, is_exec):
if not is_exception_returned:
# i.e.: by using a walrus assignment (:=), expressions can change the locals,
# so, make sure that we save the locals back to the frame.
_update_globals_and_locals(updated_globals, initial_globals, frame)
update_globals_and_locals(updated_globals, initial_globals, frame)
return ret
finally:
# Should not be kept alive if an exception happens and this frame is kept in the stack.

5
src/debugpy/_vendored/pydevd/pydevconsole.py
View file

@ -5,10 +5,7 @@ from _pydev_bundle._pydev_saved_modules import thread, _code
from _pydevd_bundle.pydevd_constants import IS_JYTHON
start_new_thread = thread.start_new_thread
try:
from code import InteractiveConsole
except ImportError:
from _pydevd_bundle.pydevconsole_code_for_ironpython import InteractiveConsole
from _pydevd_bundle.pydevconsole_code import InteractiveConsole
compile_command = _code.compile_command
InteractiveInterpreter = _code.InteractiveInterpreter

71
src/debugpy/_vendored/pydevd/tests/test_pydevconsole.py
View file

@ -4,12 +4,14 @@ import sys
import pydevconsole
from _pydev_bundle.pydev_imports import xmlrpclib, SimpleXMLRPCServer
from _pydevd_bundle import pydevd_io
from contextlib import contextmanager
import pytest
try:
raw_input
raw_input_name = 'raw_input'
except NameError:
raw_input_name = 'input'
from ast import PyCF_ALLOW_TOP_LEVEL_AWAIT # @UnusedImport
CAN_EVALUATE_TOP_LEVEL_ASYNC = True
except:
CAN_EVALUATE_TOP_LEVEL_ASYNC = False
#=======================================================================================================================
@ -17,11 +19,15 @@ except NameError:
#=======================================================================================================================
class Test(unittest.TestCase):
def test_console_hello(self):
@contextmanager
def interpreter(self):
self.original_stdout = sys.stdout
self.original_stderr = sys.stderr
sys.stdout = pydevd_io.IOBuf()
sys.stderr = pydevd_io.IOBuf()
try:
sys.stdout.encoding = sys.stdin.encoding
sys.stderr.encoding = sys.stdin.encoding
except AttributeError:
# In Python 3 encoding is not writable (whereas in Python 2 it doesn't exist).
pass
@ -34,31 +40,50 @@ class Test(unittest.TestCase):
from _pydev_bundle import pydev_localhost
interpreter = pydevconsole.InterpreterInterface(pydev_localhost.get_localhost(), client_port, threading.current_thread())
(result,) = interpreter.hello("Hello pydevconsole")
self.assertEqual(result, "Hello eclipse")
yield interpreter
except:
# if there's some error, print the output to the actual output.
self.original_stdout.write(sys.stdout.getvalue())
self.original_stderr.write(sys.stderr.getvalue())
raise
finally:
sys.stderr = self.original_stderr
sys.stdout = self.original_stdout
def test_console_requests(self):
self.original_stdout = sys.stdout
sys.stdout = pydevd_io.IOBuf()
def test_console_hello(self):
with self.interpreter() as interpreter:
(result,) = interpreter.hello("Hello pydevconsole")
self.assertEqual(result, "Hello eclipse")
try:
client_port, _server_port = self.get_free_addresses()
client_thread = self.start_client_thread(client_port) # @UnusedVariable
import time
time.sleep(.3) # let's give it some time to start the threads
from _pydev_bundle import pydev_localhost
@pytest.mark.skipif(not CAN_EVALUATE_TOP_LEVEL_ASYNC, reason='Requires top-level async.')
def test_console_async(self):
with self.interpreter() as interpreter:
from _pydev_bundle.pydev_console_utils import CodeFragment
more = interpreter.add_exec(CodeFragment('''
async def async_func(a):
return a
'''))
assert not more
assert not sys.stderr.getvalue()
assert not sys.stdout.getvalue()
interpreter = pydevconsole.InterpreterInterface(pydev_localhost.get_localhost(), client_port, threading.current_thread())
sys.stdout = pydevd_io.IOBuf()
more = interpreter.add_exec(CodeFragment('''x = await async_func(1111)'''))
assert not more
assert not sys.stderr.getvalue()
assert not sys.stdout.getvalue()
more = interpreter.add_exec(CodeFragment('''print(x)'''))
assert not more
assert not sys.stderr.getvalue()
assert '1111' in sys.stdout.getvalue()
def test_console_requests(self):
with self.interpreter() as interpreter:
from _pydev_bundle.pydev_console_utils import CodeFragment
interpreter.add_exec(CodeFragment('class Foo:\n CONSTANT=1\n'))
interpreter.add_exec(CodeFragment('foo=Foo()'))
interpreter.add_exec(CodeFragment('foo.__doc__=None'))
interpreter.add_exec(CodeFragment('val = %s()' % (raw_input_name,)))
interpreter.add_exec(CodeFragment('val = input()'))
interpreter.add_exec(CodeFragment('50'))
interpreter.add_exec(CodeFragment('print (val)'))
found = sys.stdout.getvalue().split()
@ -129,8 +154,6 @@ class Test(unittest.TestCase):
desc.find('Concatenate any number of strings.') >= 0 or
desc.find('bound method str.join') >= 0, # PyPy
"Could not recognize: %s" % (desc,))
finally:
sys.stdout = self.original_stdout
def start_client_thread(self, client_port):
@ -234,7 +257,7 @@ class Test(unittest.TestCase):
server.execLine(' pass')
server.execLine('')
server.execLine('foo = Foo()')
server.execLine('a = %s()' % (raw_input_name,))
server.execLine('a = input()')
server.execLine('print (a)')
initial = time.time()
while not client_thread.requested_input:

150
src/debugpy/_vendored/pydevd/tests_python/test_evaluate_expression.py
View file

@ -1,5 +1,7 @@
from _pydevd_bundle.pydevd_constants import IS_PY38_OR_GREATER, NULL
from _pydevd_bundle.pydevd_xml import ExceptionOnEvaluate
import sys
from _pydevd_bundle.pydevd_constants import IS_PY38_OR_GREATER
import pytest
SOME_LST = ["foo", "bar"]
@ -130,3 +132,149 @@ def test_evaluate_expression_5(disable_critical_log):
assert frame.f_locals['B'] == 6
check(next(iter(obtain_frame())))
class _DummyPyDB(object):
def __init__(self):
self.created_pydb_daemon_threads = {}
self.timeout_tracker = NULL
self.multi_threads_single_notification = False
try:
from ast import PyCF_ALLOW_TOP_LEVEL_AWAIT # @UnusedImport
CAN_EVALUATE_TOP_LEVEL_ASYNC = True
except:
CAN_EVALUATE_TOP_LEVEL_ASYNC = False
@pytest.mark.skipif(not CAN_EVALUATE_TOP_LEVEL_ASYNC, reason='Requires top-level async evaluation.')
def test_evaluate_expression_async_exec(disable_critical_log):
py_db = _DummyPyDB()
async def async_call(a):
return a
async def main():
from _pydevd_bundle.pydevd_vars import evaluate_expression
a = 10
assert async_call is not None # Make sure it's in the locals.
frame = sys._getframe()
eval_txt = 'y = await async_call(a)'
evaluate_expression(py_db, frame, eval_txt, is_exec=True)
assert frame.f_locals['y'] == a
import asyncio
asyncio.run(main())
@pytest.mark.skipif(not CAN_EVALUATE_TOP_LEVEL_ASYNC, reason='Requires top-level async evaluation.')
def test_evaluate_expression_async_exec_as_eval(disable_critical_log):
py_db = _DummyPyDB()
async def async_call(a):
return a
async def main():
from _pydevd_bundle.pydevd_vars import evaluate_expression
assert async_call is not None # Make sure it's in the locals.
frame = sys._getframe()
eval_txt = 'await async_call(10)'
from io import StringIO
_original_stdout = sys.stdout
try:
stringio = sys.stdout = StringIO()
evaluate_expression(py_db, frame, eval_txt, is_exec=True)
finally:
sys.stdout = _original_stdout
# I.e.: Check that we printed the value obtained in the exec.
assert '10\n' in stringio.getvalue()
import asyncio
asyncio.run(main())
@pytest.mark.skipif(not CAN_EVALUATE_TOP_LEVEL_ASYNC, reason='Requires top-level async evaluation.')
def test_evaluate_expression_async_exec_error(disable_critical_log):
py_db = _DummyPyDB()
async def async_call(a):
raise RuntimeError('foobar')
async def main():
from _pydevd_bundle.pydevd_vars import evaluate_expression
assert async_call is not None # Make sure it's in the locals.
frame = sys._getframe()
eval_txt = 'y = await async_call(10)'
with pytest.raises(RuntimeError) as e:
evaluate_expression(py_db, frame, eval_txt, is_exec=True)
assert 'foobar' in str(e)
assert 'y' not in frame.f_locals
import asyncio
asyncio.run(main())
@pytest.mark.skipif(not CAN_EVALUATE_TOP_LEVEL_ASYNC, reason='Requires top-level async evaluation.')
def test_evaluate_expression_async_eval(disable_critical_log):
py_db = _DummyPyDB()
async def async_call(a):
return a
async def main():
from _pydevd_bundle.pydevd_vars import evaluate_expression
a = 10
assert async_call is not None # Make sure it's in the locals.
frame = sys._getframe()
eval_txt = 'await async_call(a)'
v = evaluate_expression(py_db, frame, eval_txt, is_exec=False)
if isinstance(v, ExceptionOnEvaluate):
raise v.result.with_traceback(v.tb)
assert v == a
import asyncio
asyncio.run(main())
@pytest.mark.skipif(not CAN_EVALUATE_TOP_LEVEL_ASYNC, reason='Requires top-level async evaluation.')
def test_evaluate_expression_async_eval_error(disable_critical_log):
py_db = _DummyPyDB()
async def async_call(a):
raise RuntimeError('foobar')
async def main():
from _pydevd_bundle.pydevd_vars import evaluate_expression
a = 10
assert async_call is not None # Make sure it's in the locals.
frame = sys._getframe()
eval_txt = 'await async_call(a)'
v = evaluate_expression(py_db, frame, eval_txt, is_exec=False)
assert isinstance(v, ExceptionOnEvaluate)
assert 'foobar' in str(v.result)
import asyncio
asyncio.run(main())
def test_evaluate_expression_name_mangling(disable_critical_log):
from _pydevd_bundle.pydevd_vars import evaluate_expression
class SomeObj(object):
def __init__(self):
self.__value = 10
self.frame = sys._getframe()
obj = SomeObj()
frame = obj.frame
eval_txt = '''self.__value'''
v = evaluate_expression(None, frame, eval_txt, is_exec=False)
if isinstance(v, ExceptionOnEvaluate):
raise v.result.with_traceback(v.tb)
assert v == 10