.. versionadded:: 3.14
The *commands* argument.
+
+.. awaitablefunction:: set_trace_async(*, header=None, commands=None)
+
+ async version of :func:`set_trace`. This function should be used inside an
+ async function with :keyword:`await`.
+
+ .. code-block:: python
+
+ async def f():
+ await pdb.set_trace_async()
+
+ :keyword:`await` statements are supported if the debugger is invoked by this function.
+
+ .. versionadded:: 3.14
+
.. function:: post_mortem(t=None)
Enter post-mortem debugging of the given exception or
backend by default, which is configurable.
(Contributed by Tian Gao in :gh:`124533`.)
+* :func:`pdb.set_trace_async` is added to support debugging asyncio
+ coroutines. :keyword:`await` statements are supported with this
+ function.
+ (Contributed by Tian Gao in :gh:`132576`.)
+
pickle
------
self.commands_bnum = None # The breakpoint number for which we are
# defining a list
+ self.async_shim_frame = None
+ self.async_awaitable = None
+
self._chained_exceptions = tuple()
self._chained_exception_index = 0
super().set_trace(frame)
+ async def set_trace_async(self, frame=None, *, commands=None):
+ if self.async_awaitable is not None:
+ # We are already in a set_trace_async call, do not mess with it
+ return
+
+ if frame is None:
+ frame = sys._getframe().f_back
+
+ # We need set_trace to set up the basics, however, this will call
+ # set_stepinstr() will we need to compensate for, because we don't
+ # want to trigger on calls
+ self.set_trace(frame, commands=commands)
+ # Changing the stopframe will disable trace dispatch on calls
+ self.stopframe = frame
+ # We need to stop tracing because we don't have the privilege to avoid
+ # triggering tracing functions as normal, as we are not already in
+ # tracing functions
+ self.stop_trace()
+
+ self.async_shim_frame = sys._getframe()
+ self.async_awaitable = None
+
+ while True:
+ self.async_awaitable = None
+ # Simulate a trace event
+ # This should bring up pdb and make pdb believe it's debugging the
+ # caller frame
+ self.trace_dispatch(frame, "opcode", None)
+ if self.async_awaitable is not None:
+ try:
+ if self.breaks:
+ with self.set_enterframe(frame):
+ # set_continue requires enterframe to work
+ self.set_continue()
+ self.start_trace()
+ await self.async_awaitable
+ except Exception:
+ self._error_exc()
+ else:
+ break
+
+ self.async_shim_frame = None
+
+ # start the trace (the actual command is already set by set_* calls)
+ if self.returnframe is None and self.stoplineno == -1 and not self.breaks:
+ # This means we did a continue without any breakpoints, we should not
+ # start the trace
+ return
+
+ self.start_trace()
+
def sigint_handler(self, signum, frame):
if self.allow_kbdint:
raise KeyboardInterrupt
return True
- def default(self, line):
- if line[:1] == '!': line = line[1:].strip()
- locals = self.curframe.f_locals
- globals = self.curframe.f_globals
+ def _exec_await(self, source, globals, locals):
+ """ Run source code that contains await by playing with async shim frame"""
+ # Put the source in an async function
+ source_async = (
+ "async def __pdb_await():\n" +
+ textwrap.indent(source, " ") + '\n' +
+ " __pdb_locals.update(locals())"
+ )
+ ns = globals | locals
+ # We use __pdb_locals to do write back
+ ns["__pdb_locals"] = locals
+ exec(source_async, ns)
+ self.async_awaitable = ns["__pdb_await"]()
+
+ def _read_code(self, line):
+ buffer = line
+ is_await_code = False
+ code = None
try:
- buffer = line
if (code := codeop.compile_command(line + '\n', '<stdin>', 'single')) is None:
# Multi-line mode
with self._enable_multiline_completion():
except (EOFError, KeyboardInterrupt):
self.lastcmd = ""
print('\n')
- return
+ return None, None, False
else:
self.stdout.write(continue_prompt)
self.stdout.flush()
self.lastcmd = ""
self.stdout.write('\n')
self.stdout.flush()
- return
+ return None, None, False
else:
line = line.rstrip('\r\n')
buffer += '\n' + line
self.lastcmd = buffer
+ except SyntaxError as e:
+ # Maybe it's an await expression/statement
+ if (
+ self.async_shim_frame is not None
+ and e.msg == "'await' outside function"
+ ):
+ is_await_code = True
+ else:
+ raise
+
+ return code, buffer, is_await_code
+
+ def default(self, line):
+ if line[:1] == '!': line = line[1:].strip()
+ locals = self.curframe.f_locals
+ globals = self.curframe.f_globals
+ try:
+ code, buffer, is_await_code = self._read_code(line)
+ if buffer is None:
+ return
save_stdout = sys.stdout
save_stdin = sys.stdin
save_displayhook = sys.displayhook
sys.stdin = self.stdin
sys.stdout = self.stdout
sys.displayhook = self.displayhook
- if not self._exec_in_closure(buffer, globals, locals):
- exec(code, globals, locals)
+ if is_await_code:
+ self._exec_await(buffer, globals, locals)
+ return True
+ else:
+ if not self._exec_in_closure(buffer, globals, locals):
+ exec(code, globals, locals)
finally:
sys.stdout = save_stdout
sys.stdin = save_stdin
pdb.message(header)
pdb.set_trace(sys._getframe().f_back, commands=commands)
+async def set_trace_async(*, header=None, commands=None):
+ """Enter the debugger at the calling stack frame, but in async mode.
+
+ This should be used as await pdb.set_trace_async(). Users can do await
+ if they enter the debugger with this function. Otherwise it's the same
+ as set_trace().
+ """
+ if Pdb._last_pdb_instance is not None:
+ pdb = Pdb._last_pdb_instance
+ else:
+ pdb = Pdb(mode='inline', backend='monitoring')
+ if header is not None:
+ pdb.message(header)
+ await pdb.set_trace_async(sys._getframe().f_back, commands=commands)
+
# Remote PDB
class _PdbServer(Pdb):
# A test suite for pdb; not very comprehensive at the moment.
import doctest
+import gc
import os
import pdb
import sys
(Pdb) continue
"""
+ def test_pdb_await_support():
+ """Testing await support in pdb
+
+ >>> import asyncio
+
+ >>> async def test():
+ ... print("hello")
+ ... await asyncio.sleep(0)
+ ... print("world")
+ ... return 42
+
+ >>> async def main():
+ ... import pdb
+ ... task = asyncio.create_task(test())
+ ... await pdb.Pdb(nosigint=True, readrc=False).set_trace_async()
+ ... pass
+
+ >>> def test_function():
+ ... asyncio.run(main(), loop_factory=asyncio.EventLoop)
+
+ >>> with PdbTestInput([ # doctest: +ELLIPSIS
+ ... 'x = await task',
+ ... 'p x',
+ ... 'x = await test()',
+ ... 'p x',
+ ... 'new_task = asyncio.create_task(test())',
+ ... 'await new_task',
+ ... 'await non_exist()',
+ ... 's',
+ ... 'continue',
+ ... ]):
+ ... test_function()
+ > <doctest test.test_pdb.test_pdb_await_support[2]>(4)main()
+ -> await pdb.Pdb(nosigint=True, readrc=False).set_trace_async()
+ (Pdb) x = await task
+ hello
+ world
+ > <doctest test.test_pdb.test_pdb_await_support[2]>(4)main()
+ -> await pdb.Pdb(nosigint=True, readrc=False).set_trace_async()
+ (Pdb) p x
+ 42
+ (Pdb) x = await test()
+ hello
+ world
+ > <doctest test.test_pdb.test_pdb_await_support[2]>(4)main()
+ -> await pdb.Pdb(nosigint=True, readrc=False).set_trace_async()
+ (Pdb) p x
+ 42
+ (Pdb) new_task = asyncio.create_task(test())
+ (Pdb) await new_task
+ hello
+ world
+ > <doctest test.test_pdb.test_pdb_await_support[2]>(4)main()
+ -> await pdb.Pdb(nosigint=True, readrc=False).set_trace_async()
+ (Pdb) await non_exist()
+ *** NameError: name 'non_exist' is not defined
+ > <doctest test.test_pdb.test_pdb_await_support[2]>(4)main()
+ -> await pdb.Pdb(nosigint=True, readrc=False).set_trace_async()
+ (Pdb) s
+ > <doctest test.test_pdb.test_pdb_await_support[2]>(5)main()
+ -> pass
+ (Pdb) continue
+ """
+
+ def test_pdb_await_with_breakpoint():
+ """Testing await support with breakpoints set in tasks
+
+ >>> import asyncio
+
+ >>> async def test():
+ ... x = 2
+ ... await asyncio.sleep(0)
+ ... return 42
+
+ >>> async def main():
+ ... import pdb
+ ... task = asyncio.create_task(test())
+ ... await pdb.Pdb(nosigint=True, readrc=False).set_trace_async()
+
+ >>> def test_function():
+ ... asyncio.run(main(), loop_factory=asyncio.EventLoop)
+
+ >>> with PdbTestInput([ # doctest: +ELLIPSIS
+ ... 'b test',
+ ... 'k = await task',
+ ... 'n',
+ ... 'p x',
+ ... 'continue',
+ ... 'p k',
+ ... 'continue',
+ ... ]):
+ ... test_function()
+ > <doctest test.test_pdb.test_pdb_await_with_breakpoint[2]>(4)main()
+ -> await pdb.Pdb(nosigint=True, readrc=False).set_trace_async()
+ (Pdb) b test
+ Breakpoint 1 at <doctest test.test_pdb.test_pdb_await_with_breakpoint[1]>:2
+ (Pdb) k = await task
+ > <doctest test.test_pdb.test_pdb_await_with_breakpoint[1]>(2)test()
+ -> x = 2
+ (Pdb) n
+ > <doctest test.test_pdb.test_pdb_await_with_breakpoint[1]>(3)test()
+ -> await asyncio.sleep(0)
+ (Pdb) p x
+ 2
+ (Pdb) continue
+ > <doctest test.test_pdb.test_pdb_await_with_breakpoint[2]>(4)main()
+ -> await pdb.Pdb(nosigint=True, readrc=False).set_trace_async()
+ (Pdb) p k
+ 42
+ (Pdb) continue
+ """
+
+ def test_pdb_await_contextvar():
+ """Testing await support context vars
+
+ >>> import asyncio
+ >>> import contextvars
+
+ >>> var = contextvars.ContextVar('var')
+
+ >>> async def get_var():
+ ... return var.get()
+
+ >>> async def set_var(val):
+ ... var.set(val)
+ ... return var.get()
+
+ >>> async def main():
+ ... var.set(42)
+ ... import pdb
+ ... await pdb.Pdb(nosigint=True, readrc=False).set_trace_async()
+
+ >>> def test_function():
+ ... asyncio.run(main(), loop_factory=asyncio.EventLoop)
+
+ >>> with PdbTestInput([
+ ... 'p var.get()',
+ ... 'print(await get_var())',
+ ... 'print(await asyncio.create_task(set_var(100)))',
+ ... 'p var.get()',
+ ... 'print(await set_var(99))',
+ ... 'p var.get()',
+ ... 'print(await get_var())',
+ ... 'continue',
+ ... ]):
+ ... test_function()
+ > <doctest test.test_pdb.test_pdb_await_contextvar[5]>(4)main()
+ -> await pdb.Pdb(nosigint=True, readrc=False).set_trace_async()
+ (Pdb) p var.get()
+ 42
+ (Pdb) print(await get_var())
+ 42
+ > <doctest test.test_pdb.test_pdb_await_contextvar[5]>(4)main()
+ -> await pdb.Pdb(nosigint=True, readrc=False).set_trace_async()
+ (Pdb) print(await asyncio.create_task(set_var(100)))
+ 100
+ > <doctest test.test_pdb.test_pdb_await_contextvar[5]>(4)main()
+ -> await pdb.Pdb(nosigint=True, readrc=False).set_trace_async()
+ (Pdb) p var.get()
+ 42
+ (Pdb) print(await set_var(99))
+ 99
+ > <doctest test.test_pdb.test_pdb_await_contextvar[5]>(4)main()
+ -> await pdb.Pdb(nosigint=True, readrc=False).set_trace_async()
+ (Pdb) p var.get()
+ 99
+ (Pdb) print(await get_var())
+ 99
+ > <doctest test.test_pdb.test_pdb_await_contextvar[5]>(4)main()
+ -> await pdb.Pdb(nosigint=True, readrc=False).set_trace_async()
+ (Pdb) continue
+ """
+
def test_pdb_next_command_for_coroutine():
"""Testing skip unwinding stack on yield for coroutines for "next" command
pdb.Pdb._last_pdb_instance.stop_trace()
pdb.Pdb._last_pdb_instance = None
+ # If garbage objects are collected right after we start tracing, we
+ # could stop at __del__ of the object which would fail the test.
+ gc.collect()
+
tests.addTest(
doctest.DocTestSuite(
test_pdb,
--- /dev/null
+Add :func:`pdb.set_trace_async` function to support :keyword:`await` statements in :mod:`pdb`.
projects / thirdparty / Python / cpython.git / commitdiff
