#!/usr/bin/python

#

#             Perforce Defect Tracking Integration Project

#              <http://www.ravenbrook.com/project/p4dti/>

#

#                   COVERAGE.PY -- COVERAGE TESTING

#

#             Gareth Rees, Ravenbrook Limited, 2001-12-04

#                     Ned Batchelder, 2004-12-12

#         http://nedbatchelder.com/code/modules/coverage.html

#

#

# 1. INTRODUCTION

#

# This module provides coverage testing for Python code.

#

# The intended readership is all Python developers.

#

# This document is not confidential.

#

# See [GDR 2001-12-04a] for the command-line interface, programmatic

# interface and limitations.  See [GDR 2001-12-04b] for requirements and

# design.

r"""Usage:

coverage.py -x [-p] MODULE.py [ARG1 ARG2 ...]

    Execute module, passing the given command-line arguments, collecting

    coverage data. With the -p option, write to a temporary file containing

    the machine name and process ID.

coverage.py -e

    Erase collected coverage data.

coverage.py -c

    Collect data from multiple coverage files (as created by -p option above)

    and store it into a single file representing the union of the coverage.

coverage.py -r [-m] [-o dir1,dir2,...] FILE1 FILE2 ...

    Report on the statement coverage for the given files.  With the -m

    option, show line numbers of the statements that weren't executed.

coverage.py -a [-d dir] [-o dir1,dir2,...] FILE1 FILE2 ...

    Make annotated copies of the given files, marking statements that

    are executed with > and statements that are missed with !.  With

    the -d option, make the copies in that directory.  Without the -d

    option, make each copy in the same directory as the original.

-o dir,dir2,...

  Omit reporting or annotating files when their filename path starts with

  a directory listed in the omit list.

  e.g. python coverage.py -i -r -o c:\python23,lib\enthought\traits

Coverage data is saved in the file .coverage by default.  Set the

COVERAGE_FILE environment variable to save it somewhere else."""

__version__ = "2.85.20080914"    # see detailed history at the end of this file.

import compiler

import compiler.visitor

import glob

import os

import re

import string

import symbol

import sys

import threading

import token

import types

import zipimport

from socket import gethostname

# Python version compatibility

try:

    strclass = basestring   # new to 2.3

except:

    strclass = str

# 2. IMPLEMENTATION

#

# This uses the "singleton" pattern.

#

# The word "morf" means a module object (from which the source file can

# be deduced by suitable manipulation of the __file__ attribute) or a

# filename.

#

# When we generate a coverage report we have to canonicalize every

# filename in the coverage dictionary just in case it refers to the

# module we are reporting on.  It seems a shame to throw away this

# information so the data in the coverage dictionary is transferred to

# the 'cexecuted' dictionary under the canonical filenames.

#

# The coverage dictionary is called "c" and the trace function "t".  The

# reason for these short names is that Python looks up variables by name

# at runtime and so execution time depends on the length of variables!

# In the bottleneck of this application it's appropriate to abbreviate

# names to increase speed.

class StatementFindingAstVisitor(compiler.visitor.ASTVisitor):

    """ A visitor for a parsed Abstract Syntax Tree which finds executable

        statements.

    """

    def __init__(self, statements, excluded, suite_spots):

        compiler.visitor.ASTVisitor.__init__(self)

        self.statements = statements

        self.excluded = excluded

        self.suite_spots = suite_spots

        self.excluding_suite = 0

    def doRecursive(self, node):

        for n in node.getChildNodes():

            self.dispatch(n)

    visitStmt = visitModule = doRecursive

    def doCode(self, node):

        if hasattr(node, 'decorators') and node.decorators:

            self.dispatch(node.decorators)

            self.recordAndDispatch(node.code)

        else:

            self.doSuite(node, node.code)

    visitFunction = visitClass = doCode

    def getFirstLine(self, node):

        # Find the first line in the tree node.

        lineno = node.lineno

        for n in node.getChildNodes():

            f = self.getFirstLine(n)

            if lineno and f:

                lineno = min(lineno, f)

            else:

                lineno = lineno or f

        return lineno

    def getLastLine(self, node):

        # Find the first line in the tree node.

        lineno = node.lineno

        for n in node.getChildNodes():

            lineno = max(lineno, self.getLastLine(n))

        return lineno

    def doStatement(self, node):

        self.recordLine(self.getFirstLine(node))

    visitAssert = visitAssign = visitAssTuple = visitPrint = \

        visitPrintnl = visitRaise = visitSubscript = visitDecorators = \

        doStatement

    def visitPass(self, node):

        # Pass statements have weird interactions with docstrings.  If this

        # pass statement is part of one of those pairs, claim that the statement

        # is on the later of the two lines.

        l = node.lineno

        if l:

            lines = self.suite_spots.get(l, [l,l])

            self.statements[lines[1]] = 1

    def visitDiscard(self, node):

        # Discard nodes are statements that execute an expression, but then

        # discard the results.  This includes function calls, so we can't 

        # ignore them all.  But if the expression is a constant, the statement

        # won't be "executed", so don't count it now.

        if node.expr.__class__.__name__ != 'Const':

            self.doStatement(node)

    def recordNodeLine(self, node):

        # Stmt nodes often have None, but shouldn't claim the first line of

        # their children (because the first child might be an ignorable line

        # like "global a").

        if node.__class__.__name__ != 'Stmt':

            return self.recordLine(self.getFirstLine(node))

        else:

            return 0

    def recordLine(self, lineno):

        # Returns a bool, whether the line is included or excluded.

        if lineno:

            # Multi-line tests introducing suites have to get charged to their

            # keyword.

            if lineno in self.suite_spots:

                lineno = self.suite_spots[lineno][0]

            # If we're inside an excluded suite, record that this line was

            # excluded.

            if self.excluding_suite:

                self.excluded[lineno] = 1

                return 0

            # If this line is excluded, or suite_spots maps this line to

            # another line that is exlcuded, then we're excluded.

            elif self.excluded.has_key(lineno) or \

                 self.suite_spots.has_key(lineno) and \

                 self.excluded.has_key(self.suite_spots[lineno][1]):

                return 0

            # Otherwise, this is an executable line.

            else:

                self.statements[lineno] = 1

                return 1

        return 0

    default = recordNodeLine

    def recordAndDispatch(self, node):

        self.recordNodeLine(node)

        self.dispatch(node)

    def doSuite(self, intro, body, exclude=0):

        exsuite = self.excluding_suite

        if exclude or (intro and not self.recordNodeLine(intro)):

            self.excluding_suite = 1

        self.recordAndDispatch(body)

        self.excluding_suite = exsuite

    def doPlainWordSuite(self, prevsuite, suite):

        # Finding the exclude lines for else's is tricky, because they aren't

        # present in the compiler parse tree.  Look at the previous suite,

        # and find its last line.  If any line between there and the else's

        # first line are excluded, then we exclude the else.

        lastprev = self.getLastLine(prevsuite)

        firstelse = self.getFirstLine(suite)

        for l in range(lastprev+1, firstelse):

            if self.suite_spots.has_key(l):

                self.doSuite(None, suite, exclude=self.excluded.has_key(l))

                break

        else:

            self.doSuite(None, suite)

    def doElse(self, prevsuite, node):

        if node.else_:

            self.doPlainWordSuite(prevsuite, node.else_)

    def visitFor(self, node):

        self.doSuite(node, node.body)

        self.doElse(node.body, node)

    visitWhile = visitFor

    def visitIf(self, node):

        # The first test has to be handled separately from the rest.

        # The first test is credited to the line with the "if", but the others

        # are credited to the line with the test for the elif.

        self.doSuite(node, node.tests[0][1])

        for t, n in node.tests[1:]:

            self.doSuite(t, n)

        self.doElse(node.tests[-1][1], node)

    def visitTryExcept(self, node):

        self.doSuite(node, node.body)

        for i in range(len(node.handlers)):

            a, b, h = node.handlers[i]

            if not a:

                # It's a plain "except:".  Find the previous suite.

                if i > 0:

                    prev = node.handlers[i-1][2]

                else:

                    prev = node.body

                self.doPlainWordSuite(prev, h)

            else:

                self.doSuite(a, h)

        self.doElse(node.handlers[-1][2], node)

    def visitTryFinally(self, node):

        self.doSuite(node, node.body)

        self.doPlainWordSuite(node.body, node.final)

    def visitWith(self, node):

        self.doSuite(node, node.body)

    def visitGlobal(self, node):

        # "global" statements don't execute like others (they don't call the

        # trace function), so don't record their line numbers.

        pass

the_coverage = None

class CoverageException(Exception):

    pass

class coverage:

    # Name of the cache file (unless environment variable is set).

    cache_default = ".coverage"

    # Environment variable naming the cache file.

    cache_env = "COVERAGE_FILE"

    # A dictionary with an entry for (Python source file name, line number

    # in that file) if that line has been executed.

    c = {}

    # A map from canonical Python source file name to a dictionary in

    # which there's an entry for each line number that has been

    # executed.

    cexecuted = {}

    # Cache of results of calling the analysis2() method, so that you can

    # specify both -r and -a without doing double work.

    analysis_cache = {}

    # Cache of results of calling the canonical_filename() method, to

    # avoid duplicating work.

    canonical_filename_cache = {}

    def __init__(self):

        global the_coverage

        if the_coverage:

            raise CoverageException("Only one coverage object allowed.")

        self.usecache = 1

        self.cache = None

        self.parallel_mode = False

        self.exclude_re = ''

        self.nesting = 0

        self.cstack = []

        self.xstack = []

        self.relative_dir = self.abs_file(os.curdir)+os.sep

        self.exclude('# *pragma[: ]*[nN][oO] *[cC][oO][vV][eE][rR]')

    # t(f, x, y).  This method is passed to sys.settrace as a trace function.  

    # See [van Rossum 2001-07-20b, 9.2] for an explanation of sys.settrace and 

    # the arguments and return value of the trace function.

    # See [van Rossum 2001-07-20a, 3.2] for a description of frame and code

    # objects.

    def t(self, f, w, unused):                                 #pragma: no cover

        if w == 'line':

            self.c[(f.f_code.co_filename, f.f_lineno)] = 1

            #-for c in self.cstack:

            #-    c[(f.f_code.co_filename, f.f_lineno)] = 1

        return self.t

    def help(self, error=None):     #pragma: no cover

        if error:

            print error

            print

        print __doc__

        sys.exit(1)

    def command_line(self, argv, help_fn=None):

        import getopt

        help_fn = help_fn or self.help

        settings = {}

        optmap = {

            '-a': 'annotate',

            '-c': 'collect',

            '-d:': 'directory=',

            '-e': 'erase',

            '-h': 'help',

            '-i': 'ignore-errors',

            '-m': 'show-missing',

            '-p': 'parallel-mode',

            '-r': 'report',

            '-x': 'execute',

            '-o:': 'omit=',

            }

        short_opts = string.join(map(lambda o: o[1:], optmap.keys()), '')

        long_opts = optmap.values()

        options, args = getopt.getopt(argv, short_opts, long_opts)

        for o, a in options:

            if optmap.has_key(o):

                settings[optmap[o]] = 1

            elif optmap.has_key(o + ':'):

                settings[optmap[o + ':']] = a

            elif o[2:] in long_opts:

                settings[o[2:]] = 1

            elif o[2:] + '=' in long_opts:

                settings[o[2:]+'='] = a

            else:       #pragma: no cover

                pass    # Can't get here, because getopt won't return anything unknown.

        if settings.get('help'):

            help_fn()

        for i in ['erase', 'execute']:

            for j in ['annotate', 'report', 'collect']:

                if settings.get(i) and settings.get(j):

                    help_fn("You can't specify the '%s' and '%s' "

                              "options at the same time." % (i, j))

        args_needed = (settings.get('execute')

                       or settings.get('annotate')

                       or settings.get('report'))

        action = (settings.get('erase') 

                  or settings.get('collect')

                  or args_needed)

        if not action:

            help_fn("You must specify at least one of -e, -x, -c, -r, or -a.")

        if not args_needed and args:

            help_fn("Unexpected arguments: %s" % " ".join(args))

        self.parallel_mode = settings.get('parallel-mode')

        self.get_ready()

        if settings.get('erase'):

            self.erase()

        if settings.get('execute'):

            if not args:

                help_fn("Nothing to do.")

            sys.argv = args

            self.start()

            import __main__

            sys.path[0] = os.path.dirname(sys.argv[0])

            execfile(sys.argv[0], __main__.__dict__)

        if settings.get('collect'):

            self.collect()

        if not args:

            args = self.cexecuted.keys()

        ignore_errors = settings.get('ignore-errors')

        show_missing = settings.get('show-missing')

        directory = settings.get('directory=')

        omit = settings.get('omit=')

        if omit is not None:

            omit = [self.abs_file(p) for p in omit.split(',')]

        else:

            omit = []

        if settings.get('report'):

            self.report(args, show_missing, ignore_errors, omit_prefixes=omit)

        if settings.get('annotate'):

            self.annotate(args, directory, ignore_errors, omit_prefixes=omit)

    def use_cache(self, usecache, cache_file=None):

        self.usecache = usecache

        if cache_file and not self.cache:

            self.cache_default = cache_file

    def get_ready(self, parallel_mode=False):

        if self.usecache and not self.cache:

            self.cache = os.environ.get(self.cache_env, self.cache_default)

            if self.parallel_mode:

                self.cache += "." + gethostname() + "." + str(os.getpid())

            self.restore()

        self.analysis_cache = {}

    def start(self, parallel_mode=False):

        self.get_ready()

        if self.nesting == 0:                               #pragma: no cover

            sys.settrace(self.t)

            if hasattr(threading, 'settrace'):

                threading.settrace(self.t)

        self.nesting += 1

    def stop(self):

        self.nesting -= 1

        if self.nesting == 0:                               #pragma: no cover

            sys.settrace(None)

            if hasattr(threading, 'settrace'):

                threading.settrace(None)

    def erase(self):

        self.get_ready()

        self.c = {}

        self.analysis_cache = {}

        self.cexecuted = {}

        if self.cache and os.path.exists(self.cache):

            os.remove(self.cache)

    def exclude(self, re):

        if self.exclude_re:

            self.exclude_re += "|"

        self.exclude_re += "(" + re + ")"

    def begin_recursive(self):

        self.cstack.append(self.c)

        self.xstack.append(self.exclude_re)

    def end_recursive(self):

        self.c = self.cstack.pop()

        self.exclude_re = self.xstack.pop()

    # save().  Save coverage data to the coverage cache.

    def save(self):

        if self.usecache and self.cache:

            self.canonicalize_filenames()

            cache = open(self.cache, 'wb')

            import marshal

            marshal.dump(self.cexecuted, cache)

            cache.close()

    # restore().  Restore coverage data from the coverage cache (if it exists).

    def restore(self):

        self.c = {}

        self.cexecuted = {}

        assert self.usecache

        if os.path.exists(self.cache):

            self.cexecuted = self.restore_file(self.cache)

    def restore_file(self, file_name):

        try:

            cache = open(file_name, 'rb')

            import marshal

            cexecuted = marshal.load(cache)

            cache.close()

            if isinstance(cexecuted, types.DictType):

                return cexecuted

            else:

                return {}

        except:

            return {}

    # collect(). Collect data in multiple files produced by parallel mode

    def collect(self):

        cache_dir, local = os.path.split(self.cache)

        for f in os.listdir(cache_dir or '.'):

            if not f.startswith(local):

                continue

            full_path = os.path.join(cache_dir, f)

            cexecuted = self.restore_file(full_path)

            self.merge_data(cexecuted)

    def merge_data(self, new_data):

        for file_name, file_data in new_data.items():

            if self.cexecuted.has_key(file_name):

                self.merge_file_data(self.cexecuted[file_name], file_data)

            else:

                self.cexecuted[file_name] = file_data

    def merge_file_data(self, cache_data, new_data):

        for line_number in new_data.keys():

            if not cache_data.has_key(line_number):

                cache_data[line_number] = new_data[line_number]

    def abs_file(self, filename):

        """ Helper function to turn a filename into an absolute normalized

            filename.

        """