1# Licensed under the Apache License: http://www.apache.org/licenses/LICENSE-2.0
2# For details: https://github.com/nedbat/coveragepy/blob/master/NOTICE.txt
3
4"""Control of and utilities for debugging."""
5
6import contextlib
7import functools
8import inspect
9import itertools
10import os
11import pprint
12import sys
13try:
14    import _thread
15except ImportError:
16    import thread as _thread
17
18from coverage.backward import reprlib, StringIO
19from coverage.misc import isolate_module
20
21os = isolate_module(os)
22
23
24# When debugging, it can be helpful to force some options, especially when
25# debugging the configuration mechanisms you usually use to control debugging!
26# This is a list of forced debugging options.
27FORCED_DEBUG = []
28FORCED_DEBUG_FILE = None
29
30
31class DebugControl(object):
32    """Control and output for debugging."""
33
34    show_repr_attr = False      # For SimpleReprMixin
35
36    def __init__(self, options, output):
37        """Configure the options and output file for debugging."""
38        self.options = list(options) + FORCED_DEBUG
39        self.suppress_callers = False
40
41        filters = []
42        if self.should('pid'):
43            filters.append(add_pid_and_tid)
44        self.output = DebugOutputFile.get_one(
45            output,
46            show_process=self.should('process'),
47            filters=filters,
48        )
49        self.raw_output = self.output.outfile
50
51    def __repr__(self):
52        return "<DebugControl options=%r raw_output=%r>" % (self.options, self.raw_output)
53
54    def should(self, option):
55        """Decide whether to output debug information in category `option`."""
56        if option == "callers" and self.suppress_callers:
57            return False
58        return (option in self.options)
59
60    @contextlib.contextmanager
61    def without_callers(self):
62        """A context manager to prevent call stacks from being logged."""
63        old = self.suppress_callers
64        self.suppress_callers = True
65        try:
66            yield
67        finally:
68            self.suppress_callers = old
69
70    def write(self, msg):
71        """Write a line of debug output.
72
73        `msg` is the line to write. A newline will be appended.
74
75        """
76        self.output.write(msg+"\n")
77        if self.should('self'):
78            caller_self = inspect.stack()[1][0].f_locals.get('self')
79            if caller_self is not None:
80                self.output.write("self: {!r}\n".format(caller_self))
81        if self.should('callers'):
82            dump_stack_frames(out=self.output, skip=1)
83        self.output.flush()
84
85
86class DebugControlString(DebugControl):
87    """A `DebugControl` that writes to a StringIO, for testing."""
88    def __init__(self, options):
89        super(DebugControlString, self).__init__(options, StringIO())
90
91    def get_output(self):
92        """Get the output text from the `DebugControl`."""
93        return self.raw_output.getvalue()
94
95
96class NoDebugging(object):
97    """A replacement for DebugControl that will never try to do anything."""
98    def should(self, option):               # pylint: disable=unused-argument
99        """Should we write debug messages?  Never."""
100        return False
101
102
103def info_header(label):
104    """Make a nice header string."""
105    return "--{:-<60s}".format(" "+label+" ")
106
107
108def info_formatter(info):
109    """Produce a sequence of formatted lines from info.
110
111    `info` is a sequence of pairs (label, data).  The produced lines are
112    nicely formatted, ready to print.
113
114    """
115    info = list(info)
116    if not info:
117        return
118    label_len = 30
119    assert all(len(l) < label_len for l, _ in info)
120    for label, data in info:
121        if data == []:
122            data = "-none-"
123        if isinstance(data, (list, set, tuple)):
124            prefix = "%*s:" % (label_len, label)
125            for e in data:
126                yield "%*s %s" % (label_len+1, prefix, e)
127                prefix = ""
128        else:
129            yield "%*s: %s" % (label_len, label, data)
130
131
132def write_formatted_info(writer, header, info):
133    """Write a sequence of (label,data) pairs nicely."""
134    writer.write(info_header(header))
135    for line in info_formatter(info):
136        writer.write(" %s" % line)
137
138
139def short_stack(limit=None, skip=0):
140    """Return a string summarizing the call stack.
141
142    The string is multi-line, with one line per stack frame. Each line shows
143    the function name, the file name, and the line number:
144
145        ...
146        start_import_stop : /Users/ned/coverage/trunk/tests/coveragetest.py @95
147        import_local_file : /Users/ned/coverage/trunk/tests/coveragetest.py @81
148        import_local_file : /Users/ned/coverage/trunk/coverage/backward.py @159
149        ...
150
151    `limit` is the number of frames to include, defaulting to all of them.
152
153    `skip` is the number of frames to skip, so that debugging functions can
154    call this and not be included in the result.
155
156    """
157    stack = inspect.stack()[limit:skip:-1]
158    return "\n".join("%30s : %s:%d" % (t[3], t[1], t[2]) for t in stack)
159
160
161def dump_stack_frames(limit=None, out=None, skip=0):
162    """Print a summary of the stack to stdout, or someplace else."""
163    out = out or sys.stdout
164    out.write(short_stack(limit=limit, skip=skip+1))
165    out.write("\n")
166
167
168def clipped_repr(text, numchars=50):
169    """`repr(text)`, but limited to `numchars`."""
170    r = reprlib.Repr()
171    r.maxstring = numchars
172    return r.repr(text)
173
174
175def short_id(id64):
176    """Given a 64-bit id, make a shorter 16-bit one."""
177    id16 = 0
178    for offset in range(0, 64, 16):
179        id16 ^= id64 >> offset
180    return id16 & 0xFFFF
181
182
183def add_pid_and_tid(text):
184    """A filter to add pid and tid to debug messages."""
185    # Thread ids are useful, but too long. Make a shorter one.
186    tid = "{:04x}".format(short_id(_thread.get_ident()))
187    text = "{:5d}.{}: {}".format(os.getpid(), tid, text)
188    return text
189
190
191class SimpleReprMixin(object):
192    """A mixin implementing a simple __repr__."""
193    simple_repr_ignore = ['simple_repr_ignore', '$coverage.object_id']
194
195    def __repr__(self):
196        show_attrs = (
197            (k, v) for k, v in self.__dict__.items()
198            if getattr(v, "show_repr_attr", True)
199            and not callable(v)
200            and k not in self.simple_repr_ignore
201        )
202        return "<{klass} @0x{id:x} {attrs}>".format(
203            klass=self.__class__.__name__,
204            id=id(self),
205            attrs=" ".join("{}={!r}".format(k, v) for k, v in show_attrs),
206            )
207
208
209def simplify(v):                                            # pragma: debugging
210    """Turn things which are nearly dict/list/etc into dict/list/etc."""
211    if isinstance(v, dict):
212        return {k:simplify(vv) for k, vv in v.items()}
213    elif isinstance(v, (list, tuple)):
214        return type(v)(simplify(vv) for vv in v)
215    elif hasattr(v, "__dict__"):
216        return simplify({'.'+k: v for k, v in v.__dict__.items()})
217    else:
218        return v
219
220
221def pp(v):                                                  # pragma: debugging
222    """Debug helper to pretty-print data, including SimpleNamespace objects."""
223    # Might not be needed in 3.9+
224    pprint.pprint(simplify(v))
225
226
227def filter_text(text, filters):
228    """Run `text` through a series of filters.
229
230    `filters` is a list of functions. Each takes a string and returns a
231    string.  Each is run in turn.
232
233    Returns: the final string that results after all of the filters have
234    run.
235
236    """
237    clean_text = text.rstrip()
238    ending = text[len(clean_text):]
239    text = clean_text
240    for fn in filters:
241        lines = []
242        for line in text.splitlines():
243            lines.extend(fn(line).splitlines())
244        text = "\n".join(lines)
245    return text + ending
246
247
248class CwdTracker(object):                                   # pragma: debugging
249    """A class to add cwd info to debug messages."""
250    def __init__(self):
251        self.cwd = None
252
253    def filter(self, text):
254        """Add a cwd message for each new cwd."""
255        cwd = os.getcwd()
256        if cwd != self.cwd:
257            text = "cwd is now {!r}\n".format(cwd) + text
258            self.cwd = cwd
259        return text
260
261
262class DebugOutputFile(object):                              # pragma: debugging
263    """A file-like object that includes pid and cwd information."""
264    def __init__(self, outfile, show_process, filters):
265        self.outfile = outfile
266        self.show_process = show_process
267        self.filters = list(filters)
268
269        if self.show_process:
270            self.filters.insert(0, CwdTracker().filter)
271            self.write("New process: executable: %r\n" % (sys.executable,))
272            self.write("New process: cmd: %r\n" % (getattr(sys, 'argv', None),))
273            if hasattr(os, 'getppid'):
274                self.write("New process: pid: %r, parent pid: %r\n" % (os.getpid(), os.getppid()))
275
276    SYS_MOD_NAME = '$coverage.debug.DebugOutputFile.the_one'
277
278    @classmethod
279    def get_one(cls, fileobj=None, show_process=True, filters=(), interim=False):
280        """Get a DebugOutputFile.
281
282        If `fileobj` is provided, then a new DebugOutputFile is made with it.
283
284        If `fileobj` isn't provided, then a file is chosen
285        (COVERAGE_DEBUG_FILE, or stderr), and a process-wide singleton
286        DebugOutputFile is made.
287
288        `show_process` controls whether the debug file adds process-level
289        information, and filters is a list of other message filters to apply.
290
291        `filters` are the text filters to apply to the stream to annotate with
292        pids, etc.
293
294        If `interim` is true, then a future `get_one` can replace this one.
295
296        """
297        if fileobj is not None:
298            # Make DebugOutputFile around the fileobj passed.
299            return cls(fileobj, show_process, filters)
300
301        # Because of the way igor.py deletes and re-imports modules,
302        # this class can be defined more than once. But we really want
303        # a process-wide singleton. So stash it in sys.modules instead of
304        # on a class attribute. Yes, this is aggressively gross.
305        the_one, is_interim = sys.modules.get(cls.SYS_MOD_NAME, (None, True))
306        if the_one is None or is_interim:
307            if fileobj is None:
308                debug_file_name = os.environ.get("COVERAGE_DEBUG_FILE", FORCED_DEBUG_FILE)
309                if debug_file_name:
310                    fileobj = open(debug_file_name, "a")
311                else:
312                    fileobj = sys.stderr
313            the_one = cls(fileobj, show_process, filters)
314            sys.modules[cls.SYS_MOD_NAME] = (the_one, interim)
315        return the_one
316
317    def write(self, text):
318        """Just like file.write, but filter through all our filters."""
319        self.outfile.write(filter_text(text, self.filters))
320        self.outfile.flush()
321
322    def flush(self):
323        """Flush our file."""
324        self.outfile.flush()
325
326
327def log(msg, stack=False):                                  # pragma: debugging
328    """Write a log message as forcefully as possible."""
329    out = DebugOutputFile.get_one(interim=True)
330    out.write(msg+"\n")
331    if stack:
332        dump_stack_frames(out=out, skip=1)
333
334
335def decorate_methods(decorator, butnot=(), private=False):  # pragma: debugging
336    """A class decorator to apply a decorator to methods."""
337    def _decorator(cls):
338        for name, meth in inspect.getmembers(cls, inspect.isroutine):
339            if name not in cls.__dict__:
340                continue
341            if name != "__init__":
342                if not private and name.startswith("_"):
343                    continue
344            if name in butnot:
345                continue
346            setattr(cls, name, decorator(meth))
347        return cls
348    return _decorator
349
350
351def break_in_pudb(func):                                    # pragma: debugging
352    """A function decorator to stop in the debugger for each call."""
353    @functools.wraps(func)
354    def _wrapper(*args, **kwargs):
355        import pudb
356        sys.stdout = sys.__stdout__
357        pudb.set_trace()
358        return func(*args, **kwargs)
359    return _wrapper
360
361
362OBJ_IDS = itertools.count()
363CALLS = itertools.count()
364OBJ_ID_ATTR = "$coverage.object_id"
365
366def show_calls(show_args=True, show_stack=False, show_return=False):    # pragma: debugging
367    """A method decorator to debug-log each call to the function."""
368    def _decorator(func):
369        @functools.wraps(func)
370        def _wrapper(self, *args, **kwargs):
371            oid = getattr(self, OBJ_ID_ATTR, None)
372            if oid is None:
373                oid = "{:08d} {:04d}".format(os.getpid(), next(OBJ_IDS))
374                setattr(self, OBJ_ID_ATTR, oid)
375            extra = ""
376            if show_args:
377                eargs = ", ".join(map(repr, args))
378                ekwargs = ", ".join("{}={!r}".format(*item) for item in kwargs.items())
379                extra += "("
380                extra += eargs
381                if eargs and ekwargs:
382                    extra += ", "
383                extra += ekwargs
384                extra += ")"
385            if show_stack:
386                extra += " @ "
387                extra += "; ".join(_clean_stack_line(l) for l in short_stack().splitlines())
388            callid = next(CALLS)
389            msg = "{} {:04d} {}{}\n".format(oid, callid, func.__name__, extra)
390            DebugOutputFile.get_one(interim=True).write(msg)
391            ret = func(self, *args, **kwargs)
392            if show_return:
393                msg = "{} {:04d} {} return {!r}\n".format(oid, callid, func.__name__, ret)
394                DebugOutputFile.get_one(interim=True).write(msg)
395            return ret
396        return _wrapper
397    return _decorator
398
399
400def _clean_stack_line(s):                                   # pragma: debugging
401    """Simplify some paths in a stack trace, for compactness."""
402    s = s.strip()
403    s = s.replace(os.path.dirname(__file__) + '/', '')
404    s = s.replace(os.path.dirname(os.__file__) + '/', '')
405    s = s.replace(sys.prefix + '/', '')
406    return s
407