#!/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.

"""Usage:

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

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

    coverage data.

coverage.py -e

    Erase collected coverage data.

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.5.20051204"    # see detailed history at the end of this file.

import compiler

import compiler.visitor

import os

import re

import string

import sys

import threading

import types

# 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):

    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):

        self.recordNodeLine(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.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 = visitDiscard = visitPrint = \

        visitPrintnl = visitRaise = visitSubscript = visitDecorators = \

        doStatement

    def recordNodeLine(self, node):

        return self.recordLine(node.lineno)

    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 exluded 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)

    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 visitWhile(self, node):

        self.doSuite(node, node.body)

        self.doElse(node.body, 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 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 coverage:

    error = "coverage error"

    # 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 self.error, "Only one coverage object allowed."

        self.usecache = 1

        self.cache = None

        self.exclude_re = ''

        self.nesting = 0

        self.cstack = []

        self.xstack = []

        self.relative_dir = os.path.normcase(os.path.abspath(os.curdir)+os.path.sep)

    # 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, a):                                   #pragma: no cover

        #print w, f.f_code.co_filename, f.f_lineno

        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):

        if error:

            print error

            print

        print __doc__

        sys.exit(1)

    def command_line(self):

        import getopt

        settings = {}

        optmap = {

            '-a': 'annotate',

            '-d:': 'directory=',

            '-e': 'erase',

            '-h': 'help',

            '-i': 'ignore-errors',

            '-m': 'show-missing',

            '-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(sys.argv[1:], 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:

                self.help("Unknown option: '%s'." % o)

        if settings.get('help'):

            self.help()

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

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

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

                    self.help("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 args_needed

        if not action:

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

        if not args_needed and args:

            self.help("Unexpected arguments %s." % args)

        self.get_ready()

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

        if settings.get('erase'):

            self.erase()

        if settings.get('execute'):

            if not args:

                self.help("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 not args:

            args = self.cexecuted.keys()

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

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

        directory = settings.get('directory')

        omit = filter(None, settings.get('omit', '').split(','))

        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):

        self.usecache = usecache

    def get_ready(self):

        if self.usecache and not self.cache:

            self.cache = os.path.abspath(os.environ.get(self.cache_env,

                                                        self.cache_default))

            self.restore()

        self.analysis_cache = {}

    def start(self):

        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.c = {}

        self.analysis_cache = {}

        self.cexecuted = {}

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

            os.remove(self.cache)

        self.exclude_re = ""

    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):

        # move to directory that must exist.

        os.chdir(os.sep)

        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 not os.path.exists(self.cache):

            return

        try:

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

            import marshal

            cexecuted = marshal.load(cache)

            cache.close()

            if isinstance(cexecuted, types.DictType):

                self.cexecuted = cexecuted

        except:

            pass

    # canonical_filename(filename).  Return a canonical filename for the

    # file (that is, an absolute path with no redundant components and

    # normalized case).  See [GDR 2001-12-04b, 3.3].

    def canonical_filename(self, filename):

        if not self.canonical_filename_cache.has_key(filename):

            f = filename

            if os.path.isabs(f) and not os.path.exists(f):

                f = os.path.basename(f)

            if not os.path.isabs(f):

                for path in [os.curdir] + sys.path:

                    g = os.path.join(path, f)

                    if os.path.exists(g):

                        f = g

                        break

            cf = os.path.normcase(os.path.abspath(f))

            self.canonical_filename_cache[filename] = cf

        return self.canonical_filename_cache[filename]

    # canonicalize_filenames().  Copy results from "c" to "cexecuted", 

    # canonicalizing filenames on the way.  Clear the "c" map.

    def canonicalize_filenames(self):

        for filename, lineno in self.c.keys():

            f = self.canonical_filename(filename)

            if not self.cexecuted.has_key(f):

                self.cexecuted[f] = {}

            self.cexecuted[f][lineno] = 1

        self.c = {}

    # morf_filename(morf).  Return the filename for a module or file.

    def morf_filename(self, morf):

        if isinstance(morf, types.ModuleType):

            if not hasattr(morf, '__file__'):

                raise self.error, "Module has no __file__ attribute."

            file = morf.__file__

        else:

            file = morf

        return self.canonical_filename(file)

    # analyze_morf(morf).  Analyze the module or filename passed as

    # the argument.  If the source code can't be found, raise an error.

    # Otherwise, return a tuple of (1) the canonical filename of the

    # source code for the module, (2) a list of lines of statements

    # in the source code, and (3) a list of lines of excluded statements.

    def analyze_morf(self, morf):

        if self.analysis_cache.has_key(morf):

            return self.analysis_cache[morf]

        filename = self.morf_filename(morf)

        ext = os.path.splitext(filename)[1]

        if ext == '.pyc':

            if not os.path.exists(filename[0:-1]):

                raise self.error, ("No source for compiled code '%s'."

                                   % filename)

            filename = filename[0:-1]

        elif ext != '.py':

            raise self.error, "File '%s' not Python source." % filename

        source = open(filename, 'r')

        lines, excluded_lines = self.find_executable_statements(

            source.read(), exclude=self.exclude_re

            )

        source.close()

        result = filename, lines, excluded_lines

        self.analysis_cache[morf] = result

        return result

    def get_suite_spots(self, tree, spots):

        import symbol, token

        for i in range(1, len(tree)):

            if type(tree[i]) == type(()):

                if tree[i][0] == symbol.suite:

                    # Found a suite, look back for the colon and keyword.

                    lineno_colon = lineno_word = None

                    for j in range(i-1, 0, -1):

                        if tree[j][0] == token.COLON:

                            lineno_colon = tree[j][2]

                        elif tree[j][0] == token.NAME:

                            if tree[j][1] == 'elif':

                                # Find the line number of the first non-terminal

                                # after the keyword.

                                t = tree[j+1]

                                while t and token.ISNONTERMINAL(t[0]):

                                    t = t[1]

                                if t:

                                    lineno_word = t[2]

                            else:

                                lineno_word = tree[j][2]

                            break

                        elif tree[j][0] == symbol.except_clause:

                            # "except" clauses look like:

                            # ('except_clause', ('NAME', 'except', lineno), ...)

                            if tree[j][1][0] == token.NAME: