def RunCommand(self):
    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
Command entry point for the test command.
dataset/ETHPy150Open GoogleCloudPlatform/gsutil/gslib/commands/test.py/TestCommand.RunCommand