text: #
  # Licensed to the Apache Software Foundation (ASF) under one
  # or more contributor license agreements.  See the NOTICE file
  # distributed with this work for additional information
  # regarding copyright ownership.  The ASF licenses this file
  # to you 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.
  #
  
  # TODO: summarize, test harness preconditions (e.g. broker is alive)
  
  import optparse
  import os
  import struct
  import sys
  import time
  import traceback
  import types
  import cgi
  from fnmatch import fnmatchcase as match
  from logging import getLogger, StreamHandler, Formatter, Filter, \
      WARN, DEBUG, ERROR, INFO
  
  from .common import SkipTest
  
  if sys.version_info[0] == 3:
      CLASS_TYPES = (type,)
  else:
      CLASS_TYPES = (type, types.ClassType)
  
  levels = {
      "DEBUG": DEBUG,
      "INFO": INFO,
      "WARN": WARN,
      "ERROR": ERROR
  }
  
  sorted_levels = [k for k, v in sorted(levels.items(), key=lambda kv: kv[1])]
  
  parser = optparse.OptionParser(usage="usage: %prog [options] PATTERN ...",
                                 description="Run tests matching the specified PATTERNs.")
  parser.add_option("-l", "--list", action="store_true", default=False,
                    help="list tests instead of executing them")
  parser.add_option("-f", "--log-file", metavar="FILE", help="log output to FILE")
  parser.add_option("-v", "--log-level", metavar="LEVEL", default="WARN",
                    help="only display log messages of LEVEL or higher severity: "
                    "%s (default %%default)" % ", ".join(sorted_levels))
  parser.add_option("-c", "--log-category", metavar="CATEGORY", action="append",
                    dest="log_categories", default=[],
                    help="log only categories matching CATEGORY pattern")
  parser.add_option("-m", "--module", action="append", default=[],
                    dest="modules", help="add module to test search path")
  parser.add_option("-i", "--ignore", action="append", default=[],
                    help="ignore tests matching IGNORE pattern")
  parser.add_option("-I", "--ignore-file", metavar="IFILE", action="append",
                    default=[],
                    help="ignore tests matching patterns in IFILE")
  parser.add_option("-H", "--halt-on-error", action="store_true", default=False,
                    dest="hoe", help="halt if an error is encountered")
  parser.add_option("-t", "--time", action="store_true", default=False,
                    help="report timing information on test run")
  parser.add_option("-D", "--define", metavar="DEFINE", dest="defines",
                    action="append", default=[], help="define test parameters")
  parser.add_option("-x", "--xml", metavar="XML", dest="xml",
                    help="write test results in Junit style xml suitable for use by CI tools etc")
  parser.add_option("-a", "--always-colorize", action="store_true", dest="always_colorize", default=False,
                    help="always colorize the test results rather than relying on terminal tty detection. Useful when invoked from Jython/Maven.")
  parser.add_option("-n", metavar="count", dest="count", type=int, default=1,
                    help="run the tests <count> times")
  parser.add_option("-b", "--bare", action="store_true", default=False,
                    help="Run bare, i.e. don't capture stack traces. This is useful under Jython as " +
                    "captured stack traces do not include the Java portion of the stack," +
                    "whereas non captured stack traces do.")
  parser.add_option("-j", "--javatrace", action="store_true", default=False,
                    help="Show the full java stack trace. This disables heuristics to eliminate the " +
                    "jython portion of java stack traces.")
  
  
  class Config:
  
      def __init__(self):
          self.defines = {}
          self.log_file = None
          self.log_level = WARN
          self.log_categories = []
  
  
  opts, args = parser.parse_args()
  
  # Known bad tests, skipped unless specifically matched on the command line
  known_bad = ["proton_tests.messenger.*"]
  
  includes = []
  excludes = ["*__*__"]
  config = Config()
  list_only = opts.list
  for d in opts.defines:
      try:
          idx = d.index("=")
          name = d[:idx]
          value = d[idx + 1:]
          config.defines[name] = value
      except ValueError:
          config.defines[d] = None
  config.log_file = opts.log_file
  config.log_level = levels[opts.log_level.upper()]
  config.log_categories = opts.log_categories
  excludes.extend([v.strip() for v in opts.ignore])
  for v in opts.ignore_file:
      f = open(v)
      for line in f:
          line = line.strip()
          if line.startswith("#"):
              continue
          excludes.append(line)
      f.close()
  
  for a in args:
      includes.append(a.strip())
  
  
  def is_ignored(path):
      for p in excludes:
          if match(path, p):
              return True
      return False
  
  
  def is_included(path):
      if is_ignored(path):
          return False
      for p in includes:
          if match(path, p):
              return True
      for p in known_bad:
          if match(path, p):
              return False
      return not includes           # If includes is empty, include everything
  
  
  def is_smart():
      return sys.stdout.isatty() and os.environ.get("TERM", "dumb") != "dumb"
  
  
  try:
      import fcntl
      import termios
  
      def width():
          if is_smart():
              s = struct.pack("HHHH", 0, 0, 0, 0)
              fd_stdout = sys.stdout.fileno()
              x = fcntl.ioctl(fd_stdout, termios.TIOCGWINSZ, s)
              rows, cols, xpx, ypx = struct.unpack("HHHH", x)
              return cols
          else:
              try:
                  return int(os.environ.get("COLUMNS", "80"))
              except ValueError:
                  return 80
  
      WIDTH = width()
  
      def resize(sig, frm):
          global WIDTH
          WIDTH = width()
  
      import signal
      signal.signal(signal.SIGWINCH, resize)
  
  except ImportError:
      WIDTH = 80
  
  
  def vt100_attrs(*attrs):
      return "\x1B[%sm" % ";".join(map(str, attrs))
  
  
  vt100_reset = vt100_attrs(0)
  
  KEYWORDS = {"pass": (32,),
              "skip": (33,),
              "fail": (31,),
              "start": (34,),
              "total": (34,),
              "ignored": (33,),
              "selected": (34,),
              "elapsed": (34,),
              "average": (34,)}
  
  
  def colorize_word(word, text=None):
      if text is None:
          text = word
      return colorize(text, *KEYWORDS.get(word, ()))
  
  
  def colorize(text, *attrs):
      if attrs and (is_smart() or opts.always_colorize):
          return "%s%s%s" % (vt100_attrs(*attrs), text, vt100_reset)
      else:
          return text
  
  
  def indent(text):
      lines = text.split("\n")
      return "  %s" % "\n  ".join(lines)
  
  # Write a 'minimal' Junit xml style report file suitable for use by CI tools such as Jenkins.
  
  
  class JunitXmlStyleReporter:
  
      def __init__(self, file):
          self.f = open(file, "w")
  
      def begin(self):
          self.f.write('<?xml version="1.0" encoding="UTF-8" ?>\n')
          self.f.write('<testsuite name="pythontests">\n')
  
      def report(self, name, result):
          parts = name.split(".")
          method = parts[-1]
          module = '.'.join(parts[0:-1])
          self.f.write('<testcase classname="%s" name="%s" time="%f">\n' % (module, method, result.time))
          if result.failed:
              escaped_type = cgi.escape(str(result.exception_type))
              escaped_message = cgi.escape(str(result.exception_message))
              self.f.write('<failure type="%s" message="%s">\n' % (escaped_type, escaped_message))
              self.f.write('<![CDATA[\n')
              self.f.write(result.formatted_exception_trace)
              self.f.write(']]>\n')
              self.f.write('</failure>\n')
          if result.skipped:
              self.f.write('<skipped/>\n')
          self.f.write('</testcase>\n')
  
      def end(self):
          self.f.write('</testsuite>\n')
          self.f.close()
  
  
  class Interceptor:
  
      def __init__(self):
          self.newline = False
          self.indent = False
          self.passthrough = True
          self.dirty = False
          self.last = None
  
      def begin(self):
          self.newline = True
          self.indent = True
          self.passthrough = False
          self.dirty = False
          self.last = None
  
      def reset(self):
          self.newline = False
          self.indent = False
          self.passthrough = True
  
  
  class StreamWrapper:
  
      def __init__(self, interceptor, stream, prefix="  "):
          self.interceptor = interceptor
          self.stream = stream
          self.prefix = prefix
  
      def fileno(self):
          return self.stream.fileno()
  
      def isatty(self):
          return self.stream.isatty()
  
      def write(self, s):
          if self.interceptor.passthrough:
              self.stream.write(s)
              return
  
          if s:
              self.interceptor.dirty = True
  
          if self.interceptor.newline:
              self.interceptor.newline = False
              self.stream.write(" %s\n" % colorize_word("start"))
              self.interceptor.indent = True
          if self.interceptor.indent:
              self.stream.write(self.prefix)
          if s.endswith("\n"):
              s = s.replace("\n", "\n%s" % self.prefix)[:-2]
              self.interceptor.indent = True
          else:
              s = s.replace("\n", "\n%s" % self.prefix)
              self.interceptor.indent = False
          self.stream.write(s)
  
          if s:
              self.interceptor.last = s[-1]
  
      def flush(self):
          self.stream.flush()
  
  
  interceptor = Interceptor()
  
  out_wrp = StreamWrapper(interceptor, sys.stdout)
  err_wrp = StreamWrapper(interceptor, sys.stderr)
  
  out = sys.stdout
  err = sys.stderr
  sys.stdout = out_wrp
  sys.stderr = err_wrp
  
  
  class PatternFilter(Filter):
  
      def __init__(self, *patterns):
          Filter.__init__(self, patterns)
          self.patterns = patterns
  
      def filter(self, record):
          if not self.patterns:
              return True
          for p in self.patterns:
              if match(record.name, p):
                  return True
          return False
  
  
  root = getLogger()
  handler = StreamHandler(sys.stdout)
  filter = PatternFilter(*config.log_categories)
  handler.addFilter(filter)
  handler.setFormatter(Formatter("%(asctime)s %(levelname)s %(message)s"))
  root.addHandler(handler)
  root.setLevel(WARN)
  
  log = getLogger("proton.test")
  
  PASS = "pass"
  SKIP = "skip"
  FAIL = "fail"
  
  
  class Runner:
  
      def __init__(self):
          self.exception = None
          self.exception_phase_name = None
          self.skip = False
  
      def passed(self):
          return not self.exception
  
      def skipped(self):
          return self.skip
  
      def failed(self):
          return self.exception and not self.skip
  
      def halt(self):
          """determines if the overall test execution should be allowed to continue to the next phase"""
          return self.exception or self.skip
  
      def run(self, phase_name, phase):
          """invokes a test-phase method (which can be the test method itself or a setUp/tearDown
             method).  If the method raises an exception the exception is examined to see if the
             exception should be classified as a 'skipped' test"""
          # we don't try to catch exceptions for jython because currently a
          # jython bug will prevent the java portion of the stack being
          # stored with the exception info in the sys module
          if opts.bare:
              phase()
          else:
              try:
                  phase()
              except KeyboardInterrupt:
                  raise
              except:
                  self.exception_phase_name = phase_name
                  self.exception = sys.exc_info()
                  exception_type = self.exception[0]
                  self.skip = getattr(exception_type, "skipped", False) or issubclass(exception_type, SkipTest)
  
      def status(self):
          if self.passed():
              return PASS
          elif self.skipped():
              return SKIP
          elif self.failed():
              return FAIL
          else:
              return None
  
      def get_formatted_exception_trace(self):
          if self.exception:
              if self.skip:
                  # format skipped tests without a traceback
                  output = indent("".join(traceback.format_exception_only(*self.exception[:2]))).rstrip()
              else:
                  output = "Error during %s:" % self.exception_phase_name
                  lines = traceback.format_exception(*self.exception)
                  val = self.exception[1]
                  reflect = False
                  if val and hasattr(val, "getStackTrace"):
                      jlines = []
                      for frame in val.getStackTrace():
                          if reflect and frame.getClassName().startswith("org.python."):
                              # this is a heuristic to eliminate the jython portion of
                              # the java stack trace
                              if not opts.javatrace:
                                  break
                          jlines.append("  %s\n" % frame.toString())
                          if frame.getClassName().startswith("java.lang.reflect"):
                              reflect = True
                          else:
                              reflect = False
                      lines.extend(jlines)
                  output += indent("".join(lines)).rstrip()
              return output
  
      def get_exception_type(self):
          if self.exception:
              return self.exception[0]
          else:
              return None
  
      def get_exception_message(self):
          if self.exception:
              return self.exception[1]
          else:
              return None
  
  
  ST_WIDTH = 8
  
  
  def run_test(name, test, config):
      patterns = filter.patterns
      level = root.level
      filter.patterns = config.log_categories
      root.setLevel(config.log_level)
  
      parts = name.split(".")
      line = None
      output = ""
      for part in parts:
          if line:
              if len(line) + len(part) >= (WIDTH - ST_WIDTH - 1):
                  output += "%s. \\\n" % line
                  line = "    %s" % part
              else:
                  line = "%s.%s" % (line, part)
          else:
              line = part
  
      if line:
          output += "%s %s" % (line, (((WIDTH - ST_WIDTH) - len(line)) * "."))
      sys.stdout.write(output)
      sys.stdout.flush()
      interceptor.begin()
      start = time.time()
      try:
          runner = test()
      finally:
          interceptor.reset()
      end = time.time()
      if interceptor.dirty:
          if interceptor.last != "\n":
              sys.stdout.write("\n")
          sys.stdout.write(output)
      print(" %s" % colorize_word(runner.status()))
      if runner.failed() or runner.skipped():
          print(runner.get_formatted_exception_trace())
      root.setLevel(level)
      filter.patterns = patterns
      return TestResult(end - start,
                        runner.passed(),
                        runner.skipped(),
                        runner.failed(),
                        runner.get_exception_type(),
                        runner.get_exception_message(),
                        runner.get_formatted_exception_trace())
  
  
  class TestResult:
  
      def __init__(self, time, passed, skipped, failed, exception_type, exception_message, formatted_exception_trace):
          self.time = time
          self.passed = passed
          self.skipped = skipped
          self.failed = failed
          self.exception_type = exception_type
          self.exception_message = exception_message
          self.formatted_exception_trace = formatted_exception_trace
  
  
  class FunctionTest:
  
      def __init__(self, test):
          self.test = test
  
      def name(self):
          return "%s.%s" % (self.test.__module__, self.test.__name__)
  
      def run(self):
          return run_test(self.name(), self._run, config)
  
      def _run(self):
          runner = Runner()
          runner.run("test", lambda: self.test(config))
          return runner
  
      def __repr__(self):
          return "FunctionTest(%r)" % self.test
  
  
  class MethodTest:
  
      def __init__(self, cls, method):
          self.cls = cls
          self.method = method
  
      def name(self):
          return "%s.%s.%s" % (self.cls.__module__, self.cls.__name__, self.method)
  
      def run(self):
          return run_test(self.name(), self._run, config)
  
      def _run(self):
          runner = Runner()
          inst = self.cls(self.method)
          test = getattr(inst, self.method)
  
          if hasattr(inst, "configure"):
              runner.run("configure", lambda: inst.configure(config))
              if runner.halt():
                  return runner
          if hasattr(inst, "setUp"):
              runner.run("setup", inst.setUp)
              if runner.halt():
                  return runner
  
          runner.run("test", test)
  
          if hasattr(inst, "tearDown"):
              runner.run("teardown", inst.tearDown)
  
          return runner
  
      def __repr__(self):
          return "MethodTest(%r, %r)" % (self.cls, self.method)
  
  
  class PatternMatcher:
  
      def __init__(self, *patterns):
          self.patterns = patterns
  
      def matches(self, name):
          for p in self.patterns:
              if match(name, p):
                  return True
          return False
  
  
  class FunctionScanner(PatternMatcher):
  
      def inspect(self, obj):
          return type(obj) == types.FunctionType and self.matches(obj.__name__)
  
      def descend(self, func):
          # the None is required for older versions of python
          return
          yield None
  
      def extract(self, func):
          yield FunctionTest(func)
  
  
  class ClassScanner(PatternMatcher):
  
      def inspect(self, obj):
          return type(obj) in CLASS_TYPES and self.matches(obj.__name__)
  
      def descend(self, cls):
          # the None is required for older versions of python
          return
          yield None
  
      def extract(self, cls):
          for name in sorted(dir(cls)):
              obj = getattr(cls, name)
              if hasattr(obj, '__call__') and name.startswith("test"):
                  yield MethodTest(cls, name)
  
  
  class ModuleScanner:
  
      def inspect(self, obj):
          return type(obj) == types.ModuleType
  
      def descend(self, obj):
          for name in sorted(dir(obj)):
              yield getattr(obj, name)
  
      def extract(self, obj):
          # the None is required for older versions of python
          return
          yield None
  
  
  class Harness:
  
      def __init__(self):
          self.scanners = [
              ModuleScanner(),
              ClassScanner("*Test", "*Tests", "*TestCase"),
              FunctionScanner("test_*")
          ]
          self.tests = []
          self.scanned = []
  
      def scan(self, *roots):
          objects = list(roots)
  
          while objects:
              obj = objects.pop(0)
              for s in self.scanners:
                  if s.inspect(obj):
                      self.tests.extend(s.extract(obj))
                      for child in s.descend(obj):
                          if not (child in self.scanned or child in objects):
                              objects.append(child)
              self.scanned.append(obj)
  
  
  modules = opts.modules
  if not modules:
      modules.extend(["proton_tests"])
  h = Harness()
  for name in modules:
      m = __import__(name, None, None, ["dummy"])
      h.scan(m)
  
  filtered = [t for t in h.tests if is_included(t.name())]
  ignored = [t for t in h.tests if is_ignored(t.name())]
  total = len(filtered) + len(ignored)
  
  if opts.xml and not list_only:
      xmlr = JunitXmlStyleReporter(opts.xml)
      xmlr.begin()
  else:
      xmlr = None
  
  
  def runthrough():
      passed = 0
      failed = 0
      skipped = 0
      start = time.time()
      for t in filtered:
          if list_only:
              print(t.name())
          else:
              st = t.run()
              if xmlr:
                  xmlr.report(t.name(), st)
              if st.passed:
                  passed += 1
              elif st.skipped:
                  skipped += 1
              elif st.failed:
                  failed += 1
                  if opts.hoe:
                      break
      end = time.time()
  
      run = passed + failed
  
      if not list_only:
          if passed:
              _pass = "pass"
          else:
              _pass = "fail"
          if failed:
              outcome = "fail"
          else:
              outcome = "pass"
          if ignored:
              ign = "ignored"
          else:
              ign = "pass"
          if skipped:
              skip = "skip"
          else:
              skip = "pass"
          sys.stdout.write(colorize("Totals: ", 1))
          totals = [colorize_word("total", "%s tests" % total),
                    colorize_word(_pass, "%s passed" % passed),
                    colorize_word(skip, "%s skipped" % skipped),
                    colorize_word(ign, "%s ignored" % len(ignored)),
                    colorize_word(outcome, "%s failed" % failed)]
          sys.stdout.write(", ".join(totals))
          if opts.hoe and failed > 0:
              print(" -- (halted after %s)" % run)
          else:
              print("")
          if opts.time and run > 0:
              sys.stdout.write(colorize("Timing:", 1))
              timing = [colorize_word("elapsed", "%.2fs elapsed" % (end - start)),
                        colorize_word("average", "%.2fs average" % ((end - start) / run))]
              print(", ".join(timing))
  
      if xmlr:
          xmlr.end()
  
      return failed
  
  
  def main():
      limit = opts.count
      count = 0
      failures = False
      while limit == 0 or count < limit:
          count += 1
          if runthrough():
              failures = True
              if count > 1:
                  print(" -- (failures after %s runthroughs)" % count)
          else:
              continue
  
      if failures:
          return(1)
      else:
          return(0)
  