path: root/contrib/python/coverage/py3/coverage/debug.py

# Licensed under the Apache License: http://www.apache.org/licenses/LICENSE-2.0
# 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 pprint
import sys
try:
    import _thread
except ImportError:
    import thread as _thread

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


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.suppress_callers = False

        filters = []
        if self.should('pid'):
            filters.append(add_pid_and_tid)
        self.output = DebugOutputFile.get_one(
            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):
        """Decide whether to output debug information in category `option`."""
        if option == "callers" and self.suppress_callers:
            return False
        return (option in self.options)

    @contextlib.contextmanager
    def without_callers(self):
        """A context manager to prevent call stacks from being logged."""
        old = self.suppress_callers
        self.suppress_callers = True
        try:
            yield
        finally:
            self.suppress_callers = old

    def write(self, msg):
        """Write a line of debug output.

        `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()


class DebugControlString(DebugControl):
    """A `DebugControl` that writes to a StringIO, for testing."""
    def __init__(self, options):
        super(DebugControlString, self).__init__(options, StringIO())

    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 "--{:-<60s}".format(" "+label+" ")


def info_formatter(info):
    """Produce a sequence of formatted lines from info.

    `info` is a sequence of pairs (label, data).  The produced lines are
    nicely formatted, ready to print.

    """
    info = list(info)
    if not info:
        return
    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)
            for e in data:
                yield "%*s %s" % (label_len+1, prefix, e)
                prefix = ""
        else:
            yield "%*s: %s" % (label_len, label, data)


def write_formatted_info(writer, header, info):
    """Write a sequence of (label,data) pairs nicely."""
    writer.write(info_header(header))
    for line in info_formatter(info):
        writer.write(" %s" % line)


def short_stack(limit=None, skip=0):
    """Return a string summarizing the call stack.

    The string is multi-line, with one line per stack frame. Each line shows
    the function name, the file name, and the line number:

        ...
        start_import_stop : /Users/ned/coverage/trunk/tests/coveragetest.py @95
        import_local_file : /Users/ned/coverage/trunk/tests/coveragetest.py @81
        import_local_file : /Users/ned/coverage/trunk/coverage/backward.py @159
        ...

    `limit` is the number of frames to include, defaulting to all of them.

    `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)


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
    for offset in range(0, 64, 16):
        id16 ^= id64 >> offset
    return id16 & 0xFFFF


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 = "{:04x}".format(short_id(_thread.get_ident()))
    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.

    `filters` is a list of functions. Each takes a string and returns a
    string.  Each is run in turn.

    Returns: the final string that results after all of the filters have
    run.

    """
    clean_text = text.rstrip()
    ending = text[len(clean_text):]
    text = clean_text
    for fn in filters:
        lines = []
        for line in text.splitlines():
            lines.extend(fn(line).splitlines())
        text = "\n".join(lines)
    return text + ending


class CwdTracker(object):                                   # pragma: debugging
    """A class to add cwd info to debug messages."""
    def __init__(self):
        self.cwd = None

    def filter(self, text):
        """Add a cwd message for each new cwd."""
        cwd = os.getcwd()
        if cwd != self.cwd:
            text = "cwd is now {!r}\n".format(cwd) + text
            self.cwd = cwd
        return text


class DebugOutputFile(object):                              # pragma: debugging
    """A file-like object that includes pid and cwd information."""
    def __init__(self, outfile, show_process, filters):
        self.outfile = outfile
        self.show_process = show_process
        self.filters = list(filters)

        if self.show_process:
            self.filters.insert(0, CwdTracker().filter)
            self.write("New process: executable: %r\n" % (sys.executable,))
            self.write("New process: cmd: %r\n" % (getattr(sys, 'argv', None),))
            if hasattr(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 get_one(cls, fileobj=None, show_process=True, filters=(), interim=False):
        """Get a DebugOutputFile.

        If `fileobj` is provided, then a new DebugOutputFile is made with it.

        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, is_interim = sys.modules.get(cls.SYS_MOD_NAME, (None, True))
        if the_one is None or is_interim:
            if fileobj is None:
                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 flush(self):
        """Flush our file."""
        self.outfile.flush()


def log(msg, stack=False):                                  # pragma: debugging
    """Write a log message as forcefully as possible."""
    out = DebugOutputFile.get_one(interim=True)
    out.write(msg+"\n")
    if stack:
        dump_stack_frames(out=out, skip=1)


def decorate_methods(decorator, butnot=(), private=False):  # pragma: debugging
    """A class decorator to apply a decorator to methods."""
    def _decorator(cls):
        for name, meth in inspect.getmembers(cls, inspect.isroutine):
            if name not in cls.__dict__:
                continue
            if name != "__init__":
                if not private and name.startswith("_"):
                    continue
            if name in butnot:
                continue
            setattr(cls, name, decorator(meth))
        return cls
    return _decorator


def break_in_pudb(func):                                    # pragma: debugging
    """A function decorator to stop in the debugger for each call."""
    @functools.wraps(func)
    def _wrapper(*args, **kwargs):
        import pudb
        sys.stdout = sys.__stdout__
        pudb.set_trace()
        return func(*args, **kwargs)
    return _wrapper


OBJ_IDS = itertools.count()
CALLS = itertools.count()
OBJ_ID_ATTR = "$coverage.object_id"

def show_calls(show_args=True, show_stack=False, show_return=False):    # pragma: debugging
    """A method decorator to debug-log each call to the function."""
    def _decorator(func):
        @functools.wraps(func)
        def _wrapper(self, *args, **kwargs):
            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