text: #!/usr/bin/env python
  from __future__ import print_function, absolute_import, division
  
  import sys
  import os
  import glob
  import traceback
  import time
  import importlib
  from datetime import timedelta
  
  from multiprocessing.pool import ThreadPool
  from multiprocessing import cpu_count
  from . import util
  from .util import log
  from .sysinfo import RUNNING_ON_CI
  from .sysinfo import PYPY
  from .sysinfo import PY2
  from .sysinfo import RESOLVER_ARES
  from .sysinfo import RUN_LEAKCHECKS
  from . import six
  
  # Import this while we're probably single-threaded/single-processed
  # to try to avoid issues with PyPy 5.10.
  # See https://bitbucket.org/pypy/pypy/issues/2769/systemerror-unexpected-internal-exception
  try:
      __import__('_testcapi')
  except (ImportError, OSError, IOError):
      # This can raise a wide variety of errors
      pass
  
  TIMEOUT = 100
  NWORKERS = int(os.environ.get('NWORKERS') or max(cpu_count() - 1, 4))
  if NWORKERS > 10:
      NWORKERS = 10
  
  if RUN_LEAKCHECKS:
      # Capturing the stats takes time, and we run each
      # test at least twice
      TIMEOUT = 200
  
  DEFAULT_RUN_OPTIONS = {
      'timeout': TIMEOUT
  }
  
  
  if RUNNING_ON_CI:
      # Too many and we get spurious timeouts
      NWORKERS = 4
  
  
  
  
  def _package_relative_filename(filename, package):
      if not os.path.isfile(filename) and package:
          # Ok, try to locate it as a module in the package
          package_dir = _dir_from_package_name(package)
          return os.path.join(package_dir, filename)
      return filename
  
  def _dir_from_package_name(package):
      package_mod = importlib.import_module(package)
      package_dir = os.path.dirname(package_mod.__file__)
      return package_dir
  
  
  def run_many(tests,
               configured_failing_tests=(),
               failfast=False,
               quiet=False,
               configured_run_alone_tests=()):
      # pylint:disable=too-many-locals,too-many-statements
      global NWORKERS
      start = time.time()
      total = 0
      failed = {}
      passed = {}
      total_cases = [0]
      total_skipped = [0]
  
      NWORKERS = min(len(tests), NWORKERS) or 1
  
      pool = ThreadPool(NWORKERS)
      util.BUFFER_OUTPUT = NWORKERS > 1 or quiet
  
      def run_one(cmd, **kwargs):
          kwargs['quiet'] = quiet
          result = util.run(cmd, **kwargs)
          if result:
              if failfast:
                  sys.exit(1)
              failed[result.name] = [cmd, kwargs]
          else:
              passed[result.name] = True
          total_cases[0] += result.run_count
          total_skipped[0] += result.skipped_count
  
      results = []
  
      def reap():
          for r in results[:]:
              if not r.ready():
                  continue
              if r.successful():
                  results.remove(r)
              else:
                  r.get()
                  sys.exit('Internal error in testrunner.py: %r' % (r, ))
          return len(results)
  
      def reap_all():
          while reap() > 0:
              time.sleep(0.1)
  
      def spawn(cmd, options):
          while True:
              if reap() < NWORKERS:
                  r = pool.apply_async(run_one, (cmd, ), options or {})
                  results.append(r)
                  return
  
              time.sleep(0.05)
  
      run_alone = []
  
      try:
          try:
              log("Running tests in parallel with concurrency %s" % (NWORKERS,),)
              for cmd, options in tests:
                  total += 1
                  options = options or {}
                  if matches(configured_run_alone_tests, cmd):
                      run_alone.append((cmd, options))
                  else:
                      spawn(cmd, options)
              pool.close()
              pool.join()
  
              log("Running tests marked standalone")
              for cmd, options in run_alone:
                  run_one(cmd, **options)
  
          except KeyboardInterrupt:
              try:
                  log('Waiting for currently running to finish...')
                  reap_all()
              except KeyboardInterrupt:
                  pool.terminate()
                  report(total, failed, passed, exit=False, took=time.time() - start,
                         configured_failing_tests=configured_failing_tests,
                         total_cases=total_cases[0], total_skipped=total_skipped[0])
                  log('(partial results)\n')
                  raise
      except:
          traceback.print_exc()
          pool.terminate()
          raise
  
      reap_all()
      report(total, failed, passed, took=time.time() - start,
             configured_failing_tests=configured_failing_tests,
             total_cases=total_cases[0], total_skipped=total_skipped[0])
  
  def discover(
          tests=None, ignore_files=None,
          ignored=(), coverage=False,
          package=None,
          configured_ignore_coverage=(),
          configured_test_options=None,
  ):
      # pylint:disable=too-many-locals,too-many-branches
      configured_test_options = configured_test_options or {}
      olddir = os.getcwd()
      ignore = set(ignored or ())
  
      if ignore_files:
          ignore_files = ignore_files.split(',')
          for f in ignore_files:
              ignore.update(set(load_list_from_file(f, package)))
  
      if coverage:
          ignore.update(configured_ignore_coverage)
  
      if package:
          package_dir = _dir_from_package_name(package)
          # We need to glob relative names, our config is based on filenames still
          os.chdir(package_dir)
  
      if not tests:
          tests = set(glob.glob('test_*.py')) - set(['test_support.py'])
      else:
          tests = set(tests)
  
      if ignore:
          # Always ignore the designated list, even if tests were specified
          # on the command line. This fixes a nasty interaction with test__threading_vs_settrace.py
          # being run under coverage when 'grep -l subprocess test*py' is used to list the tests
          # to run.
          tests -= ignore
      tests = sorted(tests)
  
      to_process = []
      to_import = []
  
      for filename in tests:
          module_name = os.path.splitext(filename)[0]
          qualified_name = package + '.' + module_name if package else module_name
          with open(os.path.abspath(filename), 'rb') as f:
              # Some of the test files (e.g., test__socket_dns) are
              # UTF8 encoded. Depending on the environment, Python 3 may
              # try to decode those as ASCII, which fails with UnicodeDecodeError.
              # Thus, be sure to open and compare in binary mode.
              # Open the absolute path to make errors more clear,
              # but we can't store the absolute path, our configuration is based on
              # relative file names.
              contents = f.read()
          if b'TESTRUNNER' in contents: # test__monkey_patching.py
              # XXX: Rework this to avoid importing.
              to_import.append(qualified_name)
          else:
              cmd = [sys.executable, '-u']
              if PYPY and PY2:
                  # Doesn't seem to be an env var for this
                  cmd.extend(('-X', 'track-resources'))
              if package:
                  # Using a package is the best way to work with coverage 5
                  # when we specify 'source = <package>'
                  cmd.append('-m' + qualified_name)
              else:
                  cmd.append(filename)
  
              options = DEFAULT_RUN_OPTIONS.copy()
              options.update(configured_test_options.get(filename, {}))
              to_process.append((cmd, options))
  
      os.chdir(olddir)
      # When we actually execute, do so from the original directory,
      # this helps find setup.py
      for qualified_name in to_import:
          module = importlib.import_module(qualified_name)
          for cmd, options in module.TESTRUNNER():
              if remove_options(cmd)[-1] in ignore:
                  continue
              to_process.append((cmd, options))
  
      return to_process
  
  
  def remove_options(lst):
      return [x for x in lst if x and not x.startswith('-')]
  
  def load_list_from_file(filename, package):
      result = []
      if filename:
          with open(_package_relative_filename(filename, package)) as f:
              for x in f:
                  x = x.split('#', 1)[0].strip()
                  if x:
                      result.append(x)
      return result
  
  
  def matches(possibilities, command, include_flaky=True):
      if isinstance(command, list):
          command = ' '.join(command)
      for line in possibilities:
          if not include_flaky and line.startswith('FLAKY '):
              continue
          line = line.replace('FLAKY ', '')
          # Our configs are still mostly written in terms of file names,
          # but the non-monkey tests are now using package names.
          # Strip off '.py' from filenames to see if we match a module.
          # XXX: This could be much better. Our command needs better structure.
          if command.endswith(' ' + line) or command.endswith(line.replace(".py", '')):
              return True
      return False
  
  
  def format_seconds(seconds):
      if seconds < 20:
          return '%.1fs' % seconds
      seconds = str(timedelta(seconds=round(seconds)))
      if seconds.startswith('0:'):
          seconds = seconds[2:]
      return seconds
  
  
  def report(total, failed, passed, exit=True, took=None,
             configured_failing_tests=(),
             total_cases=0, total_skipped=0):
      # pylint:disable=redefined-builtin,too-many-branches,too-many-locals
      runtimelog = util.runtimelog
      if runtimelog:
          log('\nLongest-running tests:')
          runtimelog.sort()
          length = len('%.1f' % -runtimelog[0][0])
          frmt = '%' + str(length) + '.1f seconds: %s'
          for delta, name in runtimelog[:5]:
              log(frmt, -delta, name)
      if took:
          took = ' in %s' % format_seconds(took)
      else:
          took = ''
  
      failed_expected = []
      failed_unexpected = []
      passed_unexpected = []
  
      for name in passed:
          if matches(configured_failing_tests, name, include_flaky=False):
              passed_unexpected.append(name)
  
      if passed_unexpected:
          log('\n%s/%s unexpected passes', len(passed_unexpected), total, color='error')
          print_list(passed_unexpected)
  
      if failed:
          log('\n%s/%s tests failed%s', len(failed), total, took)
  
          for name in failed:
              if matches(configured_failing_tests, name, include_flaky=True):
                  failed_expected.append(name)
              else:
                  failed_unexpected.append(name)
  
          if failed_expected:
              log('\n%s/%s expected failures', len(failed_expected), total)
              print_list(failed_expected)
  
          if failed_unexpected:
              log('\n%s/%s unexpected failures', len(failed_unexpected), total, color='error')
              print_list(failed_unexpected)
      else:
          log(
              '\nRan %s tests%s in %s files%s',
              total_cases,
              util._colorize('skipped', " (skipped=%d)" % total_skipped) if total_skipped else '',
              total,
              took,
          )
  
      if exit:
          if failed_unexpected:
              sys.exit(min(100, len(failed_unexpected)))
          if passed_unexpected:
              sys.exit(101)
          if total <= 0:
              sys.exit('No tests found.')
  
  
  def print_list(lst):
      for name in lst:
          log(' - %s', name)
  
  def _setup_environ(debug=False):
      if 'PYTHONWARNINGS' not in os.environ and not sys.warnoptions:
  
          # action:message:category:module:line
          os.environ['PYTHONWARNINGS'] = ','.join([
              # Enable default warnings such as ResourceWarning.
              'default',
              # On Python 3[.6], the system site.py module has
              # "open(fullname, 'rU')" which produces the warning that
              # 'U' is deprecated, so ignore warnings from site.py
              'ignore:::site:',
              # pkgutil on Python 2 complains about missing __init__.py
              'ignore:::pkgutil',
              # importlib/_bootstrap.py likes to spit out "ImportWarning:
              # can't resolve package from __spec__ or __package__, falling
              # back on __name__ and __path__". I have no idea what that means, but it seems harmless
              # and is annoying.
              'ignore:::importlib._bootstrap:',
              'ignore:::importlib._bootstrap_external:',
              # importing ABCs from collections, not collections.abc
              'ignore:::pkg_resources._vendor.pyparsing:',
          ])
  
      if 'PYTHONFAULTHANDLER' not in os.environ:
          os.environ['PYTHONFAULTHANDLER'] = 'true'
  
      if 'GEVENT_DEBUG' not in os.environ and debug:
          os.environ['GEVENT_DEBUG'] = 'debug'
  
      if 'PYTHONTRACEMALLOC' not in os.environ:
          os.environ['PYTHONTRACEMALLOC'] = '10'
  
      if 'PYTHONDEVMODE' not in os.environ:
          # Python 3.7
          os.environ['PYTHONDEVMODE'] = '1'
  
      if 'PYTHONMALLOC' not in os.environ:
          # Python 3.6
          os.environ['PYTHONMALLOC'] = 'debug'
  
  
  
  def main():
      # pylint:disable=too-many-locals,too-many-statements
      import argparse
      parser = argparse.ArgumentParser()
      parser.add_argument('--ignore')
      parser.add_argument('--discover', action='store_true')
      parser.add_argument('--full', action='store_true')
      parser.add_argument('--config', default='known_failures.py')
      parser.add_argument('--failfast', action='store_true')
      parser.add_argument("--coverage", action="store_true")
      parser.add_argument("--quiet", action="store_true", default=True)
      parser.add_argument("--verbose", action="store_false", dest='quiet')
      parser.add_argument("--debug", action="store_true", default=False)
      parser.add_argument("--package", default="gevent.tests")
      parser.add_argument('tests', nargs='*')
      options = parser.parse_args()
      FAILING_TESTS = []
      IGNORED_TESTS = []
      RUN_ALONE = []
      TEST_FILE_OPTIONS = {}
  
      coverage = False
      if options.coverage or os.environ.get("GEVENTTEST_COVERAGE"):
          coverage = True
          os.environ['COVERAGE_PROCESS_START'] = os.path.abspath(".coveragerc")
          if PYPY:
              os.environ['COVERAGE_PROCESS_START'] = os.path.abspath(".coveragerc-pypy")
          this_dir = os.path.dirname(__file__)
          site_dir = os.path.join(this_dir, 'coveragesite')
          site_dir = os.path.abspath(site_dir)
          os.environ['PYTHONPATH'] = site_dir + os.pathsep + os.environ.get("PYTHONPATH", "")
          # We change directory often, use an absolute path to keep all the
          # coverage files (which will have distinct suffixes because of parallel=true in .coveragerc
          # in this directory; makes them easier to combine and use with coverage report)
          os.environ['COVERAGE_FILE'] = os.path.abspath(".") + os.sep + ".coverage"
          print("Enabling coverage to", os.environ['COVERAGE_FILE'], "with site", site_dir)
  
      _setup_environ(debug=options.debug)
  
      if options.config:
          config = {}
          options.config = _package_relative_filename(options.config, options.package)
          with open(options.config) as f:
              config_data = f.read()
          six.exec_(config_data, config)
          FAILING_TESTS = config['FAILING_TESTS']
          IGNORED_TESTS = config['IGNORED_TESTS']
          RUN_ALONE = config['RUN_ALONE']
          TEST_FILE_OPTIONS = config['TEST_FILE_OPTIONS']
          IGNORE_COVERAGE = config['IGNORE_COVERAGE']
  
  
      tests = discover(
          options.tests,
          ignore_files=options.ignore,
          ignored=IGNORED_TESTS,
          coverage=coverage,
          package=options.package,
          configured_ignore_coverage=IGNORE_COVERAGE,
          configured_test_options=TEST_FILE_OPTIONS,
      )
      if options.discover:
          for cmd, options in tests:
              print(util.getname(cmd, env=options.get('env'), setenv=options.get('setenv')))
          print('%s tests found.' % len(tests))
      else:
          if PYPY and RESOLVER_ARES:
              # XXX: Add a way to force these.
              print("Not running tests on pypy with c-ares; not a supported configuration")
              return
          if options.package:
              # Put this directory on the path so relative imports work.
              package_dir = _dir_from_package_name(options.package)
              os.environ['PYTHONPATH'] = os.environ.get('PYTHONPATH', "") + os.pathsep + package_dir
          run_many(
              tests,
              configured_failing_tests=FAILING_TESTS,
              failfast=options.failfast,
              quiet=options.quiet,
              configured_run_alone_tests=RUN_ALONE,
          )
  
  
  if __name__ == '__main__':
      main()
  