Testing throttling

We believe that throttling is a part of the business requirements, so it must be tested. To help users with it, we provide several utilities:

• For sync controllers: dmr.test.assert_throttling()

• For async controllers: dmr.test.assert_async_throttling()

Testing that an endpoint is throttled usually means driving it to its limit first. Sending the configured max_requests in a loop is slow for large rates such as 1000/hour. What do we do instead?

1. We lower the max_requests value for the first throttle and increase the rate to be a hour. This way we can reliably and fastly test the expected behavior

2. Next, we send several requests (max_requests controls this) that will hit the endpoint, assert that the response status matches success_status code

3. Lastly, we assert that the final request hits the rate limit, we also assert that the headers match our modified rate limit

Examples:

unittest pytest

Default Python’s testing framework:

tests/test_throttling.py

Show imports... 12 lines hidden

[source]

from typing import Final

from django.core.cache import cache
from django.test import TestCase
from typing_extensions import override

from dmr import Controller
from dmr.plugins.pydantic import PydanticFastSerializer
from dmr.test import DMRRequestFactory, assert_throttling
from dmr.throttling import Rate, SyncThrottle
from dmr.throttling.backends import SyncDjangoCache

_URL: Final = '/reports/'


class ReportsController(Controller[PydanticFastSerializer]):
    # A large hourly budget: reaching it with real requests would be slow.
    throttling = (
        SyncThrottle(
            1000,
            Rate.hour,
            # Tests use `LocMemCache`, so the unsafe cache check is disabled:
            backend=SyncDjangoCache(allow_unsafe_cache=True),
        ),
    )

    def get(self) -> str:
        return 'inside'


# In your test files:
class TestReportsThrottling(TestCase):
    @override
    def setUp(self) -> None:
        cache.clear()  # keep throttle state isolated between tests
        self.dmr_rf = DMRRequestFactory()

    def test_endpoint_is_throttled(self) -> None:
        # Lower the throttle, drive a couple of real requests, assert the 429 —
        # no 1000 requests needed. Returns the rejected response:
        assert_throttling(
            ReportsController,
            lambda: self.dmr_rf.get(_URL),
        )

Only the endpoint under test is affected; its throttling is restored afterwards.

There’s also a lower level API:

• dmr.test.reduced_throttling() to reduce the throttle number manually

• dmr.test.assert_throttled() to assert that the response object is from throttling middleware