concurrencytest 0.1.11

Run unittest test suites concurrently

concurrencytest

Python - Run unittest test suites concurrently

• Development: GitHub
• Download/Install: PyPI
• License: GPLv2+
• Copyright (c) 2013-2026 Corey Goldberg

---

Type	Status
Latest Version
Supported Python Versions
Build/Tests (CI)

---

About

concurrencytest allows parallel execution of unittest tests across multiple worker processes.

• Default: 1 process per CPU core using round-robin test distribution.
• Optional: specify number of processes and partition strategy.

---

Components

• ConcurrentTestSuite class: unittest-compatible TestSuite for running parallel tests.
• fork_for_tests function: fork-based make_tests implementation.
• partition_tests function: round-robin test distribution.
• partition_tests_by_class: class-local test distribution.

---

Installation

Install from PyPI:

pip install concurrencytest

---

Requirements

• Python 3.10+
• Unix-like OS with os.fork()
• Dependencies:
  • testtools
  • python-subunit

---

Usage

Basic steps:

1. write your tests in normal unittest style (test methods inside a unittest.TestCase class)
2. load a suite of tests using unittest.TestLoader (or unittest.defaultTestLoader):

• suite = unittest.TestLoader().discover("tests")
• suite = unittest.TestLoader().loadTestsFromModule(my_tests)
• suite = unittest.TestLoader().loadTestsFromTestCase(MyTests)
• suite = unittest.TestLoader().loadTestsFromName("MyTests.test_1")
• suite = unittest.TestLoader().loadTestsFromNames("MyTests.test_1", "MyTests.test_2")

3. Wrap with ConcurrentTestSuite:

• concurrent_suite = ConcurrentTestSuite(suite)
• concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(4))

4. Run the suite using a unittest-compatible runner:

• unittest.TextTestRunner().run(concurrent_suite)

Configuring number of processes and partition strategy:

The concurrencytest module provides a make_tests implementation (fork_for_tests). This allows you to specify the number of worker processes to use and a partition strategy for specifying how tests are distributed to workers. If ConcurrentTestSuite is instantiated without a make_tests argument, it defaults to forking one process per available CPU core, and distributing tests using a round-robin strategy.

The fork_for_tests function is called with positional or keyword arguments like this:

fork_for_tests(num_processes, partition_func)

• num_processes (optional): Number of worker processes to spawn
  • Defaults to the number of CPUs on the system.
• partition_func (optional): Function used to partition tests across workers.
  • Defaults to partition_tests (round-robin partition strategy).

Available partition functions:

• partition_tests (round-robin):

  This is the default strategy.

  This function splits a test suite into its individual test cases and assigns them in a round-robin fashion to distribute load evenly across workers. This helps avoid situations where one worker gets all slow tests while others finish quickly. One potential drawback is that if you have a setUpClass/tearDownClass defined in a TestCase, it may be run multiple times if tests from the same class are run on different workers.

• partition_tests_by_class (class-local):

  This function groups all tests belonging to the same test case class and assigns them as a block to the worker with the current smallest number of tests already assigned. This ensures that all tests from a single class run in the same worker, which preserves setUpClass/tearDownClass lifecycle semantics.

Examples of creating a ConcurrentTestSuite:

• default concurrency and round-robin partition strategy:

  ConcurrentTestSuite(suite)

• 4 worker processes and round-robin partition strategy:

  ConcurrentTestSuite(suite, fork_for_tests(4))

• default concurrency and class-local partition strategy:

  ConcurrentTestSuite(suite, fork_for_tests(partition_func=partition_tests_by_class))

• 4 worker processes and class-local partition strategy:

  ConcurrentTestSuite(suite, fork_for_tests(4, partition_tests_by_class))

---

Examples

Basic example:

import time
import unittest

from concurrencytest import ConcurrentTestSuite

"""Tests just sleep for demo."""


class ExampleTestCase(unittest.TestCase):

    def test_1(self):
        time.sleep(1)

    def test_2(self):
        time.sleep(1)

    def test_3(self):
        time.sleep(1)

    def test_4(self):
        time.sleep(1)


runner = unittest.TextTestRunner()

# Run the tests from above sequentially
suite = unittest.defaultTestLoader.loadTestsFromTestCase(ExampleTestCase)
print("running sequential (without concurrencytest):")
runner.run(suite)

print()

# Run same tests concurrently across multiple processes
# (1 process per available CPU core)
suite = unittest.defaultTestLoader.loadTestsFromTestCase(ExampleTestCase)
print("running parallel:")
concurrent_suite = ConcurrentTestSuite(suite)
runner.run(concurrent_suite)

Output:

running sequential (without concurrencytest):
....
----------------------------------------------------------------------
Ran 4 tests in 4.002s

OK

running parallel:
....
----------------------------------------------------------------------
Ran 4 tests in 1.009s

OK

Advanced example:

import time
import unittest

from concurrencytest import (
    ConcurrentTestSuite,
    fork_for_tests,
    partition_tests_by_class,
)

"""Tests just sleep for demo."""


class ExampleTestCase1(unittest.TestCase):

    def test_1(self):
        time.sleep(1)

    def test_2(self):
        time.sleep(1)


class ExampleTestCase2(unittest.TestCase):
    """Dummy tests that sleep for demo."""

    def test_3(self):
        time.sleep(1)

    def test_4(self):
        time.sleep(1)


def load_test_suite(*test_cases):
    suite = unittest.TestSuite()
    for cls in test_cases:
        suite.addTests(unittest.defaultTestLoader.loadTestsFromTestCase(cls))
    return suite


runner = unittest.TextTestRunner()

# Run the tests from above sequentially
suite = load_test_suite(ExampleTestCase1, ExampleTestCase2)
print("running sequential (without concurrencytest):")
runner.run(suite)

print()

# Run same tests concurrently across multiple processes
# (1 process per available CPU core)
suite = load_test_suite(ExampleTestCase1, ExampleTestCase2)
concurrent_suite = ConcurrentTestSuite(suite)
print("running parallel:")
runner.run(concurrent_suite)

print()

# Run same tests concurrently across 4 processes
suite = load_test_suite(ExampleTestCase1, ExampleTestCase2)
concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(2))
print("running parallel (2 processes):")
runner.run(concurrent_suite)

print()

# Run same tests concurrently across multiple processes
# (1 process per available CPU core), keeping tests class-local
suite = load_test_suite(ExampleTestCase1, ExampleTestCase2)
concurrent_suite = ConcurrentTestSuite(
    suite, fork_for_tests(partition_func=partition_tests_by_class)
)
print("running parallel (grouped by class):")
runner.run(concurrent_suite)

Output:

running sequential (without concurrencytest):
....
----------------------------------------------------------------------
Ran 4 tests in 4.002s

OK

running parallel:
....
----------------------------------------------------------------------
Ran 4 tests in 1.010s

OK

running parallel (2 processes):
....
----------------------------------------------------------------------
Ran 4 tests in 2.006s

OK

running parallel (grouped by class):
....
----------------------------------------------------------------------
Ran 4 tests in 2.008s

OK

Notes

For more info about writing/running tests with the unittest testing framework, see the official documentation.