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