eric: comparison eric6/DebugClients/Python/coverage/debug.py

-:dc171b1d8261
+:362cd1b6f81a
 # Licensed under the Apache License: http://www.apache.org/licenses/LICENSE-2.0
-# For details: https://bitbucket.org/ned/coveragepy/src/default/NOTICE.txt
+# For details: https://github.com/nedbat/coveragepy/blob/master/NOTICE.txt
 """Control of and utilities for debugging."""
 import contextlib
+import functools
 import inspect
+import itertools
 import os
-import re
+import pprint
 import sys
 try:
 import _thread
 except ImportError:
 import thread as _thread
-from coverage.backward import StringIO
+from coverage.backward import reprlib, StringIO
 from coverage.misc import isolate_module
 os = isolate_module(os)
 # When debugging, it can be helpful to force some options, especially when
 # debugging the configuration mechanisms you usually use to control debugging!
 # This is a list of forced debugging options.
 FORCED_DEBUG = []
+FORCED_DEBUG_FILE = None
-# A hack for debugging testing in sub-processes.
-_TEST_NAME_FILE = ""    # "/tmp/covtest.txt"
 class DebugControl(object):
 """Control and output for debugging."""
+show_repr_attr = False      # For SimpleReprMixin
 def __init__(self, options, output):
 """Configure the options and output file for debugging."""
 self.options = list(options) + FORCED_DEBUG
-self.raw_output = output
 self.suppress_callers = False
 filters = []
 if self.should('pid'):
 filters.append(add_pid_and_tid)
-self.output = DebugOutputFile(
+self.output = DebugOutputFile.get_one(
-self.raw_output,
+output,
 show_process=self.should('process'),
 filters=filters,
 )
+self.raw_output = self.output.outfile
 def __repr__(self):
 return "<DebugControl options=%r raw_output=%r>" % (self.options, self.raw_output)
 def should(self, option):
 `msg` is the line to write. A newline will be appended.
 """
 self.output.write(msg+"\n")
+if self.should('self'):
+caller_self = inspect.stack()[1][0].f_locals.get('self')
+if caller_self is not None:
+self.output.write("self: {!r}\n".format(caller_self))
 if self.should('callers'):
 dump_stack_frames(out=self.output, skip=1)
 self.output.flush()
 def get_output(self):
 """Get the output text from the `DebugControl`."""
 return self.raw_output.getvalue()
+class NoDebugging(object):
+"""A replacement for DebugControl that will never try to do anything."""
+def should(self, option):               # pylint: disable=unused-argument
+"""Should we write debug messages?  Never."""
+return False
 def info_header(label):
 """Make a nice header string."""
-return "--{0:-<60s}".format(" "+label+" ")
+return "--{:-<60s}".format(" "+label+" ")
 def info_formatter(info):
 """Produce a sequence of formatted lines from info.
 """
 info = list(info)
 if not info:
 return
-label_len = max(len(l) for l, _d in info)
+label_len = 30
+assert all(len(l) < label_len for l, _ in info)
 for label, data in info:
 if data == []:
 data = "-none-"
 if isinstance(data, (list, set, tuple)):
 prefix = "%*s:" % (label_len, label)
 `skip` is the number of frames to skip, so that debugging functions can
 call this and not be included in the result.
 """
 stack = inspect.stack()[limit:skip:-1]
-return "\n".join("%30s : %s @%d" % (t[3], t[1], t[2]) for t in stack)
+return "\n".join("%30s : %s:%d" % (t[3], t[1], t[2]) for t in stack)
 def dump_stack_frames(limit=None, out=None, skip=0):
 """Print a summary of the stack to stdout, or someplace else."""
 out = out or sys.stdout
 out.write(short_stack(limit=limit, skip=skip+1))
 out.write("\n")
+def clipped_repr(text, numchars=50):
+"""`repr(text)`, but limited to `numchars`."""
+r = reprlib.Repr()
+r.maxstring = numchars
+return r.repr(text)
 def short_id(id64):
 """Given a 64-bit id, make a shorter 16-bit one."""
 id16 = 0
 def add_pid_and_tid(text):
 """A filter to add pid and tid to debug messages."""
 # Thread ids are useful, but too long. Make a shorter one.
-tid = "{0:04x}".format(short_id(_thread.get_ident()))
+tid = "{:04x}".format(short_id(_thread.get_ident()))
-text = "{0:5d}.{1}: {2}".format(os.getpid(), tid, text)
+text = "{:5d}.{}: {}".format(os.getpid(), tid, text)
 return text
+class SimpleReprMixin(object):
+"""A mixin implementing a simple __repr__."""
+simple_repr_ignore = ['simple_repr_ignore', '$coverage.object_id']
+def __repr__(self):
+show_attrs = (
+(k, v) for k, v in self.__dict__.items()
+if getattr(v, "show_repr_attr", True)
+and not callable(v)
+and k not in self.simple_repr_ignore
+)
+return "<{klass} @0x{id:x} {attrs}>".format(
+klass=self.__class__.__name__,
+id=id(self),
+attrs=" ".join("{}={!r}".format(k, v) for k, v in show_attrs),
+)
+def simplify(v):                                            # pragma: debugging
+"""Turn things which are nearly dict/list/etc into dict/list/etc."""
+if isinstance(v, dict):
+return {k:simplify(vv) for k, vv in v.items()}
+elif isinstance(v, (list, tuple)):
+return type(v)(simplify(vv) for vv in v)
+elif hasattr(v, "__dict__"):
+return simplify({'.'+k: v for k, v in v.__dict__.items()})
+else:
+return v
+def pp(v):                                                  # pragma: debugging
+"""Debug helper to pretty-print data, including SimpleNamespace objects."""
+# Might not be needed in 3.9+
+pprint.pprint(simplify(v))
 def filter_text(text, filters):
 """Run `text` through a series of filters.
 def filter(self, text):
 """Add a cwd message for each new cwd."""
 cwd = os.getcwd()
 if cwd != self.cwd:
-text = "cwd is now {0!r}\n".format(cwd) + text
+text = "cwd is now {!r}\n".format(cwd) + text
 self.cwd = cwd
 return text
 class DebugOutputFile(object):                              # pragma: debugging
 self.outfile = outfile
 self.show_process = show_process
 self.filters = list(filters)
 if self.show_process:
-self.filters.append(CwdTracker().filter)
+self.filters.insert(0, CwdTracker().filter)
-cmd = " ".join(getattr(sys, 'argv', ['???']))
+self.write("New process: executable: %r\n" % (sys.executable,))
-self.write("New process: executable: %s\n" % (sys.executable,))
+self.write("New process: cmd: %r\n" % (getattr(sys, 'argv', None),))
-self.write("New process: cmd: %s\n" % (cmd,))
 if hasattr(os, 'getppid'):
-self.write("New process: parent pid: %s\n" % (os.getppid(),))
+self.write("New process: pid: %r, parent pid: %r\n" % (os.getpid(), os.getppid()))
 SYS_MOD_NAME = '$coverage.debug.DebugOutputFile.the_one'
 @classmethod
-def the_one(cls, fileobj=None, show_process=True, filters=()):
+def get_one(cls, fileobj=None, show_process=True, filters=(), interim=False):
-"""Get the process-wide singleton DebugOutputFile.
+"""Get a DebugOutputFile.
-If it doesn't exist yet, then create it as a wrapper around the file
+If `fileobj` is provided, then a new DebugOutputFile is made with it.
-object `fileobj`. `show_process` controls whether the debug file adds
-process-level information.
+If `fileobj` isn't provided, then a file is chosen
+(COVERAGE_DEBUG_FILE, or stderr), and a process-wide singleton
+DebugOutputFile is made.
+`show_process` controls whether the debug file adds process-level
+information, and filters is a list of other message filters to apply.
+`filters` are the text filters to apply to the stream to annotate with
+pids, etc.
+If `interim` is true, then a future `get_one` can replace this one.
 """
+if fileobj is not None:
+# Make DebugOutputFile around the fileobj passed.
+return cls(fileobj, show_process, filters)
 # Because of the way igor.py deletes and re-imports modules,
 # this class can be defined more than once. But we really want
 # a process-wide singleton. So stash it in sys.modules instead of
 # on a class attribute. Yes, this is aggressively gross.
-the_one = sys.modules.get(cls.SYS_MOD_NAME)
+the_one, is_interim = sys.modules.get(cls.SYS_MOD_NAME, (None, True))
-if the_one is None:
+if the_one is None or is_interim:
-assert fileobj is not None
+if fileobj is None:
-sys.modules[cls.SYS_MOD_NAME] = the_one = cls(fileobj, show_process, filters)
+debug_file_name = os.environ.get("COVERAGE_DEBUG_FILE", FORCED_DEBUG_FILE)
+if debug_file_name:
+fileobj = open(debug_file_name, "a")
+else:
+fileobj = sys.stderr
+the_one = cls(fileobj, show_process, filters)
+sys.modules[cls.SYS_MOD_NAME] = (the_one, interim)
 return the_one
 def write(self, text):
 """Just like file.write, but filter through all our filters."""
 self.outfile.write(filter_text(text, self.filters))
 self.outfile.flush()
 def log(msg, stack=False):                                  # pragma: debugging
 """Write a log message as forcefully as possible."""
-out = DebugOutputFile.the_one()
+out = DebugOutputFile.get_one(interim=True)
 out.write(msg+"\n")
 if stack:
 dump_stack_frames(out=out, skip=1)
-def filter_aspectlib_frames(text):                          # pragma: debugging
+def decorate_methods(decorator, butnot=(), private=False):  # pragma: debugging
-"""Aspectlib prints stack traces, but includes its own frames.  Scrub those out."""
+"""A class decorator to apply a decorator to methods."""
-# <<< aspectlib/__init__.py:257:function_wrapper < igor.py:143:run_tests < ...
+def _decorator(cls):
-text = re.sub(r"(?<= )aspectlib/[^.]+\.py:\d+:\w+ < ", "", text)
+for name, meth in inspect.getmembers(cls, inspect.isroutine):
-return text
+if name not in cls.__dict__:
+continue
+if name != "__init__":
-def enable_aspectlib_maybe():                               # pragma: debugging
+if not private and name.startswith("_"):
-"""For debugging, we can use aspectlib to trace execution.
+continue
+if name in butnot:
-Define COVERAGE_ASPECTLIB to enable and configure aspectlib to trace
+continue
-execution::
+setattr(cls, name, decorator(meth))
+return cls
-$ export COVERAGE_LOG=covaspect.txt
+return _decorator
-$ export COVERAGE_ASPECTLIB=coverage.Coverage:coverage.data.CoverageData
-$ coverage run blah.py ...
+def break_in_pudb(func):                                    # pragma: debugging
-This will trace all the public methods on Coverage and CoverageData,
+"""A function decorator to stop in the debugger for each call."""
-writing the information to covaspect.txt.
+@functools.wraps(func)
+def _wrapper(*args, **kwargs):
-"""
+import pudb
-aspects = os.environ.get("COVERAGE_ASPECTLIB", "")
+sys.stdout = sys.__stdout__
-if not aspects:
+pudb.set_trace()
-return
+return func(*args, **kwargs)
+return _wrapper
-import aspectlib                            # pylint: disable=import-error
-import aspectlib.debug                      # pylint: disable=import-error
+OBJ_IDS = itertools.count()
-filename = os.environ.get("COVERAGE_LOG", "/tmp/covlog.txt")
+CALLS = itertools.count()
-filters = [add_pid_and_tid, filter_aspectlib_frames]
+OBJ_ID_ATTR = "$coverage.object_id"
-aspects_file = DebugOutputFile.the_one(open(filename, "a"), show_process=True, filters=filters)
-aspect_log = aspectlib.debug.log(
+def show_calls(show_args=True, show_stack=False, show_return=False):    # pragma: debugging
-print_to=aspects_file, attributes=['id'], stacktrace=30, use_logging=False
+"""A method decorator to debug-log each call to the function."""
-)
+def _decorator(func):
-public_methods = re.compile(r'^(__init__|[a-zA-Z].*)$')
+@functools.wraps(func)
-for aspect in aspects.split(':'):
+def _wrapper(self, *args, **kwargs):
-aspectlib.weave(aspect, aspect_log, methods=public_methods)
+oid = getattr(self, OBJ_ID_ATTR, None)
+if oid is None:
+oid = "{:08d} {:04d}".format(os.getpid(), next(OBJ_IDS))
+setattr(self, OBJ_ID_ATTR, oid)
+extra = ""
+if show_args:
+eargs = ", ".join(map(repr, args))
+ekwargs = ", ".join("{}={!r}".format(*item) for item in kwargs.items())
+extra += "("
+extra += eargs
+if eargs and ekwargs:
+extra += ", "
+extra += ekwargs
+extra += ")"
+if show_stack:
+extra += " @ "
+extra += "; ".join(_clean_stack_line(l) for l in short_stack().splitlines())
+callid = next(CALLS)
+msg = "{} {:04d} {}{}\n".format(oid, callid, func.__name__, extra)
+DebugOutputFile.get_one(interim=True).write(msg)
+ret = func(self, *args, **kwargs)
+if show_return:
+msg = "{} {:04d} {} return {!r}\n".format(oid, callid, func.__name__, ret)
+DebugOutputFile.get_one(interim=True).write(msg)
+return ret
+return _wrapper
+return _decorator
+def _clean_stack_line(s):                                   # pragma: debugging
+"""Simplify some paths in a stack trace, for compactness."""
+s = s.strip()
+s = s.replace(os.path.dirname(__file__) + '/', '')
+s = s.replace(os.path.dirname(os.__file__) + '/', '')
+s = s.replace(sys.prefix + '/', '')
+return s

Mercurial Repositories > eric / file comparison

comparison: eric6/DebugClients/Python/coverage/debug.py

eric6/DebugClients/Python/coverage/debug.py