path: root/build/scripts/coverage-info.py

import argparse
import os
import sys
import tarfile
import collections
import subprocess
import re


GCDA_EXT = '.gcda'
GCNO_EXT = '.gcno'


def suffixes(path):
    """
    >>> list(suffixes('/a/b/c'))
    ['c', 'b/c', '/a/b/c']
    >>> list(suffixes('/a/b/c/'))
    ['c', 'b/c', '/a/b/c']
    >>> list(suffixes('/a'))
    ['/a']
    >>> list(suffixes('/a/'))
    ['/a']
    >>> list(suffixes('/'))
    []
    """
    path = os.path.normpath(path)

    def up_dirs(cur_path):
        while os.path.dirname(cur_path) != cur_path:
            cur_path = os.path.dirname(cur_path)
            yield cur_path

    for x in up_dirs(path):
        yield path.replace(x + os.path.sep, '')


def recast(in_file, out_file, probe_path, update_stat):
    PREFIX = 'SF:'

    probed_path = None

    any_payload = False

    with open(in_file, 'r') as input, open(out_file, 'w') as output:
        active = True
        for line in input:
            line = line.rstrip('\n')
            if line.startswith('TN:'):
                output.write(line + '\n')
            elif line.startswith(PREFIX):
                path = line[len(PREFIX) :]
                probed_path = probe_path(path)
                if probed_path:
                    output.write(PREFIX + probed_path + '\n')
                active = bool(probed_path)
            else:
                if active:
                    update_stat(probed_path, line)
                    output.write(line + '\n')
                    any_payload = True

    return any_payload


def print_stat(da, fnda, teamcity_stat_output):
    lines_hit = sum(map(bool, da.values()))
    lines_total = len(da.values())
    lines_coverage = 100.0 * lines_hit / lines_total if lines_total else 0

    func_hit = sum(map(bool, fnda.values()))
    func_total = len(fnda.values())
    func_coverage = 100.0 * func_hit / func_total if func_total else 0

    print >> sys.stderr, '[[imp]]Lines[[rst]]     {: >16} {: >16} {: >16.1f}%'.format(
        lines_hit, lines_total, lines_coverage
    )
    print >> sys.stderr, '[[imp]]Functions[[rst]] {: >16} {: >16} {: >16.1f}%'.format(
        func_hit, func_total, func_coverage
    )

    if teamcity_stat_output:
        with open(teamcity_stat_output, 'w') as tc_file:
            tc_file.write("##teamcity[blockOpened name='Code Coverage Summary']\n")
            tc_file.write(
                "##teamcity[buildStatisticValue key=\'CodeCoverageAbsLTotal\' value='{}']\n".format(lines_total)
            )
            tc_file.write(
                "##teamcity[buildStatisticValue key=\'CodeCoverageAbsLCovered\' value='{}']\n".format(lines_hit)
            )
            tc_file.write(
                "##teamcity[buildStatisticValue key=\'CodeCoverageAbsMTotal\' value='{}']\n".format(func_total)
            )
            tc_file.write(
                "##teamcity[buildStatisticValue key=\'CodeCoverageAbsMCovered\' value='{}']\n".format(func_hit)
            )
            tc_file.write("##teamcity[blockClosed name='Code Coverage Summary']\n")


def chunks(l, n):
    """
    >>> list(chunks(range(10), 3))
    [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9]]
    >>> list(chunks(range(10), 5))
    [[0, 1, 2, 3, 4], [5, 6, 7, 8, 9]]
    """
    for i in xrange(0, len(l), n):
        yield l[i : i + n]


def combine_info_files(lcov, files, out_file):
    chunk_size = 50
    files = list(set(files))

    for chunk in chunks(files, chunk_size):
        combine_cmd = [lcov]
        if os.path.exists(out_file):
            chunk.append(out_file)
        for trace in chunk:
            assert os.path.exists(trace), "Trace file does not exist: {} (cwd={})".format(trace, os.getcwd())
            combine_cmd += ["-a", os.path.abspath(trace)]
        print >> sys.stderr, '## lcov', ' '.join(combine_cmd[1:])
        out_file_tmp = "combined.tmp"
        with open(out_file_tmp, "w") as stdout:
            subprocess.check_call(combine_cmd, stdout=stdout)
        if os.path.exists(out_file):
            os.remove(out_file)
        os.rename(out_file_tmp, out_file)


def probe_path_global(path, source_root, prefix_filter, exclude_files):
    if path.endswith('_ut.cpp'):
        return None

    for suff in reversed(list(suffixes(path))):
        if (not prefix_filter or suff.startswith(prefix_filter)) and (
            not exclude_files or not exclude_files.match(suff)
        ):
            full_path = source_root + os.sep + suff
            if os.path.isfile(full_path):
                return full_path

    return None


def update_stat_global(src_file, line, fnda, da):
    if line.startswith("FNDA:"):
        visits, func_name = line[len("FNDA:") :].split(',')
        fnda[src_file + func_name] += int(visits)

    if line.startswith("DA"):
        line_number, visits = line[len("DA:") :].split(',')
        if visits == '=====':
            visits = 0

        da[src_file + line_number] += int(visits)


def gen_info_global(cmd, cov_info, probe_path, update_stat, lcov_args):
    print >> sys.stderr, '## geninfo', ' '.join(cmd)
    subprocess.check_call(cmd)
    if recast(cov_info + '.tmp', cov_info, probe_path, update_stat):
        lcov_args.append(cov_info)


def init_all_coverage_files(
    gcno_archive, fname2gcno, fname2info, geninfo_executable, gcov_tool, gen_info, prefix_filter, exclude_files
):
    with tarfile.open(gcno_archive) as gcno_tf:
        for gcno_item in gcno_tf:
            if gcno_item.isfile() and gcno_item.name.endswith(GCNO_EXT):
                gcno_tf.extract(gcno_item)

                gcno_name = gcno_item.name
                source_fname = gcno_name[: -len(GCNO_EXT)]
                if prefix_filter and not source_fname.startswith(prefix_filter):
                    sys.stderr.write("Skipping {} (doesn't match prefix '{}')\n".format(source_fname, prefix_filter))
                    continue
                if exclude_files and exclude_files.search(source_fname):
                    sys.stderr.write(
                        "Skipping {} (matched exclude pattern '{}')\n".format(source_fname, exclude_files.pattern)
                    )
                    continue

                fname2gcno[source_fname] = gcno_name

                if os.path.getsize(gcno_name) > 0:
                    coverage_info = source_fname + '.' + str(len(fname2info[source_fname])) + '.info'
                    fname2info[source_fname].append(coverage_info)
                    geninfo_cmd = [
                        geninfo_executable,
                        '--gcov-tool',
                        gcov_tool,
                        '-i',
                        gcno_name,
                        '-o',
                        coverage_info + '.tmp',
                    ]
                    gen_info(geninfo_cmd, coverage_info)


def process_all_coverage_files(gcda_archive, fname2gcno, fname2info, geninfo_executable, gcov_tool, gen_info):
    with tarfile.open(gcda_archive) as gcda_tf:
        for gcda_item in gcda_tf:
            if gcda_item.isfile() and gcda_item.name.endswith(GCDA_EXT):
                gcda_name = gcda_item.name
                source_fname = gcda_name[: -len(GCDA_EXT)]
                for suff in suffixes(source_fname):
                    if suff in fname2gcno:
                        gcda_new_name = suff + GCDA_EXT
                        gcda_item.name = gcda_new_name
                        gcda_tf.extract(gcda_item)
                        if os.path.getsize(gcda_new_name) > 0:
                            coverage_info = suff + '.' + str(len(fname2info[suff])) + '.info'
                            fname2info[suff].append(coverage_info)
                            geninfo_cmd = [
                                geninfo_executable,
                                '--gcov-tool',
                                gcov_tool,
                                gcda_new_name,
                                '-o',
                                coverage_info + '.tmp',
                            ]
                            gen_info(geninfo_cmd, coverage_info)


def gen_cobertura(tool, output, combined_info):
    cmd = [tool, combined_info, '-b', '#hamster#', '-o', output]
    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    out, err = p.communicate()
    if p.returncode:
        raise Exception(
            'lcov_cobertura failed with exit code {}\nstdout: {}\nstderr: {}'.format(p.returncode, out, err)
        )


def main(
    source_root,
    output,
    gcno_archive,
    gcda_archive,
    gcov_tool,
    prefix_filter,
    exclude_regexp,
    teamcity_stat_output,
    coverage_report_path,
    gcov_report,
    lcov_cobertura,
):
    exclude_files = re.compile(exclude_regexp) if exclude_regexp else None

    fname2gcno = {}
    fname2info = collections.defaultdict(list)
    lcov_args = []
    geninfo_executable = os.path.join(source_root, 'devtools', 'lcov', 'geninfo')

    def probe_path(path):
        return probe_path_global(path, source_root, prefix_filter, exclude_files)

    fnda = collections.defaultdict(int)
    da = collections.defaultdict(int)

    def update_stat(src_file, line):
        update_stat_global(src_file, line, da, fnda)

    def gen_info(cmd, cov_info):
        gen_info_global(cmd, cov_info, probe_path, update_stat, lcov_args)

    init_all_coverage_files(
        gcno_archive, fname2gcno, fname2info, geninfo_executable, gcov_tool, gen_info, prefix_filter, exclude_files
    )
    process_all_coverage_files(gcda_archive, fname2gcno, fname2info, geninfo_executable, gcov_tool, gen_info)

    if coverage_report_path:
        output_dir = coverage_report_path
    else:
        output_dir = output + '.dir'

    if not os.path.exists(output_dir):
        os.makedirs(output_dir)

    teamcity_stat_file = None
    if teamcity_stat_output:
        teamcity_stat_file = os.path.join(output_dir, 'teamcity.out')
    print_stat(da, fnda, teamcity_stat_file)

    if lcov_args:
        output_trace = "combined.info"
        combine_info_files(os.path.join(source_root, 'devtools', 'lcov', 'lcov'), lcov_args, output_trace)
        cmd = [
            os.path.join(source_root, 'devtools', 'lcov', 'genhtml'),
            '-p',
            source_root,
            '--ignore-errors',
            'source',
            '-o',
            output_dir,
            output_trace,
        ]
        print >> sys.stderr, '## genhtml', ' '.join(cmd)
        subprocess.check_call(cmd)
        if lcov_cobertura:
            gen_cobertura(lcov_cobertura, gcov_report, output_trace)

    with tarfile.open(output, 'w') as tar:
        tar.add(output_dir, arcname='.')


if __name__ == '__main__':
    parser = argparse.ArgumentParser()

    parser.add_argument('--source-root', action='store')
    parser.add_argument('--output', action='store')
    parser.add_argument('--gcno-archive', action='store')
    parser.add_argument('--gcda-archive', action='store')
    parser.add_argument('--gcov-tool', action='store')
    parser.add_argument('--prefix-filter', action='store')
    parser.add_argument('--exclude-regexp', action='store')
    parser.add_argument('--teamcity-stat-output', action='store_const', const=True)
    parser.add_argument('--coverage-report-path', action='store')
    parser.add_argument('--gcov-report', action='store')
    parser.add_argument('--lcov-cobertura', action='store')

    args = parser.parse_args()
    main(**vars(args))