Edit File: pytracer.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 """Raw data collector for coverage.py.""" from __future__ import annotations import atexit import dis import sys import threading from types import FrameType, ModuleType from typing import Any, Callable, Dict, List, Optional, Set, Tuple, cast from coverage import env from coverage.types import ( TArc, TFileDisposition, TLineNo, TTraceData, TTraceFileData, TTraceFn, TTracer, TWarnFn, ) # We need the YIELD_VALUE opcode below, in a comparison-friendly form. RESUME = dis.opmap.get("RESUME") RETURN_VALUE = dis.opmap["RETURN_VALUE"] if RESUME is None: YIELD_VALUE = dis.opmap["YIELD_VALUE"] YIELD_FROM = dis.opmap["YIELD_FROM"] YIELD_FROM_OFFSET = 0 if env.PYPY else 2 # When running meta-coverage, this file can try to trace itself, which confuses # everything. Don't trace ourselves. THIS_FILE = __file__.rstrip("co") class PyTracer(TTracer): """Python implementation of the raw data tracer.""" # Because of poor implementations of trace-function-manipulating tools, # the Python trace function must be kept very simple. In particular, there # must be only one function ever set as the trace function, both through # sys.settrace, and as the return value from the trace function. Put # another way, the trace function must always return itself. It cannot # swap in other functions, or return None to avoid tracing a particular # frame. # # The trace manipulator that introduced this restriction is DecoratorTools, # which sets a trace function, and then later restores the pre-existing one # by calling sys.settrace with a function it found in the current frame. # # Systems that use DecoratorTools (or similar trace manipulations) must use # PyTracer to get accurate results. The command-line --timid argument is # used to force the use of this tracer. def __init__(self) -> None: # pylint: disable=super-init-not-called # Attributes set from the collector: self.data: TTraceData self.trace_arcs = False self.should_trace: Callable[[str, FrameType], TFileDisposition] self.should_trace_cache: Dict[str, Optional[TFileDisposition]] self.should_start_context: Optional[Callable[[FrameType], Optional[str]]] = None self.switch_context: Optional[Callable[[Optional[str]], None]] = None self.warn: TWarnFn # The threading module to use, if any. self.threading: Optional[ModuleType] = None self.cur_file_data: Optional[TTraceFileData] = None self.last_line: TLineNo = 0 self.cur_file_name: Optional[str] = None self.context: Optional[str] = None self.started_context = False self.data_stack: List[Tuple[Optional[TTraceFileData], Optional[str], TLineNo, bool]] = [] self.thread: Optional[threading.Thread] = None self.stopped = False self._activity = False self.in_atexit = False # On exit, self.in_atexit = True atexit.register(setattr, self, "in_atexit", True) # Cache a bound method on the instance, so that we don't have to # re-create a bound method object all the time. self._cached_bound_method_trace: TTraceFn = self._trace def __repr__(self) -> str: me = id(self) points = sum(len(v) for v in self.data.values()) files = len(self.data) return f"<PyTracer at 0x{me:x}: {points} data points in {files} files>" def log(self, marker: str, *args: Any) -> None: """For hard-core logging of what this tracer is doing.""" with open("/tmp/debug_trace.txt", "a") as f: f.write("{} {}[{}]".format( marker, id(self), len(self.data_stack), )) if 0: # if you want thread ids.. f.write(".{:x}.{:x}".format( # type: ignore[unreachable] self.thread.ident, self.threading.current_thread().ident, )) f.write(" {}".format(" ".join(map(str, args)))) if 0: # if you want callers.. f.write(" | ") # type: ignore[unreachable] stack = " / ".join( (fname or "???").rpartition("/")[-1] for _, fname, _, _ in self.data_stack ) f.write(stack) f.write("\n") def _trace( self, frame: FrameType, event: str, arg: Any, # pylint: disable=unused-argument lineno: Optional[TLineNo] = None, # pylint: disable=unused-argument ) -> Optional[TTraceFn]: """The trace function passed to sys.settrace.""" if THIS_FILE in frame.f_code.co_filename: return None #self.log(":", frame.f_code.co_filename, frame.f_lineno, frame.f_code.co_name + "()", event) if (self.stopped and sys.gettrace() == self._cached_bound_method_trace): # pylint: disable=comparison-with-callable # The PyTrace.stop() method has been called, possibly by another # thread, let's deactivate ourselves now. if 0: f = frame # type: ignore[unreachable] self.log("---\nX", f.f_code.co_filename, f.f_lineno) while f: self.log(">", f.f_code.co_filename, f.f_lineno, f.f_code.co_name, f.f_trace) f = f.f_back sys.settrace(None) try: self.cur_file_data, self.cur_file_name, self.last_line, self.started_context = ( self.data_stack.pop() ) except IndexError: self.log( "Empty stack!", frame.f_code.co_filename, frame.f_lineno, frame.f_code.co_name ) return None # if event != "call" and frame.f_code.co_filename != self.cur_file_name: # self.log("---\n*", frame.f_code.co_filename, self.cur_file_name, frame.f_lineno) if event == "call": # Should we start a new context? if self.should_start_context and self.context is None: context_maybe = self.should_start_context(frame) if context_maybe is not None: self.context = context_maybe started_context = True assert self.switch_context is not None self.switch_context(self.context) else: started_context = False else: started_context = False self.started_context = started_context # Entering a new frame. Decide if we should trace in this file. self._activity = True self.data_stack.append( ( self.cur_file_data, self.cur_file_name, self.last_line, started_context, ) ) # Improve tracing performance: when calling a function, both caller # and callee are often within the same file. if that's the case, we # don't have to re-check whether to trace the corresponding # function (which is a little bit expensive since it involves # dictionary lookups). This optimization is only correct if we # didn't start a context. filename = frame.f_code.co_filename if filename != self.cur_file_name or started_context: self.cur_file_name = filename disp = self.should_trace_cache.get(filename) if disp is None: disp = self.should_trace(filename, frame) self.should_trace_cache[filename] = disp self.cur_file_data = None if disp.trace: tracename = disp.source_filename assert tracename is not None if tracename not in self.data: self.data[tracename] = set() # type: ignore[assignment] self.cur_file_data = self.data[tracename] else: frame.f_trace_lines = False elif not self.cur_file_data: frame.f_trace_lines = False # The call event is really a "start frame" event, and happens for # function calls and re-entering generators. The f_lasti field is # -1 for calls, and a real offset for generators. Use <0 as the # line number for calls, and the real line number for generators. if RESUME is not None: # The current opcode is guaranteed to be RESUME. The argument # determines what kind of resume it is. oparg = frame.f_code.co_code[frame.f_lasti + 1] real_call = (oparg == 0) else: real_call = (getattr(frame, "f_lasti", -1) < 0) if real_call: self.last_line = -frame.f_code.co_firstlineno else: self.last_line = frame.f_lineno elif event == "line": # Record an executed line. if self.cur_file_data is not None: flineno: TLineNo = frame.f_lineno if self.trace_arcs: cast(Set[TArc], self.cur_file_data).add((self.last_line, flineno)) else: cast(Set[TLineNo], self.cur_file_data).add(flineno) self.last_line = flineno elif event == "return": if self.trace_arcs and self.cur_file_data: # Record an arc leaving the function, but beware that a # "return" event might just mean yielding from a generator. code = frame.f_code.co_code lasti = frame.f_lasti if RESUME is not None: if len(code) == lasti + 2: # A return from the end of a code object is a real return. real_return = True else: # it's a real return. real_return = (code[lasti + 2] != RESUME) else: if code[lasti] == RETURN_VALUE: real_return = True elif code[lasti] == YIELD_VALUE: real_return = False elif len(code) <= lasti + YIELD_FROM_OFFSET: real_return = True elif code[lasti + YIELD_FROM_OFFSET] == YIELD_FROM: real_return = False else: real_return = True if real_return: first = frame.f_code.co_firstlineno cast(Set[TArc], self.cur_file_data).add((self.last_line, -first)) # Leaving this function, pop the filename stack. self.cur_file_data, self.cur_file_name, self.last_line, self.started_context = ( self.data_stack.pop() ) # Leaving a context? if self.started_context: assert self.switch_context is not None self.context = None self.switch_context(None) return self._cached_bound_method_trace def start(self) -> TTraceFn: """Start this Tracer. Return a Python function suitable for use with sys.settrace(). """ self.stopped = False if self.threading: if self.thread is None: self.thread = self.threading.current_thread() else: if self.thread.ident != self.threading.current_thread().ident: # Re-starting from a different thread!? Don't set the trace # function, but we are marked as running again, so maybe it # will be ok? #self.log("~", "starting on different threads") return self._cached_bound_method_trace sys.settrace(self._cached_bound_method_trace) return self._cached_bound_method_trace def stop(self) -> None: """Stop this Tracer.""" # Get the active tracer callback before setting the stop flag to be # able to detect if the tracer was changed prior to stopping it. tf = sys.gettrace() # Set the stop flag. The actual call to sys.settrace(None) will happen # in the self._trace callback itself to make sure to call it from the # right thread. self.stopped = True if self.threading: assert self.thread is not None if self.thread.ident != self.threading.current_thread().ident: # Called on a different thread than started us: we can't unhook # ourselves, but we've set the flag that we should stop, so we # won't do any more tracing. #self.log("~", "stopping on different threads") return if self.warn: # PyPy clears the trace function before running atexit functions, # so don't warn if we are in atexit on PyPy and the trace function # has changed to None. dont_warn = (env.PYPY and env.PYPYVERSION >= (5, 4) and self.in_atexit and tf is None) if (not dont_warn) and tf != self._cached_bound_method_trace: # pylint: disable=comparison-with-callable self.warn( "Trace function changed, data is likely wrong: " + f"{tf!r} != {self._cached_bound_method_trace!r}", slug="trace-changed", ) def activity(self) -> bool: """Has there been any activity?""" return self._activity def reset_activity(self) -> None: """Reset the activity() flag.""" self._activity = False def get_stats(self) -> Optional[Dict[str, int]]: """Return a dictionary of statistics, or None.""" return None