in src/python/pants/backend/python/tasks/pytest_run.py [0:0]
def _maybe_emit_coverage_data(self, workdirs, test_targets, pex):
coverage = self.get_options().coverage
if coverage is None:
yield []
return
pex_src_root = os.path.relpath(self._source_chroot_path, get_buildroot())
src_to_target_base = {}
for target in test_targets:
libs = (tgt for tgt in target.closure()
if tgt.has_sources('.py') and not isinstance(tgt, PythonTests))
for lib in libs:
for src in lib.sources_relative_to_source_root():
src_to_target_base[src] = lib.target_base
def ensure_trailing_sep(path):
return path if path.endswith(os.path.sep) else path + os.path.sep
if coverage == 'auto':
def compute_coverage_pkgs(tgt):
if tgt.coverage:
return tgt.coverage
else:
# This makes the assumption that tests/python/<tgt> will be testing src/python/<tgt>.
# Note in particular that this doesn't work for pants' own tests, as those are under
# the top level package 'pants_tests', rather than just 'pants'.
# TODO(John Sirois): consider failing fast if there is no explicit coverage scheme;
# but also consider supporting configuration of a global scheme whether that be parallel
# dirs/packages or some arbitrary function that can be registered that takes a test target
# and hands back the source packages or paths under test.
def package(test_source_path):
return os.path.dirname(test_source_path).replace(os.sep, '.')
def packages():
for test_source_path in tgt.sources_relative_to_source_root():
pkg = package(test_source_path)
if pkg:
yield pkg
return packages()
coverage_morfs = set(itertools.chain(*[compute_coverage_pkgs(t) for t in test_targets]))
else:
coverage_morfs = []
for morf in coverage.split(','):
if os.path.isdir(morf):
# The source is a dir, so correct its prefix for the chroot.
# E.g. if source is /path/to/src/python/foo/bar or src/python/foo/bar then
# rel_source is src/python/foo/bar, and ...
rel_source = os.path.relpath(morf, get_buildroot())
rel_source = ensure_trailing_sep(rel_source)
found_target_base = False
for target_base in set(src_to_target_base.values()):
prefix = ensure_trailing_sep(target_base)
if rel_source.startswith(prefix):
# ... rel_source will match on prefix=src/python/ ...
suffix = rel_source[len(prefix):]
# ... suffix will equal foo/bar ...
coverage_morfs.append(os.path.join(get_buildroot(), pex_src_root, suffix))
found_target_base = True
# ... and we end up appending <pex_src_root>/foo/bar to the coverage_sources.
break
if not found_target_base:
self.context.log.warn('Coverage path {} is not in any target. Skipping.'.format(morf))
else:
# The source is to be interpreted as a package name.
coverage_morfs.append(morf)
with self._cov_setup(workdirs,
coverage_morfs=coverage_morfs,
src_to_target_base=src_to_target_base) as (args, coverage_rc):
try:
yield args
finally:
env = {
'PEX_MODULE': 'coverage.cmdline:main'
}
def coverage_run(subcommand, arguments):
return self._pex_run(pex,
workunit_name='coverage-{}'.format(subcommand),
args=[subcommand] + arguments,
env=env)
# The '.coverage' data file is output in the CWD of the test run above; so we make sure to
# look for it there.
with self._maybe_run_in_chroot():
# On failures or timeouts, the .coverage file won't be written.
if not os.path.exists('.coverage'):
self.context.log.warn('No .coverage file was found! Skipping coverage reporting.')
else:
coverage_run('report', ['-i', '--rcfile', coverage_rc])
coverage_workdir = workdirs.coverage_path
coverage_run('html', ['-i', '--rcfile', coverage_rc, '-d', coverage_workdir])
coverage_xml = os.path.join(coverage_workdir, 'coverage.xml')
coverage_run('xml', ['-i', '--rcfile', coverage_rc, '-o', coverage_xml])