go/test.py

#!/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([
  'build_tags',
  'comment',
  'expected_coverage_max',
  'expected_coverage_min',
  'skip_checks',
  'skip_testing',
])

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