mirror of
https://github.com/RustPython/RustPython.git
synced 2026-10-28 10:44:36 +08:00
Added pdb.py from Python 3.5
Also got its pure-python dependencies bdb.py and cmd.py Also stubbed the C-based gc.py using `pystubber` (https://pypi.org/project/pystubber/)
This commit is contained in:
4 files changed
+2960
No files matched your search
Vendored
+689
@@ -0,0 +1,689 @@
|
||||
"""Debugger basics"""
|
||||
|
||||
import fnmatch
|
||||
import sys
|
||||
import os
|
||||
from inspect import CO_GENERATOR
|
||||
|
||||
__all__ = ["BdbQuit", "Bdb", "Breakpoint"]
|
||||
|
||||
|
||||
class BdbQuit(Exception):
|
||||
"""Exception to give up completely."""
|
||||
|
||||
|
||||
class Bdb:
|
||||
"""Generic Python debugger base class.
|
||||
|
||||
This class takes care of details of the trace facility;
|
||||
a derived class should implement user interaction.
|
||||
The standard debugger class (pdb.Pdb) is an example.
|
||||
"""
|
||||
def __init__(self, skip=None):
|
||||
self.skip = set(skip) if skip else None
|
||||
self.breaks = {}
|
||||
self.fncache = {}
|
||||
self.frame_returning = None
|
||||
|
||||
def canonic(self, filename):
|
||||
if filename == "<" + filename[1:-1] + ">":
|
||||
return filename
|
||||
canonic = self.fncache.get(filename)
|
||||
if not canonic:
|
||||
canonic = os.path.abspath(filename)
|
||||
canonic = os.path.normcase(canonic)
|
||||
self.fncache[filename] = canonic
|
||||
return canonic
|
||||
|
||||
def reset(self):
|
||||
import linecache
|
||||
linecache.checkcache()
|
||||
self.botframe = None
|
||||
self._set_stopinfo(None, None)
|
||||
|
||||
def trace_dispatch(self, frame, event, arg):
|
||||
if self.quitting:
|
||||
return # None
|
||||
if event == 'line':
|
||||
return self.dispatch_line(frame)
|
||||
if event == 'call':
|
||||
return self.dispatch_call(frame, arg)
|
||||
if event == 'return':
|
||||
return self.dispatch_return(frame, arg)
|
||||
if event == 'exception':
|
||||
return self.dispatch_exception(frame, arg)
|
||||
if event == 'c_call':
|
||||
return self.trace_dispatch
|
||||
if event == 'c_exception':
|
||||
return self.trace_dispatch
|
||||
if event == 'c_return':
|
||||
return self.trace_dispatch
|
||||
print('bdb.Bdb.dispatch: unknown debugging event:', repr(event))
|
||||
return self.trace_dispatch
|
||||
|
||||
def dispatch_line(self, frame):
|
||||
if self.stop_here(frame) or self.break_here(frame):
|
||||
self.user_line(frame)
|
||||
if self.quitting: raise BdbQuit
|
||||
return self.trace_dispatch
|
||||
|
||||
def dispatch_call(self, frame, arg):
|
||||
# XXX 'arg' is no longer used
|
||||
if self.botframe is None:
|
||||
# First call of dispatch since reset()
|
||||
self.botframe = frame.f_back # (CT) Note that this may also be None!
|
||||
return self.trace_dispatch
|
||||
if not (self.stop_here(frame) or self.break_anywhere(frame)):
|
||||
# No need to trace this function
|
||||
return # None
|
||||
# Ignore call events in generator except when stepping.
|
||||
if self.stopframe and frame.f_code.co_flags & CO_GENERATOR:
|
||||
return self.trace_dispatch
|
||||
self.user_call(frame, arg)
|
||||
if self.quitting: raise BdbQuit
|
||||
return self.trace_dispatch
|
||||
|
||||
def dispatch_return(self, frame, arg):
|
||||
if self.stop_here(frame) or frame == self.returnframe:
|
||||
# Ignore return events in generator except when stepping.
|
||||
if self.stopframe and frame.f_code.co_flags & CO_GENERATOR:
|
||||
return self.trace_dispatch
|
||||
try:
|
||||
self.frame_returning = frame
|
||||
self.user_return(frame, arg)
|
||||
finally:
|
||||
self.frame_returning = None
|
||||
if self.quitting: raise BdbQuit
|
||||
# The user issued a 'next' or 'until' command.
|
||||
if self.stopframe is frame and self.stoplineno != -1:
|
||||
self._set_stopinfo(None, None)
|
||||
return self.trace_dispatch
|
||||
|
||||
def dispatch_exception(self, frame, arg):
|
||||
if self.stop_here(frame):
|
||||
# When stepping with next/until/return in a generator frame, skip
|
||||
# the internal StopIteration exception (with no traceback)
|
||||
# triggered by a subiterator run with the 'yield from' statement.
|
||||
if not (frame.f_code.co_flags & CO_GENERATOR
|
||||
and arg[0] is StopIteration and arg[2] is None):
|
||||
self.user_exception(frame, arg)
|
||||
if self.quitting: raise BdbQuit
|
||||
# Stop at the StopIteration or GeneratorExit exception when the user
|
||||
# has set stopframe in a generator by issuing a return command, or a
|
||||
# next/until command at the last statement in the generator before the
|
||||
# exception.
|
||||
elif (self.stopframe and frame is not self.stopframe
|
||||
and self.stopframe.f_code.co_flags & CO_GENERATOR
|
||||
and arg[0] in (StopIteration, GeneratorExit)):
|
||||
self.user_exception(frame, arg)
|
||||
if self.quitting: raise BdbQuit
|
||||
|
||||
return self.trace_dispatch
|
||||
|
||||
# Normally derived classes don't override the following
|
||||
# methods, but they may if they want to redefine the
|
||||
# definition of stopping and breakpoints.
|
||||
|
||||
def is_skipped_module(self, module_name):
|
||||
for pattern in self.skip:
|
||||
if fnmatch.fnmatch(module_name, pattern):
|
||||
return True
|
||||
return False
|
||||
|
||||
def stop_here(self, frame):
|
||||
# (CT) stopframe may now also be None, see dispatch_call.
|
||||
# (CT) the former test for None is therefore removed from here.
|
||||
if self.skip and \
|
||||
self.is_skipped_module(frame.f_globals.get('__name__')):
|
||||
return False
|
||||
if frame is self.stopframe:
|
||||
if self.stoplineno == -1:
|
||||
return False
|
||||
return frame.f_lineno >= self.stoplineno
|
||||
if not self.stopframe:
|
||||
return True
|
||||
return False
|
||||
|
||||
def break_here(self, frame):
|
||||
filename = self.canonic(frame.f_code.co_filename)
|
||||
if filename not in self.breaks:
|
||||
return False
|
||||
lineno = frame.f_lineno
|
||||
if lineno not in self.breaks[filename]:
|
||||
# The line itself has no breakpoint, but maybe the line is the
|
||||
# first line of a function with breakpoint set by function name.
|
||||
lineno = frame.f_code.co_firstlineno
|
||||
if lineno not in self.breaks[filename]:
|
||||
return False
|
||||
|
||||
# flag says ok to delete temp. bp
|
||||
(bp, flag) = effective(filename, lineno, frame)
|
||||
if bp:
|
||||
self.currentbp = bp.number
|
||||
if (flag and bp.temporary):
|
||||
self.do_clear(str(bp.number))
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
def do_clear(self, arg):
|
||||
raise NotImplementedError("subclass of bdb must implement do_clear()")
|
||||
|
||||
def break_anywhere(self, frame):
|
||||
return self.canonic(frame.f_code.co_filename) in self.breaks
|
||||
|
||||
# Derived classes should override the user_* methods
|
||||
# to gain control.
|
||||
|
||||
def user_call(self, frame, argument_list):
|
||||
"""This method is called when there is the remote possibility
|
||||
that we ever need to stop in this function."""
|
||||
pass
|
||||
|
||||
def user_line(self, frame):
|
||||
"""This method is called when we stop or break at this line."""
|
||||
pass
|
||||
|
||||
def user_return(self, frame, return_value):
|
||||
"""This method is called when a return trap is set here."""
|
||||
pass
|
||||
|
||||
def user_exception(self, frame, exc_info):
|
||||
"""This method is called if an exception occurs,
|
||||
but only if we are to stop at or just below this level."""
|
||||
pass
|
||||
|
||||
def _set_stopinfo(self, stopframe, returnframe, stoplineno=0):
|
||||
self.stopframe = stopframe
|
||||
self.returnframe = returnframe
|
||||
self.quitting = False
|
||||
# stoplineno >= 0 means: stop at line >= the stoplineno
|
||||
# stoplineno -1 means: don't stop at all
|
||||
self.stoplineno = stoplineno
|
||||
|
||||
# Derived classes and clients can call the following methods
|
||||
# to affect the stepping state.
|
||||
|
||||
def set_until(self, frame, lineno=None):
|
||||
"""Stop when the line with the line no greater than the current one is
|
||||
reached or when returning from current frame"""
|
||||
# the name "until" is borrowed from gdb
|
||||
if lineno is None:
|
||||
lineno = frame.f_lineno + 1
|
||||
self._set_stopinfo(frame, frame, lineno)
|
||||
|
||||
def set_step(self):
|
||||
"""Stop after one line of code."""
|
||||
# Issue #13183: pdb skips frames after hitting a breakpoint and running
|
||||
# step commands.
|
||||
# Restore the trace function in the caller (that may not have been set
|
||||
# for performance reasons) when returning from the current frame.
|
||||
if self.frame_returning:
|
||||
caller_frame = self.frame_returning.f_back
|
||||
if caller_frame and not caller_frame.f_trace:
|
||||
caller_frame.f_trace = self.trace_dispatch
|
||||
self._set_stopinfo(None, None)
|
||||
|
||||
def set_next(self, frame):
|
||||
"""Stop on the next line in or below the given frame."""
|
||||
self._set_stopinfo(frame, None)
|
||||
|
||||
def set_return(self, frame):
|
||||
"""Stop when returning from the given frame."""
|
||||
if frame.f_code.co_flags & CO_GENERATOR:
|
||||
self._set_stopinfo(frame, None, -1)
|
||||
else:
|
||||
self._set_stopinfo(frame.f_back, frame)
|
||||
|
||||
def set_trace(self, frame=None):
|
||||
"""Start debugging from `frame`.
|
||||
|
||||
If frame is not specified, debugging starts from caller's frame.
|
||||
"""
|
||||
if frame is None:
|
||||
frame = sys._getframe().f_back
|
||||
self.reset()
|
||||
while frame:
|
||||
frame.f_trace = self.trace_dispatch
|
||||
self.botframe = frame
|
||||
frame = frame.f_back
|
||||
self.set_step()
|
||||
sys.settrace(self.trace_dispatch)
|
||||
|
||||
def set_continue(self):
|
||||
# Don't stop except at breakpoints or when finished
|
||||
self._set_stopinfo(self.botframe, None, -1)
|
||||
if not self.breaks:
|
||||
# no breakpoints; run without debugger overhead
|
||||
sys.settrace(None)
|
||||
frame = sys._getframe().f_back
|
||||
while frame and frame is not self.botframe:
|
||||
print(f'Frame: {frame}')
|
||||
del frame.f_trace
|
||||
frame = frame.f_back
|
||||
|
||||
def set_quit(self):
|
||||
self.stopframe = self.botframe
|
||||
self.returnframe = None
|
||||
self.quitting = True
|
||||
sys.settrace(None)
|
||||
|
||||
# Derived classes and clients can call the following methods
|
||||
# to manipulate breakpoints. These methods return an
|
||||
# error message is something went wrong, None if all is well.
|
||||
# Set_break prints out the breakpoint line and file:lineno.
|
||||
# Call self.get_*break*() to see the breakpoints or better
|
||||
# for bp in Breakpoint.bpbynumber: if bp: bp.bpprint().
|
||||
|
||||
def set_break(self,
|
||||
filename,
|
||||
lineno,
|
||||
temporary=False,
|
||||
cond=None,
|
||||
funcname=None):
|
||||
filename = self.canonic(filename)
|
||||
import linecache # Import as late as possible
|
||||
line = linecache.getline(filename, lineno)
|
||||
if not line:
|
||||
return 'Line %s:%d does not exist' % (filename, lineno)
|
||||
list = self.breaks.setdefault(filename, [])
|
||||
if lineno not in list:
|
||||
list.append(lineno)
|
||||
bp = Breakpoint(filename, lineno, temporary, cond, funcname)
|
||||
|
||||
def _prune_breaks(self, filename, lineno):
|
||||
if (filename, lineno) not in Breakpoint.bplist:
|
||||
self.breaks[filename].remove(lineno)
|
||||
if not self.breaks[filename]:
|
||||
del self.breaks[filename]
|
||||
|
||||
def clear_break(self, filename, lineno):
|
||||
filename = self.canonic(filename)
|
||||
if filename not in self.breaks:
|
||||
return 'There are no breakpoints in %s' % filename
|
||||
if lineno not in self.breaks[filename]:
|
||||
return 'There is no breakpoint at %s:%d' % (filename, lineno)
|
||||
# If there's only one bp in the list for that file,line
|
||||
# pair, then remove the breaks entry
|
||||
for bp in Breakpoint.bplist[filename, lineno][:]:
|
||||
bp.deleteMe()
|
||||
self._prune_breaks(filename, lineno)
|
||||
|
||||
def clear_bpbynumber(self, arg):
|
||||
try:
|
||||
bp = self.get_bpbynumber(arg)
|
||||
except ValueError as err:
|
||||
return str(err)
|
||||
bp.deleteMe()
|
||||
self._prune_breaks(bp.file, bp.line)
|
||||
|
||||
def clear_all_file_breaks(self, filename):
|
||||
filename = self.canonic(filename)
|
||||
if filename not in self.breaks:
|
||||
return 'There are no breakpoints in %s' % filename
|
||||
for line in self.breaks[filename]:
|
||||
blist = Breakpoint.bplist[filename, line]
|
||||
for bp in blist:
|
||||
bp.deleteMe()
|
||||
del self.breaks[filename]
|
||||
|
||||
def clear_all_breaks(self):
|
||||
if not self.breaks:
|
||||
return 'There are no breakpoints'
|
||||
for bp in Breakpoint.bpbynumber:
|
||||
if bp:
|
||||
bp.deleteMe()
|
||||
self.breaks = {}
|
||||
|
||||
def get_bpbynumber(self, arg):
|
||||
if not arg:
|
||||
raise ValueError('Breakpoint number expected')
|
||||
try:
|
||||
number = int(arg)
|
||||
except ValueError:
|
||||
raise ValueError('Non-numeric breakpoint number %s' % arg)
|
||||
try:
|
||||
bp = Breakpoint.bpbynumber[number]
|
||||
except IndexError:
|
||||
raise ValueError('Breakpoint number %d out of range' % number)
|
||||
if bp is None:
|
||||
raise ValueError('Breakpoint %d already deleted' % number)
|
||||
return bp
|
||||
|
||||
def get_break(self, filename, lineno):
|
||||
filename = self.canonic(filename)
|
||||
return filename in self.breaks and \
|
||||
lineno in self.breaks[filename]
|
||||
|
||||
def get_breaks(self, filename, lineno):
|
||||
filename = self.canonic(filename)
|
||||
return filename in self.breaks and \
|
||||
lineno in self.breaks[filename] and \
|
||||
Breakpoint.bplist[filename, lineno] or []
|
||||
|
||||
def get_file_breaks(self, filename):
|
||||
filename = self.canonic(filename)
|
||||
if filename in self.breaks:
|
||||
return self.breaks[filename]
|
||||
else:
|
||||
return []
|
||||
|
||||
def get_all_breaks(self):
|
||||
return self.breaks
|
||||
|
||||
# Derived classes and clients can call the following method
|
||||
# to get a data structure representing a stack trace.
|
||||
|
||||
def get_stack(self, f, t):
|
||||
stack = []
|
||||
if t and t.tb_frame is f:
|
||||
t = t.tb_next
|
||||
while f is not None:
|
||||
stack.append((f, f.f_lineno))
|
||||
if f is self.botframe:
|
||||
break
|
||||
f = f.f_back
|
||||
stack.reverse()
|
||||
i = max(0, len(stack) - 1)
|
||||
while t is not None:
|
||||
stack.append((t.tb_frame, t.tb_lineno))
|
||||
t = t.tb_next
|
||||
if f is None:
|
||||
i = max(0, len(stack) - 1)
|
||||
return stack, i
|
||||
|
||||
def format_stack_entry(self, frame_lineno, lprefix=': '):
|
||||
import linecache, reprlib
|
||||
frame, lineno = frame_lineno
|
||||
filename = self.canonic(frame.f_code.co_filename)
|
||||
s = '%s(%r)' % (filename, lineno)
|
||||
if frame.f_code.co_name:
|
||||
s += frame.f_code.co_name
|
||||
else:
|
||||
s += "<lambda>"
|
||||
if '__args__' in frame.f_locals:
|
||||
args = frame.f_locals['__args__']
|
||||
else:
|
||||
args = None
|
||||
if args:
|
||||
s += reprlib.repr(args)
|
||||
else:
|
||||
s += '()'
|
||||
if '__return__' in frame.f_locals:
|
||||
rv = frame.f_locals['__return__']
|
||||
s += '->'
|
||||
s += reprlib.repr(rv)
|
||||
line = linecache.getline(filename, lineno, frame.f_globals)
|
||||
if line:
|
||||
s += lprefix + line.strip()
|
||||
return s
|
||||
|
||||
# The following methods can be called by clients to use
|
||||
# a debugger to debug a statement or an expression.
|
||||
# Both can be given as a string, or a code object.
|
||||
|
||||
def run(self, cmd, globals=None, locals=None):
|
||||
if globals is None:
|
||||
import __main__
|
||||
globals = __main__.__dict__
|
||||
if locals is None:
|
||||
locals = globals
|
||||
self.reset()
|
||||
if isinstance(cmd, str):
|
||||
cmd = compile(cmd, "<string>", "exec")
|
||||
sys.settrace(self.trace_dispatch)
|
||||
try:
|
||||
exec(cmd, globals, locals)
|
||||
except BdbQuit:
|
||||
pass
|
||||
finally:
|
||||
self.quitting = True
|
||||
sys.settrace(None)
|
||||
|
||||
def runeval(self, expr, globals=None, locals=None):
|
||||
if globals is None:
|
||||
import __main__
|
||||
globals = __main__.__dict__
|
||||
if locals is None:
|
||||
locals = globals
|
||||
self.reset()
|
||||
sys.settrace(self.trace_dispatch)
|
||||
try:
|
||||
return eval(expr, globals, locals)
|
||||
except BdbQuit:
|
||||
pass
|
||||
finally:
|
||||
self.quitting = True
|
||||
sys.settrace(None)
|
||||
|
||||
def runctx(self, cmd, globals, locals):
|
||||
# B/W compatibility
|
||||
self.run(cmd, globals, locals)
|
||||
|
||||
# This method is more useful to debug a single function call.
|
||||
|
||||
def runcall(self, func, *args, **kwds):
|
||||
self.reset()
|
||||
sys.settrace(self.trace_dispatch)
|
||||
res = None
|
||||
try:
|
||||
res = func(*args, **kwds)
|
||||
except BdbQuit:
|
||||
pass
|
||||
finally:
|
||||
self.quitting = True
|
||||
sys.settrace(None)
|
||||
return res
|
||||
|
||||
|
||||
def set_trace():
|
||||
Bdb().set_trace()
|
||||
|
||||
|
||||
class Breakpoint:
|
||||
"""Breakpoint class.
|
||||
|
||||
Implements temporary breakpoints, ignore counts, disabling and
|
||||
(re)-enabling, and conditionals.
|
||||
|
||||
Breakpoints are indexed by number through bpbynumber and by
|
||||
the file,line tuple using bplist. The former points to a
|
||||
single instance of class Breakpoint. The latter points to a
|
||||
list of such instances since there may be more than one
|
||||
breakpoint per line.
|
||||
|
||||
"""
|
||||
|
||||
# XXX Keeping state in the class is a mistake -- this means
|
||||
# you cannot have more than one active Bdb instance.
|
||||
|
||||
next = 1 # Next bp to be assigned
|
||||
bplist = {} # indexed by (file, lineno) tuple
|
||||
bpbynumber = [None] # Each entry is None or an instance of Bpt
|
||||
|
||||
# index 0 is unused, except for marking an
|
||||
# effective break .... see effective()
|
||||
|
||||
def __init__(self, file, line, temporary=False, cond=None, funcname=None):
|
||||
self.funcname = funcname
|
||||
# Needed if funcname is not None.
|
||||
self.func_first_executable_line = None
|
||||
self.file = file # This better be in canonical form!
|
||||
self.line = line
|
||||
self.temporary = temporary
|
||||
self.cond = cond
|
||||
self.enabled = True
|
||||
self.ignore = 0
|
||||
self.hits = 0
|
||||
self.number = Breakpoint.next
|
||||
Breakpoint.next += 1
|
||||
# Build the two lists
|
||||
self.bpbynumber.append(self)
|
||||
if (file, line) in self.bplist:
|
||||
self.bplist[file, line].append(self)
|
||||
else:
|
||||
self.bplist[file, line] = [self]
|
||||
|
||||
def deleteMe(self):
|
||||
index = (self.file, self.line)
|
||||
self.bpbynumber[self.number] = None # No longer in list
|
||||
self.bplist[index].remove(self)
|
||||
if not self.bplist[index]:
|
||||
# No more bp for this f:l combo
|
||||
del self.bplist[index]
|
||||
|
||||
def enable(self):
|
||||
self.enabled = True
|
||||
|
||||
def disable(self):
|
||||
self.enabled = False
|
||||
|
||||
def bpprint(self, out=None):
|
||||
if out is None:
|
||||
out = sys.stdout
|
||||
print(self.bpformat(), file=out)
|
||||
|
||||
def bpformat(self):
|
||||
if self.temporary:
|
||||
disp = 'del '
|
||||
else:
|
||||
disp = 'keep '
|
||||
if self.enabled:
|
||||
disp = disp + 'yes '
|
||||
else:
|
||||
disp = disp + 'no '
|
||||
ret = '%-4dbreakpoint %s at %s:%d' % (self.number, disp, self.file,
|
||||
self.line)
|
||||
if self.cond:
|
||||
ret += '\n\tstop only if %s' % (self.cond, )
|
||||
if self.ignore:
|
||||
ret += '\n\tignore next %d hits' % (self.ignore, )
|
||||
if self.hits:
|
||||
if self.hits > 1:
|
||||
ss = 's'
|
||||
else:
|
||||
ss = ''
|
||||
ret += '\n\tbreakpoint already hit %d time%s' % (self.hits, ss)
|
||||
return ret
|
||||
|
||||
def __str__(self):
|
||||
return 'breakpoint %s at %s:%s' % (self.number, self.file, self.line)
|
||||
|
||||
|
||||
# -----------end of Breakpoint class----------
|
||||
|
||||
|
||||
def checkfuncname(b, frame):
|
||||
"""Check whether we should break here because of `b.funcname`."""
|
||||
if not b.funcname:
|
||||
# Breakpoint was set via line number.
|
||||
if b.line != frame.f_lineno:
|
||||
# Breakpoint was set at a line with a def statement and the function
|
||||
# defined is called: don't break.
|
||||
return False
|
||||
return True
|
||||
|
||||
# Breakpoint set via function name.
|
||||
|
||||
if frame.f_code.co_name != b.funcname:
|
||||
# It's not a function call, but rather execution of def statement.
|
||||
return False
|
||||
|
||||
# We are in the right frame.
|
||||
if not b.func_first_executable_line:
|
||||
# The function is entered for the 1st time.
|
||||
b.func_first_executable_line = frame.f_lineno
|
||||
|
||||
if b.func_first_executable_line != frame.f_lineno:
|
||||
# But we are not at the first line number: don't break.
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
# Determines if there is an effective (active) breakpoint at this
|
||||
# line of code. Returns breakpoint number or 0 if none
|
||||
def effective(file, line, frame):
|
||||
"""Determine which breakpoint for this file:line is to be acted upon.
|
||||
|
||||
Called only if we know there is a bpt at this
|
||||
location. Returns breakpoint that was triggered and a flag
|
||||
that indicates if it is ok to delete a temporary bp.
|
||||
|
||||
"""
|
||||
possibles = Breakpoint.bplist[file, line]
|
||||
for b in possibles:
|
||||
if not b.enabled:
|
||||
continue
|
||||
if not checkfuncname(b, frame):
|
||||
continue
|
||||
# Count every hit when bp is enabled
|
||||
b.hits += 1
|
||||
if not b.cond:
|
||||
# If unconditional, and ignoring go on to next, else break
|
||||
if b.ignore > 0:
|
||||
b.ignore -= 1
|
||||
continue
|
||||
else:
|
||||
# breakpoint and marker that it's ok to delete if temporary
|
||||
return (b, True)
|
||||
else:
|
||||
# Conditional bp.
|
||||
# Ignore count applies only to those bpt hits where the
|
||||
# condition evaluates to true.
|
||||
try:
|
||||
val = eval(b.cond, frame.f_globals, frame.f_locals)
|
||||
if val:
|
||||
if b.ignore > 0:
|
||||
b.ignore -= 1
|
||||
# continue
|
||||
else:
|
||||
return (b, True)
|
||||
# else:
|
||||
# continue
|
||||
except:
|
||||
# if eval fails, most conservative thing is to stop on
|
||||
# breakpoint regardless of ignore count. Don't delete
|
||||
# temporary, as another hint to user.
|
||||
return (b, False)
|
||||
return (None, None)
|
||||
|
||||
|
||||
# -------------------- testing --------------------
|
||||
|
||||
|
||||
class Tdb(Bdb):
|
||||
def user_call(self, frame, args):
|
||||
name = frame.f_code.co_name
|
||||
if not name: name = '???'
|
||||
print('+++ call', name, args)
|
||||
|
||||
def user_line(self, frame):
|
||||
import linecache
|
||||
name = frame.f_code.co_name
|
||||
if not name: name = '???'
|
||||
fn = self.canonic(frame.f_code.co_filename)
|
||||
line = linecache.getline(fn, frame.f_lineno, frame.f_globals)
|
||||
print('+++', fn, frame.f_lineno, name, ':', line.strip())
|
||||
|
||||
def user_return(self, frame, retval):
|
||||
print('+++ return', retval)
|
||||
|
||||
def user_exception(self, frame, exc_stuff):
|
||||
print('+++ exception', exc_stuff)
|
||||
self.set_continue()
|
||||
|
||||
|
||||
def foo(n):
|
||||
print('foo(', n, ')')
|
||||
x = bar(n * 10)
|
||||
print('bar returned', x)
|
||||
|
||||
|
||||
def bar(a):
|
||||
print('bar(', a, ')')
|
||||
return a / 2
|
||||
|
||||
|
||||
def test():
|
||||
t = Tdb()
|
||||
t.run('import bdb; bdb.foo(10)')
|
||||
Vendored
+401
@@ -0,0 +1,401 @@
|
||||
"""A generic class to build line-oriented command interpreters.
|
||||
|
||||
Interpreters constructed with this class obey the following conventions:
|
||||
|
||||
1. End of file on input is processed as the command 'EOF'.
|
||||
2. A command is parsed out of each line by collecting the prefix composed
|
||||
of characters in the identchars member.
|
||||
3. A command `foo' is dispatched to a method 'do_foo()'; the do_ method
|
||||
is passed a single argument consisting of the remainder of the line.
|
||||
4. Typing an empty line repeats the last command. (Actually, it calls the
|
||||
method `emptyline', which may be overridden in a subclass.)
|
||||
5. There is a predefined `help' method. Given an argument `topic', it
|
||||
calls the command `help_topic'. With no arguments, it lists all topics
|
||||
with defined help_ functions, broken into up to three topics; documented
|
||||
commands, miscellaneous help topics, and undocumented commands.
|
||||
6. The command '?' is a synonym for `help'. The command '!' is a synonym
|
||||
for `shell', if a do_shell method exists.
|
||||
7. If completion is enabled, completing commands will be done automatically,
|
||||
and completing of commands args is done by calling complete_foo() with
|
||||
arguments text, line, begidx, endidx. text is string we are matching
|
||||
against, all returned matches must begin with it. line is the current
|
||||
input line (lstripped), begidx and endidx are the beginning and end
|
||||
indexes of the text being matched, which could be used to provide
|
||||
different completion depending upon which position the argument is in.
|
||||
|
||||
The `default' method may be overridden to intercept commands for which there
|
||||
is no do_ method.
|
||||
|
||||
The `completedefault' method may be overridden to intercept completions for
|
||||
commands that have no complete_ method.
|
||||
|
||||
The data member `self.ruler' sets the character used to draw separator lines
|
||||
in the help messages. If empty, no ruler line is drawn. It defaults to "=".
|
||||
|
||||
If the value of `self.intro' is nonempty when the cmdloop method is called,
|
||||
it is printed out on interpreter startup. This value may be overridden
|
||||
via an optional argument to the cmdloop() method.
|
||||
|
||||
The data members `self.doc_header', `self.misc_header', and
|
||||
`self.undoc_header' set the headers used for the help function's
|
||||
listings of documented functions, miscellaneous topics, and undocumented
|
||||
functions respectively.
|
||||
"""
|
||||
|
||||
import string, sys
|
||||
|
||||
__all__ = ["Cmd"]
|
||||
|
||||
PROMPT = '(Cmd) '
|
||||
IDENTCHARS = string.ascii_letters + string.digits + '_'
|
||||
|
||||
class Cmd:
|
||||
"""A simple framework for writing line-oriented command interpreters.
|
||||
|
||||
These are often useful for test harnesses, administrative tools, and
|
||||
prototypes that will later be wrapped in a more sophisticated interface.
|
||||
|
||||
A Cmd instance or subclass instance is a line-oriented interpreter
|
||||
framework. There is no good reason to instantiate Cmd itself; rather,
|
||||
it's useful as a superclass of an interpreter class you define yourself
|
||||
in order to inherit Cmd's methods and encapsulate action methods.
|
||||
|
||||
"""
|
||||
prompt = PROMPT
|
||||
identchars = IDENTCHARS
|
||||
ruler = '='
|
||||
lastcmd = ''
|
||||
intro = None
|
||||
doc_leader = ""
|
||||
doc_header = "Documented commands (type help <topic>):"
|
||||
misc_header = "Miscellaneous help topics:"
|
||||
undoc_header = "Undocumented commands:"
|
||||
nohelp = "*** No help on %s"
|
||||
use_rawinput = 1
|
||||
|
||||
def __init__(self, completekey='tab', stdin=None, stdout=None):
|
||||
"""Instantiate a line-oriented interpreter framework.
|
||||
|
||||
The optional argument 'completekey' is the readline name of a
|
||||
completion key; it defaults to the Tab key. If completekey is
|
||||
not None and the readline module is available, command completion
|
||||
is done automatically. The optional arguments stdin and stdout
|
||||
specify alternate input and output file objects; if not specified,
|
||||
sys.stdin and sys.stdout are used.
|
||||
|
||||
"""
|
||||
if stdin is not None:
|
||||
self.stdin = stdin
|
||||
else:
|
||||
self.stdin = sys.stdin
|
||||
if stdout is not None:
|
||||
self.stdout = stdout
|
||||
else:
|
||||
self.stdout = sys.stdout
|
||||
self.cmdqueue = []
|
||||
self.completekey = completekey
|
||||
|
||||
def cmdloop(self, intro=None):
|
||||
"""Repeatedly issue a prompt, accept input, parse an initial prefix
|
||||
off the received input, and dispatch to action methods, passing them
|
||||
the remainder of the line as argument.
|
||||
|
||||
"""
|
||||
|
||||
self.preloop()
|
||||
if self.use_rawinput and self.completekey:
|
||||
try:
|
||||
import readline
|
||||
self.old_completer = readline.get_completer()
|
||||
readline.set_completer(self.complete)
|
||||
readline.parse_and_bind(self.completekey+": complete")
|
||||
except ImportError:
|
||||
pass
|
||||
try:
|
||||
if intro is not None:
|
||||
self.intro = intro
|
||||
if self.intro:
|
||||
self.stdout.write(str(self.intro)+"\n")
|
||||
stop = None
|
||||
while not stop:
|
||||
if self.cmdqueue:
|
||||
line = self.cmdqueue.pop(0)
|
||||
else:
|
||||
if self.use_rawinput:
|
||||
try:
|
||||
line = input(self.prompt)
|
||||
except EOFError:
|
||||
line = 'EOF'
|
||||
else:
|
||||
self.stdout.write(self.prompt)
|
||||
self.stdout.flush()
|
||||
line = self.stdin.readline()
|
||||
if not len(line):
|
||||
line = 'EOF'
|
||||
else:
|
||||
line = line.rstrip('\r\n')
|
||||
line = self.precmd(line)
|
||||
stop = self.onecmd(line)
|
||||
stop = self.postcmd(stop, line)
|
||||
self.postloop()
|
||||
finally:
|
||||
if self.use_rawinput and self.completekey:
|
||||
try:
|
||||
import readline
|
||||
readline.set_completer(self.old_completer)
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
|
||||
def precmd(self, line):
|
||||
"""Hook method executed just before the command line is
|
||||
interpreted, but after the input prompt is generated and issued.
|
||||
|
||||
"""
|
||||
return line
|
||||
|
||||
def postcmd(self, stop, line):
|
||||
"""Hook method executed just after a command dispatch is finished."""
|
||||
return stop
|
||||
|
||||
def preloop(self):
|
||||
"""Hook method executed once when the cmdloop() method is called."""
|
||||
pass
|
||||
|
||||
def postloop(self):
|
||||
"""Hook method executed once when the cmdloop() method is about to
|
||||
return.
|
||||
|
||||
"""
|
||||
pass
|
||||
|
||||
def parseline(self, line):
|
||||
"""Parse the line into a command name and a string containing
|
||||
the arguments. Returns a tuple containing (command, args, line).
|
||||
'command' and 'args' may be None if the line couldn't be parsed.
|
||||
"""
|
||||
line = line.strip()
|
||||
if not line:
|
||||
return None, None, line
|
||||
elif line[0] == '?':
|
||||
line = 'help ' + line[1:]
|
||||
elif line[0] == '!':
|
||||
if hasattr(self, 'do_shell'):
|
||||
line = 'shell ' + line[1:]
|
||||
else:
|
||||
return None, None, line
|
||||
i, n = 0, len(line)
|
||||
while i < n and line[i] in self.identchars: i = i+1
|
||||
cmd, arg = line[:i], line[i:].strip()
|
||||
return cmd, arg, line
|
||||
|
||||
def onecmd(self, line):
|
||||
"""Interpret the argument as though it had been typed in response
|
||||
to the prompt.
|
||||
|
||||
This may be overridden, but should not normally need to be;
|
||||
see the precmd() and postcmd() methods for useful execution hooks.
|
||||
The return value is a flag indicating whether interpretation of
|
||||
commands by the interpreter should stop.
|
||||
|
||||
"""
|
||||
cmd, arg, line = self.parseline(line)
|
||||
if not line:
|
||||
return self.emptyline()
|
||||
if cmd is None:
|
||||
return self.default(line)
|
||||
self.lastcmd = line
|
||||
if line == 'EOF' :
|
||||
self.lastcmd = ''
|
||||
if cmd == '':
|
||||
return self.default(line)
|
||||
else:
|
||||
try:
|
||||
func = getattr(self, 'do_' + cmd)
|
||||
except AttributeError:
|
||||
return self.default(line)
|
||||
return func(arg)
|
||||
|
||||
def emptyline(self):
|
||||
"""Called when an empty line is entered in response to the prompt.
|
||||
|
||||
If this method is not overridden, it repeats the last nonempty
|
||||
command entered.
|
||||
|
||||
"""
|
||||
if self.lastcmd:
|
||||
return self.onecmd(self.lastcmd)
|
||||
|
||||
def default(self, line):
|
||||
"""Called on an input line when the command prefix is not recognized.
|
||||
|
||||
If this method is not overridden, it prints an error message and
|
||||
returns.
|
||||
|
||||
"""
|
||||
self.stdout.write('*** Unknown syntax: %s\n'%line)
|
||||
|
||||
def completedefault(self, *ignored):
|
||||
"""Method called to complete an input line when no command-specific
|
||||
complete_*() method is available.
|
||||
|
||||
By default, it returns an empty list.
|
||||
|
||||
"""
|
||||
return []
|
||||
|
||||
def completenames(self, text, *ignored):
|
||||
dotext = 'do_'+text
|
||||
return [a[3:] for a in self.get_names() if a.startswith(dotext)]
|
||||
|
||||
def complete(self, text, state):
|
||||
"""Return the next possible completion for 'text'.
|
||||
|
||||
If a command has not been entered, then complete against command list.
|
||||
Otherwise try to call complete_<command> to get list of completions.
|
||||
"""
|
||||
if state == 0:
|
||||
import readline
|
||||
origline = readline.get_line_buffer()
|
||||
line = origline.lstrip()
|
||||
stripped = len(origline) - len(line)
|
||||
begidx = readline.get_begidx() - stripped
|
||||
endidx = readline.get_endidx() - stripped
|
||||
if begidx>0:
|
||||
cmd, args, foo = self.parseline(line)
|
||||
if cmd == '':
|
||||
compfunc = self.completedefault
|
||||
else:
|
||||
try:
|
||||
compfunc = getattr(self, 'complete_' + cmd)
|
||||
except AttributeError:
|
||||
compfunc = self.completedefault
|
||||
else:
|
||||
compfunc = self.completenames
|
||||
self.completion_matches = compfunc(text, line, begidx, endidx)
|
||||
try:
|
||||
return self.completion_matches[state]
|
||||
except IndexError:
|
||||
return None
|
||||
|
||||
def get_names(self):
|
||||
# This method used to pull in base class attributes
|
||||
# at a time dir() didn't do it yet.
|
||||
return dir(self.__class__)
|
||||
|
||||
def complete_help(self, *args):
|
||||
commands = set(self.completenames(*args))
|
||||
topics = set(a[5:] for a in self.get_names()
|
||||
if a.startswith('help_' + args[0]))
|
||||
return list(commands | topics)
|
||||
|
||||
def do_help(self, arg):
|
||||
'List available commands with "help" or detailed help with "help cmd".'
|
||||
if arg:
|
||||
# XXX check arg syntax
|
||||
try:
|
||||
func = getattr(self, 'help_' + arg)
|
||||
except AttributeError:
|
||||
try:
|
||||
doc=getattr(self, 'do_' + arg).__doc__
|
||||
if doc:
|
||||
self.stdout.write("%s\n"%str(doc))
|
||||
return
|
||||
except AttributeError:
|
||||
pass
|
||||
self.stdout.write("%s\n"%str(self.nohelp % (arg,)))
|
||||
return
|
||||
func()
|
||||
else:
|
||||
names = self.get_names()
|
||||
cmds_doc = []
|
||||
cmds_undoc = []
|
||||
help = {}
|
||||
for name in names:
|
||||
if name[:5] == 'help_':
|
||||
help[name[5:]]=1
|
||||
names.sort()
|
||||
# There can be duplicates if routines overridden
|
||||
prevname = ''
|
||||
for name in names:
|
||||
if name[:3] == 'do_':
|
||||
if name == prevname:
|
||||
continue
|
||||
prevname = name
|
||||
cmd=name[3:]
|
||||
if cmd in help:
|
||||
cmds_doc.append(cmd)
|
||||
del help[cmd]
|
||||
elif getattr(self, name).__doc__:
|
||||
cmds_doc.append(cmd)
|
||||
else:
|
||||
cmds_undoc.append(cmd)
|
||||
self.stdout.write("%s\n"%str(self.doc_leader))
|
||||
self.print_topics(self.doc_header, cmds_doc, 15,80)
|
||||
self.print_topics(self.misc_header, list(help.keys()),15,80)
|
||||
self.print_topics(self.undoc_header, cmds_undoc, 15,80)
|
||||
|
||||
def print_topics(self, header, cmds, cmdlen, maxcol):
|
||||
if cmds:
|
||||
self.stdout.write("%s\n"%str(header))
|
||||
if self.ruler:
|
||||
self.stdout.write("%s\n"%str(self.ruler * len(header)))
|
||||
self.columnize(cmds, maxcol-1)
|
||||
self.stdout.write("\n")
|
||||
|
||||
def columnize(self, list, displaywidth=80):
|
||||
"""Display a list of strings as a compact set of columns.
|
||||
|
||||
Each column is only as wide as necessary.
|
||||
Columns are separated by two spaces (one was not legible enough).
|
||||
"""
|
||||
if not list:
|
||||
self.stdout.write("<empty>\n")
|
||||
return
|
||||
|
||||
nonstrings = [i for i in range(len(list))
|
||||
if not isinstance(list[i], str)]
|
||||
if nonstrings:
|
||||
raise TypeError("list[i] not a string for i in %s"
|
||||
% ", ".join(map(str, nonstrings)))
|
||||
size = len(list)
|
||||
if size == 1:
|
||||
self.stdout.write('%s\n'%str(list[0]))
|
||||
return
|
||||
# Try every row count from 1 upwards
|
||||
for nrows in range(1, len(list)):
|
||||
ncols = (size+nrows-1) // nrows
|
||||
colwidths = []
|
||||
totwidth = -2
|
||||
for col in range(ncols):
|
||||
colwidth = 0
|
||||
for row in range(nrows):
|
||||
i = row + nrows*col
|
||||
if i >= size:
|
||||
break
|
||||
x = list[i]
|
||||
colwidth = max(colwidth, len(x))
|
||||
colwidths.append(colwidth)
|
||||
totwidth += colwidth + 2
|
||||
if totwidth > displaywidth:
|
||||
break
|
||||
if totwidth <= displaywidth:
|
||||
break
|
||||
else:
|
||||
nrows = len(list)
|
||||
ncols = 1
|
||||
colwidths = [0]
|
||||
for row in range(nrows):
|
||||
texts = []
|
||||
for col in range(ncols):
|
||||
i = row + nrows*col
|
||||
if i >= size:
|
||||
x = ""
|
||||
else:
|
||||
x = list[i]
|
||||
texts.append(x)
|
||||
while texts and not texts[-1]:
|
||||
del texts[-1]
|
||||
for col in range(len(texts)):
|
||||
texts[col] = texts[col].ljust(colwidths[col])
|
||||
self.stdout.write("%s\n"%str(" ".join(texts)))
|
||||
@@ -0,0 +1,178 @@
|
||||
#!/usr/bin/env python # [built-in module gc]
|
||||
"""
|
||||
This module provides access to the garbage collector for reference cycles.
|
||||
|
||||
enable() -- Enable automatic garbage collection.
|
||||
disable() -- Disable automatic garbage collection.
|
||||
isenabled() -- Returns true if automatic collection is enabled.
|
||||
collect() -- Do a full collection right now.
|
||||
get_count() -- Return the current collection counts.
|
||||
get_stats() -- Return list of dictionaries containing per-generation stats.
|
||||
set_debug() -- Set debugging flags.
|
||||
get_debug() -- Get debugging flags.
|
||||
set_threshold() -- Set the collection thresholds.
|
||||
get_threshold() -- Return the current the collection thresholds.
|
||||
get_objects() -- Return a list of all objects tracked by the collector.
|
||||
is_tracked() -- Returns true if a given object is tracked.
|
||||
get_referrers() -- Return the list of objects that refer to an object.
|
||||
get_referents() -- Return the list of objects that an object refers to.
|
||||
"""
|
||||
|
||||
## DATA ##
|
||||
|
||||
DEBUG_COLLECTABLE = 2
|
||||
# None
|
||||
DEBUG_LEAK = 38
|
||||
# None
|
||||
DEBUG_SAVEALL = 32
|
||||
# None
|
||||
DEBUG_STATS = 1
|
||||
# None
|
||||
DEBUG_UNCOLLECTABLE = 4
|
||||
# None
|
||||
callbacks = []
|
||||
# None
|
||||
garbage = []
|
||||
# None
|
||||
|
||||
## FUNCTIONS ##
|
||||
|
||||
|
||||
def collect(*args, **kwargs): # unknown args #
|
||||
"""
|
||||
collect([generation]) -> n
|
||||
|
||||
With no arguments, run a full collection. The optional argument
|
||||
may be an integer specifying which generation to collect. A ValueError
|
||||
is raised if the generation number is invalid.
|
||||
|
||||
The number of unreachable objects is returned.
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
def disable(*args, **kwargs): # unknown args #
|
||||
"""
|
||||
disable() -> None
|
||||
|
||||
Disable automatic garbage collection.
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
def enable(*args, **kwargs): # unknown args #
|
||||
"""
|
||||
enable() -> None
|
||||
|
||||
Enable automatic garbage collection.
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
def get_count(*args, **kwargs): # unknown args #
|
||||
"""
|
||||
get_count() -> (count0, count1, count2)
|
||||
|
||||
Return the current collection counts
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
def get_debug(*args, **kwargs): # unknown args #
|
||||
"""
|
||||
get_debug() -> flags
|
||||
|
||||
Get the garbage collection debugging flags.
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
def get_objects(*args, **kwargs): # unknown args #
|
||||
"""
|
||||
get_objects() -> [...]
|
||||
|
||||
Return a list of objects tracked by the collector (excluding the list
|
||||
returned).
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
def get_referents(*args, **kwargs): # unknown args #
|
||||
"""
|
||||
get_referents(*objs) -> list
|
||||
Return the list of objects that are directly referred to by objs.
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
def get_referrers(*args, **kwargs): # unknown args #
|
||||
"""
|
||||
get_referrers(*objs) -> list
|
||||
Return the list of objects that directly refer to any of objs.
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
def get_stats(*args, **kwargs): # unknown args #
|
||||
"""
|
||||
get_stats() -> [...]
|
||||
|
||||
Return a list of dictionaries containing per-generation statistics.
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
def get_threshold(*args, **kwargs): # unknown args #
|
||||
"""
|
||||
get_threshold() -> (threshold0, threshold1, threshold2)
|
||||
|
||||
Return the current collection thresholds
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
def is_tracked(*args, **kwargs): # unknown args #
|
||||
"""
|
||||
is_tracked(obj) -> bool
|
||||
|
||||
Returns true if the object is tracked by the garbage collector.
|
||||
Simple atomic objects will return false.
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
def isenabled(*args, **kwargs): # unknown args #
|
||||
"""
|
||||
isenabled() -> status
|
||||
|
||||
Returns true if automatic garbage collection is enabled.
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
def set_debug(*args, **kwargs): # unknown args #
|
||||
"""
|
||||
set_debug(flags) -> None
|
||||
|
||||
Set the garbage collection debugging flags. Debugging information is
|
||||
written to sys.stderr.
|
||||
|
||||
flags is an integer and can have the following bits turned on:
|
||||
|
||||
DEBUG_STATS - Print statistics during collection.
|
||||
DEBUG_COLLECTABLE - Print collectable objects found.
|
||||
DEBUG_UNCOLLECTABLE - Print unreachable but uncollectable objects found.
|
||||
DEBUG_SAVEALL - Save objects to gc.garbage rather than freeing them.
|
||||
DEBUG_LEAK - Debug leaking programs (everything but STATS).
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
def set_threshold(*args, **kwargs): # unknown args #
|
||||
"""
|
||||
set_threshold(threshold0, [threshold1, threshold2]) -> None
|
||||
|
||||
Sets the collection thresholds. Setting threshold0 to zero disables
|
||||
collection.
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
+1692
File diff suppressed because it is too large.
Load diff
Reference in new issue
Block a user