Move rate limiting and ShareJson to another plugin

Author: xverges

README.rst

@@ -140,95 +140,27 @@ Examples
         """10% of requests"""
         assert api.delete("/data/old").status_code == 204
 
-**Conditional stop with shared state across workers:**
+**Conditional stop:**
 
 .. code-block:: python
 
-    import pytest
     from pytest_load_testing import weight, stop_load_testing
 
-    @pytest.fixture(scope="session")
-    def error_tracker(shared_json_fixture_factory):
-        """Track errors across all workers."""
-        return shared_json_fixture_factory(
-            name="errors",
-            on_first_worker={'count': 0}
-        )
-
     @weight(1)
-    def test_health_check(request, error_tracker):
+    def test_health_check(request):
         response = api.get("/health")
         if response.json()["status"] == "critical":
             stop_load_testing(request, "System health critical")
         assert response.status_code == 200
 
-    @weight(10)
-    def test_normal_operation(error_tracker):
-        try:
-            assert api.get("/api/data").status_code == 200
-        except AssertionError:
-            with error_tracker.locked_dict() as data:
-                data['count'] += 1
-
-See the full documentation for more details on concurrent fixtures and shared state.
-
-
-Rate Limiting
-~~~~~~~~~~~~~
-
-The plugin provides a ``rate_limiter_fixture_factory`` for enforcing rate limits across workers:
-
-.. code-block:: python
-
-    import pytest
-    from pytest_load_testing import weight, stop_load_testing
-    from pytest_load_testing.token_bucket_rate_limiter import RateLimit
-
-    @pytest.fixture(scope="session")
-    def api_limiter(rate_limiter_fixture_factory, request):
-        """Rate limiter that stops tests if rate drift exceeds 20%."""
-
-        def on_drift(limiter_id, current_rate, target_rate, drift):
-            message = (
-                f"Rate drift for {limiter_id}: "
-                f"current={current_rate:.2f}/hr, target={target_rate}/hr, "
-                f"drift={drift:.2%}"
-            )
-            stop_load_testing(request, message)
-
-        return rate_limiter_fixture_factory(
-            name="api_limiter",
-            hourly_rate=RateLimit.per_second(10),  # 10 calls/second
-            max_drift=0.2,  # 20% tolerance
-            on_drift_callback=on_drift
-        )
-
-    @weight(1)
-    def test_api_call(api_limiter):
-        with api_limiter.rate_limited_context() as ctx:
-            # Context entry waits if rate limit would be exceeded
-            response = api.get("/data")
-            assert response.status_code == 200
-            assert ctx.call_count >= 1
-
-**Key Features:**
-
-* **Token Bucket Algorithm**: Allows controlled bursts while maintaining average rate
-* **Shared State**: Rate limiting coordinated across all workers
-* **Drift Detection**: Monitors actual vs. target rate and triggers callbacks
-* **Max Calls**: Optional limit on total calls with callback
-* **Dynamic Rates**: Support for callable rate functions
 
-**Rate Limit Helpers:**
-
-.. code-block:: python
+Rate Limiting and Shared State
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-    RateLimit.per_second(10)   # 10 calls per second
-    RateLimit.per_minute(600)  # 600 calls per minute
-    RateLimit.per_hour(3600)   # 3600 calls per hour
-    RateLimit.per_day(86400)   # 86400 calls per day
+For rate limiting and shared state management across pytest-xdist workers,
+see the companion package `pytest-xdist-rate-limit`_.
 
-See the full documentation for more examples and advanced usage.
+.. _`pytest-xdist-rate-limit`: https://github.com/xverges/pytest-xdist-rate-limit
 
 
 License

examples/conftest.py

@@ -1,9 +1,3 @@
-"""Conftest for examples to ensure fixtures work with xdist."""
+"""Conftest for examples."""
 
-from pytest_load_testing.concurrent_fixtures import (
-    rate_limiter_fixture_factory,
-    shared_json_fixture_factory,
-)
-
-# Re-export the fixtures so they're available in examples
-__all__ = ["shared_json_fixture_factory", "rate_limiter_fixture_factory"]
+# No special fixtures needed for basic load testing examples

examples/test_load_example.py

@@ -0,0 +1,60 @@
+"""
+Example: Load testing with weighted distribution
+
+Run with: pytest -n auto --load-test examples/test_load_example.py
+
+This demonstrates:
+1. Weighted test selection (tests run with different frequencies)
+2. Conditional stopping based on iteration count
+
+TEST_CODE:
+```python
+result = pytester.runpytest('--load-test', '-n', '2', '-v')
+result.stdout.fnmatch_lines([
+    '*Interrupted: Test session completed*',
+])
+assert result.ret == pytest.ExitCode.INTERRUPTED
+```
+"""
+
+import pytest
+
+from pytest_load_testing import stop_load_testing, weight
+
+# Simple counter without shared state
+_iteration_count = 0
+
+
+@pytest.fixture
+def iteration_counter(request):
+    """Common fixture that tracks iterations and stops after threshold."""
+    global _iteration_count
+    _iteration_count += 1
+
+    # Stop after 100 total test executions
+    if _iteration_count >= 100:
+        stop_load_testing(request, "Test session completed")
+
+
+@weight(70)
+def test_read_heavy(iteration_counter):
+    """70% of requests - simulates read-heavy operations."""
+    assert True
+
+
+@weight(20)
+def test_write_operations(iteration_counter):
+    """20% of requests - simulates write operations."""
+    assert True
+
+
+@weight(10)
+def test_admin_operations(iteration_counter):
+    """10% of requests - simulates admin operations."""
+    assert True
+
+
+@weight(1)
+def test_health_check(iteration_counter):
+    """1% of requests - simulates health check."""
+    assert True

examples/test_rate_limiter_example.py

@@ -37,7 +37,6 @@ classifiers = [
 dependencies = [
     "pytest>=6.2.0",
     "pytest-xdist>=2.0.0",
-    "filelock>=3.0.0",
 ]
 [project.urls]
 Repository = "https://github.com/xverges/pytest-xdist-load-testing"

pyproject.toml

@@ -1,15 +1,9 @@
 """pytest-xdist-load-testing: Load testing scheduler for pytest-xdist."""
 
 from .api import stop_load_testing, weight
-from .concurrent_fixtures import SharedJson, shared_json_fixture_factory
-from .token_bucket_rate_limiter import RateLimit, TokenBucketRateLimiter
 
 __version__ = "0.1.0"
 __all__ = [
     "weight",
     "stop_load_testing",
-    "shared_json_fixture_factory",
-    "SharedJson",
-    "RateLimit",
-    "TokenBucketRateLimiter",
 ]