Upgrade to coverage 3.7.1 and have it auto-build itself on first use.

third_party/coverage-3.6/coverage/collector.py

-"""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
-
-    def _trace(self, frame, event, arg_unused):
-        """The trace function passed to sys.settrace."""
-
-        #print("trace event: %s %r @%d" % (
-        #           event, frame.f_code.co_filename, frame.f_lineno),
-        #      file=sys.stderr)
-
-        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
-            tracename = self.should_trace_cache.get(filename)
-            if tracename is None:
-                tracename = self.should_trace(filename, frame)
-                self.should_trace_cache[filename] = tracename
-            #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().
-
-        """
-        sys.settrace(self._trace)
-        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(),))
-                #--debug
-                #from coverage.misc import short_stack
-                #self.warn(msg % (sys.gettrace()))#, short_stack()))
-        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()
-
-        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
-        # 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("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 {}