pytest-xdist-rate-limit 1.0.0

Shared state management and rate limiting for pytest-xdist workers

pytest-xdist-rate-limit

Set the rate at which pytest-xdist workers can run tests.

Features

• Call pacing: Define how often tests hit the System Under Test. Setup flexible load testing scenarios when used in conjunction with pytest-xdist-load-testing.
• Rate drift detection: Set callbacks when the test system cannot adhere to the intended rate.
• Shared state across workers: Have session-scoped fixtures share state across pytest-xdist workers, transparently using file-based JSON storage
• Setup/teardown on first/last worker: First/last worker callbacks for setup and teardown, transparently using the pattern in Making session-scoped fixtures execute only once

Requirements

• Python 3.9+
• pytest >= 8.4.2
• pytest-xdist >= 3.8.0
• filelock >= 3.0.0

Installation

pip install pytest-xdist-rate-limit

Examples

See the examples/ folder for working examples.

Rate Limiting

Use make_rate_limiter to enforce rate limits across workers.

Example using pytest_xdist_load_testing to run ~80% of test_get and ~20% test_put for 10K calls, or until we detect that we cannot keep with the requested rate 10 calls per second.

import pytest
from pytest_xdist_load_testing import stop_load_testing, weight
from pytest_xdist_rate_limit import RateLimit

@pytest.fixture(scope="session")
def pacer(request, make_rate_limiter):
    """Rate limiter for calls to SUT"""

    def on_drift(event):
        """Called when rate drift exceeds max_drift threshold.
        """
        msg = (f"Rate drift for {event.limiter_id}: "
               f"current={event.current_rate:.2f}/hr, target={event.target_rate}/hr, "
               f"drift={event.drift:.2%}")
        stop_load_testing(msg)

    def on_max_calls(event):
        """Called when max_calls limit is reached.
        """
        msg = f"Reached {event.max_calls} calls in {event.elapsed_time:.1f}s"
        stop_load_testing(msg)

    return make_rate_limiter(
        name="pacer",
        hourly_rate=RateLimit.per_second(10),  # 10 calls/second
        max_drift=0.2,  # 20% tolerance
        on_drift_callback=on_drift,
        max_calls=10_000,
        on_max_calls_callback=on_max_calls
    )

@weight(80)
def test_get(pacer):
    with pacer():
        # Context entry waits if rate limit would be exceeded
        response = api.get("/data")

@weight(20)
def test_put(pacer):
    with pacer() as ctx:
        # Context entry waits if rate limit would be exceeded
        response = api.put(f"/data/{ctx.call_count}")

Timeout Support

You can specify a timeout to prevent tests from waiting too long:

def test_with_timeout(pacer):
    try:
        with pacer(timeout=5.0) as ctx:
            # Will raise RateLimitTimeout if wait exceeds 5 seconds
            ...
    except RateLimitTimeout as e:
        ...

Shared Session state

import pytest

@pytest.fixture(scope="session")
def shared_resource(make_shared_json):
    """Shared resource with setup and teardown."""

    def setup():
        # Called by first worker only
        # Other workers have to wait for completion
        return {'initialized': True, 'counter': 0}

    def teardown(data):
        # Called by last worker only
        print(f"Final counter value: {data['counter']}")

    return make_shared_json(
        name="resource",
        on_first_worker=setup,
        on_last_worker=teardown
    )

def test_with_resource(shared_resource):
    with shared_resource.locked_dict() as data:
        data['counter'] += 1

Documentation

📚 Full Documentation

• API Reference - Complete API documentation

License

Distributed under the terms of the MIT license, "pytest-xdist-rate-limit" is free and open source software

Issues

If you encounter any problems, please file an issue along with a detailed description.

---

This pytest plugin was generated with Cookiecutter along with @hackebrot's cookiecutter-pytest-plugin template.