"""Raw data collector for Coverage.""" import sys, threading try: # Use the C extension code when we can, for speed. from coverage.tracer import CTracer except ImportError: # Couldn't import the C extension, maybe it isn't built. 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.previous_file = None self.previous_covered_lines = None def start(self): """Start this Tracer. Return a Python function suitable for use with sys.settrace(). """ sys.settrace(self._trace) return self._trace def _get_covered_lines(self, frame): filename = frame.f_code.co_filename if filename==self.previous_file: return self.previous_covered_lines self.previous_file = filename tracename = self.should_trace_cache.get(filename) if tracename is None: tracename = self.should_trace(filename, frame) self.should_trace_cache[filename] = tracename if tracename: if tracename not in self.data: self.data[tracename] = {} self.previous_covered_lines = self.data[tracename] else: self.previous_covered_lines = None return self.previous_covered_lines def _trace(self, frame, event, arg_unused): if event=='line': covered_lines = self._get_covered_lines(frame) if covered_lines is not None: covered_lines[frame.f_lineno] = None return self._trace def stop(self): """Stop this Tracer.""" # 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()) 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 False 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() self._trace_class = PyTracer def __repr__(self): return "" % 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 # False). 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 >>sys.stderr, "Started: %r" % self._collectors # Check to see whether we had a fullcoverage tracer installed. traces0 = None if hasattr(sys, "gettrace"): fn0 = sys.gettrace() if fn0: tracer0 = getattr(fn0, '__self__', None) if tracer0: traces0 = getattr(tracer0, 'traces', None) # Install the tracer on this thread. fn = self._start_tracer() if traces0: for args in traces0: (frame, event, arg), lineno = args fn(frame, event, arg, lineno=lineno) # 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 {}