"""Generic Python debugger base class.

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

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,

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

# 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 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:

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)

"""Breakpoint class.

# 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):

"""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

"""Determine which breakpoint for this file:line is to be acted upon.

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)

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)