Welcome, guest | Sign In | My Account | Store | Cart

Note: This recipe is superceded by TestOOB, a Python unit testing framework that extends unittest and provides many new features - including running tests in threads! http://testoob.sourceforge.net

Trying to extend unittest to provide extra features wasn't easy. This scheme allows easy extensions for running existing test suites.

Python, 183 lines
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
"""
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"))

See the 'examples' function for simple usage similar to unittest's text runner. For running tests with different output schemes, see the 'text_run' function's code.

Implementation notes: * Note the ease of plugging in a new runner (see ThreadRunner) and of using test extractors (see regexp_extractor for filtering tests based on a name regexp).

  • class ThreadedRunner uses TwistedPython's thread pool. It can easily be replaced with any other thread pool.

  • The test fixtures can run in parallel even in the presence of setUp/tearDown, because each fixture has an independent instance of the relevant TestCase. Thanks to Steve Purcell for pointing this out (and for unittest itself :-)

I'd love to hear any comments and ideas.

2 comments

Ori Peleg (author) 16 years, 8 months ago  # | flag

High-latency test suites. Using the threaded runner is great for suites of high-latency tests that aren't CPU-bound.

An example suite is below.

Output for running examples(suite()):

---- begin output ----

== sequential ==
.......
----------------------------------------------------------------------
Ran 7 tests in 16.415s

OK
== threaded ==
.......
----------------------------------------------------------------------
Ran 7 tests in 5.869s

OK
== filtered ==
..
----------------------------------------------------------------------
Ran 2 tests in 2.444s

OK

---- end output ----

The suite:

import unittest, urllib

class NewsSitesTestCase(unittest.TestCase):
    def testSlashdot(self):
        urllib.urlopen("http://www.slashdot.org").read()
    def testWired(self):
        urllib.urlopen("http://www.wired.com").read()
    def testTheOnion(self):
        urllib.urlopen("http://www.theonion.com").read()

class OtherSitesTestCase(unittest.TestCase):
    def testYahoo(self):
        urllib.urlopen("http://www.yahoo.com").read()
    def testGoogle(self):
        urllib.urlopen("http://www.google.com").read()
    def testPython(self):
        urllib.urlopen("http://www.python.org").read()
    def testThinlet(self):
        urllib.urlopen("http://thinlet.sourceforge.net").read()

def suite():
    result = unittest.TestSuite()
    result.addTest( unittest.makeSuite(NewsSitesTestCase) )
    result.addTest( unittest.makeSuite(OtherSitesTestCase) )
    return result
Siva 4 years, 10 months ago  # | flag

Hi Ori Peleg, We are using the above recipe but we have not used extract_fixtures, instead we are using entire suite like below

for fixture in test_extractor(suite):

for fixture in suite:

The method which was given by you executes only setupmethod,testmethod and teardownmethods, but it will not execute setupclass and teardownclass. Hence we have used suite instead of test_extractor and unittest2 instead of unittest for this. But the issue that we are facing currently is setupclass and teardownclass are getting executed multiple times if we are using threaded runner(i.e. when we are trying to execute unittest classes parallely).

Please suggest some solution for this as soon as possible.