def _start_tor(self, tor_cmd):
    """
    Initializes a tor process. This blocks until initialization completes or we
    error out.

    :param str tor_cmd: command to start tor with

    :raises: OSError if we either fail to create the tor process or reached a timeout without success
    """

    println("Starting tor...\n", STATUS)
    start_time = time.time()

    try:
      # wait to fully complete if we're running tests with network activity,
      # otherwise finish after local bootstraping
      complete_percent = 100 if Target.ONLINE in self.attribute_targets else 5

      # prints output from tor's stdout while it starts up
      print_init_line = lambda line: println("  %s" % line, SUBSTATUS)

      torrc_dst = os.path.join(self._test_dir, "torrc")
      self._tor_process = stem.process.launch_tor(tor_cmd, None, torrc_dst, complete_percent, print_init_line)

      runtime = time.time() - start_time
      println("  done (%i seconds)\n" % runtime, STATUS)
    except OSError, exc:
      test.output.print_error("  failed to start tor: %s\n" % exc)
      raise exc

def python3_run_tests():
  println()
  println()

  python3_runner = os.path.join(get_python3_destination(), "run_tests.py")
  exit_status = os.system("python3 %s %s" % (python3_runner, " ".join(sys.argv[1:])))
  sys.exit(exit_status)

  def run(self):
    println('  %s...' % self.label, STATUS, NO_NL)

    padding = 50 - len(self.label)
    println(' ' * padding, NO_NL)

    try:
      if self.args:
        self.result = self.runner(*self.args)
      else:
        self.result = self.runner()

      self.is_successful = True
      output_msg = 'done'

      if self.print_result and isinstance(self.result, str):
        output_msg = self.result

      println(output_msg, STATUS)

      if self.print_result and isinstance(self.result, (list, tuple)):
        for line in self.result:
          println('    %s' % line, STATUS)
    except Exception as exc:
      output_msg = str(exc)

      if not output_msg or self.is_required:
        output_msg = 'failed'

      println(output_msg, ERROR)
      self.error = exc

  def run(self):
    println("  %s..." % self.label, STATUS, NO_NL)

    padding = 50 - len(self.label)
    println(" " * padding, NO_NL)

    try:
      if self.args:
        result = self.runner(*self.args)
      else:
        result = self.runner()

      output_msg = "done"

      if isinstance(result, str):
        output_msg = result

      println(output_msg, STATUS)

      if isinstance(result, (list, tuple)):
        for line in result:
          println("    %s" % line, STATUS)
    except Exception as exc:
      output_msg = str(exc)

      if not output_msg or self.is_required:
        output_msg = "failed"

      println(output_msg, ERROR)
      self.error = exc

def main():
  start_time = time.time()

  try:
    stem.prereq.check_requirements()
  except ImportError, exc:
    println("%s\n" % exc)
    sys.exit(1)

def _run_test(test_class, output_filters, logging_buffer):
  test.output.print_divider(test_class.__module__)
  suite = unittest.TestLoader().loadTestsFromTestCase(test_class)

  test_results = StringIO.StringIO()
  run_result = unittest.TextTestRunner(test_results, verbosity=2).run(suite)

  sys.stdout.write(test.output.apply_filters(test_results.getvalue(), *output_filters))
  println()
  test.output.print_logging(logging_buffer)

  return run_result

    def is_running(self):
        """
    Checks if we're running a tor test instance and that it's alive.

    :returns: True if we have a running tor test instance, False otherwise
    """

        with self._runner_lock:
            # Check for an unexpected shutdown by calling subprocess.Popen.poll(),
            # which returns the exit code or None if we're still running.

            if self._tor_process and self._tor_process.poll() is not None:
                # clean up the temporary resources and note the unexpected shutdown
                self.stop()
                println("tor shut down unexpectedly", ERROR)

            return bool(self._tor_process)

  def stop(self):
    """
    Stops our tor test instance and cleans up any temporary resources.
    """

    with self._runner_lock:
      println('Shutting down tor... ', STATUS, NO_NL)

      if self._owner_controller:
        self._owner_controller.close()
        self._owner_controller = None

      if self._tor_process:
        # if the tor process has stopped on its own then the following raises
        # an OSError ([Errno 3] No such process)

        try:
          self._tor_process.kill()
        except OSError:
          pass

        self._tor_process.wait()  # blocks until the process is done

      # if we've made a temporary data directory then clean it up
      if self._test_dir and CONFIG['integ.test_directory'] == '':
        shutil.rmtree(self._test_dir, ignore_errors = True)

      # reverts any mocking of stem.socket.recv_message
      if self._original_recv_message:
        stem.socket.recv_message = self._original_recv_message
        self._original_recv_message = None

      # clean up our socket directory if we made one
      socket_dir = os.path.dirname(CONTROL_SOCKET_PATH)

      if os.path.exists(socket_dir):
        shutil.rmtree(socket_dir, ignore_errors = True)

      self._test_dir = ''
      self._tor_cmd = None
      self._tor_cwd = ''
      self._torrc_contents = ''
      self._custom_opts = None
      self._tor_process = None

      println('done', STATUS)

  def _start_tor(self, tor_cmd):
    """
    Initializes a tor process. This blocks until initialization completes or we
    error out.

    :param str tor_cmd: command to start tor with

    :raises: OSError if we either fail to create the tor process or reached a timeout without success
    """

    println('Starting %s...\n' % tor_cmd, STATUS)
    start_time = time.time()

    try:
      self._tor_process = stem.process.launch_tor(
        tor_cmd = tor_cmd,
        torrc_path = os.path.join(self._test_dir, 'torrc'),
        completion_percent = 100 if test.Target.ONLINE in self.attribute_targets else 5,
        init_msg_handler = lambda line: println('  %s' % line, SUBSTATUS),
        take_ownership = True,
      )

      runtime = time.time() - start_time
      println('  done (%i seconds)\n' % runtime, STATUS)
    except OSError as exc:
      println('  failed to start tor: %s\n' % exc, ERROR)
      raise exc

def run_tasks(category, *tasks):
  """
  Runs a series of :class:`test.util.Task` instances. This simply prints 'done'
  or 'failed' for each unless we fail one that is marked as being required. If
  that happens then we print its error message and call sys.exit().

  :param str category: label for the series of tasks
  :param list tasks: **Task** instances to be ran
  """

  test.output.print_divider(category, True)

  for task in tasks:
    task.run()

    if task.is_required and task.error:
      println("\n%s\n" % task.error, ERROR)
      sys.exit(1)

  println()

  def run(self):
    start_time = time.time()
    println('  %s...' % self.label, STATUS, NO_NL)

    padding = 50 - len(self.label)
    println(' ' * padding, NO_NL)

    try:
      if self._is_background_task:
        def _run_wrapper(conn, runner, args):
          os.nice(15)
          conn.send(runner(*args) if args else runner())
          conn.close()

        self._background_pipe, child_pipe = multiprocessing.Pipe()
        self._background_process = multiprocessing.Process(target = _run_wrapper, args = (child_pipe, self.runner, self.args))
        self._background_process.start()
      else:
        self.result = self.runner(*self.args) if self.args else self.runner()

      self.is_successful = True
      output_msg = 'running' if self._is_background_task else 'done'

      if self.result and self.print_result and isinstance(self.result, str):
        output_msg = self.result
      elif self.print_runtime:
        output_msg += ' (%0.1fs)' % (time.time() - start_time)

      println(output_msg, STATUS)

      if self.print_result and isinstance(self.result, (list, tuple)):
        for line in self.result:
          println('    %s' % line, STATUS)
    except Exception as exc:
      output_msg = str(exc)

      if not output_msg or self.is_required:
        output_msg = 'failed'

      println(output_msg, ERROR)
      self.error = exc

  def _start_tor(self, tor_cmd):
    """
    Initializes a tor process. This blocks until initialization completes or we
    error out.

    :param str tor_cmd: command to start tor with

    :raises: OSError if we either fail to create the tor process or reached a timeout without success
    """

    println('Starting %s...\n' % tor_cmd, STATUS)
    start_time = time.time()

    try:
      # wait to fully complete if we're running tests with network activity,
      # otherwise finish after local bootstraping

      complete_percent = 100 if Target.ONLINE in self.attribute_targets else 5

      def print_init_line(line):
        println('  %s' % line, SUBSTATUS)

      torrc_dst = os.path.join(self._test_dir, 'torrc')
      self._tor_process = stem.process.launch_tor(tor_cmd, None, torrc_dst, complete_percent, print_init_line, take_ownership = True)

      runtime = time.time() - start_time
      println('  done (%i seconds)\n' % runtime, STATUS)
    except OSError as exc:
      println('  failed to start tor: %s\n' % exc, ERROR)
      raise exc

  def stop(self):
    """
    Stops our tor test instance and cleans up any temporary resources.
    """

    with self._runner_lock:
      println("Shutting down tor... ", STATUS, NO_NL)

      if self._tor_process:
        # if the tor process has stopped on its own then the following raises
        # an OSError ([Errno 3] No such process)

        try:
          self._tor_process.kill()
        except OSError:
          pass

        self._tor_process.communicate()  # blocks until the process is done

      # if we've made a temporary data directory then clean it up
      if self._test_dir and CONFIG["integ.test_directory"] == "":
        shutil.rmtree(self._test_dir, ignore_errors = True)

      # reverts any mocking of stem.socket.recv_message
      if self._original_recv_message:
        stem.socket.recv_message = self._original_recv_message
        self._original_recv_message = None

      self._test_dir = ""
      self._tor_cmd = None
      self._tor_cwd = ""
      self._torrc_contents = ""
      self._custom_opts = None
      self._tor_process = None

      println("done", STATUS)

def _print_static_issues(static_check_issues):
  if static_check_issues:
    println('STATIC CHECKS', STATUS)

    for file_path in static_check_issues:
      println('* %s' % file_path, STATUS)

      # Make a dict of line numbers to its issues. This is so we can both sort
      # by the line number and clear any duplicate messages.

      line_to_issues = {}

      for issue in static_check_issues[file_path]:
        line_to_issues.setdefault(issue.line_number, set()).add((issue.message, issue.line))

      for line_number in sorted(line_to_issues.keys()):
        for msg, line in line_to_issues[line_number]:
          line_count = '%-4s' % line_number
          println('  line %s - %-40s %s' % (line_count, msg, line.strip()))

      println()

  def _run_setup(self):
    """
    Makes a temporary runtime resources of our integration test instance.

    :raises: OSError if unsuccessful
    """

    # makes a temporary data directory if needed
    try:
      println("  making test directory (%s)... " % self._test_dir, STATUS, NO_NL)

      if os.path.exists(self._test_dir):
        println("skipped", STATUS)
      else:
        os.makedirs(self._test_dir)
        println("done", STATUS)
    except OSError, exc:
      test.output.print_error("failed (%s)" % exc)
      raise exc

def main():
  start_time = time.time()

  try:
    stem.prereq.check_requirements()
  except ImportError as exc:
    println('%s\n' % exc)
    sys.exit(1)

  test_config = stem.util.conf.get_config('test')
  test_config.load(os.path.join(STEM_BASE, 'test', 'settings.cfg'))

  try:
    args = test.arguments.parse(sys.argv[1:])
  except ValueError as exc:
    println(str(exc))
    sys.exit(1)

  if args.quiet:
    test.output.SUPPRESS_STDOUT = True

  if args.print_help:
    println(test.arguments.get_help())
    sys.exit()
  elif not args.run_unit and not args.run_integ:
    println('Nothing to run (for usage provide --help)\n')
    sys.exit()

  if not stem.prereq.is_mock_available():
    try:
      try:
        import unittest.mock
      except ImportError:
        import mock

      println(MOCK_OUT_OF_DATE_MSG % mock.__version__)
    except ImportError:
      println(MOCK_UNAVAILABLE_MSG)

    if stem.util.system.is_available('pip'):
      println("You can get it by running 'sudo pip install mock'.")
    elif stem.util.system.is_available('apt-get'):
      println("You can get it by running 'sudo apt-get install python-mock'.")

    sys.exit(1)

  pyflakes_task, pep8_task = None, None

  if not args.specific_test:
    if stem.util.test_tools.is_pyflakes_available():
      pyflakes_task = PYFLAKES_TASK

    if stem.util.test_tools.is_pep8_available():
      pep8_task = PEP8_TASK

  test.util.run_tasks(
    'INITIALISING',
    Task('checking stem version', test.util.check_stem_version),
    Task('checking python version', test.util.check_python_version),
    Task('checking pycrypto version', test.util.check_pycrypto_version),
    Task('checking mock version', test.util.check_mock_version),
    Task('checking pyflakes version', test.util.check_pyflakes_version),
    Task('checking pep8 version', test.util.check_pep8_version),
    Task('checking for orphaned .pyc files', test.util.clean_orphaned_pyc, (SRC_PATHS,)),
    Task('checking for unused tests', test.util.check_for_unused_tests, ((os.path.join(STEM_BASE, 'test'),),)),
    pyflakes_task,
    pep8_task,
  )

  # buffer that we log messages into so they can be printed after a test has finished

  logging_buffer = stem.util.log.LogBuffer(args.logging_runlevel)
  stem.util.log.get_logger().addHandler(logging_buffer)

  # filters for how testing output is displayed

  error_tracker = test.output.ErrorTracker()

  output_filters = (
    error_tracker.get_filter(),
    test.output.strip_module,
    test.output.align_results,
    test.output.colorize,
  )

  # Number of tests that we have skipped. This is only available with python
  # 2.7 or later because before that test results didn't have a 'skipped'
  # attribute.

  skipped_tests = 0

  if args.run_unit:
    test.output.print_divider('UNIT TESTS', True)
    error_tracker.set_category('UNIT TEST')

    for test_class in test.util.get_unit_tests(args.specific_test):
      run_result = _run_test(args, test_class, output_filters, logging_buffer)
      skipped_tests += len(getattr(run_result, 'skipped', []))

    println()
#.........这里部分代码省略.........

 def print_init_line(line):
   println('  %s' % line, SUBSTATUS)

        println("skipped", STATUS)
      else:
        os.makedirs(self._test_dir)
        println("done", STATUS)
    except OSError, exc:
      test.output.print_error("failed (%s)" % exc)
      raise exc

    # Tor checks during startup that the directory a control socket resides in
    # is only accessible by the tor user (and refuses to finish starting if it
    # isn't).

    if Torrc.SOCKET in self._custom_opts:
      try:
        socket_dir = os.path.dirname(CONTROL_SOCKET_PATH)
        println("  making control socket directory (%s)... " % socket_dir, STATUS, NO_NL)

        if os.path.exists(socket_dir) and stat.S_IMODE(os.stat(socket_dir).st_mode) == 0700:
          println("skipped", STATUS)
        else:
          if not os.path.exists(socket_dir):
            os.makedirs(socket_dir)

          os.chmod(socket_dir, 0700)
          println("done", STATUS)
      except OSError, exc:
        test.output.print_error("failed (%s)" % exc)
        raise exc

    # configures logging
    logging_path = CONFIG["integ.log"]

def main():
  start_time = time.time()

  try:
    stem.prereq.check_requirements()
  except ImportError as exc:
    println("%s\n" % exc)
    sys.exit(1)

  test_config = stem.util.conf.get_config("test")
  test_config.load(os.path.join(STEM_BASE, "test", "settings.cfg"))

  try:
    args = _get_args(sys.argv[1:])
  except getopt.GetoptError as exc:
    println("%s (for usage provide --help)" % exc)
    sys.exit(1)
  except ValueError as exc:
    println(str(exc))
    sys.exit(1)

  if args.print_help:
    println(test.util.get_help_message())
    sys.exit()
  elif not args.run_unit and not args.run_integ and not args.run_style:
    println("Nothing to run (for usage provide --help)\n")
    sys.exit()

  if not stem.prereq.is_mock_available():
    try:
      import mock
      println(MOCK_OUT_OF_DATE_MSG % mock.__version__)
    except ImportError:
      println(MOCK_UNAVAILABLE_MSG)

    if stem.util.system.is_available('pip'):
      println("You can get it by running 'sudo pip install mock'.")
    elif stem.util.system.is_available('apt-get'):
      println("You can get it by running 'sudo apt-get install python-mock'.")

    sys.exit(1)

  test.util.run_tasks(
    "INITIALISING",
    Task("checking stem version", test.util.check_stem_version),
    Task("checking python version", test.util.check_python_version),
    Task("checking pycrypto version", test.util.check_pycrypto_version),
    Task("checking mock version", test.util.check_mock_version),
    Task("checking pyflakes version", test.util.check_pyflakes_version),
    Task("checking pep8 version", test.util.check_pep8_version),
    Task("checking for orphaned .pyc files", test.util.clean_orphaned_pyc, (SRC_PATHS,)),
    Task("checking for unused tests", test.util.check_for_unused_tests, ((os.path.join(STEM_BASE, 'test'),),)),
  )

  if args.run_python3 and sys.version_info[0] != 3:
    test.util.run_tasks(
      "EXPORTING TO PYTHON 3",
      Task("checking requirements", test.util.python3_prereq),
      Task("cleaning prior export", test.util.python3_clean, (not args.run_python3_clean,)),
      Task("exporting python 3 copy", test.util.python3_copy_stem),
      Task("running tests", test.util.python3_run_tests),
    )

    println("BUG: python3_run_tests() should have terminated our process", ERROR)
    sys.exit(1)

  # buffer that we log messages into so they can be printed after a test has finished

  logging_buffer = stem.util.log.LogBuffer(args.logging_runlevel)
  stem.util.log.get_logger().addHandler(logging_buffer)

  # filters for how testing output is displayed

  error_tracker = test.output.ErrorTracker()

  output_filters = (
    error_tracker.get_filter(),
    test.output.strip_module,
    test.output.align_results,
    test.output.colorize,
  )

  # Number of tests that we have skipped. This is only available with python
  # 2.7 or later because before that test results didn't have a 'skipped'
  # attribute.

  skipped_tests = 0

  if args.run_unit:
    test.output.print_divider("UNIT TESTS", True)
    error_tracker.set_category("UNIT TEST")

    for test_class in test.util.get_unit_tests(args.test_prefix):
      run_result = _run_test(test_class, output_filters, logging_buffer)
      skipped_tests += len(getattr(run_result, 'skipped', []))

    println()

  if args.run_integ:
    test.output.print_divider("INTEGRATION TESTS", True)
#.........这里部分代码省略.........

def _run_test(args, test_class, output_filters, logging_buffer):
  start_time = time.time()

  if args.verbose:
    test.output.print_divider(test_class)
  else:
    # Test classes look like...
    #
    #   test.unit.util.conf.TestConf.test_parse_enum_csv
    #
    # We want to strip the 'test.unit.' or 'test.integ.' prefix since it's
    # redundant. We also want to drop the test class name. The individual test
    # name at the end it optional (only present if we used the '--test'
    # argument).

    label_comp = test_class.split('.')[2:]
    del label_comp[-1 if label_comp[-1][0].isupper() else -2]
    label = '.'.join(label_comp)

    label = '  %s...' % label
    label = '%-54s' % label

    println(label, STATUS, NO_NL)

  try:
    suite = unittest.TestLoader().loadTestsFromName(test_class)
  except AttributeError:
    # should only come up if user provided '--test' for something that doesn't exist
    println(' no such test', ERROR)
    return None
  except Exception as exc:
    println(' failed', ERROR)
    traceback.print_exc(exc)
    return None

  test_results = StringIO()
  run_result = unittest.TextTestRunner(test_results, verbosity=2).run(suite)

  if args.verbose:
    println(test.output.apply_filters(test_results.getvalue(), *output_filters))
  elif not run_result.failures and not run_result.errors:
    println(' success (%0.2fs)' % (time.time() - start_time), SUCCESS)
  else:
    if args.quiet:
      println(label, STATUS, NO_NL, STDERR)
      println(' failed (%0.2fs)' % (time.time() - start_time), ERROR, STDERR)
      println(test.output.apply_filters(test_results.getvalue(), *output_filters), STDERR)
    else:
      println(' failed (%0.2fs)' % (time.time() - start_time), ERROR)
      println(test.output.apply_filters(test_results.getvalue(), *output_filters), NO_NL)

  test.output.print_logging(logging_buffer)

  return run_result