mirror of
https://gitflic.ru/project/openide/openide.git
synced 2026-01-05 01:50:56 +07:00
Coverage.py to separate dir
This commit is contained in:
Dmitry Trofimov
353
python/helpers/coveragepy/coverage/collector.py
Normal file
353
python/helpers/coveragepy/coverage/collector.py
Normal file
@@ -0,0 +1,353 @@
|
||||
"""Raw data collector for Coverage."""
|
||||
|
||||
import os, sys, threading
|
||||
|
||||
try:
|
||||
# Use the C extension code when we can, for speed.
|
||||
from coverage.tracer import CTracer # pylint: disable=F0401,E0611
|
||||
except ImportError:
|
||||
# Couldn't import the C extension, maybe it isn't built.
|
||||
if os.getenv('COVERAGE_TEST_TRACER') == 'c':
|
||||
# During testing, we use the COVERAGE_TEST_TRACER env var to indicate
|
||||
# that we've fiddled with the environment to test this fallback code.
|
||||
# If we thought we had a C tracer, but couldn't import it, then exit
|
||||
# quickly and clearly instead of dribbling confusing errors. I'm using
|
||||
# sys.exit here instead of an exception because an exception here
|
||||
# causes all sorts of other noise in unittest.
|
||||
sys.stderr.write(
|
||||
"*** COVERAGE_TEST_TRACER is 'c' but can't import CTracer!\n"
|
||||
)
|
||||
sys.exit(1)
|
||||
CTracer = None
|
||||
|
||||
|
||||
class PyTracer(object):
|
||||
"""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):
|
||||
self.data = None
|
||||
self.should_trace = None
|
||||
self.should_trace_cache = None
|
||||
self.warn = None
|
||||
self.cur_file_data = None
|
||||
self.last_line = 0
|
||||
self.data_stack = []
|
||||
self.last_exc_back = None
|
||||
self.last_exc_firstlineno = 0
|
||||
self.arcs = False
|
||||
self.thread = None
|
||||
self.stopped = False
|
||||
|
||||
def _trace(self, frame, event, arg_unused):
|
||||
"""The trace function passed to sys.settrace."""
|
||||
|
||||
if self.stopped:
|
||||
return
|
||||
|
||||
if 0:
|
||||
sys.stderr.write("trace event: %s %r @%d\n" % (
|
||||
event, frame.f_code.co_filename, frame.f_lineno
|
||||
))
|
||||
|
||||
if self.last_exc_back:
|
||||
if frame == self.last_exc_back:
|
||||
# Someone forgot a return event.
|
||||
if self.arcs and self.cur_file_data:
|
||||
pair = (self.last_line, -self.last_exc_firstlineno)
|
||||
self.cur_file_data[pair] = None
|
||||
self.cur_file_data, self.last_line = self.data_stack.pop()
|
||||
self.last_exc_back = None
|
||||
|
||||
if event == 'call':
|
||||
# Entering a new function context. Decide if we should trace
|
||||
# in this file.
|
||||
self.data_stack.append((self.cur_file_data, self.last_line))
|
||||
filename = frame.f_code.co_filename
|
||||
if filename not in self.should_trace_cache:
|
||||
tracename = self.should_trace(filename, frame)
|
||||
self.should_trace_cache[filename] = tracename
|
||||
else:
|
||||
tracename = self.should_trace_cache[filename]
|
||||
#print("called, stack is %d deep, tracename is %r" % (
|
||||
# len(self.data_stack), tracename))
|
||||
if tracename:
|
||||
if tracename not in self.data:
|
||||
self.data[tracename] = {}
|
||||
self.cur_file_data = self.data[tracename]
|
||||
else:
|
||||
self.cur_file_data = None
|
||||
# Set the last_line to -1 because the next arc will be entering a
|
||||
# code block, indicated by (-1, n).
|
||||
self.last_line = -1
|
||||
elif event == 'line':
|
||||
# Record an executed line.
|
||||
if self.cur_file_data is not None:
|
||||
if self.arcs:
|
||||
#print("lin", self.last_line, frame.f_lineno)
|
||||
self.cur_file_data[(self.last_line, frame.f_lineno)] = None
|
||||
else:
|
||||
#print("lin", frame.f_lineno)
|
||||
self.cur_file_data[frame.f_lineno] = None
|
||||
self.last_line = frame.f_lineno
|
||||
elif event == 'return':
|
||||
if self.arcs and self.cur_file_data:
|
||||
first = frame.f_code.co_firstlineno
|
||||
self.cur_file_data[(self.last_line, -first)] = None
|
||||
# Leaving this function, pop the filename stack.
|
||||
self.cur_file_data, self.last_line = self.data_stack.pop()
|
||||
#print("returned, stack is %d deep" % (len(self.data_stack)))
|
||||
elif event == 'exception':
|
||||
#print("exc", self.last_line, frame.f_lineno)
|
||||
self.last_exc_back = frame.f_back
|
||||
self.last_exc_firstlineno = frame.f_code.co_firstlineno
|
||||
return self._trace
|
||||
|
||||
def start(self):
|
||||
"""Start this Tracer.
|
||||
|
||||
Return a Python function suitable for use with sys.settrace().
|
||||
|
||||
"""
|
||||
self.thread = threading.currentThread()
|
||||
sys.settrace(self._trace)
|
||||
return self._trace
|
||||
|
||||
def stop(self):
|
||||
"""Stop this Tracer."""
|
||||
self.stopped = True
|
||||
if self.thread != threading.currentThread():
|
||||
# Called on a different thread than started us: we can't unhook
|
||||
# ourseves, but we've set the flag that we should stop, so we won't
|
||||
# do any more tracing.
|
||||
return
|
||||
|
||||
if hasattr(sys, "gettrace") and self.warn:
|
||||
if sys.gettrace() != self._trace:
|
||||
msg = "Trace function changed, measurement is likely wrong: %r"
|
||||
self.warn(msg % (sys.gettrace(),))
|
||||
#print("Stopping tracer on %s" % threading.current_thread().ident)
|
||||
sys.settrace(None)
|
||||
|
||||
def get_stats(self):
|
||||
"""Return a dictionary of statistics, or None."""
|
||||
return None
|
||||
|
||||
|
||||
class Collector(object):
|
||||
"""Collects trace data.
|
||||
|
||||
Creates a Tracer object for each thread, since they track stack
|
||||
information. Each Tracer points to the same shared data, contributing
|
||||
traced data points.
|
||||
|
||||
When the Collector is started, it creates a Tracer for the current thread,
|
||||
and installs a function to create Tracers for each new thread started.
|
||||
When the Collector is stopped, all active Tracers are stopped.
|
||||
|
||||
Threads started while the Collector is stopped will never have Tracers
|
||||
associated with them.
|
||||
|
||||
"""
|
||||
|
||||
# The stack of active Collectors. Collectors are added here when started,
|
||||
# and popped when stopped. Collectors on the stack are paused when not
|
||||
# the top, and resumed when they become the top again.
|
||||
_collectors = []
|
||||
|
||||
def __init__(self, should_trace, timid, branch, warn):
|
||||
"""Create a collector.
|
||||
|
||||
`should_trace` is a function, taking a filename, and returning a
|
||||
canonicalized filename, or None depending on whether the file should
|
||||
be traced or not.
|
||||
|
||||
If `timid` is true, then a slower simpler trace function will be
|
||||
used. This is important for some environments where manipulation of
|
||||
tracing functions make the faster more sophisticated trace function not
|
||||
operate properly.
|
||||
|
||||
If `branch` is true, then branches will be measured. This involves
|
||||
collecting data on which statements followed each other (arcs). Use
|
||||
`get_arc_data` to get the arc data.
|
||||
|
||||
`warn` is a warning function, taking a single string message argument,
|
||||
to be used if a warning needs to be issued.
|
||||
|
||||
"""
|
||||
self.should_trace = should_trace
|
||||
self.warn = warn
|
||||
self.branch = branch
|
||||
self.reset()
|
||||
|
||||
if timid:
|
||||
# Being timid: use the simple Python trace function.
|
||||
self._trace_class = PyTracer
|
||||
else:
|
||||
# Being fast: use the C Tracer if it is available, else the Python
|
||||
# trace function.
|
||||
self._trace_class = CTracer or PyTracer
|
||||
|
||||
def __repr__(self):
|
||||
return "<Collector at 0x%x>" % id(self)
|
||||
|
||||
def tracer_name(self):
|
||||
"""Return the class name of the tracer we're using."""
|
||||
return self._trace_class.__name__
|
||||
|
||||
def reset(self):
|
||||
"""Clear collected data, and prepare to collect more."""
|
||||
# A dictionary mapping filenames to dicts with linenumber keys,
|
||||
# or mapping filenames to dicts with linenumber pairs as keys.
|
||||
self.data = {}
|
||||
|
||||
# A cache of the results from should_trace, the decision about whether
|
||||
# to trace execution in a file. A dict of filename to (filename or
|
||||
# None).
|
||||
self.should_trace_cache = {}
|
||||
|
||||
# Our active Tracers.
|
||||
self.tracers = []
|
||||
|
||||
def _start_tracer(self):
|
||||
"""Start a new Tracer object, and store it in self.tracers."""
|
||||
tracer = self._trace_class()
|
||||
tracer.data = self.data
|
||||
tracer.arcs = self.branch
|
||||
tracer.should_trace = self.should_trace
|
||||
tracer.should_trace_cache = self.should_trace_cache
|
||||
tracer.warn = self.warn
|
||||
fn = tracer.start()
|
||||
self.tracers.append(tracer)
|
||||
return fn
|
||||
|
||||
# The trace function has to be set individually on each thread before
|
||||
# execution begins. Ironically, the only support the threading module has
|
||||
# for running code before the thread main is the tracing function. So we
|
||||
# install this as a trace function, and the first time it's called, it does
|
||||
# the real trace installation.
|
||||
|
||||
def _installation_trace(self, frame_unused, event_unused, arg_unused):
|
||||
"""Called on new threads, installs the real tracer."""
|
||||
# Remove ourselves as the trace function
|
||||
sys.settrace(None)
|
||||
# Install the real tracer.
|
||||
fn = self._start_tracer()
|
||||
# Invoke the real trace function with the current event, to be sure
|
||||
# not to lose an event.
|
||||
if fn:
|
||||
fn = fn(frame_unused, event_unused, arg_unused)
|
||||
# Return the new trace function to continue tracing in this scope.
|
||||
return fn
|
||||
|
||||
def start(self):
|
||||
"""Start collecting trace information."""
|
||||
if self._collectors:
|
||||
self._collectors[-1].pause()
|
||||
self._collectors.append(self)
|
||||
#print("Started: %r" % self._collectors, file=sys.stderr)
|
||||
|
||||
# Check to see whether we had a fullcoverage tracer installed.
|
||||
traces0 = []
|
||||
if hasattr(sys, "gettrace"):
|
||||
fn0 = sys.gettrace()
|
||||
if fn0:
|
||||
tracer0 = getattr(fn0, '__self__', None)
|
||||
if tracer0:
|
||||
traces0 = getattr(tracer0, 'traces', [])
|
||||
|
||||
# Install the tracer on this thread.
|
||||
fn = self._start_tracer()
|
||||
|
||||
for args in traces0:
|
||||
(frame, event, arg), lineno = args
|
||||
try:
|
||||
fn(frame, event, arg, lineno=lineno)
|
||||
except TypeError:
|
||||
raise Exception(
|
||||
"fullcoverage must be run with the C trace function."
|
||||
)
|
||||
|
||||
# Install our installation tracer in threading, to jump start other
|
||||
# threads.
|
||||
threading.settrace(self._installation_trace)
|
||||
|
||||
def stop(self):
|
||||
"""Stop collecting trace information."""
|
||||
#print >>sys.stderr, "Stopping: %r" % self._collectors
|
||||
assert self._collectors
|
||||
assert self._collectors[-1] is self
|
||||
|
||||
self.pause()
|
||||
self.tracers = []
|
||||
|
||||
# Remove this Collector from the stack, and resume the one underneath
|
||||
# (if any).
|
||||
self._collectors.pop()
|
||||
if self._collectors:
|
||||
self._collectors[-1].resume()
|
||||
|
||||
def pause(self):
|
||||
"""Pause tracing, but be prepared to `resume`."""
|
||||
for tracer in self.tracers:
|
||||
tracer.stop()
|
||||
stats = tracer.get_stats()
|
||||
if stats:
|
||||
print("\nCoverage.py tracer stats:")
|
||||
for k in sorted(stats.keys()):
|
||||
print("%16s: %s" % (k, stats[k]))
|
||||
threading.settrace(None)
|
||||
|
||||
def resume(self):
|
||||
"""Resume tracing after a `pause`."""
|
||||
for tracer in self.tracers:
|
||||
tracer.start()
|
||||
threading.settrace(self._installation_trace)
|
||||
|
||||
def get_line_data(self):
|
||||
"""Return the line data collected.
|
||||
|
||||
Data is { filename: { lineno: None, ...}, ...}
|
||||
|
||||
"""
|
||||
if self.branch:
|
||||
# If we were measuring branches, then we have to re-build the dict
|
||||
# to show line data.
|
||||
line_data = {}
|
||||
for f, arcs in self.data.items():
|
||||
line_data[f] = ldf = {}
|
||||
for l1, _ in list(arcs.keys()):
|
||||
if l1:
|
||||
ldf[l1] = None
|
||||
return line_data
|
||||
else:
|
||||
return self.data
|
||||
|
||||
def get_arc_data(self):
|
||||
"""Return the arc data collected.
|
||||
|
||||
Data is { filename: { (l1, l2): None, ...}, ...}
|
||||
|
||||
Note that no data is collected or returned if the Collector wasn't
|
||||
created with `branch` true.
|
||||
|
||||
"""
|
||||
if self.branch:
|
||||
return self.data
|
||||
else:
|
||||
return {}
|
||||
Reference in New Issue
Block a user