338 lines
14 KiB
Python
338 lines
14 KiB
Python
|
# 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
|