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
- fastdigest >= 0.3.2
- filelock >= 3.0.0
Installation
Examples
See the examples/
folder for working examples.
Call Pacing
Use make_pacer to generate load at a controlled rate 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 Rate
@pytest.fixture(scope="session")
def pacer(request, make_pacer):
"""Pacer for generating load to SUT"""
def on_periodic_check(event):
"""Called every N calls to monitor performance metrics.
Provides detailed metrics including:
- Current vs target rate and drift
- Call duration percentiles (p50, p90, p99)
- Wait time percentiles
- Worker count and sample statistics
"""
# Check drift manually
if event.drift is not None and event.drift > 0.2:
msg = (f"Rate drift detected: current={event.current_rate:.2f}/hr, "
f"target={event.target_rate}/hr, drift={event.drift:.2%}")
stop_load_testing(msg)
# Log performance metrics
if event.duration_p50 is not None:
print(f"Performance: p50={event.duration_p50*1000:.1f}ms, "
f"p90={event.duration_p90*1000:.1f}ms, "
f"p99={event.duration_p99*1000:.1f}ms")
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_pacer(
name="pacer",
hourly_rate=Rate.per_second(10), # 10 calls/second
num_calls_between_checks=50, # Check every 50 calls
seconds_before_first_check=10.0, # Wait 10s before first check
on_periodic_check_callback=on_periodic_check,
max_calls=10_000,
on_max_calls_callback=on_max_calls
)
@weight(80)
def test_get(pacer):
with pacer():
# Context entry waits to maintain target rate
response = api.get("/data")
@weight(20)
def test_put(pacer):
with pacer() as ctx:
# Context entry waits to maintain target rate
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
- 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.