/usr/lib/python2.7/dist-packages/coverage/collector.py is in python-coverage 3.7.1+dfsg.1-1ubuntu2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 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 {}
|