RELAND of Cross-repo recipe package system.

third_party/recipe_engine/expect_tests/pipeline.py

-# Copyright 2014 The Chromium Authors. All rights reserved.
-# Use of this source code is governed by a BSD-style license that can be
-# found in the LICENSE file.
-
-import Queue
-import glob
-import logging
-import multiprocessing
-import re
-import signal
-import traceback
-
-from cStringIO import StringIO
-
-from .type_definitions import (
-    Test, UnknownError, TestError, NoMatchingTestsError,
-    Result, ResultStageAbort)
-
-from . import util
-
-
-class ResetableStringIO(object):
-  def __init__(self):
-    self._stream = StringIO()
-
-  def reset(self):
-    self._stream = StringIO()
-
-  def __getattr__(self, key):
-    return getattr(self._stream, key)
-
-
-def gen_loop_process(gen, test_queue, result_queue, opts, kill_switch,
-                     cover_ctx):
-  """Generate `Test`'s from |gen|, and feed them into |test_queue|.
-
-  Non-Test instances will be translated into `UnknownError` objects.
-
-  On completion, feed |opts.jobs| None objects into |test_queue|.
-
-  @param gen: generator yielding Test() instances.
-  @type test_queue: multiprocessing.Queue()
-  @type result_queue: multiprocessing.Queue()
-  @type opts: argparse.Namespace
-  @type kill_switch: multiprocessing.Event()
-  @type cover_ctx: cover.CoverageContext().create_subprocess_context()
-  """
-  # Implicitly append '*'' to globs that don't specify it.
-  globs = ['%s%s' % (g, '*' if '*' not in g else '') for g in opts.test_glob]
-
-  matcher = re.compile(
-      '^%s$' % '|'.join('(?:%s)' % glob.fnmatch.translate(g)
-                        for g in globs if g[0] != '-'))
-  if matcher.pattern == '^$':
-    matcher = re.compile('^.*$')
-
-  neg_matcher = re.compile(
-      '^%s$' % '|'.join('(?:%s)' % glob.fnmatch.translate(g[1:])
-                        for g in globs if g[0] == '-'))
-
-  def generate_tests():
-    paths_seen = set()
-    seen_tests = False
-    try:
-      with cover_ctx:
-        gen_inst = gen()
-
-      while not kill_switch.is_set():
-        with cover_ctx:
-          root_test = next(gen_inst)
-
-        if kill_switch.is_set():
-          break
-
-        ok_tests = []
-        subtests = root_test.tests
-
-        for subtest in subtests:
-          if not isinstance(subtest, Test):
-            result_queue.put_nowait(
-                UnknownError('Got non-[Multi]Test isinstance from generator: %r'
-                             % subtest))
-            continue
-
-          test_path = subtest.expect_path()
-          if test_path is not None and test_path in paths_seen:
-            result_queue.put_nowait(
-                TestError(subtest, 'Duplicate expectation path!'))
-          else:
-            if test_path is not None:
-              paths_seen.add(test_path)
-            name = subtest.name
-            if not neg_matcher.match(name) and matcher.match(name):
-              ok_tests.append(subtest)
-
-        if ok_tests:
-          seen_tests = True
-          yield root_test.restrict(ok_tests)
-
-      if not seen_tests:
-        result_queue.put_nowait(NoMatchingTestsError())
-    except StopIteration:
-      pass
-    except KeyboardInterrupt:
-      pass
-    finally:
-      for _ in xrange(opts.jobs):
-        test_queue.put_nowait(None)
-
-
-  next_stage = (result_queue if opts.handler.SKIP_RUNLOOP else test_queue)
-  opts.handler.gen_stage_loop(opts, generate_tests(), next_stage.put_nowait,
-                              result_queue.put_nowait)
-
-
-def run_loop_process(test_queue, result_queue, opts, kill_switch, cover_ctx):
-  """Consume `Test` instances from |test_queue|, run them, and yield the results
-  into opts.run_stage_loop().
-
-  Generates coverage data as a side-effect.
-
-  @type test_queue: multiprocessing.Queue()
-  @type result_queue: multiprocessing.Queue()
-  @type opts: argparse.Namespace
-  @type kill_switch: multiprocessing.Event()
-  @type cover_ctx: cover.CoverageContext().create_subprocess_context()
-  """
-  logstream = ResetableStringIO()
-  logger = logging.getLogger()
-  logger.setLevel(logging.DEBUG)
-  shandler = logging.StreamHandler(logstream)
-  shandler.setFormatter(
-      logging.Formatter('%(levelname)s: %(message)s'))
-  logger.addHandler(shandler)
-
-  SKIP = object()
-  def process_test(subtest):
-    logstream.reset()
-    with cover_ctx(include=subtest.coverage_includes()):
-      subresult = subtest.run()
-    if isinstance(subresult, TestError):
-      result_queue.put_nowait(subresult)
-      return SKIP
-    elif not isinstance(subresult, Result):
-      result_queue.put_nowait(
-          TestError(
-              subtest,
-              'Got non-Result instance from test: %r' % subresult))
-      return SKIP
-    return subresult
-
-  def generate_tests_results():
-    try:
-      while not kill_switch.is_set():
-        try:
-          test = test_queue.get(timeout=0.1)
-          if test is None:
-            break
-        except Queue.Empty:
-          continue
-
-        try:
-          for subtest, subresult in test.process(process_test):
-            if subresult is not SKIP:
-              yield subtest, subresult, logstream.getvalue().splitlines()
-        except Exception:
-          result_queue.put_nowait(
-              TestError(test, traceback.format_exc(),
-                        logstream.getvalue().splitlines()))
-    except KeyboardInterrupt:
-      pass
-
-  opts.handler.run_stage_loop(opts, generate_tests_results(),
-                              result_queue.put_nowait)
-
-
-def result_loop(test_gen, cover_ctx, opts):
-  kill_switch = multiprocessing.Event()
-  def handle_killswitch(*_):
-    kill_switch.set()
-    # Reset the signal to DFL so that double ctrl-C kills us for sure.
-    signal.signal(signal.SIGINT, signal.SIG_DFL)
-    signal.signal(signal.SIGTERM, signal.SIG_DFL)
-  signal.signal(signal.SIGINT, handle_killswitch)
-  signal.signal(signal.SIGTERM, handle_killswitch)
-
-  test_queue = multiprocessing.Queue()
-  result_queue = multiprocessing.Queue()
-
-  gen_cover_ctx = cover_ctx
-  if cover_ctx.enabled:
-    gen_cover_ctx = cover_ctx(include=util.get_cover_list(test_gen))
-
-  test_gen_args = (
-      test_gen, test_queue, result_queue, opts, kill_switch, gen_cover_ctx)
-
-  procs = []
-  if opts.handler.SKIP_RUNLOOP:
-    gen_loop_process(*test_gen_args)
-  else:
-    procs = [multiprocessing.Process(
-        target=gen_loop_process, args=test_gen_args)]
-
-    procs += [
-        multiprocessing.Process(
-            target=run_loop_process, args=(
-                test_queue, result_queue, opts, kill_switch, cover_ctx))
-        for _ in xrange(opts.jobs)
-    ]
-
-    for p in procs:
-      p.daemon = True
-      p.start()
-
-  error = False
-  try:
-    def generate_objects():
-      while not kill_switch.is_set():
-        while not kill_switch.is_set():
-          try:
-            yield result_queue.get(timeout=0.1)
-          except Queue.Empty:
-            break
-
-        if not any(p.is_alive() for p in procs):
-          break
-
-      # Get everything still in the queue. Still need timeout, but since nothing
-      # is going to be adding stuff to the queue, use a very short timeout.
-      while not kill_switch.is_set():
-        try:
-          yield result_queue.get(timeout=0.00001)
-        except Queue.Empty:
-          break
-
-      if kill_switch.is_set():
-        raise ResultStageAbort()
-    error = opts.handler.result_stage_loop(opts, generate_objects())
-  except ResultStageAbort:
-    pass
-
-  for p in procs:
-    p.join()
-
-  if not kill_switch.is_set() and not result_queue.empty():
-    error = True
-
-  return error, kill_switch.is_set()