path: root/git-coverage

#!/usr/bin/env python

import fnmatch
import getopt
import os
import re
import subprocess
import sys
import tempfile

SKIP_PATTERNS = [
        'assert_not_reached',
        'return_val_if_reached',
        'return_if_reached',
        'UNREACHABLE:'
]

def subprocess_lines(argv):
    proc = subprocess.Popen(argv, stdout=subprocess.PIPE)
    while True:
        line = proc.stdout.readline()
        if line != "":
            yield line
        else:
            return

def compile_all_re(patterns):
    regexps = []
    for pattern in patterns:
        regexp = re.compile(pattern)
        regexps.append(regexp)
    return regexps

def search_any_re(regexps, line):
    for regexp in regexps:
        match = regexp.search(line)
        if match:
            return match
    return None

def match_any_re(regexps, line):
    for regexp in regexps:
        match = regexp.match(line)
        if match:
            return match
    return None

def warning(string):
    print >> sys.stderr, string

# ----------------------------------------------------------------------------
# PATCH PARSING
#
# The patch parsing code, heavily modified originated from:
#
# Copyright (C) 2005-2010 Aaron Bentley, Canonical Ltd
# <aaron.bentley@utoronto.ca>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
#

class BadPatch(Exception):
    def __init(self, message):
        self.message = message

    def __str__(self):
        return self.message

class Line:
    def __init__(self, old, new, data):
        self.old = old
        self.new = new
        self.data = data
        self.covered = False

class Hunk:
    def __init__(self, trailer=None, lines=None):
        self.trailer = trailer
        self.lines = lines or []

    def ranges(self):
        orig_pos = 0
        orig_range = 0
        mod_pos = 0
        mod_range = 0

        for line in self.lines:
            if line.old:
                if not orig_pos:
                    orig_pos = line.old
                orig_range += 1
            if line.new:
                if not mod_pos:
                    mod_pos = line.new
                mod_range += 1
        trailer = self.trailer
        if trailer:
            trailer = trailer.strip()
        else:
            trailer = ""
        return "@@ -%d,%d +%d,%d @@" % (orig_pos, orig_range, mod_pos, mod_range)

    @staticmethod
    def _parse_range(textrange):
        tmp = textrange.split(',')
        if len(tmp) == 1:
            pos = tmp[0]
            range = "1"
        else:
            (pos, range) = tmp
        pos = int(pos)
        range = int(range)
        return (pos, range)

    @staticmethod
    def from_header(line):
        matches = re.match(r'\@\@ ([^@]*) \@\@( (.*))?\n', line)
        if matches is None:
            raise BadPatch("Does not match format.", line)
        try:
            (orig, mod) = matches.group(1).split(" ")
        except (ValueError, IndexError), e:
            raise BadPatch(str(e), line)
        if not orig.startswith('-') or not mod.startswith('+'):
            raise BadPatch("Positions don't start with + or -.", line)
        try:
            (orig_pos, orig_range) = Hunk._parse_range(orig[1:])
            (mod_pos, mod_range) = Hunk._parse_range(mod[1:])
        except (ValueError, IndexError), e:
            raise BadPatch(str(e), line)
        if mod_range < 0 or orig_range < 0:
            raise BadPatch("Hunk range is negative", line)
        trailer = matches.group(3)
        return (Hunk(trailer), orig_pos, orig_range, mod_pos, mod_range)

    @staticmethod
    def parse(lines):
        hunk = None
        lines = iter(lines)
        for line in lines:
            if line == "\n":
                if hunk is not None:
                    yield hunk
                    hunk = None
                continue
            if hunk is not None:
                yield hunk
            try:
                (hunk, orig_pos, orig_range, mod_pos, mod_range) = Hunk.from_header(line)
            except BadPatch:
                # If the line isn't a hunk header, then we've reached the end
                # of this patch and there's "junk" at the end.  Ignore the
                # rest of this patch.
                return
            orig_size = 0
            mod_size = 0
            offset = mod_pos
            while orig_size < orig_range or mod_size < mod_range:
                hunk_line = lines.next()
                if hunk_line.startswith("-"):
                    hunk.lines.append(Line(offset, 0, hunk_line))
                    orig_size += 1
                elif hunk_line.startswith("\n") or hunk_line.startswith(" "):
                    hunk.lines.append(Line(offset, offset, hunk_line))
                    orig_size += 1
                    mod_size += 1
                    offset += 1
                elif hunk_line.startswith("+"):
                    hunk.lines.append(Line(0, offset, hunk_line))
                    mod_size += 1
                    offset += 1
                elif hunk_line.startswith("\\"):
                    pass # No newline at end of file
                else:
                    raise BadPatch("Unknown line type: %s" % hunk_line)
        if hunk is not None:
            yield hunk


class Patch(object):
    BINARY_FILES_RE = re.compile('Binary files (.*) and (.*) differ\n')

    def __init__(self, oldname, newname):
        self.oldname = oldname.split("/", 1)[1]
        self.newname = newname.split("/", 1)[1]
        self.prefix = []
        self.hunks = []

    @staticmethod
    def parse_one(lines):
        try:
            first = lines.next()
            if not first.startswith("--- "):
                raise BadPatch("No orig name: %s" % first)
            else:
                orig_name = first[4:].rstrip("\n")
        except StopIteration:
            raise BadPatch("No orig line")
        try:
            second = lines.next()
            if not second.startswith("+++ "):
                raise BadPatch("No mod name")
            else:
                mod_name = second[4:].rstrip("\n")
        except StopIteration:
            raise BadPatch("No mod line")

        patch = Patch(orig_name, mod_name)
        for hunk in Hunk.parse(lines):
            patch.hunks.append(hunk)
        patch.prefix = [first, second]
        return patch

    @staticmethod
    def parse(lines):
        saved_lines = []
        orig_range = 0
        beginning = True
        for line in lines:
            if Patch.BINARY_FILES_RE.match(line):
                continue
            if line.startswith('=== ') or line.startswith('*** '):
                continue
            if line.startswith('#'):
                continue
            elif orig_range > 0:
                if line.startswith('-') or line.startswith(' '):
                    orig_range -= 1
            elif line.startswith('--- '):
                if beginning:
                    # Patches can have "junk" at the beginning
                    # Stripping junk from the end of patches is handled when we
                    # parse the patch
                    beginning = False
                elif len(saved_lines) > 0:
                    yield Patch.parse_one(iter(saved_lines))
                saved_lines = []
            elif line.startswith('@@'):
                (hunk, orig_pos, orig_range, mod_pos, mod_range) = Hunk.from_header(line)
            saved_lines.append(line)
        if len(saved_lines) > 0:
            yield Patch.parse_one(iter(saved_lines))


# ----------------------------------------------------------------------------
# COVERAGE PARSERS

class GccCoverage:
    extensions = [".c", ".cpp", ".cc"]

    def __init__(self, skips):
        self._gcno_cache = { }
        self._gcno_unresolved = [ ]
        self._creating_re = re.compile("^.*'(.+\.gcov)'$")
        self._file_re = re.compile("^File.*'(.+)'$")
        self._skips = skips
        self._trailer_res = compile_all_re([
            # C/++ functions/methods at top level
            r"^\b[\w_]+\b(\s*::)?[\s*]*\([\w_*\s,\)\{]*$",
            # compound type at top level
            r"^(struct|class|enum)[^;]*$"
        ])

        def visit(paths, dirname, names):
            for name in names:
                path = os.path.normpath(os.path.join(dirname, name))
                if os.path.isdir(path):
                    continue
                if not fnmatch.fnmatch(name, "*.gcno"):
                    continue
                paths.append(os.path.abspath(path))
        os.path.walk(".", visit, self._gcno_unresolved)

    def _directory_gcno_compiled_in(self, gcno):
        (directory, base) = os.path.split(gcno)

        # libtool always gets it's grubby little fingers involved
        (parent, last) = os.path.split(directory)
        if last == ".libs":
            return parent

        return directory

    def _add_to_gcno_cache(self, gcno, source):
        if source not in self._gcno_cache:
            self._gcno_cache[source] = []
        self._gcno_cache[source].append(gcno)

    def _lookup_gcno_files(self, filename):
        source = os.path.abspath(os.path.normpath(filename))

        # Find gcno files that haven't been run through gcov yet
        # Look for likely candidates that match the source file's
        # base name.

        (directory, base) = os.path.split(source)
        (base, ext) = os.path.splitext(base)
        match = "*%s.gcno" % (base, )

        resolve = []
        for gcno in self._gcno_unresolved:
            if fnmatch.fnmatch(gcno, match):
                resolve.append(gcno)

        no_gcda = False
        cmd = ['gcov', '--preserve-paths', '--no-output']
        for gcno in resolve:
            self._gcno_unresolved.remove(gcno)

            # Check if there is a .gcda file for this gcno file
            # If not, then the compilation unit that created the .gcno
            # If we don't find any other run .gcno files then we'll
            # warn about this below, using the flag
            (base, ext) = os.path.splitext(gcno)
            if not os.path.exists(base + ".gcda"):
                no_gcda = True
                continue

            # Run the gcno file through gcov in the --no-output mode
            # which will tell us the source file(s) it represents
            directory = self._directory_gcno_compiled_in(gcno)
            for line in subprocess_lines(cmd + [gcno]):
                match = self._file_re.match(line.strip())
                if not match:
                    continue
                filename = match.group(1)

                # We've found a gcno/source combination, make note of it
                path = os.path.join(directory, filename)
                self._add_to_gcno_cache(gcno, os.path.normpath(path))

        # Now look through our cache of gcno files that have been run
        # through gcov, for gcno files that represent the source file

        matches = []
        bad_mtime = False
        gcnos = self._gcno_cache.get(source, [])
        mtime = os.path.getmtime(source)
        for gcno in gcnos:

            # If the source file has been modified later than the
            # gcno file this is an indication of not being built
            # correctly, so get ready to complain about that
            if os.path.getctime(gcno) < mtime:
                bad_mtime = True
                continue

            matches.append(gcno)

        if not matches:
            if bad_mtime:
                warning("%s: Found old coverage data, likely not built" % filename)
            elif no_gcda:
                warning("%s: No gcda coverage data found, likely not run" % filename)
            else:
                warning("%s: Found no coverage data" % filename)

        return matches

    def _gcov_lines_for_files(self, filename):
        # We scrape the output of the command for the names of the
        # gcov files created, which we process, and then remove

        for gcno in self._lookup_gcno_files(filename):
            # Need to run gcov in the same directory compiled in
            directory = self._directory_gcno_compiled_in(gcno)
            oldcwd = os.getcwd()
            os.chdir(directory)

            gcovs = []
            cmd = ['gcov', '--preserve-paths']
            for line in subprocess_lines(cmd + [gcno]):
                match = self._creating_re.match(line.strip())
                if not match:
                    continue
                gcov = match.group(1)
                if os.path.exists(gcov):
                    gcovs.append(os.path.abspath(gcov))

            # Because we change the directory, we have to take care not
            # to yield while the current directory is changed

            if oldcwd:
                os.chdir(oldcwd)

            for gcov in gcovs:
                with open(gcov, 'r') as f:
                    for l in f:
                        yield l
                    os.unlink(gcov)

    def coverage(self, filename):
        coverage = { }
        for line in self._gcov_lines_for_files(filename):
            # Each gcov coverage output line looks something like this
            #  coverage:   lineno:  remainder is actual line content
            parts = line.split(':', 2)
            if len(parts) != 3:
                continue

            covered = parts[0].strip()
            try:
                no = int(parts[1].strip())
                if covered == '-':
                    count = 0
                else:
                    count = int(covered)
                coverage[no] = parts[2]
            except ValueError:
                if search_any_re(self._skips, parts[2]):
                    coverage[no] = parts[2]
        return coverage

    def trailer(self, string):
        match = match_any_re(self._trailer_res, string)
        return match and match.group(0) or None

    def usage(self, output):
        string = """GCC gcov C code coverage

             Used with: %s

             The program should be (re)built with the specicial GCC options
             '-fprofile-arcs -ftest-coverage'. Run the C applications as you
             normally would to create test coverage data.
        """

        message = string % ", ".join(self.extensions)
        message = message.replace("\t", "")
        output.write(message)


class PythonCoverage:
    extensions = [".py"]

    def __init__(self, skips):
        self._temp_dir = tempfile.mkdtemp(prefix='git-coverage')
        self._skips = skips
        self._trailer_re = re.compile("^[ \t]*((class|def)[ \t].*)$")

    def __del__(self):
        for path in self._list_files():
            os.unlink(path)
        os.rmdir(self._temp_dir)

    def _list_files(self):
        for name in os.listdir(self._temp_dir):
            if fnmatch.fnmatch(name, "*,cover"):
                yield os.path.join(self._temp_dir, name)

    def _read_coverage(self, filename):
        coverage = { }
        no = 1
        for line in open(filename, 'r'):
            if not line.startswith("!"):
                coverage[no] = line
            elif search_any_re(self._skips, line[1:]):
                coverage[no] = line
            no += 1
        return coverage

    def coverage(self, filename):
        cmd = ["coverage", "annotate", "--directory", self._temp_dir, filename]
        subprocess.check_call(cmd)

        coverage = { }
        base = os.path.basename(filename)

        for path in self._list_files():
            if not coverage and fnmatch.fnmatch(path, "*%s*" % base):
                coverage = self._read_coverage(path)
            os.unlink(path)

        return coverage

    def trailer(self, string):
        match = self._trailer_re.match(string)
        return match and match.group(0) or None

    def usage(self, output):
        string = """Python code coverage

             Used with: %s

             This requires the python-coverage module. The program should be
             run with 'coverage run my_program.py' which produces test coverage
             data in the current directory.
        """

        message = string % ", ".join(self.extensions)
        message = message.replace("\t", "")
        output.write(message)


class Output:
    defaults = {
            'diff.new': 'green',
            'diff.meta': 'bold',
            'diff.plain': 'normal',
            'diff.frag': 'cyan',
            'diff.old': 'red',
            'diff.whitespace': 'normal red',
            'diff.problem': 'normal red'
    }

    def __init__(self, output, with_colors, context=3, cover_context=False):
        self.output = output
        self.escapes = { }
        self.context = context
        self.cover_context = cover_context

        if with_colors is None:
            cmd = ['git', 'config', '--get-colorbool',
                   'color.diff', output.isatty() and 'true' or 'false']
            self.with_colors = subprocess.check_output(cmd).strip() == 'true'
        else:
            self.with_colors = with_colors

    def write_meta(self, data, meta):
        if not self.with_colors:
            pass
        elif not meta:
            self.output.write('\033[0m')
        elif meta in self.escapes:
            self.output.write(self.escapes[meta])
        else:
            default = self.defaults.get(meta, 'normal')
            cmd = ['git', 'config', '--get-color', "color.%s" % meta, default]
            escape = subprocess.check_output(cmd)
            self.output.write(escape)
            self.escapes[meta] = escape
        self.write(data)
        if self.with_colors:
            self.output.write('\033[0m')

    def write(self, data):
        self.output.write(data)


# ----------------------------------------------------------------------------

def prepare_uncovered_hunks(ihunk, parser, coverage, output):
    # for line in ihunk.lines:
    #     line.covered = False
    # yield ihunk
    # return

    context = output.context

    lines = []
    since = 0
    have = False
    trailer = ihunk.trailer

    for line in ihunk.lines:

        # We look for function name frag trailers for each line so that
        # can use them for chunks that start after this line
        line.trailer = parser.trailer(line.data[1:])
        line.covered = True

        # In cover context mode we check all the lines in
        # the patch for coverage, regardless of whether they
        # were added, removed or not touched
        if output.cover_context:
            if line.old and line.old not in coverage:
                line.covered = False
            elif line.new and line.new not in coverage:
                line.covered = False

        # In the default mode we only check new lines for coverage
        else:
            if not line.old and line.new and line.new not in coverage:
                line.covered = False

        lines.append(line)

        if not line.covered:
            # If a line is uncovered then add, and start counting the lines
            # that come after this one as trailing context
            since = 0
            have = True

        elif have:
            since += 1

            # So once we have more than twice the trailing context as necessary
            # then we break this off into its own chuck, only consuming as
            # half of the trailing context lines added.
            if since > context * 2:
                pos = len(lines) - (since - context)
                hunk = Hunk(trailer, lines[0:pos])

                # Choose a function name frag from within this hunk
                # for the next hunk, if we found a new one for a certain line
                for line in hunk.lines:
                    if line.trailer:
                        trailer = line.trailer

                yield hunk
                lines = lines[pos:]
                since = 0
                have = False

        if not have:
            # If there are too many prefix context lines, then go ahead and
            # drop one, looking for the function frag name to use for next hunk
            if len(lines) > context:
                drop = lines.pop(0)
                if drop.trailer:
                    trailer = drop.trailer

    if have:
        if since > context:
            pos = len(lines) - (since - context)
            hunk = Hunk(trailer, lines[0:pos])
        else:
            hunk = Hunk(trailer, lines)
        yield hunk

def print_hunk_lines(hunk, coverage, output):
    output.write_meta(hunk.ranges(), 'diff.frag')
    output.write_meta(" %s\n" % (hunk.trailer or "").strip(), 'diff.plain')
    for line in hunk.lines:
        if line.covered:
            output.write(" ")
        else:
            output.write_meta("!", 'diff.problem')
        if not line.new:
            output.write_meta(line.data, 'diff.old')
        elif not line.old:
            output.write_meta(line.data, 'diff.new')
        else:
            output.write_meta(line.data, 'diff.plain')

def print_patch_hunks(patch, hunks, coverage, output):
    for line in patch.prefix:
        output.write_meta(line, 'diff.meta')
    for hunk in hunks:
        print_hunk_lines(hunk, coverage, output)

def usage(parsers, output=sys.stdout):
    string = """usage: git coverage [diff-options] commit

    Shows the code coverage for code changed between the specified commit and
    the latest code. Use standard git diff options to specify which commits
    to include in the code.
    """
    message = string.replace("\t", "")
    output.write(message)
    for parser in parsers:
        output.write("\n")
        parser.usage(output=output)

def getopt_keep_dashes(argv, short, long):
    # Split off everything after the -- argument
    try:
        double_dash = argv.index("--", 1)
        before_dashes = argv[0:double_dash]
        after_dashes = argv[double_dash:]
    except ValueError:
        before_dashes = argv
        after_dashes = []

    opts, args = getopt.getopt(before_dashes, short, long)
    args += after_dashes
    return opts, args

def main(argv):
    skips = compile_all_re(SKIP_PATTERNS)

    parsers = (
            GccCoverage(skips),
            PythonCoverage(skips)
    )

    short = "abC:G:l:pO:M:S:uU:W"

    long = (
        "help",
        "cover-context",

        "color",
        "color=",
        "diff-filter=",
        "exit-code",
        "find-renames",
        "find-renames=",
        "find-copies",
        "find-copies=",
        "find-copies-hander",
        "function-context",
        "histogram",
        "ignore-all-space",
        "ignore-space-at-eol",
        "ignore-space-change",
        "inter-hunk-context",
        "minimal",
        "no-color",
        "patch",
        "patience",
        "pickaxe-all",
        "pickaxe-regex",
        "unified=",
        "text",
    )

    try:
        opts, args = getopt_keep_dashes(argv[1:], short, long)
    except getopt.GetoptError as err:
        print str(err)
        return 2

    have_target = False
    context = 3
    cover_context = False
    with_colors = None

    cmd = [
        'git', '--no-pager', 'diff', '--relative', '--no-color',
    ]

    for o, a in opts:
        # git-coverage specific options
        if o in ('-h', '--help'):
            usage(parsers)
            return 0
        elif o in ("--cover-context"):
            cover_context = True
        elif o in ("--color"):
            if not a or a in ("always"):
                with_colors = True
            elif a in ("never"):
                with_colors = False
            elif a in ("auto"):
                with_colors = None
            continue
        elif o in ("--no-color"):
            with_colors = False
            continue

        # Take note of these, and then pass through
        if o in ('-U', "--unified"):
            context = int(a)
        elif o in ("-W", "--function-context"):
            context = 1024 * 1024 * 1024  # Big but still usable in calculations

        # Pass through all other options
        if not a:
            cmd += [o]
        else:
            cmd += [o, a]

    if args:
        cmd += args
    else:
        cmd += ['HEAD']

    output = Output(sys.stdout, with_colors, context, cover_context)

    printed_any = 0
    patches_by_filename = { }

    # Pull all the patches appart into the hunks that we need
    for patch in Patch.parse(subprocess_lines(cmd)):
        filename = os.path.normpath(patch.newname)
        if filename not in patches_by_filename:
            patches_by_filename[filename] = []
        patches_by_filename[filename].append(patch)

    # Now go through and calculate coverage
    for (filename, patches) in patches_by_filename.items():
        (name, ext) = os.path.splitext(filename)
        for parser in parsers:
            if ext in parser.extensions:
                coverage = parser.coverage(filename)
                break
        else:
            continue

        for patch in patches:
            to_print = []
            for ihunk in patch.hunks:
                for hunk in prepare_uncovered_hunks(ihunk, parser, coverage, output):
                    to_print.append(hunk)
            if to_print:
                print_patch_hunks(patch, to_print, coverage, output)
                printed_any = 1

    return printed_any


if __name__ == '__main__':
    sys.exit(main(sys.argv))