Python - concurrencytest: Running Concurrent Tests

Add parallel testing to your unit test framework.

In my previous post, I described running concurrent tests using nose as a loader and runner.

On a similar note, let's look at building concurrency into your own test framework built on Python's unittest.

Have a look at this module: concurrencytest

(Thanks to bits and concepts taken from testtools and bzrlib)

An Example:

Say you have a 'TestSuite' of tests loaded. You could run them with the standard 'TextTestRunner' like this:

runner = unittest.TextTestRunner()
runner.run(suite)

That would run the tests in your suite sequentially in a single process.

By adding the concurrencytest module, you can use a 'ConcurrentTestSuite' instead, by adding:

from concurrencytest import ConcurrentTestSuite, fork_for_tests

concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(4))
runner.run(concurrent_suite)

That would run the same tests split across 4 processes (workers).

Note: this relies on 'os.fork()' which only works on Unix systems.

There's no way to understand this better than looking at some contrived examples!

This first example is totally unrealistic, but shows off concurrency perfectly. The test cases it loads each sleep for 0.5 seconds and then exit.

The Code:

#!/usr/bin/env python
#
# Example using `concurrencytest`:
#   https://github.com/cgoldberg/concurrencytest
 
import time
import unittest
 
from concurrencytest import ConcurrentTestSuite, fork_for_tests
 
 
class SampleTestCase(unittest.TestCase):
    def test_it(self):
        time.sleep(0.5)
 
 
if __name__ == '__main__':
 
    # load a TestSuite with 50x TestCases for demo
    loader = unittest.TestLoader()
    suite = unittest.TestSuite()
    for _ in range(50):
        suite.addTests(loader.loadTestsFromTestCase(SampleTestCase))
    print('Loaded %d test cases...' % suite.countTestCases())
 
    runner = unittest.TextTestRunner()
 
    print('\nRun tests sequentially:')
    runner.run(suite)
 
    print('\nRun same tests across 50 processes:')
    concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(50))
    runner.run(concurrent_suite)

Output:

Loaded 50 test cases...

Run tests sequentially:
..................................................
----------------------------------------------------------------------
Ran 50 tests in 25.031s

OK

Run same tests across 50 processes:
..................................................
----------------------------------------------------------------------
Ran 50 tests in 0.525s

OK

nice!

Now another example that shows concurrency with CPU-bound test cases. The test cases it loads each calculate fibonacci of 31 (recursively!) and then exit. We can see how it performs on my 8-core machine (Core2 i7 quad, hyperthreaded).

The Code:

#!/usr/bin/env python
#
# Example using `concurrencytest`:
#   https://github.com/cgoldberg/concurrencytest
 
import unittest
 
from concurrencytest import ConcurrentTestSuite, fork_for_tests
 
 
def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    return fib(n - 1) + fib(n - 2)
 
 
class FibonacciTestCase(unittest.TestCase):
    def test_fib(self):
        self.assertEqual(fib(31), 1346269)
 
 
if __name__ == '__main__':
 
    # load a TestSuite with 50x TestCases for demo
    loader = unittest.TestLoader()
    suite = unittest.TestSuite()
    for _ in range(50):
        suite.addTests(loader.loadTestsFromTestCase(FibonacciTestCase))
    print('Loaded %d test cases...' % suite.countTestCases())
 
    runner = unittest.TextTestRunner()
 
    print('\nRun tests sequentially:')
    runner.run(suite)
 
    print('\nRun same tests with 2 processes:')
    concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(2))
    runner.run(concurrent_suite)
 
    print('\nRun same tests with 4 processes:')
    concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(4))
    runner.run(concurrent_suite)
 
    print('\nRun same tests with 8 processes:')
    concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(8))
    runner.run(concurrent_suite)

Output:

Loaded 50 test cases...

Run tests sequentially:
..................................................
----------------------------------------------------------------------
Ran 50 tests in 21.941s

OK

Run same tests with 2 processes:
..................................................
----------------------------------------------------------------------
Ran 50 tests in 11.081s

OK

Run same tests with 4 processes:
..................................................
----------------------------------------------------------------------
Ran 50 tests in 5.862s

OK

Run same tests with 8 processes:
..................................................
----------------------------------------------------------------------
Ran 50 tests in 4.743s

OK

happy hacking.