gslib/commands/test.py

# -*- coding: utf-8 -*-
# Copyright 2011 Google Inc. All Rights Reserved.
#
# 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.
"""Implementation of gsutil test command."""

from __future__ import absolute_import

from collections import namedtuple
import logging
import os
import subprocess
import sys
import tempfile
import textwrap
import time

import gslib
from gslib.command import Command
from gslib.command import ResetFailureCount
from gslib.exception import CommandException
from gslib.project_id import PopulateProjectId
import gslib.tests as tests
from gslib.util import IS_WINDOWS
from gslib.util import NO_MAX


# For Python 2.6, unittest2 is required to run the tests. If it's not available,
# display an error if the test command is run instead of breaking the whole
# program.
# pylint: disable=g-import-not-at-top
try:
  from gslib.tests.util import GetTestNames
  from gslib.tests.util import unittest
except ImportError as e:
  if 'unittest2' in str(e):
    unittest = None
    GetTestNames = None  # pylint: disable=invalid-name
  else:
    raise


try:
  import coverage
except ImportError:
  coverage = None


_DEFAULT_TEST_PARALLEL_PROCESSES = 5
_DEFAULT_S3_TEST_PARALLEL_PROCESSES = 50
_SEQUENTIAL_ISOLATION_FLAG = 'sequential_only'


_SYNOPSIS = """
  gsutil test [-l] [-u] [-f] [command command...]
"""

_DETAILED_HELP_TEXT = ("""
<B>SYNOPSIS</B>
""" + _SYNOPSIS + """


<B>DESCRIPTION</B>
  The gsutil test command runs the gsutil unit tests and integration tests.
  The unit tests use an in-memory mock storage service implementation, while
  the integration tests send requests to the production service using the
  preferred API set in the boto configuration file (see "gsutil help apis" for
  details).

  To run both the unit tests and integration tests, run the command with no
  arguments:

    gsutil test

  To run the unit tests only (which run quickly):

    gsutil test -u

  Tests run in parallel regardless of whether the top-level -m flag is
  present. To limit the number of tests run in parallel to 10 at a time:

    gsutil test -p 10

  To force tests to run sequentially:

    gsutil test -p 1

  To have sequentially-run tests stop running immediately when an error occurs:

    gsutil test -f

  To run tests for one or more individual commands add those commands as
  arguments. For example, the following command will run the cp and mv command
  tests:

    gsutil test cp mv

  To list available tests, run the test command with the -l argument:

    gsutil test -l

  The tests are defined in the code under the gslib/tests module. Each test
  file is of the format test_[name].py where [name] is the test name you can
  pass to this command. For example, running "gsutil test ls" would run the
  tests in "gslib/tests/test_ls.py".

  You can also run an individual test class or function name by passing the
  test module followed by the class name and optionally a test name. For
  example, to run the an entire test class by name:

    gsutil test naming.GsutilNamingTests

  or an individual test function:

    gsutil test cp.TestCp.test_streaming

  You can list the available tests under a module or class by passing arguments
  with the -l option. For example, to list all available test functions in the
  cp module:

    gsutil test -l cp

  To output test coverage:

    gsutil test -c -p 500
    coverage html

  This will output an HTML report to a directory named 'htmlcov'.


<B>OPTIONS</B>
  -c          Output coverage information.

  -f          Exit on first sequential test failure.

  -l          List available tests.

  -p N        Run at most N tests in parallel. The default value is %d.

  -s          Run tests against S3 instead of GS.

  -u          Only run unit tests.
""" % _DEFAULT_TEST_PARALLEL_PROCESSES)


TestProcessData = namedtuple('TestProcessData',
                             'name return_code stdout stderr')


def MakeCustomTestResultClass(total_tests):
  """Creates a closure of CustomTestResult.

  Args:
    total_tests: The total number of tests being run.

  Returns:
    An instance of CustomTestResult.
  """

  class CustomTestResult(unittest.TextTestResult):
    """A subclass of unittest.TextTestResult that prints a progress report."""

    def startTest(self, test):
      super(CustomTestResult, self).startTest(test)
      if self.dots:
        test_id = '.'.join(test.id().split('.')[-2:])
        message = ('\r%d/%d finished - E[%d] F[%d] s[%d] - %s' % (
            self.testsRun, total_tests, len(self.errors),
            len(self.failures), len(self.skipped), test_id))
        message = message[:73]
        message = message.ljust(73)
        self.stream.write('%s - ' % message)

  return CustomTestResult


def GetTestNamesFromSuites(test_suite):
  """Takes a list of test suites and returns a list of contained test names."""
  suites = [test_suite]
  test_names = []
  while suites:
    suite = suites.pop()
    for test in suite:
      if isinstance(test, unittest.TestSuite):
        suites.append(test)
      else:
        test_names.append(test.id()[len('gslib.tests.test_'):])
  return test_names


# pylint: disable=protected-access
# Need to get into the guts of unittest to evaluate test cases for parallelism.
def TestCaseToName(test_case):
  """Converts a python.unittest to its gsutil test-callable name."""
  return (str(test_case.__class__).split('\'')[1] + '.' +
          test_case._testMethodName)


# pylint: disable=protected-access
# Need to get into the guts of unittest to evaluate test cases for parallelism.
def SplitParallelizableTestSuite(test_suite):
  """Splits a test suite into groups with different running properties.

  Args:
    test_suite: A python unittest test suite.

  Returns:
    4-part tuple of lists of test names:
    (tests that must be run sequentially,
     tests that must be isolated in a separate process but can be run either
         sequentially or in parallel,
     unit tests that can be run in parallel,
     integration tests that can run in parallel)
  """
  # pylint: disable=import-not-at-top
  # Need to import this after test globals are set so that skip functions work.
  from gslib.tests.testcase.unit_testcase import GsUtilUnitTestCase
  isolated_tests = []
  sequential_tests = []
  parallelizable_integration_tests = []
  parallelizable_unit_tests = []

  items_to_evaluate = [test_suite]
  cases_to_evaluate = []
  # Expand the test suites into individual test cases:
  while items_to_evaluate:
    suite_or_case = items_to_evaluate.pop()
    if isinstance(suite_or_case, unittest.suite.TestSuite):
      for item in suite_or_case._tests:
        items_to_evaluate.append(item)
    elif isinstance(suite_or_case, unittest.TestCase):
      cases_to_evaluate.append(suite_or_case)

  for test_case in cases_to_evaluate:
    test_method = getattr(test_case, test_case._testMethodName, None)
    if getattr(test_method, 'requires_isolation', False):
      # Test must be isolated to a separate process, even it if is being
      # run sequentially.
      isolated_tests.append(TestCaseToName(test_case))
    elif not getattr(test_method, 'is_parallelizable', True):
      sequential_tests.append(TestCaseToName(test_case))
    elif isinstance(test_case, GsUtilUnitTestCase):
      parallelizable_unit_tests.append(TestCaseToName(test_case))
    else:
      parallelizable_integration_tests.append(TestCaseToName(test_case))

  return (sorted(sequential_tests),
          sorted(isolated_tests),
          sorted(parallelizable_unit_tests),
          sorted(parallelizable_integration_tests))


def CountFalseInList(input_list):
  """Counts number of falses in the input list."""
  num_false = 0
  for item in input_list:
    if not item:
      num_false += 1
  return num_false


def CreateTestProcesses(parallel_tests, test_index, process_list, process_done,
                        max_parallel_tests, root_coverage_file=None):
  """Creates test processes to run tests in parallel.

  Args:
    parallel_tests: List of all parallel tests.
    test_index: List index of last created test before this function call.
    process_list: List of running subprocesses. Created processes are appended
                  to this list.
    process_done: List of booleans indicating process completion. One 'False'
                  will be added per process created.
    max_parallel_tests: Maximum number of tests to run in parallel.
    root_coverage_file: The root .coverage filename if coverage is requested.

  Returns:
    Index of last created test.
  """
  orig_test_index = test_index
  executable_prefix = [sys.executable] if sys.executable and IS_WINDOWS else []
  s3_argument = ['-s'] if tests.util.RUN_S3_TESTS else []

  process_create_start_time = time.time()
  last_log_time = process_create_start_time
  while (CountFalseInList(process_done) < max_parallel_tests and
         test_index < len(parallel_tests)):
    env = os.environ.copy()
    if root_coverage_file:
      env['GSUTIL_COVERAGE_OUTPUT_FILE'] = root_coverage_file
    process_list.append(subprocess.Popen(
        executable_prefix + [gslib.GSUTIL_PATH] +
        ['-o', 'GSUtil:default_project_id=' + PopulateProjectId()] +
        ['test'] + s3_argument +
        ['--' + _SEQUENTIAL_ISOLATION_FLAG] +
        [parallel_tests[test_index][len('gslib.tests.test_'):]],
        stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env))
    test_index += 1
    process_done.append(False)
    if time.time() - last_log_time > 5:
      print ('Created %d new processes (total %d/%d created)' %
             (test_index - orig_test_index, len(process_list),
              len(parallel_tests)))
      last_log_time = time.time()
  if test_index == len(parallel_tests):
    print ('Test process creation finished (%d/%d created)' %
           (len(process_list), len(parallel_tests)))
  return test_index


class TestCommand(Command):
  """Implementation of gsutil test command."""

  # Command specification. See base class for documentation.
  command_spec = Command.CreateCommandSpec(
      'test',
      command_name_aliases=[],
      usage_synopsis=_SYNOPSIS,
      min_args=0,
      max_args=NO_MAX,
      supported_sub_args='uflp:sc',
      file_url_ok=True,
      provider_url_ok=False,
      urls_start_arg=0,
      supported_private_args=[_SEQUENTIAL_ISOLATION_FLAG]
  )
  # Help specification. See help_provider.py for documentation.
  help_spec = Command.HelpSpec(
      help_name='test',
      help_name_aliases=[],
      help_type='command_help',
      help_one_line_summary='Run gsutil tests',
      help_text=_DETAILED_HELP_TEXT,
      subcommand_help_text={},
  )

  def RunParallelTests(self, parallel_integration_tests,
                       max_parallel_tests, coverage_filename):
    """Executes the parallel/isolated portion of the test suite.

    Args:
      parallel_integration_tests: List of tests to execute.
      max_parallel_tests: Maximum number of parallel tests to run at once.
      coverage_filename: If not None, filename for coverage output.

    Returns:
      (int number of test failures, float elapsed time)
    """
    process_list = []
    process_done = []
    process_results = []  # Tuples of (name, return code, stdout, stderr)
    num_parallel_failures = 0
    # Number of logging cycles we ran with no progress.
    progress_less_logging_cycles = 0
    completed_as_of_last_log = 0
    num_parallel_tests = len(parallel_integration_tests)
    parallel_start_time = last_log_time = time.time()
    test_index = CreateTestProcesses(
        parallel_integration_tests, 0, process_list, process_done,
        max_parallel_tests, root_coverage_file=coverage_filename)
    while len(process_results) < num_parallel_tests:
      for proc_num in xrange(len(process_list)):
        if process_done[proc_num] or process_list[proc_num].poll() is None:
          continue
        process_done[proc_num] = True
        stdout, stderr = process_list[proc_num].communicate()
        process_list[proc_num].stdout.close()
        process_list[proc_num].stderr.close()
        # TODO: Differentiate test failures from errors.
        if process_list[proc_num].returncode != 0:
          num_parallel_failures += 1
        process_results.append(TestProcessData(
            name=parallel_integration_tests[proc_num],
            return_code=process_list[proc_num].returncode,
            stdout=stdout, stderr=stderr))
      if len(process_list) < num_parallel_tests:
        test_index = CreateTestProcesses(
            parallel_integration_tests, test_index, process_list,
            process_done, max_parallel_tests,
            root_coverage_file=coverage_filename)
      if len(process_results) < num_parallel_tests:
        if time.time() - last_log_time > 5:
          print '%d/%d finished - %d failures' % (
              len(process_results), num_parallel_tests, num_parallel_failures)
          if len(process_results) == completed_as_of_last_log:
            progress_less_logging_cycles += 1
          else:
            completed_as_of_last_log = len(process_results)
            # A process completed, so we made progress.
            progress_less_logging_cycles = 0
          if progress_less_logging_cycles > 4:
            # Ran 5 or more logging cycles with no progress, let the user
            # know which tests are running slowly or hanging.
            still_running = []
            for proc_num in xrange(len(process_list)):
              if not process_done[proc_num]:
                still_running.append(parallel_integration_tests[proc_num])
            print 'Still running: %s' % still_running
            # TODO: Terminate still-running processes if they
            # hang for a long time.
          last_log_time = time.time()
        time.sleep(1)
    process_run_finish_time = time.time()
    if num_parallel_failures:
      for result in process_results:
        if result.return_code != 0:
          new_stderr = result.stderr.split('\n')
          print 'Results for failed test %s:' % result.name
          for line in new_stderr:
            print line

    return (num_parallel_failures,
            (process_run_finish_time - parallel_start_time))

  def PrintTestResults(self, num_sequential_tests, sequential_success,
                       sequential_time_elapsed,
                       num_parallel_tests, num_parallel_failures,
                       parallel_time_elapsed):
    """Prints test results for parallel and sequential tests."""
    # TODO: Properly track test skips.
    print 'Parallel tests complete. Success: %s Fail: %s' % (
        num_parallel_tests - num_parallel_failures, num_parallel_failures)
    print (
        'Ran %d tests in %.3fs (%d sequential in %.3fs, %d parallel in %.3fs)'
        % (num_parallel_tests + num_sequential_tests,
           float(sequential_time_elapsed + parallel_time_elapsed),
           num_sequential_tests,
           float(sequential_time_elapsed),
           num_parallel_tests,
           float(parallel_time_elapsed)))
    print

    if not num_parallel_failures and sequential_success:
      print 'OK'
    else:
      if num_parallel_failures:
        print 'FAILED (parallel tests)'
      if not sequential_success:
        print 'FAILED (sequential tests)'

  def RunCommand(self):
    """Command entry point for the test command."""
    if not unittest:
      raise CommandException('On Python 2.6, the unittest2 module is required '
                             'to run the gsutil tests.')

    failfast = False
    list_tests = False
    max_parallel_tests = _DEFAULT_TEST_PARALLEL_PROCESSES
    perform_coverage = False
    sequential_only = False
    if self.sub_opts:
      for o, a in self.sub_opts:
        if o == '-c':
          perform_coverage = True
        elif o == '-f':
          failfast = True
        elif o == '-l':
          list_tests = True
        elif o == ('--' + _SEQUENTIAL_ISOLATION_FLAG):
          # Called to isolate a single test in a separate process.
          # Don't try to isolate it again (would lead to an infinite loop).
          sequential_only = True
        elif o == '-p':
          max_parallel_tests = long(a)
        elif o == '-s':
          if not tests.util.HAS_S3_CREDS:
            raise CommandException('S3 tests require S3 credentials. Please '
                                   'add appropriate credentials to your .boto '
                                   'file and re-run.')
          tests.util.RUN_S3_TESTS = True
        elif o == '-u':
          tests.util.RUN_INTEGRATION_TESTS = False

    if perform_coverage and not coverage:
      raise CommandException(
          'Coverage has been requested but the coverage module was not found. '
          'You can install it with "pip install coverage".')

    if (tests.util.RUN_S3_TESTS and
          max_parallel_tests > _DEFAULT_S3_TEST_PARALLEL_PROCESSES):
      self.logger.warn(
          'Reducing parallel tests to %d due to S3 maximum bucket '
          'limitations.', _DEFAULT_S3_TEST_PARALLEL_PROCESSES)
      max_parallel_tests = _DEFAULT_S3_TEST_PARALLEL_PROCESSES

    test_names = sorted(GetTestNames())
    if list_tests and not self.args:
      print 'Found %d test names:' % len(test_names)
      print ' ', '\n  '.join(sorted(test_names))
      return 0

    # Set list of commands to test if supplied.
    if self.args:
      commands_to_test = []
      for name in self.args:
        if name in test_names or name.split('.')[0] in test_names:
          commands_to_test.append('gslib.tests.test_%s' % name)
        else:
          commands_to_test.append(name)
    else:
      commands_to_test = ['gslib.tests.test_%s' % name for name in test_names]

    # Installs a ctrl-c handler that tries to cleanly tear down tests.
    unittest.installHandler()

    loader = unittest.TestLoader()

    if commands_to_test:
      try:
        suite = loader.loadTestsFromNames(commands_to_test)
      except (ImportError, AttributeError) as e:
        raise CommandException('Invalid test argument name: %s' % e)

    if list_tests:
      test_names = GetTestNamesFromSuites(suite)
      print 'Found %d test names:' % len(test_names)
      print ' ', '\n  '.join(sorted(test_names))
      return 0

    if logging.getLogger().getEffectiveLevel() <= logging.INFO:
      verbosity = 1
    else:
      verbosity = 2
      logging.disable(logging.ERROR)

    if perform_coverage:
      # We want to run coverage over the gslib module, but filter out the test
      # modules and any third-party code. We also filter out anything under the
      # temporary directory. Otherwise, the gsutil update test (which copies
      # code to the temporary directory) gets included in the output.
      coverage_controller = coverage.coverage(
          source=['gslib'], omit=['gslib/third_party/*', 'gslib/tests/*',
                                  tempfile.gettempdir() + '*'])
      coverage_controller.erase()
      coverage_controller.start()

    num_parallel_failures = 0
    sequential_success = False

    (sequential_tests, isolated_tests,
     parallel_unit_tests, parallel_integration_tests) = (
         SplitParallelizableTestSuite(suite))

    # Since parallel integration tests are run in a separate process, they
    # won't get the override to tests.util, so skip them here.
    if not tests.util.RUN_INTEGRATION_TESTS:
      parallel_integration_tests = []

    logging.debug('Sequential tests to run: %s', sequential_tests)
    logging.debug('Isolated tests to run: %s', isolated_tests)
    logging.debug('Parallel unit tests to run: %s', parallel_unit_tests)
    logging.debug('Parallel integration tests to run: %s',
                  parallel_integration_tests)

    # If we're running an already-isolated test (spawned in isolation by a
    # previous test process), or we have no parallel tests to run,
    # just run sequentially. For now, unit tests are always run sequentially.
    run_tests_sequentially = (sequential_only or
                              (len(parallel_integration_tests) <= 1
                               and not isolated_tests))

    if run_tests_sequentially:
      total_tests = suite.countTestCases()
      resultclass = MakeCustomTestResultClass(total_tests)

      runner = unittest.TextTestRunner(verbosity=verbosity,
                                       resultclass=resultclass,
                                       failfast=failfast)
      ret = runner.run(suite)
      sequential_success = ret.wasSuccessful()
    else:
      if max_parallel_tests == 1:
        # We can't take advantage of parallelism, though we may have tests that
        # need isolation.
        sequential_tests += parallel_integration_tests
        parallel_integration_tests = []

      sequential_start_time = time.time()
      # TODO: For now, run unit tests sequentially because they are fast.
      # We could potentially shave off several seconds of execution time
      # by executing them in parallel with the integration tests.
      if len(sequential_tests) + len(parallel_unit_tests):
        print 'Running %d tests sequentially.' % (len(sequential_tests) +
                                                  len(parallel_unit_tests))
        sequential_tests_to_run = sequential_tests + parallel_unit_tests
        suite = loader.loadTestsFromNames(
            sorted([test_name for test_name in sequential_tests_to_run]))
        num_sequential_tests = suite.countTestCases()
        resultclass = MakeCustomTestResultClass(num_sequential_tests)
        runner = unittest.TextTestRunner(verbosity=verbosity,
                                         resultclass=resultclass,
                                         failfast=failfast)

        ret = runner.run(suite)
        sequential_success = ret.wasSuccessful()
      else:
        num_sequential_tests = 0
        sequential_success = True
      sequential_time_elapsed = time.time() - sequential_start_time

      # At this point, all tests get their own process so just treat the
      # isolated tests as parallel tests.
      parallel_integration_tests += isolated_tests
      num_parallel_tests = len(parallel_integration_tests)

      if not num_parallel_tests:
        pass
      else:
        num_processes = min(max_parallel_tests, num_parallel_tests)
        if num_parallel_tests > 1 and max_parallel_tests > 1:
          message = 'Running %d tests in parallel mode (%d processes).'
          if num_processes > _DEFAULT_TEST_PARALLEL_PROCESSES:
            message += (
                ' Please be patient while your CPU is incinerated. '
                'If your machine becomes unresponsive, consider reducing '
                'the amount of parallel test processes by running '
                '\'gsutil test -p <num_processes>\'.')
          print ('\n'.join(textwrap.wrap(
              message % (num_parallel_tests, num_processes))))
        else:
          print ('Running %d tests sequentially in isolated processes.' %
                 num_parallel_tests)
        (num_parallel_failures, parallel_time_elapsed) = self.RunParallelTests(
            parallel_integration_tests, max_parallel_tests,
            coverage_controller.data.filename if perform_coverage else None)
        self.PrintTestResults(
            num_sequential_tests, sequential_success,
            sequential_time_elapsed,
            num_parallel_tests, num_parallel_failures,
            parallel_time_elapsed)

    if perform_coverage:
      coverage_controller.stop()
      coverage_controller.combine()
      coverage_controller.save()
      print ('Coverage information was saved to: %s' %
             coverage_controller.data.filename)

    if sequential_success and not num_parallel_failures:
      ResetFailureCount()
      return 0
    return 1