ActiveState Code

"""
An alternative running scheme for unittest test suites.

Superceded by the TestOOB Python unit testing framework,
http://testoob.sourceforge.net
"""

__author__ = "Ori Peleg"

import unittest, sys
from itertools import ifilter

###############################################################################
# apply_runner
###############################################################################
# David Eppstein's breadth_first
# http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/231503
def _breadth_first(tree,children=iter):
    """Traverse the nodes of a tree in breadth-first order.
    The first argument should be the tree root; children
    should be a function taking as argument a tree node and
    returning an iterator of the node's children.
    """
    yield tree
    last = tree
    for node in _breadth_first(tree,children):
	for child in children(node):
	    yield child
	    last = child
	if last == node:
	    return

def extract_fixtures(suite, recursive_iterator=_breadth_first):
    """Extract the text fixtures from a suite.
    Descends recursively into sub-suites."""
    def test_children(node):
        if isinstance(node, unittest.TestSuite): return iter(node)
        return []

    return ifilter(lambda test: isinstance(test, unittest.TestCase),
                   recursive_iterator(suite, children=test_children))

def apply_runner(suite, runner_class, result_class=unittest.TestResult,
          test_extractor=extract_fixtures):
    """Runs the suite."""
    runner = runner_class(result_class)

    for fixture in test_extractor(suite):

        runner.run(fixture)

    return runner.result()

###############################################################################
# Runners
###############################################################################

class SimpleRunner:
    def __init__(self, result_class):
        self._result = result_class()
        self._done = False

    def run(self, fixture):
        assert not self._done
        fixture(self._result)

    def result(self):
        self._done = True
        return self._result

# Connelly Barnes's (connellybarnes at yahoo.com) threadclass
# http://mail.python.org/pipermail/python-list/2004-June/225478.html
import types, threading
def _threadclass(C):
  """Returns a 'threadsafe' copy of class C.
     All public methods are modified to lock the
     object when called."""

  class D(C):
    def __init__(self, *args, **kwargs):
      self.lock = threading.RLock()
      C.__init__(self, *args, **kwargs)

  def ubthreadfunction(f):
    def g(self, *args, **kwargs):
      self.lock.acquire()
      try:
          return f(self, *args, **kwargs)
      finally:
          self.lock.release()
    return g

  for a in dir(D):
    f = getattr(D, a)
    if isinstance(f, types.UnboundMethodType) and a[:2] != '__':
      setattr(D, a, ubthreadfunction(f))
  return D

class ThreadedRunner(SimpleRunner):
    """Run tests using a threadpool.
    Uses TwistedPython's thread pool"""
    def __init__(self, result_class):
        from twisted.python.threadpool import ThreadPool

        SimpleRunner.__init__(self, _threadclass(result_class))
        
        self._pool = ThreadPool()
        self._pool.start()

    def run(self, fixture):
        assert not self._done
        self._pool.dispatch(None, fixture, self._result)

    def result(self):
        self._pool.stop()
        return SimpleRunner.result(self)

###############################################################################
# text_run
###############################################################################

def _print_results(result, timeTaken):
    # code modified from Python 2.4's standard unittest module
    stream = result.stream
    result.printErrors()
    stream.writeln(result.separator2)
    run = result.testsRun
    stream.writeln("Ran %d test%s in %.3fs" %
                   (run, run != 1 and "s" or "", timeTaken))
    stream.writeln()
    if not result.wasSuccessful():
        stream.write("FAILED (")
        failed, errored = map(len, (result.failures, result.errors))
        if failed:
            stream.write("failures=%d" % failed)
        if errored:
            if failed: stream.write(", ")
            stream.write("errors=%d" % errored)
        stream.writeln(")")
    else:
        stream.writeln("OK")

class _TextTestResult(unittest._TextTestResult):
    """provide defaults for unittest._TextTestResult"""
    def __init__(self, stream = sys.stderr, descriptions=1, verbosity=1):
        stream = unittest._WritelnDecorator(stream)
        unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)

def text_run(suite, runner_class=SimpleRunner, **kwargs):
    """Run a suite and generate output similar to unittest.TextTestRunner's"""
    import time
    start = time.time()
    result = apply_runner(suite, runner_class, result_class=_TextTestResult,
                          **kwargs)
    timeTaken = time.time() - start
    
    _print_results(result, timeTaken)

###############################################################################
# Test extractors
###############################################################################
def regexp_extractor(regexp):
    """Filter tests based on matching a regexp to their id.
    Matching is performed with re.search"""
    import re
    compiled = re.compile(regexp)
    def pred(test): return compiled.search(test.id())
    def wrapper(suite):
        return ifilter(pred, extract_fixtures(suite))
    return wrapper

###############################################################################
# examples
###############################################################################
def examples(suite):
    print "== sequential =="
    text_run(suite)

    print "== threaded =="
    text_run(suite, ThreadedRunner)

    print "== filtered =="
    text_run(suite, test_extractor = regexp_extractor("Th"))