test.py

#!/usr/bin/env vpython3
#
# Copyright 2017 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Run all tests for Franky source code.

This test harness looks for .coveragerc files in the current subtree, and runs
the tests in the corresponding subtrees with the specified coverage
configurations.

Usage:

    ./test.py list                                # List all the tests
    ./test.py run                                 # Run all the tests
    ./test.py run franky/<specific-file>_test.py  # Run specific test files.
"""

from __future__ import absolute_import
from __future__ import print_function
import argparse
import logging
import os
import shutil
import subprocess
import sys
import six
from six.moves import range


# Folder for HTML coverage report.
HTML_DIR = 'htmlcov'

EXCLUDE_DIRS = [
    '.git',
    'build',
    'dist',
    'infra',
    'recipes',
    HTML_DIR,
]

TEST_SUFFIXES = [
    '_test.py',
]

# A config file defining a subtree as a module for tests.
COVERAGERC = 'coveragerc'
# Data file generated by the coverage tool.
COVERAGE_DATA = '.coverage'
# Root of the repository.
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))


class TestException(Exception):
  pass


def parse_args(argv):
  parser = argparse.ArgumentParser(description='Run or list Franky tests')
  parser.add_argument('-v', '--verbose', action='store_true',
                      help='Print debugging info.')
  subparsers = parser.add_subparsers(
      dest='subparser_name',
      help='Run ./test.py <subcommand> --help for more info')
  list_parser = subparsers.add_parser('list', help='List all tests.')
  run_parser = subparsers.add_parser('run', help='Run all tests.')

  list_parser.add_argument('modules', nargs='*',
                           help='Optional modules to list the tests from.')
  run_parser.add_argument('tests', nargs='*', metavar='TEST',
                          help='Optional modules or test files to run. '
                          'By default, run all tests.')
  run_parser.add_argument('--enable-coverage', action='store_true',
                          help='Print detailed coverage info (default when '
                          'running all tests)')
  run_parser.add_argument('--disable-coverage', action='store_true',
                          help='Suppress detailed coverage info (default '
                          'when tests are specified)')

  args = parser.parse_args(argv)
  if args.subparser_name == 'run':
    # By default, report coverage if no tests are specified.
    args.coverage = not args.tests
    if args.enable_coverage and args.disable_coverage:
      logging.error('Only one of --enable-coverage or --disable-coverage '
                    'is allowed.')
      sys.exit(1)
    if args.enable_coverage:
      args.coverage = True
    if args.disable_coverage:
      args.coverage = False
  return args


def filter_dirs(dirs):
  """Remove EXCLUDE_DIRS from the list of dirs."""
  return [d for d in dirs if d not in EXCLUDE_DIRS]


def find_modules():
  """Look for top-level directories with COVERAGERC file.

  Returns:
    list of paths to modules relative to the root (strings).
  """
  _, dirs, _ = next(os.walk(ROOT_DIR))
  dirs = filter_dirs(dirs)
  modules = []
  while dirs:
    curr_dir = dirs.pop()
    _, subdirs, files = next(os.walk(curr_dir))
    if COVERAGERC in files:
      modules.append(curr_dir)
    else:
      dirs.extend(filter_dirs(os.path.join(curr_dir, d) for d in subdirs))
  return modules


def is_valid_module(module):
  """A valid module is a directory with COVERAGERC file in it."""
  return os.path.isfile(os.path.join(module, COVERAGERC))


def find_bad_modules(modules):
  """Checks for invalid modules in the given list.

  Args:
    modules: (list of strings) list of module names.
  Returns:
    list of invalid modules, per is_valid_module().
  """
  return [m for m in modules if not is_valid_module(m)]


def validate_modules(modules):
  if not modules:
    modules = find_modules()
  else:
    bad_modules = find_bad_modules(modules)
    if bad_modules:
      raise TestException('Invalid modules: %s' % ', '.join(bad_modules))
  return modules


def find_tests_in_modules(modules, tests=None):
  """Populate tests dict with tests indexed by module."""
  if tests is None:
    tests = {}
  for module in modules:
    for curr_dir, _, files in os.walk(module):
      tests.setdefault(module, [])
      tests[module].extend(os.path.join(curr_dir, f) for f in files
                           if any(f.endswith(s) for s in TEST_SUFFIXES))
  return tests


def module_for_test(test):
  """Returns a module the test belongs to, or None."""
  components = test.split(os.sep)
  for i in range(1, len(components)):
    module = os.path.join(*components[:i])
    if is_valid_module(module):
      return module


def expand_tests(tests_or_modules):
  """Given a list of tests or modules, produce a {module -> tests} dict."""
  testsets = {}
  for test_or_module in tests_or_modules:
    if is_valid_module(test_or_module):
      find_tests_in_modules([test_or_module], testsets)
    elif os.path.isfile(test_or_module):
      module = module_for_test(test_or_module)
      if not module:
        raise TestException(
            'Test does not belong to any module: %s' % test_or_module)
      testsets.setdefault(module, [])
      testsets[module].append(test_or_module)
    else:
      raise TestException('Invalid test or module: %s' % test_or_module)
  return testsets


def list_tests(modules):
  modules = validate_modules(modules)
  testsets = find_tests_in_modules(modules)
  for module, tests in six.iteritems(testsets):
    print('\nTests in module %s:\n  %s' % (
        module, '\n  '.join(tests)))


def delete_path(path):
  """Safely deletes a file or a directory recursively."""
  try:
    if os.path.isfile(path):
      os.remove(path)
    elif os.path.isdir(path):
      shutil.rmtree(path)
  except OSError as e:
    logging.warning('Failed to delete existing %s: %s', path, e)


class FailedModules(dict):
  """Map of {module -> list of failure reasons}."""
  def add_reason(self, module, reason):
    self.setdefault(module, [])
    self[module].append(reason)


def run_tests(tests_or_modules, report_coverage):
  if not tests_or_modules:
    tests_or_modules = find_modules()
  testsets = expand_tests(tests_or_modules)
  failed_modules = FailedModules()
  for module, tests in six.iteritems(testsets):
    delete_path(os.path.join(ROOT_DIR, COVERAGE_DATA))
    env = {}
    env.update(os.environ)
    rcfile_arg = '--rcfile=%s' % (os.path.join(module, COVERAGERC))
    module_failed = False
    for test in tests:
      print('Running %s...' % test)
      module_failed = subprocess.call([
        'coverage', 'run',
        rcfile_arg,
        '-a', test
      ]) or module_failed
    if module_failed:
      failed_modules.add_reason(module, 'Tests failed')
    if report_coverage:
      subprocess.call(['coverage', 'report', '--skip-covered', rcfile_arg])
    html_dir = os.path.join(module, HTML_DIR)
    delete_path(html_dir)
    if subprocess.call(['coverage', 'html', rcfile_arg, '-d', html_dir]):
      if report_coverage:
        logging.error('Insufficient coverage.')
        failed_modules.add_reason(module, 'Insufficient coverage')
    print('Detailed coverage report is in file:///%s' % (
        os.path.join(ROOT_DIR, html_dir, 'index.html')))
  if failed_modules:
    print('\nFailed tests in:')
    for module, reasons in six.iteritems(failed_modules):
      print('  %s: %s' % (module, '; '.join(reasons)))
  else:
    print('\nAll tests pass.')
  return bool(failed_modules)


def main(argv):
  args = parse_args(argv)
  logging.basicConfig(level=logging.DEBUG if args.verbose else logging.WARNING)
  os.chdir(ROOT_DIR)
  try:
    if args.subparser_name == 'list':
      return list_tests(args.modules)
    if args.subparser_name == 'run':
      return run_tests(args.tests, args.coverage)
  except TestException as e:
    logging.error('ERROR: %s', e)
    return 1


if __name__ == "__main__":
  sys.exit(main(sys.argv[1:]))