yes...again Python3....

1 files changed, 835 insertions, 0 deletions

diff --git a/scripts/external_libs/nose-1.3.4/python2/nose/plugins/multiprocess.py b/scripts/external_libs/nose-1.3.4/python2/nose/plugins/multiprocess.py
new file mode 100755
index 00000000..2cae744a
--- /dev/null
+++ b/scripts/external_libs/nose-1.3.4/python2/nose/plugins/multiprocess.py
@@ -0,0 +1,835 @@
+"""
+Overview
+========
+
+The multiprocess plugin enables you to distribute your test run among a set of
+worker processes that run tests in parallel. This can speed up CPU-bound test
+runs (as long as the number of work processeses is around the number of
+processors or cores available), but is mainly useful for IO-bound tests that
+spend most of their time waiting for data to arrive from someplace else.
+
+.. note ::
+
+   See :doc:`../doc_tests/test_multiprocess/multiprocess` for
+   additional documentation and examples. Use of this plugin on python
+   2.5 or earlier requires the multiprocessing_ module, also available
+   from PyPI.
+
+.. _multiprocessing : http://code.google.com/p/python-multiprocessing/
+
+How tests are distributed
+=========================
+
+The ideal case would be to dispatch each test to a worker process
+separately. This ideal is not attainable in all cases, however, because many
+test suites depend on context (class, module or package) fixtures.
+
+The plugin can't know (unless you tell it -- see below!) if a context fixture
+can be called many times concurrently (is re-entrant), or if it can be shared
+among tests running in different processes. Therefore, if a context has
+fixtures, the default behavior is to dispatch the entire suite to a worker as
+a unit.
+
+Controlling distribution
+^^^^^^^^^^^^^^^^^^^^^^^^
+
+There are two context-level variables that you can use to control this default
+behavior.
+
+If a context's fixtures are re-entrant, set ``_multiprocess_can_split_ = True``
+in the context, and the plugin will dispatch tests in suites bound to that
+context as if the context had no fixtures. This means that the fixtures will
+execute concurrently and multiple times, typically once per test.
+
+If a context's fixtures can be shared by tests running in different processes
+-- such as a package-level fixture that starts an external http server or
+initializes a shared database -- then set ``_multiprocess_shared_ = True`` in
+the context. These fixtures will then execute in the primary nose process, and
+tests in those contexts will be individually dispatched to run in parallel.
+
+How results are collected and reported
+======================================
+
+As each test or suite executes in a worker process, results (failures, errors,
+and specially handled exceptions like SkipTest) are collected in that
+process. When the worker process finishes, it returns results to the main
+nose process. There, any progress output is printed (dots!), and the
+results from the test run are combined into a consolidated result
+set. When results have been received for all dispatched tests, or all
+workers have died, the result summary is output as normal.
+
+Beware!
+=======
+
+Not all test suites will benefit from, or even operate correctly using, this
+plugin. For example, CPU-bound tests will run more slowly if you don't have
+multiple processors. There are also some differences in plugin
+interactions and behaviors due to the way in which tests are dispatched and
+loaded. In general, test loading under this plugin operates as if it were
+always in directed mode instead of discovered mode. For instance, doctests
+in test modules will always be found when using this plugin with the doctest
+plugin.
+
+But the biggest issue you will face is probably concurrency. Unless you
+have kept your tests as religiously pure unit tests, with no side-effects, no
+ordering issues, and no external dependencies, chances are you will experience
+odd, intermittent and unexplainable failures and errors when using this
+plugin. This doesn't necessarily mean the plugin is broken; it may mean that
+your test suite is not safe for concurrency.
+
+New Features in 1.1.0
+=====================
+
+* functions generated by test generators are now added to the worker queue
+  making them multi-threaded.
+* fixed timeout functionality, now functions will be terminated with a
+  TimedOutException exception when they exceed their execution time. The
+  worker processes are not terminated.
+* added ``--process-restartworker`` option to restart workers once they are
+  done, this helps control memory usage. Sometimes memory leaks can accumulate
+  making long runs very difficult.
+* added global _instantiate_plugins to configure which plugins are started
+  on the worker processes.
+
+"""
+
+import logging
+import os
+import sys
+import time
+import traceback
+import unittest
+import pickle
+import signal
+import nose.case
+from nose.core import TextTestRunner
+from nose import failure
+from nose import loader
+from nose.plugins.base import Plugin
+from nose.pyversion import bytes_
+from nose.result import TextTestResult
+from nose.suite import ContextSuite
+from nose.util import test_address
+try:
+    # 2.7+
+    from unittest.runner import _WritelnDecorator
+except ImportError:
+    from unittest import _WritelnDecorator
+from Queue import Empty
+from warnings import warn
+try:
+    from cStringIO import StringIO
+except ImportError:
+    import StringIO
+
+# this is a list of plugin classes that will be checked for and created inside 
+# each worker process
+_instantiate_plugins = None
+
+log = logging.getLogger(__name__)
+
+Process = Queue = Pool = Event = Value = Array = None
+
+# have to inherit KeyboardInterrupt to it will interrupt process properly
+class TimedOutException(KeyboardInterrupt):
+    def __init__(self, value = "Timed Out"):
+        self.value = value
+    def __str__(self):
+        return repr(self.value)
+
+def _import_mp():
+    global Process, Queue, Pool, Event, Value, Array
+    try:
+        from multiprocessing import Manager, Process
+        #prevent the server process created in the manager which holds Python 
+        #objects and allows other processes to manipulate them using proxies
+        #to interrupt on SIGINT (keyboardinterrupt) so that the communication
+        #channel between subprocesses and main process is still usable after
+        #ctrl+C is received in the main process.
+        old=signal.signal(signal.SIGINT, signal.SIG_IGN)
+        m = Manager()
+        #reset it back so main process will receive a KeyboardInterrupt
+        #exception on ctrl+c
+        signal.signal(signal.SIGINT, old)
+        Queue, Pool, Event, Value, Array = (
+                m.Queue, m.Pool, m.Event, m.Value, m.Array
+        )
+    except ImportError:
+        warn("multiprocessing module is not available, multiprocess plugin "
+             "cannot be used", RuntimeWarning)
+
+
+class TestLet:
+    def __init__(self, case):
+        try:
+            self._id = case.id()
+        except AttributeError:
+            pass
+        self._short_description = case.shortDescription()
+        self._str = str(case)
+
+    def id(self):
+        return self._id
+
+    def shortDescription(self):
+        return self._short_description
+
+    def __str__(self):
+        return self._str
+
+class MultiProcess(Plugin):
+    """
+    Run tests in multiple processes. Requires processing module.
+    """
+    score = 1000
+    status = {}
+
+    def options(self, parser, env):
+        """
+        Register command-line options.
+        """
+        parser.add_option("--processes", action="store",
+                          default=env.get('NOSE_PROCESSES', 0),
+                          dest="multiprocess_workers",
+                          metavar="NUM",
+                          help="Spread test run among this many processes. "
+                          "Set a number equal to the number of processors "
+                          "or cores in your machine for best results. "
+                          "Pass a negative number to have the number of "
+                          "processes automatically set to the number of "
+                          "cores. Passing 0 means to disable parallel "
+                          "testing. Default is 0 unless NOSE_PROCESSES is "
+                          "set. "
+                          "[NOSE_PROCESSES]")
+        parser.add_option("--process-timeout", action="store",
+                          default=env.get('NOSE_PROCESS_TIMEOUT', 10),
+                          dest="multiprocess_timeout",
+                          metavar="SECONDS",
+                          help="Set timeout for return of results from each "
+                          "test runner process. Default is 10. "
+                          "[NOSE_PROCESS_TIMEOUT]")
+        parser.add_option("--process-restartworker", action="store_true",
+                          default=env.get('NOSE_PROCESS_RESTARTWORKER', False),
+                          dest="multiprocess_restartworker",
+                          help="If set, will restart each worker process once"
+                          " their tests are done, this helps control memory "
+                          "leaks from killing the system. "
+                          "[NOSE_PROCESS_RESTARTWORKER]")
+
+    def configure(self, options, config):
+        """
+        Configure plugin.
+        """
+        try:
+            self.status.pop('active')
+        except KeyError:
+            pass
+        if not hasattr(options, 'multiprocess_workers'):
+            self.enabled = False
+            return
+        # don't start inside of a worker process
+        if config.worker:
+            return
+        self.config = config
+        try:
+            workers = int(options.multiprocess_workers)
+        except (TypeError, ValueError):
+            workers = 0
+        if workers:
+            _import_mp()
+            if Process is None:
+                self.enabled = False
+                return
+            # Negative number of workers will cause multiprocessing to hang.
+            # Set the number of workers to the CPU count to avoid this.
+            if workers < 0:
+                try:
+                    import multiprocessing
+                    workers = multiprocessing.cpu_count()
+                except NotImplementedError:
+                    self.enabled = False
+                    return
+            self.enabled = True
+            self.config.multiprocess_workers = workers
+            t = float(options.multiprocess_timeout)
+            self.config.multiprocess_timeout = t
+            r = int(options.multiprocess_restartworker)
+            self.config.multiprocess_restartworker = r
+            self.status['active'] = True
+
+    def prepareTestLoader(self, loader):
+        """Remember loader class so MultiProcessTestRunner can instantiate
+        the right loader.
+        """
+        self.loaderClass = loader.__class__
+
+    def prepareTestRunner(self, runner):
+        """Replace test runner with MultiProcessTestRunner.
+        """
+        # replace with our runner class
+        return MultiProcessTestRunner(stream=runner.stream,
+                                      verbosity=self.config.verbosity,
+                                      config=self.config,
+                                      loaderClass=self.loaderClass)
+
+def signalhandler(sig, frame):
+    raise TimedOutException()
+
+class MultiProcessTestRunner(TextTestRunner):
+    waitkilltime = 5.0 # max time to wait to terminate a process that does not
+                       # respond to SIGILL
+    def __init__(self, **kw):
+        self.loaderClass = kw.pop('loaderClass', loader.defaultTestLoader)
+        super(MultiProcessTestRunner, self).__init__(**kw)
+
+    def collect(self, test, testQueue, tasks, to_teardown, result):
+        # dispatch and collect results
+        # put indexes only on queue because tests aren't picklable
+        for case in self.nextBatch(test):
+            log.debug("Next batch %s (%s)", case, type(case))
+            if (isinstance(case, nose.case.Test) and
+                isinstance(case.test, failure.Failure)):
+                log.debug("Case is a Failure")
+                case(result) # run here to capture the failure
+                continue
+            # handle shared fixtures
+            if isinstance(case, ContextSuite) and case.context is failure.Failure:
+                log.debug("Case is a Failure")
+                case(result) # run here to capture the failure
+                continue
+            elif isinstance(case, ContextSuite) and self.sharedFixtures(case):
+                log.debug("%s has shared fixtures", case)
+                try:
+                    case.setUp()
+                except (KeyboardInterrupt, SystemExit):
+                    raise
+                except:
+                    log.debug("%s setup failed", sys.exc_info())
+                    result.addError(case, sys.exc_info())
+                else:
+                    to_teardown.append(case)
+                    if case.factory:
+                        ancestors=case.factory.context.get(case, [])
+                        for an in ancestors[:2]:
+                            #log.debug('reset ancestor %s', an)
+                            if getattr(an, '_multiprocess_shared_', False):
+                                an._multiprocess_can_split_=True
+                            #an._multiprocess_shared_=False
+                    self.collect(case, testQueue, tasks, to_teardown, result)
+
+            else:
+                test_addr = self.addtask(testQueue,tasks,case)
+                log.debug("Queued test %s (%s) to %s",
+                          len(tasks), test_addr, testQueue)
+
+    def startProcess(self, iworker, testQueue, resultQueue, shouldStop, result):
+        currentaddr = Value('c',bytes_(''))
+        currentstart = Value('d',time.time())
+        keyboardCaught = Event()
+        p = Process(target=runner,
+                   args=(iworker, testQueue,
+                         resultQueue,
+                         currentaddr,
+                         currentstart,
+                         keyboardCaught,
+                         shouldStop,
+                         self.loaderClass,
+                         result.__class__,
+                         pickle.dumps(self.config)))
+        p.currentaddr = currentaddr
+        p.currentstart = currentstart
+        p.keyboardCaught = keyboardCaught
+        old = signal.signal(signal.SIGILL, signalhandler)
+        p.start()
+        signal.signal(signal.SIGILL, old)
+        return p
+
+    def run(self, test):
+        """
+        Execute the test (which may be a test suite). If the test is a suite,
+        distribute it out among as many processes as have been configured, at
+        as fine a level as is possible given the context fixtures defined in
+        the suite or any sub-suites.
+
+        """
+        log.debug("%s.run(%s) (%s)", self, test, os.getpid())
+        wrapper = self.config.plugins.prepareTest(test)
+        if wrapper is not None:
+            test = wrapper
+
+        # plugins can decorate or capture the output stream
+        wrapped = self.config.plugins.setOutputStream(self.stream)
+        if wrapped is not None:
+            self.stream = wrapped
+
+        testQueue = Queue()
+        resultQueue = Queue()
+        tasks = []
+        completed = []
+        workers = []
+        to_teardown = []
+        shouldStop = Event()
+
+        result = self._makeResult()
+        start = time.time()
+
+        self.collect(test, testQueue, tasks, to_teardown, result)
+
+        log.debug("Starting %s workers", self.config.multiprocess_workers)
+        for i in range(self.config.multiprocess_workers):
+            p = self.startProcess(i, testQueue, resultQueue, shouldStop, result)
+            workers.append(p)
+            log.debug("Started worker process %s", i+1)
+
+        total_tasks = len(tasks)
+        # need to keep track of the next time to check for timeouts in case
+        # more than one process times out at the same time.
+        nexttimeout=self.config.multiprocess_timeout
+        thrownError = None
+
+        try:
+            while tasks:
+                log.debug("Waiting for results (%s/%s tasks), next timeout=%.3fs",
+                          len(completed), total_tasks,nexttimeout)
+                try:
+                    iworker, addr, newtask_addrs, batch_result = resultQueue.get(
+                                                            timeout=nexttimeout)
+                    log.debug('Results received for worker %d, %s, new tasks: %d',
+                              iworker,addr,len(newtask_addrs))
+                    try:
+                        try:
+                            tasks.remove(addr)
+                        except ValueError:
+                            log.warn('worker %s failed to remove from tasks: %s',
+                                     iworker,addr)
+                        total_tasks += len(newtask_addrs)
+                        tasks.extend(newtask_addrs)
+                    except KeyError:
+                        log.debug("Got result for unknown task? %s", addr)
+                        log.debug("current: %s",str(list(tasks)[0]))
+                    else:
+                        completed.append([addr,batch_result])
+                    self.consolidate(result, batch_result)
+                    if (self.config.stopOnError
+                        and not result.wasSuccessful()):
+                        # set the stop condition
+                        shouldStop.set()
+                        break
+                    if self.config.multiprocess_restartworker:
+                        log.debug('joining worker %s',iworker)
+                        # wait for working, but not that important if worker
+                        # cannot be joined in fact, for workers that add to
+                        # testQueue, they will not terminate until all their
+                        # items are read
+                        workers[iworker].join(timeout=1)
+                        if not shouldStop.is_set() and not testQueue.empty():
+                            log.debug('starting new process on worker %s',iworker)
+                            workers[iworker] = self.startProcess(iworker, testQueue, resultQueue, shouldStop, result)
+                except Empty:
+                    log.debug("Timed out with %s tasks pending "
+                              "(empty testQueue=%r): %s",
+                              len(tasks),testQueue.empty(),str(tasks))
+                    any_alive = False
+                    for iworker, w in enumerate(workers):
+                        if w.is_alive():
+                            worker_addr = bytes_(w.currentaddr.value,'ascii')
+                            timeprocessing = time.time() - w.currentstart.value
+                            if ( len(worker_addr) == 0
+                                    and timeprocessing > self.config.multiprocess_timeout-0.1):
+                                log.debug('worker %d has finished its work item, '
+                                          'but is not exiting? do we wait for it?',
+                                          iworker)
+                            else:
+                                any_alive = True
+                            if (len(worker_addr) > 0
+                                and timeprocessing > self.config.multiprocess_timeout-0.1):
+                                log.debug('timed out worker %s: %s',
+                                          iworker,worker_addr)
+                                w.currentaddr.value = bytes_('')
+                                # If the process is in C++ code, sending a SIGILL
+                                # might not send a python KeybordInterrupt exception
+                                # therefore, send multiple signals until an
+                                # exception is caught. If this takes too long, then
+                                # terminate the process
+                                w.keyboardCaught.clear()
+                                startkilltime = time.time()
+                                while not w.keyboardCaught.is_set() and w.is_alive():
+                                    if time.time()-startkilltime > self.waitkilltime:
+                                        # have to terminate...
+                                        log.error("terminating worker %s",iworker)
+                                        w.terminate()
+                                        # there is a small probability that the
+                                        # terminated process might send a result,
+                                        # which has to be specially handled or
+                                        # else processes might get orphaned.
+                                        workers[iworker] = w = self.startProcess(iworker, testQueue, resultQueue, shouldStop, result)
+                                        break
+                                    os.kill(w.pid, signal.SIGILL)
+                                    time.sleep(0.1)
+                    if not any_alive and testQueue.empty():
+                        log.debug("All workers dead")
+                        break
+                nexttimeout=self.config.multiprocess_timeout
+                for w in workers:
+                    if w.is_alive() and len(w.currentaddr.value) > 0:
+                        timeprocessing = time.time()-w.currentstart.value
+                        if timeprocessing <= self.config.multiprocess_timeout:
+                            nexttimeout = min(nexttimeout,
+                                self.config.multiprocess_timeout-timeprocessing)
+            log.debug("Completed %s tasks (%s remain)", len(completed), len(tasks))
+
+        except (KeyboardInterrupt, SystemExit), e:
+            log.info('parent received ctrl-c when waiting for test results')
+            thrownError = e
+            #resultQueue.get(False)
+                
+            result.addError(test, sys.exc_info())
+
+        try:
+            for case in to_teardown:
+                log.debug("Tearing down shared fixtures for %s", case)
+                try:
+                    case.tearDown()
+                except (KeyboardInterrupt, SystemExit):
+                    raise
+                except:
+                    result.addError(case, sys.exc_info())
+
+            stop = time.time()
+
+            # first write since can freeze on shutting down processes
+            result.printErrors()
+            result.printSummary(start, stop)
+            self.config.plugins.finalize(result)
+
+            if thrownError is None:
+                log.debug("Tell all workers to stop")
+                for w in workers:
+                    if w.is_alive():
+                        testQueue.put('STOP', block=False)
+
+            # wait for the workers to end
+            for iworker,worker in enumerate(workers):
+                if worker.is_alive():
+                    log.debug('joining worker %s',iworker)
+                    worker.join()
+                    if worker.is_alive():
+                        log.debug('failed to join worker %s',iworker)
+        except (KeyboardInterrupt, SystemExit):
+            log.info('parent received ctrl-c when shutting down: stop all processes')
+            for worker in workers:
+                if worker.is_alive():
+                    worker.terminate()
+
+            if thrownError: raise thrownError
+            else: raise
+
+        return result
+
+    def addtask(testQueue,tasks,case):
+        arg = None
+        if isinstance(case,nose.case.Test) and hasattr(case.test,'arg'):
+            # this removes the top level descriptor and allows real function
+            # name to be returned
+            case.test.descriptor = None
+            arg = case.test.arg
+        test_addr = MultiProcessTestRunner.address(case)
+        testQueue.put((test_addr,arg), block=False)
+        if arg is not None:
+            test_addr += str(arg)
+        if tasks is not None:
+            tasks.append(test_addr)
+        return test_addr
+    addtask = staticmethod(addtask)
+
+    def address(case):
+        if hasattr(case, 'address'):
+            file, mod, call = case.address()
+        elif hasattr(case, 'context'):
+            file, mod, call = test_address(case.context)
+        else:
+            raise Exception("Unable to convert %s to address" % case)
+        parts = []
+        if file is None:
+            if mod is None:
+                raise Exception("Unaddressable case %s" % case)
+            else:
+                parts.append(mod)
+        else:
+            # strip __init__.py(c) from end of file part
+            # if present, having it there confuses loader
+            dirname, basename = os.path.split(file)
+            if basename.startswith('__init__'):
+                file = dirname
+            parts.append(file)
+        if call is not None:
+            parts.append(call)
+        return ':'.join(map(str, parts))
+    address = staticmethod(address)
+
+    def nextBatch(self, test):
+        # allows tests or suites to mark themselves as not safe
+        # for multiprocess execution
+        if hasattr(test, 'context'):
+            if not getattr(test.context, '_multiprocess_', True):
+                return
+
+        if ((isinstance(test, ContextSuite)
+             and test.hasFixtures(self.checkCanSplit))
+            or not getattr(test, 'can_split', True)
+            or not isinstance(test, unittest.TestSuite)):
+            # regular test case, or a suite with context fixtures
+
+            # special case: when run like nosetests path/to/module.py
+            # the top-level suite has only one item, and it shares
+            # the same context as that item. In that case, we want the
+            # item, not the top-level suite
+            if isinstance(test, ContextSuite):
+                contained = list(test)
+                if (len(contained) == 1
+                    and getattr(contained[0],
+                                'context', None) == test.context):
+                    test = contained[0]
+            yield test
+        else:
+            # Suite is without fixtures at this level; but it may have
+            # fixtures at any deeper level, so we need to examine it all
+            # the way down to the case level
+            for case in test:
+                for batch in self.nextBatch(case):
+                    yield batch
+
+    def checkCanSplit(context, fixt):
+        """
+        Callback that we use to check whether the fixtures found in a
+        context or ancestor are ones we care about.
+
+        Contexts can tell us that their fixtures are reentrant by setting
+        _multiprocess_can_split_. So if we see that, we return False to
+        disregard those fixtures.
+        """
+        if not fixt:
+            return False
+        if getattr(context, '_multiprocess_can_split_', False):
+            return False
+        return True
+    checkCanSplit = staticmethod(checkCanSplit)
+
+    def sharedFixtures(self, case):
+        context = getattr(case, 'context', None)
+        if not context:
+            return False
+        return getattr(context, '_multiprocess_shared_', False)
+
+    def consolidate(self, result, batch_result):
+        log.debug("batch result is %s" , batch_result)
+        try:
+            output, testsRun, failures, errors, errorClasses = batch_result
+        except ValueError:
+            log.debug("result in unexpected format %s", batch_result)
+            failure.Failure(*sys.exc_info())(result)
+            return
+        self.stream.write(output)
+        result.testsRun += testsRun
+        result.failures.extend(failures)
+        result.errors.extend(errors)
+        for key, (storage, label, isfail) in errorClasses.items():
+            if key not in result.errorClasses:
+                # Ordinarily storage is result attribute
+                # but it's only processed through the errorClasses
+                # dict, so it's ok to fake it here
+                result.errorClasses[key] = ([], label, isfail)
+            mystorage, _junk, _junk = result.errorClasses[key]
+            mystorage.extend(storage)
+        log.debug("Ran %s tests (total: %s)", testsRun, result.testsRun)
+
+
+def runner(ix, testQueue, resultQueue, currentaddr, currentstart,
+           keyboardCaught, shouldStop, loaderClass, resultClass, config):
+    try:
+        try:
+            return __runner(ix, testQueue, resultQueue, currentaddr, currentstart,
+                    keyboardCaught, shouldStop, loaderClass, resultClass, config)
+        except KeyboardInterrupt:
+            log.debug('Worker %s keyboard interrupt, stopping',ix)
+    except Empty:
+        log.debug("Worker %s timed out waiting for tasks", ix)
+
+def __runner(ix, testQueue, resultQueue, currentaddr, currentstart,
+           keyboardCaught, shouldStop, loaderClass, resultClass, config):
+
+    config = pickle.loads(config)
+    dummy_parser = config.parserClass()
+    if _instantiate_plugins is not None:
+        for pluginclass in _instantiate_plugins:
+            plugin = pluginclass()
+            plugin.addOptions(dummy_parser,{})
+            config.plugins.addPlugin(plugin)
+    config.plugins.configure(config.options,config)
+    config.plugins.begin()
+    log.debug("Worker %s executing, pid=%d", ix,os.getpid())
+    loader = loaderClass(config=config)
+    loader.suiteClass.suiteClass = NoSharedFixtureContextSuite
+
+    def get():
+        return testQueue.get(timeout=config.multiprocess_timeout)
+
+    def makeResult():
+        stream = _WritelnDecorator(StringIO())
+        result = resultClass(stream, descriptions=1,
+                             verbosity=config.verbosity,
+                             config=config)
+        plug_result = config.plugins.prepareTestResult(result)
+        if plug_result:
+            return plug_result
+        return result
+
+    def batch(result):
+        failures = [(TestLet(c), err) for c, err in result.failures]
+        errors = [(TestLet(c), err) for c, err in result.errors]
+        errorClasses = {}
+        for key, (storage, label, isfail) in result.errorClasses.items():
+            errorClasses[key] = ([(TestLet(c), err) for c, err in storage],
+                                 label, isfail)
+        return (
+            result.stream.getvalue(),
+            result.testsRun,
+            failures,
+            errors,
+            errorClasses)
+    for test_addr, arg in iter(get, 'STOP'):
+        if shouldStop.is_set():
+            log.exception('Worker %d STOPPED',ix)
+            break
+        result = makeResult()
+        test = loader.loadTestsFromNames([test_addr])
+        test.testQueue = testQueue
+        test.tasks = []
+        test.arg = arg
+        log.debug("Worker %s Test is %s (%s)", ix, test_addr, test)
+        try:
+            if arg is not None:
+                test_addr = test_addr + str(arg)
+            currentaddr.value = bytes_(test_addr)
+            currentstart.value = time.time()
+            test(result)
+            currentaddr.value = bytes_('')
+            resultQueue.put((ix, test_addr, test.tasks, batch(result)))
+        except KeyboardInterrupt, e: #TimedOutException:
+            timeout = isinstance(e, TimedOutException)
+            if timeout:
+                keyboardCaught.set()
+            if len(currentaddr.value):
+                if timeout:
+                    msg = 'Worker %s timed out, failing current test %s'
+                else:
+                    msg = 'Worker %s keyboard interrupt, failing current test %s'
+                log.exception(msg,ix,test_addr)
+                currentaddr.value = bytes_('')
+                failure.Failure(*sys.exc_info())(result)
+                resultQueue.put((ix, test_addr, test.tasks, batch(result)))
+            else:
+                if timeout:
+                    msg = 'Worker %s test %s timed out'
+                else:
+                    msg = 'Worker %s test %s keyboard interrupt'
+                log.debug(msg,ix,test_addr)
+                resultQueue.put((ix, test_addr, test.tasks, batch(result)))
+            if not timeout:
+                raise
+        except SystemExit:
+            currentaddr.value = bytes_('')
+            log.exception('Worker %s system exit',ix)
+            raise
+        except:
+            currentaddr.value = bytes_('')
+            log.exception("Worker %s error running test or returning "
+                            "results",ix)
+            failure.Failure(*sys.exc_info())(result)
+            resultQueue.put((ix, test_addr, test.tasks, batch(result)))
+        if config.multiprocess_restartworker:
+            break
+    log.debug("Worker %s ending", ix)
+
+
+class NoSharedFixtureContextSuite(ContextSuite):
+    """
+    Context suite that never fires shared fixtures.
+
+    When a context sets _multiprocess_shared_, fixtures in that context
+    are executed by the main process. Using this suite class prevents them
+    from executing in the runner process as well.
+
+    """
+    testQueue = None
+    tasks = None
+    arg = None
+    def setupContext(self, context):
+        if getattr(context, '_multiprocess_shared_', False):
+            return
+        super(NoSharedFixtureContextSuite, self).setupContext(context)
+
+    def teardownContext(self, context):
+        if getattr(context, '_multiprocess_shared_', False):
+            return
+        super(NoSharedFixtureContextSuite, self).teardownContext(context)
+    def run(self, result):
+        """Run tests in suite inside of suite fixtures.
+        """
+        # proxy the result for myself
+        log.debug("suite %s (%s) run called, tests: %s",
+                  id(self), self, self._tests)
+        if self.resultProxy:
+            result, orig = self.resultProxy(result, self), result
+        else:
+            result, orig = result, result
+        try:
+            #log.debug('setUp for %s', id(self));
+            self.setUp()
+        except KeyboardInterrupt:
+            raise
+        except:
+            self.error_context = 'setup'
+            result.addError(self, self._exc_info())
+            return
+        try:
+            for test in self._tests:
+                if (isinstance(test,nose.case.Test)
+                    and self.arg is not None):
+                    test.test.arg = self.arg
+                else:
+                    test.arg = self.arg
+                test.testQueue = self.testQueue
+                test.tasks = self.tasks
+                if result.shouldStop:
+                    log.debug("stopping")
+                    break
+                # each nose.case.Test will create its own result proxy
+                # so the cases need the original result, to avoid proxy
+                # chains
+                #log.debug('running test %s in suite %s', test, self);
+                try:
+                    test(orig)
+                except KeyboardInterrupt, e:
+                    timeout = isinstance(e, TimedOutException)
+                    if timeout:
+                        msg = 'Timeout when running test %s in suite %s'
+                    else:
+                        msg = 'KeyboardInterrupt when running test %s in suite %s'
+                    log.debug(msg, test, self)
+                    err = (TimedOutException,TimedOutException(str(test)),
+                           sys.exc_info()[2])
+                    test.config.plugins.addError(test,err)
+                    orig.addError(test,err)
+                    if not timeout:
+                        raise
+        finally:
+            self.has_run = True
+            try:
+                #log.debug('tearDown for %s', id(self));
+                self.tearDown()
+            except KeyboardInterrupt:
+                raise
+            except:
+                self.error_context = 'teardown'
+                result.addError(self, self._exc_info())