pytest-race

0.2.0 · active · verified Mon Apr 13

pytest-race is a plugin for the pytest testing framework that helps detect and test race conditions in your Python codebase. It introduces a `start_race` fixture, allowing you to easily run a target callable function multiple times concurrently in separate threads. The current version is 0.2.0, with a slow release cadence, last updated in June 2022.

Warnings

gotcha When writing concurrent tests, be extremely careful with shared mutable state (like global variables or shared objects). The `pytest-race` fixture merely facilitates running code in multiple threads; it's up to the user to correctly design the test to expose and assert against race conditions, which often involves shared state that isn't properly synchronized.
Fix: Ensure that shared resources in your `target` callable are either properly locked/synchronized or designed to be thread-safe. Alternatively, craft your assertions to reflect the non-deterministic but acceptable outcomes of concurrent operations, or to specifically catch the race condition you expect.
gotcha Pytest itself is single-threaded. Avoid using pytest-specific assertion helpers or primitives (e.g., `pytest.warns()`, `pytest.raises()`, `caplog`, `capsys`) directly from multiple threads spawned by `start_race`'s `target` callable, as these are not designed to be thread-safe.
Fix: Perform thread-safe logging or accumulate results within the `target` callable, and then perform final assertions on the collected data in the main test function (after `start_race` completes).
gotcha Race condition tests can be inherently flaky due to timing. Overly strict or poorly timed assertions can lead to intermittent failures that mask the actual race condition or lead to false positives.
Fix: Design assertions to be robust against expected timing variations, and consider using techniques like `pytest.approx()` for numerical comparisons or implementing retries/waits with timeouts within the target function before making assertions, if appropriate for your use case. Reproduce flaky tests repeatedly to confirm the issue.

Install

pip install pytest-race Install with pip

Imports

start_race
```
def test_my_race_condition(start_race):
```
start_race is a pytest fixture and is injected directly into test functions as an argument.

Quickstart

This quickstart demonstrates how to use the `start_race` fixture to test a simple race condition involving a global accumulator. The `actual_test_target` function is executed concurrently by multiple threads, and an assertion is used to detect if the race condition leads to unexpected results. The `sleep` call helps simulate conditions where race conditions are more likely to occur.

import pytest
from time import sleep
from random import randint

ACCUMULATOR = 0 # This global var is race conditions prone.

def test_race_condition_example(start_race):
    """Demonstrates testing a race condition with pytest-race."""
    global ACCUMULATOR
    ACCUMULATOR = 0 # Reset for each test run

    def actual_test_target():
        global ACCUMULATOR
        increment = randint(1, 10000)
        accumulator_before = ACCUMULATOR
        sleep(0.01) # Simulate some lag for race condition to manifest
        ACCUMULATOR += increment

        # Assert that the accumulator increased by exactly 'increment' 
        # (this assertion will likely fail due to race conditions).
        # In a real test, you'd assert against expected concurrent behavior.
        assert accumulator_before + increment == ACCUMULATOR

    # Run 'actual_test_target' in 2 threads simultaneously
    start_race(threads_num=2, target=actual_test_target)

# To run this test:
# 1. Save it as e.g., test_race_example.py
# 2. Run `pytest -s` (the -s flag is useful to see print output if you add any)

view raw JSON →