# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
import json
import os
import sys
from argparse
import ArgumentParser
try:
import urlparse
except ImportError:
import urllib.parse
as urlparse
import mozpack.path
as mozpath
from mozpack.chrome.manifest
import parse_manifest
from six
import viewitems
from .manifest_handler
import ChromeManifestHandler
class LcovRecord(object):
__slots__ = (
"test_name",
"source_file",
"functions",
"function_exec_counts",
"function_count",
"covered_function_count",
"branches",
"branch_count",
"covered_branch_count",
"lines",
"line_count",
"covered_line_count",
)
def __init__(self):
self.functions = {}
self.function_exec_counts = {}
self.branches = {}
self.lines = {}
def __iadd__(self, other):
# These shouldn't differ.
self.source_file = other.source_file
if hasattr(other,
"test_name"):
self.test_name = other.test_name
self.functions.update(other.functions)
for name, count
in viewitems(other.function_exec_counts):
self.function_exec_counts[name] = count + self.function_exec_counts.get(
name, 0
)
for key, taken
in viewitems(other.branches):
self.branches[key] = taken + self.branches.get(key, 0)
for line, (exec_count, checksum)
in viewitems(other.lines):
new_exec_count = exec_count
if line
in self.lines:
old_exec_count, _ = self.lines[line]
new_exec_count += old_exec_count
self.lines[line] = new_exec_count, checksum
self.resummarize()
return self
def resummarize(self):
# Re-calculate summaries after generating or splitting a record.
self.function_count = len(self.functions.keys())
# Function records may have moved between files, so filter here.
self.function_exec_counts = {
fn_name: count
for fn_name, count
in viewitems(self.function_exec_counts)
if fn_name
in self.functions.values()
}
self.covered_function_count = len(
[c
for c
in self.function_exec_counts.values()
if c]
)
self.line_count = len(self.lines)
self.covered_line_count = len([c
for c, _
in self.lines.values()
if c])
self.branch_count = len(self.branches)
self.covered_branch_count = len([c
for c
in self.branches.values()
if c])
class RecordRewriter(object):
# Helper class for rewriting/spliting individual lcov records according
# to what the preprocessor did.
def __init__(self):
self._ranges =
None
def _get_range(self, line):
for start, end
in self._ranges:
if line < start:
return None
if line < end:
return start, end
return None
def _get_mapped_line(self, line, r):
inc_source, inc_start = self._current_pp_info[r]
start, end = r
offs = line - start
return inc_start + offs
def _get_record(self, inc_source):
if inc_source
in self._additions:
gen_rec = self._additions[inc_source]
else:
gen_rec = LcovRecord()
gen_rec.source_file = inc_source
self._additions[inc_source] = gen_rec
return gen_rec
def _rewrite_lines(self, record):
rewritten_lines = {}
for ln, line_info
in viewitems(record.lines):
r = self._get_range(ln)
if r
is None:
rewritten_lines[ln] = line_info
continue
new_ln = self._get_mapped_line(ln, r)
inc_source, _ = self._current_pp_info[r]
if inc_source != record.source_file:
gen_rec = self._get_record(inc_source)
gen_rec.lines[new_ln] = line_info
continue
# Move exec_count to the new lineno.
rewritten_lines[new_ln] = line_info
record.lines = rewritten_lines
def _rewrite_functions(self, record):
rewritten_fns = {}
# Sometimes we get multiple entries for a named function ("top-level", for
# instance). It's not clear the records that result are well-formed, but
# we act as though if a function has multiple FN's, the corresponding
# FNDA's are all the same.
for ln, fn_name
in viewitems(record.functions):
r = self._get_range(ln)
if r
is None:
rewritten_fns[ln] = fn_name
continue
new_ln = self._get_mapped_line(ln, r)
inc_source, _ = self._current_pp_info[r]
if inc_source != record.source_file:
gen_rec = self._get_record(inc_source)
gen_rec.functions[new_ln] = fn_name
if fn_name
in record.function_exec_counts:
gen_rec.function_exec_counts[fn_name] = record.function_exec_counts[
fn_name
]
continue
rewritten_fns[new_ln] = fn_name
record.functions = rewritten_fns
def _rewrite_branches(self, record):
rewritten_branches = {}
for (ln, block_number, branch_number), taken
in viewitems(record.branches):
r = self._get_range(ln)
if r
is None:
rewritten_branches[ln, block_number, branch_number] = taken
continue
new_ln = self._get_mapped_line(ln, r)
inc_source, _ = self._current_pp_info[r]
if inc_source != record.source_file:
gen_rec = self._get_record(inc_source)
gen_rec.branches[(new_ln, block_number, branch_number)] = taken
continue
rewritten_branches[(new_ln, block_number, branch_number)] = taken
record.branches = rewritten_branches
def rewrite_record(self, record, pp_info):
# Rewrite the lines in the given record according to preprocessor info
# and split to additional records when pp_info has included file info.
self._current_pp_info = dict(
[(tuple([int(l)
for l
in k.split(
",")]), v)
for k, v
in pp_info.items()]
)
self._ranges = sorted(self._current_pp_info.keys())
self._additions = {}
self._rewrite_lines(record)
self._rewrite_functions(record)
self._rewrite_branches(record)
record.resummarize()
generated_records = self._additions.values()
for r
in generated_records:
r.resummarize()
return generated_records
class LcovFile(object):
# Simple parser/pretty-printer for lcov format.
# lcov parsing based on http://ltp.sourceforge.net/coverage/lcov/geninfo.1.php
# TN:<test name>
# SF:<absolute path to the source file>
# FN:<line number of function start>,<function name>
# FNDA:<execution count>,<function name>
# FNF:<number of functions found>
# FNH:<number of function hit>
# BRDA:<line number>,<block number>,<branch number>,<taken>
# BRF:<number of branches found>
# BRH:<number of branches hit>
# DA:<line number>,<execution count>[,<checksum>]
# LF:<number of instrumented lines>
# LH:<number of lines with a non-zero execution count>
# end_of_record
PREFIX_TYPES = {
"TN": 0,
"SF": 0,
"FN": 1,
"FNDA": 1,
"FNF": 0,
"FNH": 0,
"BRDA": 3,
"BRF": 0,
"BRH": 0,
"DA": 2,
"LH": 0,
"LF": 0,
}
def __init__(self, lcov_paths):
self.lcov_paths = lcov_paths
def iterate_records(self, rewrite_source=
None):
current_source_file =
None
current_pp_info =
None
current_lines = []
for lcov_path
in self.lcov_paths:
with open(lcov_path,
"r", encoding=
"utf-8")
as lcov_fh:
for line
in lcov_fh:
line = line.rstrip()
if not line:
continue
if line ==
"end_of_record":
# We skip records that we couldn't rewrite, that is records for which
# rewrite_url returns None.
if current_source_file
is not None:
yield (current_source_file, current_pp_info, current_lines)
current_source_file =
None
current_pp_info =
None
current_lines = []
continue
colon = line.find(
":")
prefix = line[:colon]
if prefix ==
"SF":
sf = line[(colon + 1) :]
res = (
rewrite_source(sf)
if rewrite_source
is not None
else (sf,
None)
)
if res
is None:
current_lines.append(line)
else:
current_source_file, current_pp_info = res
current_lines.append(
"SF:" + current_source_file)
else:
current_lines.append(line)
def parse_record(self, record_content):
self.current_record = LcovRecord()
for line
in record_content:
colon = line.find(
":")
prefix = line[:colon]
# We occasionally end up with multi-line scripts in data:
# uris that will trip up the parser, just skip them for now.
if colon < 0
or prefix
not in self.PREFIX_TYPES:
continue
args = line[(colon + 1) :].split(
",", self.PREFIX_TYPES[prefix])
def try_convert(a):
try:
return int(a)
except ValueError:
return a
args = [try_convert(a)
for a
in args]
try:
LcovFile.__dict__[
"parse_" + prefix](self, *args)
except ValueError:
print(
"Encountered an error in %s:\n%s" % (self.lcov_fh.name, line))
raise
except KeyError:
print(
"Invalid lcov line start in %s:\n%s" % (self.lcov_fh.name, line))
raise
except TypeError:
print(
"Invalid lcov line start in %s:\n%s" % (self.lcov_fh.name, line))
raise
ret = self.current_record
self.current_record = LcovRecord()
return ret
def print_file(self, fh, rewrite_source, rewrite_record):
for source_file, pp_info, record_content
in self.iterate_records(
rewrite_source
):
if pp_info
is not None:
record = self.parse_record(record_content)
for r
in rewrite_record(record, pp_info):
fh.write(self.format_record(r))
fh.write(self.format_record(record))
else:
fh.write(
"\n".join(record_content) +
"\nend_of_record\n")
def format_record(self, record):
out_lines = []
for name
in LcovRecord.__slots__:
if hasattr(record, name):
out_lines.append(LcovFile.__dict__[
"format_" + name](self, record))
return "\n".join(out_lines) +
"\nend_of_record\n"
def format_test_name(self, record):
return "TN:%s" % record.test_name
def format_source_file(self, record):
return "SF:%s" % record.source_file
def format_functions(self, record):
# Sorting results gives deterministic output (and is a lot faster than
# using OrderedDict).
fns = []
for start_lineno, fn_name
in sorted(viewitems(record.functions)):
fns.append(
"FN:%s,%s" % (start_lineno, fn_name))
return "\n".join(fns)
def format_function_exec_counts(self, record):
fndas = []
for name, exec_count
in sorted(viewitems(record.function_exec_counts)):
fndas.append(
"FNDA:%s,%s" % (exec_count, name))
return "\n".join(fndas)
def format_function_count(self, record):
return "FNF:%s" % record.function_count
def format_covered_function_count(self, record):
return "FNH:%s" % record.covered_function_count
def format_branches(self, record):
brdas = []
for key
in sorted(record.branches):
taken = record.branches[key]
taken =
"-" if taken == 0
else taken
brdas.append(
"BRDA:%s" %
",".join(map(str, list(key) + [taken])))
return "\n".join(brdas)
def format_branch_count(self, record):
return "BRF:%s" % record.branch_count
def format_covered_branch_count(self, record):
return "BRH:%s" % record.covered_branch_count
def format_lines(self, record):
das = []
for line_no, (exec_count, checksum)
in sorted(viewitems(record.lines)):
s =
"DA:%s,%s" % (line_no, exec_count)
if checksum:
s +=
",%s" % checksum
das.append(s)
return "\n".join(das)
def format_line_count(self, record):
return "LF:%s" % record.line_count
def format_covered_line_count(self, record):
return "LH:%s" % record.covered_line_count
def parse_TN(self, test_name):
self.current_record.test_name = test_name
def parse_SF(self, source_file):
self.current_record.source_file = source_file
def parse_FN(self, start_lineno, fn_name):
self.current_record.functions[start_lineno] = fn_name
def parse_FNDA(self, exec_count, fn_name):
self.current_record.function_exec_counts[fn_name] = exec_count
def parse_FNF(self, function_count):
self.current_record.function_count = function_count
def parse_FNH(self, covered_function_count):
self.current_record.covered_function_count = covered_function_count
def parse_BRDA(self, line_number, block_number, branch_number, taken):
taken = 0
if taken ==
"-" else taken
self.current_record.branches[(line_number, block_number, branch_number)] = taken
def parse_BRF(self, branch_count):
self.current_record.branch_count = branch_count
def parse_BRH(self, covered_branch_count):
self.current_record.covered_branch_count = covered_branch_count
def parse_DA(self, line_number, execution_count, checksum=
None):
self.current_record.lines[line_number] = (execution_count, checksum)
def parse_LH(self, covered_line_count):
self.current_record.covered_line_count = covered_line_count
def parse_LF(self, line_count):
self.current_record.line_count = line_count
class UrlFinderError(Exception):
pass
class UrlFinder(object):
# Given a "chrome://" or "resource://" url, uses data from the UrlMapBackend
# and install manifests to find a path to the source file and the corresponding
# (potentially pre-processed) file in the objdir.
def __init__(self, chrome_map_path, appdir, gredir, extra_chrome_manifests):
# Cached entries
self._final_mapping = {}
try:
with open(chrome_map_path,
"r", encoding=
"utf-8")
as fh:
url_prefixes, overrides, install_info, buildconfig = json.load(fh)
except IOError:
print(
"Error reading %s. Run |./mach build-backend -b ChromeMap| to "
"populate the ChromeMap backend." % chrome_map_path
)
raise
self.topobjdir = buildconfig[
"topobjdir"]
self.MOZ_APP_NAME = buildconfig[
"MOZ_APP_NAME"]
self.OMNIJAR_NAME = buildconfig[
"OMNIJAR_NAME"]
# These are added dynamically in nsIResProtocolHandler, we might
# need to get them at run time.
if "resource:///" not in url_prefixes:
url_prefixes[
"resource:///"] = [appdir]
if "resource://gre/" not in url_prefixes:
url_prefixes[
"resource://gre/"] = [gredir]
self._url_prefixes = url_prefixes
self._url_overrides = overrides
self._respath =
None
mac_bundle_name = buildconfig[
"MOZ_MACBUNDLE_NAME"]
if mac_bundle_name:
self._respath = mozpath.join(
"dist", mac_bundle_name,
"Contents",
"Resources"
)
if not extra_chrome_manifests:
extra_path = os.path.join(self.topobjdir,
"_tests",
"extra.manifest")
if os.path.isfile(extra_path):
extra_chrome_manifests = [extra_path]
if extra_chrome_manifests:
self._populate_chrome(extra_chrome_manifests)
self._install_mapping = install_info
def _populate_chrome(self, manifests):
handler = ChromeManifestHandler()
for m
in manifests:
path = os.path.abspath(m)
for e
in parse_manifest(
None, path):
handler.handle_manifest_entry(e)
self._url_overrides.update(handler.overrides)
self._url_prefixes.update(handler.chrome_mapping)
def _find_install_prefix(self, objdir_path):
def _prefix(s):
for p
in mozpath.split(s):
if "*" not in p:
yield p +
"/"
offset = 0
for leaf
in reversed(mozpath.split(objdir_path)):
offset += len(leaf)
if objdir_path[:-offset]
in self._install_mapping:
pattern_prefix, is_pp = self._install_mapping[objdir_path[:-offset]]
full_leaf = objdir_path[len(objdir_path) - offset :]
src_prefix =
"".join(_prefix(pattern_prefix))
self._install_mapping[objdir_path] = (
mozpath.join(src_prefix, full_leaf),
is_pp,
)
break
offset += 1
def _install_info(self, objdir_path):
if objdir_path
not in self._install_mapping:
# If our path is missing, some prefix of it may be in the install
# mapping mapped to a wildcard.
self._find_install_prefix(objdir_path)
if objdir_path
not in self._install_mapping:
raise UrlFinderError(
"Couldn't find entry in manifest for %s" % objdir_path)
return self._install_mapping[objdir_path]
def _abs_objdir_install_info(self, term):
obj_relpath = term[len(self.topobjdir) + 1 :]
res = self._install_info(obj_relpath)
# Some urls on osx will refer to paths in the mac bundle, so we
# re-interpret them as being their original location in dist/bin.
if not res
and self._respath
and obj_relpath.startswith(self._respath):
obj_relpath = obj_relpath.replace(self._respath,
"dist/bin")
res = self._install_info(obj_relpath)
if not res:
raise UrlFinderError(
"Couldn't find entry in manifest for %s" % obj_relpath)
return res
def find_files(self, url):
# Returns a tuple of (source file, pp_info)
# for the given "resource:", "chrome:", or "file:" uri.
term = url
if term
in self._url_overrides:
term = self._url_overrides[term]
if os.path.isabs(term)
and term.startswith(self.topobjdir):
source_path, pp_info = self._abs_objdir_install_info(term)
return source_path, pp_info
for prefix, dests
in viewitems(self._url_prefixes):
if term.startswith(prefix):
for dest
in dests:
if not dest.endswith(
"/"):
dest +=
"/"
objdir_path = term.replace(prefix, dest)
while objdir_path.startswith(
"//"):
# The mochitest harness produces some wonky file:// uris
# that need to be fixed.
objdir_path = objdir_path[1:]
try:
if os.path.isabs(objdir_path)
and objdir_path.startswith(
self.topobjdir
):
return self._abs_objdir_install_info(objdir_path)
else:
src_path, pp_info = self._install_info(objdir_path)
return mozpath.normpath(src_path), pp_info
except UrlFinderError:
pass
if dest.startswith(
"resource://")
or dest.startswith(
"chrome://"):
result = self.find_files(term.replace(prefix, dest))
if result:
return result
raise UrlFinderError(
"No objdir path for %s" % term)
def rewrite_url(self, url):
# This applies one-off rules and returns None for urls that we aren't
# going to be able to resolve to a source file ("about:" urls, for
# instance).
if url
in self._final_mapping:
return self._final_mapping[url]
if url.endswith(
"> eval"):
return None
if url.endswith(
"> Function"):
return None
if " -> " in url:
url = url.split(
" -> ")[1].rstrip()
if "?" in url:
url = url.split(
"?")[0]
url_obj = urlparse.urlparse(url)
if url_obj.scheme ==
"jar":
app_name = self.MOZ_APP_NAME
omnijar_name = self.OMNIJAR_NAME
if app_name
in url:
if omnijar_name
in url:
# e.g. file:///home/worker/workspace/build/application/firefox/omni.ja!/components/MainProcessSingleton.js # noqa
parts = url_obj.path.split(omnijar_name +
"!", 1)
elif ".xpi!" in url:
# e.g. file:///home/worker/workspace/build/application/firefox/browser/features/e10srollout@mozilla.org.xpi!/bootstrap.js # noqa
parts = url_obj.path.split(
".xpi!", 1)
else:
# We don't know how to handle this jar: path, so return it to the
# caller to make it print a warning.
return url_obj.path,
None
dir_parts = parts[0].rsplit(app_name +
"/", 1)
url = mozpath.normpath(
mozpath.join(
self.topobjdir,
"dist",
"bin",
dir_parts[1].lstrip(
"/"),
parts[1].lstrip(
"/"),
)
)
elif ".xpi!" in url:
# This matching mechanism is quite brittle and based on examples seen in the wild.
# There's no rule to match the XPI name to the path in dist/xpi-stage.
parts = url_obj.path.split(
".xpi!", 1)
addon_name = os.path.basename(parts[0])
if "-test@mozilla.org" in addon_name:
addon_name = addon_name[: -len(
"-test@mozilla.org")]
elif addon_name.endswith(
"@mozilla.org"):
addon_name = addon_name[: -len(
"@mozilla.org")]
url = mozpath.normpath(
mozpath.join(
self.topobjdir,
"dist",
"xpi-stage",
addon_name,
parts[1].lstrip(
"/"),
)
)
elif url_obj.scheme ==
"file" and os.path.isabs(url_obj.path):
path = url_obj.path
if not os.path.isfile(path):
# This may have been in a profile directory that no
# longer exists.
return None
if not path.startswith(self.topobjdir):
return path,
None
url = url_obj.path
elif url_obj.scheme
in (
"http",
"https",
"javascript",
"data",
"about"):
return None
result = self.find_files(url)
self._final_mapping[url] = result
return result
class LcovFileRewriter(object):
# Class for partial parses of LCOV format and rewriting to resolve urls
# and preprocessed file lines.
def __init__(
self,
chrome_map_path,
appdir=
"dist/bin/browser/",
gredir=
"dist/bin/",
extra_chrome_manifests=[],
):
self.url_finder = UrlFinder(
chrome_map_path, appdir, gredir, extra_chrome_manifests
)
self.pp_rewriter = RecordRewriter()
def rewrite_files(self, in_paths, output_file, output_suffix):
unknowns = set()
found_valid = [
False]
def rewrite_source(url):
try:
res = self.url_finder.rewrite_url(url)
if res
is None:
return None
except Exception
as e:
if url
not in unknowns:
# The exception can contain random filename used by
# test cases, and there can be character that cannot be
# encoded with the stdout encoding.
sys.stdout.buffer.write(
(
"Error: %s.\nCouldn't find source info for %s, removing record\n"
% (e, url)
).encode(sys.stdout.encoding, errors=
"replace")
)
unknowns.add(url)
return None
source_file, pp_info = res
# We can't assert that the file exists here, because we don't have the source
# checkout available on test machines. We can bring back this assertion when
# bug 1432287 is fixed.
# assert os.path.isfile(source_file), "Couldn't find mapped source file %s at %s!" % (
# url, source_file)
found_valid[0] =
True
return res
in_paths = [os.path.abspath(in_path)
for in_path
in in_paths]
if output_file:
lcov_file = LcovFile(in_paths)
with open(output_file,
"w+", encoding=
"utf-8")
as out_fh:
lcov_file.print_file(
out_fh, rewrite_source, self.pp_rewriter.rewrite_record
)
else:
for in_path
in in_paths:
lcov_file = LcovFile([in_path])
with open(in_path + output_suffix,
"w+", encoding=
"utf-8")
as out_fh:
lcov_file.print_file(
out_fh, rewrite_source, self.pp_rewriter.rewrite_record
)
if not found_valid[0]:
print(
"WARNING: No valid records found in %s" % in_paths)
return
def main():
parser = ArgumentParser(
description=
"Given a set of gcov .info files produced "
"by spidermonkey's code coverage, re-maps file urls "
"back to source files and lines in preprocessed files "
"back to their original locations."
)
parser.add_argument(
"--chrome-map-path",
default=
"chrome-map.json",
help=
"Path to the chrome-map.json file.",
)
parser.add_argument(
"--app-dir",
default=
"dist/bin/browser/",
help=
"Prefix of the appdir in use. This is used to map "
"urls starting with resource:///. It may differ by "
"app, but defaults to the valid value for firefox.",
)
parser.add_argument(
"--gre-dir",
default=
"dist/bin/",
help=
"Prefix of the gre dir in use. This is used to map "
"urls starting with resource://gre. It may differ by "
"app, but defaults to the valid value for firefox.",
)
parser.add_argument(
"--output-suffix", default=
".out", help=
"The suffix to append to output files."
)
parser.add_argument(
"--extra-chrome-manifests",
nargs=
"+",
help=
"Paths to files containing extra chrome registration.",
)
parser.add_argument(
"--output-file",
default=
"",
help=
"The output file where the results are merged. Leave empty to make the rewriter not "
"merge files.",
)
parser.add_argument(
"files", nargs=
"+", help=
"The set of files to process.")
args = parser.parse_args()
rewriter = LcovFileRewriter(
args.chrome_map_path, args.app_dir, args.gre_dir, args.extra_chrome_manifests
)
files = []
for f
in args.files:
if os.path.isdir(f):
files += [os.path.join(f, e)
for e
in os.listdir(f)]
else:
files.append(f)
rewriter.rewrite_files(files, args.output_file, args.output_suffix)
if __name__ ==
"__main__":
main()
