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:'
]

GIT_DIFF = [
        'git', 'diff', '--relative'
]

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

def match_any_re(regexps, line):
    for regexp in regexps:
        if regexp.search(line.strip()):
            return True
    return False

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 Hunk:
    def __init__(self, orig_pos, orig_range, mod_pos, mod_range, header=None):
        self.orig_pos = orig_pos
        self.orig_range = orig_range
        self.mod_pos = mod_pos
        self.mod_range = mod_range
        self.header = header
        self.lines = []

    @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)
        tail = matches.group(3)
        return Hunk(orig_pos, orig_range, mod_pos, mod_range, line)

    @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 = 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 = hunk.mod_pos
            while orig_size < hunk.orig_range or mod_size < hunk.mod_range:
                hunk_line = lines.next()
                if hunk_line.startswith("-"):
                    orig_size += 1
                elif hunk_line.startswith("\n"):
                    orig_size += 1
                    mod_size += 1
                elif hunk_line.startswith(" "):
                    orig_size += 1
                    mod_size += 1
                elif hunk_line.startswith("+"):
                    mod_size += 1
                else:
                    raise BadPatch("Unknown line type: %s" % hunk_line)
                hunk.lines.append((offset, hunk_line))
                if not hunk_line.startswith("-"):
                    offset += 1
        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 = Hunk.from_header(line)
                orig_range = hunk.orig_range
            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._creating_re = re.compile("^.*'(.+\.gcov)'$")
        self._file_re = re.compile("^File.*'(.+)'$")
        self._skips = skips

        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
                # Skip if no gcda file for this gcno file
                (base, ext) = os.path.splitext(path)
                if os.path.exists(base + ".gcda"):
                    paths.append(path)
        os.path.walk(".", visit, self._gcno_cache)

    def _match_gcno_files(self, filename):
        matches = []
        (directory, base) = os.path.split(filename)
        (base, ext) = os.path.splitext(base)
        if directory:
            match = "%s/*%s.gcno" % (directory, base)
        else:
            match = "*%s.gcno" % (base, )
        bad_mtime = False
        mtime = os.path.getmtime(filename)
        for gcno in self._gcno_cache:
            if fnmatch.fnmatch(gcno, match):
                if os.path.getctime(gcno) < mtime:
                    bad_mtime = True
                else:
                    matches.append(gcno)

        if not matches:
            if bad_mtime:
                warning("%s: Found old coverage data, likely not built" % filename)
            else:
                warning("%s: Found no coverage data" % filename)
        return matches

    def _find_directory_gcno_compiled_in(self, gcno, filename):
        cmd = ['gcov', '--preserve-paths', '--relative-only', '--no-output']
        for line in subprocess_lines(cmd + gcno):
            match = self._file_re.match(line.strip())
            if not match:
                continue
            expected = match.group(1)
            if filename.endswith(expected):
                extra = filename[:-len(expected)]
                if os.path.exists(extra):
                    return extra
            elif expected.endswith(filename):
                extra = expected[:-len(filename)]
                up = "../" * len(extra.strip(os.sep).split(os.sep))
                if os.path.exists(up):
                    return up
        return None

    def _gcov_lines_for_files(self, filename):
        gcno = self._match_gcno_files(filename)
        if not gcno:
            return

        absgcno = [os.path.abspath(path) for path in gcno]

        # gcov wants to be in the directory that gcc was executed
        # from. We don't know which directory that is. So we run
        # gcov once to figure out the file path it thinks the source
        # is at.

        directory = self._find_directory_gcno_compiled_in(absgcno, filename)

        oldcwd = None
        if directory:
            oldcwd = os.getcwd()
            os.chdir(directory)

        # We scrape the output of the command for the names of the
        # gcov files created, which we process, and then remove
        gcovs = []

        cmd = ['gcov', '--preserve-paths', '--relative-only']
        for line in subprocess_lines(cmd + absgcno):
            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 match_any_re(self._skips, parts[2]):
                    coverage[no] = parts[2]
        return coverage

    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

    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 match_any_re(self._skips, line[1:]):
                covarege[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,cover" % base):
                coverage = self._read_coverage(path)
            os.unlink(path)

        return coverage

    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):
        self.output = output
        self.escapes = { }

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

    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 print_patch_hunks(patch, hunks, coverage, output):
    for line in patch.prefix:
        output.write_meta(line, 'diff.meta')
    for hunk in hunks:
        output.write_meta(hunk.header, 'diff.frag')
        for (no, line) in hunk.lines:
            if no in coverage:
                output.write(" ")
            else:
                output.write_meta("!", 'diff.problem')
            if line.startswith("-"):
                output.write_meta(line, 'diff.old')
            elif line.startswith("+"):
                output.write_meta(line, 'diff.new')
            else:
                output.write_meta(line, 'diff.plain')

def is_hunk_covered(hunk, coverage):
    for (no, contents) in hunk.lines:
        if no not in coverage:
            return False
    return True


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 main(argv):
    skips = [re.compile(p) for p in SKIP_PATTERNS]

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

    have_target = False
    for arg in argv[1:]:
        if arg in ('-h', '--help'):
            usage(parsers)
            return 0

        if not arg.startswith("-"):
            have_target = True
            break

    cmd = GIT_DIFF + argv[1:]
    if not have_target:
        cmd += ['HEAD']

    output = Output(sys.stdout)

    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 hunk in patch.hunks:
                if not is_hunk_covered(hunk, coverage):
                    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))