Over a million developers have joined DZone.

Python - concurrencytest: Running Concurrent Tests

· DevOps Zone

The DevOps zone is brought to you in partnership with Sonatype Nexus. The Nexus suite helps scale your DevOps delivery with continuous component intelligence integrated into development tools, including Eclipse, IntelliJ, Jenkins, Bamboo, SonarQube and more. Schedule a demo today

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.

The DevOps zone is brought to you in partnership with Sonatype Nexus. Use the Nexus Suite to automate your software supply chain and ensure you're using the highest quality open source components at every step of the development lifecycle. Get Nexus today

Topics:

Published at DZone with permission of Corey Goldberg, DZone MVB. See the original article here.

Opinions expressed by DZone contributors are their own.

The best of DZone straight to your inbox.

SEE AN EXAMPLE
Please provide a valid email address.

Thanks for subscribing!

Awesome! Check your inbox to verify your email so you can start receiving the latest in tech news and resources.
Subscribe

{{ parent.title || parent.header.title}}

{{ parent.tldr }}

{{ parent.urlSource.name }}