blob: 577293e80d6249aec7f1f44888beb840737cc3a6 [file] [log] [blame]
#!/usr/bin/env python
# Copyright 2014 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Runs Go unit tests verifying code coverage.
Expects Go toolset to be in PATH, GOPATH and GOROOT correctly set. Use ./env.py
to set them up.
Usage:
test.py [root package path]
By default runs all tests for infra/*.
"""
import collections
import errno
import json
import os
import re
import shutil
import subprocess
import sys
import time
from multiprocessing.pool import ThreadPool
# infra/go/
INFRA_GO_DIR = os.path.dirname(os.path.abspath(__file__))
# Allowed keys in *.infra_testing dict.
EXPECTED_INFO_KEYS = frozenset([
'skip_testing',
'expected_coverage_min',
'expected_coverage_max',
'build_tags',
'skip_checks',
])
MAGIC_FILE_EXT = 'infra_testing'
# Return value of run_package_tests.
TestResults = collections.namedtuple('TestResults', [
# Name of the package being tested.
'package',
# True if all unit tests pass.
'tests_pass',
# Percentage of source code covered, or None if not available.
'coverage_percent',
# Expected coverage to pass code coverage test, tuple (min, max) or None.
'coverage_expected',
# True if coverage report is enabled and coverage is acceptable.
'coverage_pass',
# Path to HTML file with coverage report.
'coverage_html',
# Std output of the tests.
'stdout',
# Std error of the tests.
'stderr',
])
def check_go_available():
"""Returns True if go executable is in the PATH."""
try:
subprocess.check_output(['go', 'version'], stderr=subprocess.STDOUT)
return True
except subprocess.CalledProcessError:
return False
except OSError as err:
if err.errno == errno.ENOENT:
return False
raise
def makedirs(path):
"""Same as os.makedirs, but doesn't fail if path exists."""
try:
os.makedirs(path)
except OSError as err:
if err.errno != errno.EEXIST:
raise
def get_goos():
"""Converts sys.platform to GOOS value."""
if sys.platform.startswith('win'):
return 'windows'
if sys.platform.startswith('darwin'):
return 'darwin'
if sys.platform.startswith('linux'):
return 'linux'
raise ValueError('Unrecognized platform: %s' % sys.platform)
def parse_info_file(info_file):
"""Returns contents of <package>/<name>.infra_testing file or {} if missing.
*.infra_testing contains a JSON dict with the following keys:
{
// Do not run tests in this package at all. Default 'false'.
"skip_testing": a list of platforms (GOOS) to skip tests on,
// Minimum allowed code coverage percentage, see below. Default '100'.
"expected_coverage_min": number,
// Maximum allowed code coverage percentage, see below. Default '100'.
"expected_coverage_max": number
}
expected_coverage_min and expected_coverage_max set a boundary on what the
code coverage percentage of the package is expected to be. test.py will fail
if code coverage is less than 'expected_coverage_min' (meaning the package
code has degraded), or larger than 'expected_coverage_max' (meaning the
package code has improved and expected_coverage_min should probably be
changed too).
Setting expected_coverage_min=0 and expected_coverage_max=100 effectively
disables code coverage checks for a package (test.py still will generate
HTML coverage reports though).
"""
try:
with open(info_file, 'r') as f:
info = json.load(f)
if not isinstance(info, dict):
print >> sys.stderr, 'Expecting to find dict in %s' % info_file
return {}
if not EXPECTED_INFO_KEYS.issuperset(info):
print >> sys.stderr, 'Unexpected keys found in %s: %s' % (
info_file, set(info) - EXPECTED_INFO_KEYS)
return info
except IOError:
return {}
except ValueError:
print >> sys.stderr, 'Not a valid JSON file: %s' % info_file
return {}
class SkipCache(object):
"""Extremely basic cache for tracking the skipped-ness of files. Used for
check_*.py scripts in this folder, but since it deals with the infra_testing
magic files, its implementation belongs here."""
def __init__(self, check_name):
"""check_name is the name that would show up in the 'skip_checks' field of
an infra_testing magic file."""
self.cache = {}
self.check_name = check_name
def is_skipped(self, file_or_dirpath):
dirname = os.path.abspath(file_or_dirpath)
if os.path.isfile(dirname):
dirname = os.path.dirname(dirname)
if dirname not in self.cache:
base = os.path.basename(dirname)
fullpath = os.path.join(dirname, base+"."+MAGIC_FILE_EXT)
info = parse_info_file(fullpath)
self.cache[dirname] = self.check_name in info.get("skip_checks", ())
return self.cache[dirname]
class PackageBundle(object):
"""Bunch of packages rooted at a single package."""
def __init__(self, root):
out = subprocess.check_output(
['go', 'list', '-f', '{{.ImportPath}} {{.Dir}}', '%s/...' % root])
self.packages = {}
for line in out.splitlines():
line = line.strip()
if not line:
continue
pkg, path = line.split(' ', 1)
self.packages[pkg] = path
def get_magic_file_path(self, package, extension):
"""Full package name -> path to <package>/<package name>.<extension>."""
# Name of the package is the last component.
if '/' not in package:
name = package
else:
name = package.split('/')[-1]
return os.path.join(self.packages[package], '%s.%s' % (name, extension))
def get_package_info(self, package):
return parse_info_file(self.get_magic_file_path(package, MAGIC_FILE_EXT))
def should_skip(self, package):
"""True to skip package tests, reads 'skip_testing' from *.infra_testing."""
skip = self.get_package_info(package).get('skip_testing', [])
if not isinstance(skip, list):
raise TypeError(
'%s: "skip_testing" should be a list of platforms to skip tests on, '
'got %r instead' % (package, skip))
return get_goos() in skip
def get_build_tags(self, package):
"""Build tags to use when building a package, read from *.infra_testing."""
tags = self.get_package_info(package).get('build_tags', ())
if tags:
return '-tags='+(','.join(tags))
return None
def get_expected_coverage(self, package):
"""Returns allowed code coverage percentage as a pair (min, max)."""
info = self.get_package_info(package)
min_cover = info.get('expected_coverage_min', 100.0)
max_cover = info.get('expected_coverage_max', 100.0)
if max_cover < min_cover:
max_cover = min_cover
return (min_cover, max_cover)
def get_goconvey_test_flags(self, package):
"""Reads <package>/*.goconvey file with flags for "go test" if it exists."""
path = self.get_magic_file_path(package, 'goconvey')
if os.path.exists(path):
with open(path, 'r') as f:
return f.read().strip().split()
return []
def build_install_package(self, package):
"""Builds+Installs a single package.
'go install' is used to avoid having it dump superfluous binaries into the
source tree (e.g. they go to the bin/ folder which is at least marginally
useful).
Returns:
TestResults tuple.
"""
cmd = ['go', 'install']
build_tags = self.get_build_tags(package)
if build_tags:
cmd.append(build_tags)
cmd.append(package)
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
out, err = proc.communicate()
passed = proc.returncode==0
return TestResults(
package=package,
tests_pass=passed,
coverage_percent=None,
coverage_expected=None,
coverage_pass=passed,
coverage_html=None,
stdout=out,
stderr=err)
def run_package_tests(self, package, coverage_file):
"""Runs unit tests for a single package.
Also perform code coverage check for that package. Tests from other packages
do not contribute to total code coverage for the package. Required code
coverage percentage is read from <package>/<name>.infra_testing file.
Args:
package: package name (e.g. infra/package).
coverage_file: prefix for code coverage files (*.out and *.html).
Returns:
TestResults tuple.
"""
assert os.path.isabs(coverage_file), coverage_file
# Ask go test to collect coverage to a file, to convert it to HTML later.
cmd = ['go', 'test', package]
build_tags = self.get_build_tags(package)
if build_tags:
cmd.append(build_tags)
cmd.extend(self.get_goconvey_test_flags(package))
makedirs(os.path.dirname(coverage_file))
coverage_out = '%s.out' % coverage_file
cmd.extend(['-coverprofile', coverage_out])
# Run the test.
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
out, err = proc.communicate()
if proc.returncode:
return TestResults(
package=package,
tests_pass=False,
coverage_percent=None,
coverage_expected=None,
coverage_pass=False,
coverage_html=None,
stdout=out,
stderr=err)
# Convert coverage report to fancy HTML. Coverage report may be missing if
# package has no tests at all.
if os.path.exists(coverage_out):
coverage_html = '%s.html' % coverage_file
subprocess.check_output(
['go', 'tool', 'cover', '-html', coverage_out, '-o', coverage_html],
stderr=subprocess.STDOUT)
else:
coverage_html = None
# Assume 0% coverage if package does not define any tests.
match = re.search('coverage: ([\d\.]+)% of statements', out)
coverage = float(match.group(1)) if match else 0
coverage_expected = self.get_expected_coverage(package)
coverage_pass = (
coverage >= coverage_expected[0] and coverage <= coverage_expected[1])
return TestResults(
package=package,
tests_pass=True,
coverage_percent=coverage,
coverage_expected=coverage_expected,
coverage_pass=coverage_pass,
coverage_html=coverage_html,
stdout=out,
stderr=err)
def reinstall_packages(packages, silent=False):
"""Rebuilds and reinstalls all given packages."""
proc = subprocess.Popen(
['go', 'install', '-a'] + sorted(packages),
stdout=subprocess.PIPE if silent else None,
stderr=subprocess.PIPE if silent else None)
proc.communicate()
def run_tests(package_root, coverage_dir):
"""Runs an equivalent of 'go test <package_root>/...'.
Collects code coverage. Prints reports about failed tests or non sufficiently
covered packages to stdout.
Args:
package_root: base go package path to run test on.
coverage_dir: base directory to put coverage reports to, will be completely
overwritten.
Returns:
0 if all tests pass with acceptable code coverage, 1 otherwise.
"""
if not check_go_available():
print 'Can\'t find Go executable in PATH.'
print 'Use ./env.py python test.py'
return 1
# Coverage dir is always overwritten with new coverage reports.
if os.path.exists(coverage_dir):
shutil.rmtree(coverage_dir)
makedirs(coverage_dir)
# PackageBundle is used to batch "go list ..." calls since they are too slow.
bundle = PackageBundle(package_root)
# Code coverage report requires tests to be run against a single package, so
# discover all individual packages.
skipped = []
packages = []
for p in sorted(bundle.packages):
if bundle.should_skip(p):
skipped.append(p)
else:
packages.append(p)
# Silently rebuild everything (in particular recursively imported packages
# from non-main GOPATH). Do not abort here yet on failure, since we want to
# split failures by packages (it is done below). Note that we issue a full
# rebuild (not just clean), because otherwise parallel 'go test' calls below
# will bump into each other while trying to rebuild commonly references
# dependencies.
print 'Rebuilding dependencies...'
started = time.time()
reinstall_packages(skipped + packages, silent=True)
print 'Finished in %.1f sec.' % (time.time() - started)
print
if skipped:
print 'Build-only (see "skip_testing" in *.infra_testing):'
for p in skipped:
print ' %s' % p
print
if packages:
print 'About to run tests for: '
for p in packages:
print ' %s' % p
print '-' * 80
# TODO(vadimsh): Windows seems to have problems building & testing go packages
# in parallel (file locking issues). So don't build stuff in parallel on
# Windows for now.
tpool_size = 1 if sys.platform == 'win32' else len(packages)
failed = []
bad_cover = []
tpool = ThreadPool(tpool_size)
def run((pkg, build_only)):
if not build_only:
coverage_file = os.path.join(coverage_dir, pkg.replace('/', os.sep))
return bundle.run_package_tests(pkg, coverage_file)
else:
return bundle.build_install_package(pkg)
work = [(p, False) for p in packages]+[(p, True) for p in skipped]
for result in tpool.imap_unordered(run, work):
if result.tests_pass and result.coverage_pass:
sys.stdout.write('.')
elif not result.tests_pass:
sys.stdout.write('F')
failed.append(result)
else:
sys.stdout.write('C')
bad_cover.append(result)
sys.stdout.flush()
print
if bad_cover:
print
print 'Code coverage is not what it is expected to be.'
for i, r in enumerate(bad_cover):
coverage_str = 'missing, no tests?'
coverage_expected_str = '???'
if r.coverage_percent is not None:
if r.coverage_percent == 0.0:
coverage_str = '0.0% no tests?'
else:
coverage_str = '%.1f%%' % r.coverage_percent
if r.coverage_expected is not None:
coverage_expected_str = '[%.1f%%, %.1f%%]' % r.coverage_expected
report_url = 'none'
if r.coverage_html:
report_url = 'file://%s' % r.coverage_html
print '-' * 80
print 'PACKAGE: %s' % r.package
print '-' * 80
print ' coverage: %s' % coverage_str
print ' expected: %s' % coverage_expected_str
print ' report: %s' % report_url
print '-' * 80
if i != len(bad_cover) - 1:
print
if failed:
print
for i, r in enumerate(failed):
print '-' * 80
print 'PACKAGE: %s' % r.package
print '-' * 80
print r.stdout,
if r.stderr.strip():
print r.stderr.strip()
print '-' * 80
if i != len(failed) - 1:
print
return int(bool(bad_cover or failed))
def main(args):
if not args:
package_root = 'infra'
elif len(args) == 1:
package_root = args[0]
else:
print >> sys.stderr, sys.modules['__main__'].__doc__.strip()
return 1
return run_tests(package_root, os.path.join(INFRA_GO_DIR, 'coverage'))
if __name__ == '__main__':
sys.exit(main(sys.argv[1:]))