concurrencytest

0.1.11 · active · verified Thu Apr 16

The `concurrencytest` library, currently at version 0.1.11, enables parallel execution of standard Python `unittest` test suites across multiple worker processes. It aims to speed up test execution by leveraging CPU cores. The library is actively maintained, with its latest release in March 2026, and provides mechanisms to control the number of worker processes and how tests are distributed among them.

Common errors

```
AttributeError: 'module' object has no attribute 'fork'
```
cause Attempting to run `concurrencytest` on a Windows operating system.

fix The library relies on `os.fork()`, which is not available on Windows. Run tests on Linux or macOS, or within a WSL (Windows Subsystem for Linux) environment.
```
setUpClass or tearDownClass runs unexpectedly multiple times or causes incorrect test state.
```
cause The default test partitioning strategy (`partition_tests`) distributes individual test methods across workers, potentially splitting tests from a single `TestCase` class. This causes `setUpClass` and `tearDownClass` to be executed by each worker that gets a test from that class.

fix Initialize `ConcurrentTestSuite` with the `partition_tests_by_class` strategy to ensure all tests from a given `TestCase` class run on the same worker process. Example: `ConcurrentTestSuite(suite, fork_for_tests(partition_func=partition_tests_by_class))`.
```
Tests fail intermittently or produce inconsistent results when run concurrently.
```
cause Likely due to race conditions or shared mutable state between test processes that is not properly synchronized. Each test is expected to run in its own isolated process.

fix Review test logic for any reliance on global variables, shared file system resources, or database state that is not reset or managed in a process-safe manner between tests. Make tests entirely independent of each other's execution order or state.

Warnings

gotcha The `fork_for_tests` implementation, and thus `concurrencytest` itself, relies on `os.fork()`. This means it is only compatible with Unix-like operating systems (Linux, macOS) and will not work on Windows.
Fix: Use a Unix-like environment or consider alternative parallel testing frameworks if Windows compatibility is critical.
gotcha When using the default `partition_tests` strategy (round-robin), `unittest.TestCase` methods like `setUpClass` and `tearDownClass` may run multiple times if tests from the same class are distributed to different worker processes. This can lead to unexpected test state or resource management issues.
Fix: To preserve `setUpClass`/`tearDownClass` lifecycle semantics, explicitly use `partition_tests_by_class` when initializing `ConcurrentTestSuite`, e.g., `ConcurrentTestSuite(suite, fork_for_tests(num_workers, partition_tests_by_class))`.
gotcha Test cases must be designed to be independent and self-contained when run concurrently. Shared mutable state between tests or processes without proper synchronization can lead to race conditions, flakiness, or incorrect results that are hard to debug.
Fix: Ensure each test creates its own resources and cleans them up. Avoid shared global state or use process-safe synchronization primitives (e.g., multiprocessing.Lock, Queue) if inter-process communication is absolutely necessary.

Install

pip install concurrencytest

Imports

ConcurrentTestSuite
```
from concurrencytest import ConcurrentTestSuite
```
The core class for wrapping a unittest.TestSuite to enable concurrent execution.
fork_for_tests
```
from concurrencytest import fork_for_tests
```
The default make_tests implementation, used to configure fork-based worker processes. It is typically passed as an argument to ConcurrentTestSuite.
partition_tests
```
from concurrencytest import partition_tests
```
The default test partitioning strategy (round-robin), which distributes individual test cases evenly across workers. Can cause setUpClass/tearDownClass to run multiple times.
partition_tests_by_class
```
from concurrencytest import partition_tests_by_class
```
An alternative partitioning strategy that groups all tests from the same TestCase class onto a single worker, preserving setUpClass/tearDownClass semantics.

Quickstart

This quickstart demonstrates how to set up a `unittest.TestCase` and run its tests concurrently using `concurrencytest`. It shows how to load a test suite and wrap it with `ConcurrentTestSuite`, specifying the number of worker processes.

import time
import unittest
from concurrencytest import ConcurrentTestSuite, fork_for_tests

class ExampleTestCase(unittest.TestCase):
    def test_slow_1(self):
        time.sleep(0.1)
        self.assertTrue(True)

    def test_slow_2(self):
        time.sleep(0.1)
        self.assertEqual(1 + 1, 2)

    def test_fast_3(self):
        self.assertFalse(False)

if __name__ == '__main__':
    # Load tests from current file
    suite = unittest.defaultTestLoader.loadTestsFromTestCase(ExampleTestCase)
    
    # Run tests concurrently using 2 processes
    # The default is fork_for_tests() which uses os.cpu_count() processes
    concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(2))
    
    runner = unittest.TextTestRunner(verbosity=2)
    runner.run(concurrent_suite)

view raw JSON →