在 DRF 中测试节流的正确方法是什么?

What is the proper way of testing throttling in DRF?

在 DRF 中测试节流的正确方法是什么?我在网上找不到这个问题的任何答案。我想对每个端点进行单独测试,因为每个端点都有自定义请求限制 (ScopedRateThrottle)。

重要的是它不能影响其他测试 - 它们必须以某种方式 运行 没有节流和限制。

就像人们已经提到的,这并不完全属于单元测试的范围,但是,简单地做这样的事情怎么样:

from django.core.urlresolvers import reverse
from django.test import override_settings
from rest_framework.test import APITestCase, APIClient


class ThrottleApiTests(APITestCase):
    # make sure to override your settings for testing
    TESTING_THRESHOLD = '5/min'
    # THROTTLE_THRESHOLD is the variable that you set for DRF DEFAULT_THROTTLE_RATES
    @override_settings(THROTTLE_THRESHOLD=TESTING_THRESHOLD)
    def test_check_health(self):
        client = APIClient()
        # some end point you want to test (in this case it's a public enpoint that doesn't require authentication
        _url = reverse('check-health')
        # this is probably set in settings in you case
        for i in range(0, self.TESTING_THRESHOLD):
            client.get(_url)

        # this call should err
        response = client.get(_url)
        # 429 - too many requests
        self.assertEqual(response.status_code, 429)

此外,关于您对副作用的担忧,只要您在 setUpsetUpTestData 中创建用户,测试将被隔离(它们应该被隔离),因此无需担心关于 'dirty' 这个意义上的数据或范围。

关于 cache clearing between tests, I would just add cache.clear() in tearDown or try and clear the specific key defined for throttling

一个简单的解决方案是 patch get_rate 油门 class 方法。感谢 tprestegard for this comment!

我有一个自定义 class 我的案例:

from rest_framework.throttling import UserRateThrottle

class AuthRateThrottle(UserRateThrottle):
    scope = 'auth'

在你的测试中:

from unittest.mock import patch
from django.core.cache import cache
from rest_framework import status

class Tests(SimpleTestCase):
    def setUp(self):
        cache.clear()

    @patch('path.to.AuthRateThrottle.get_rate')
    def test_throttling(self, mock):
        mock.return_value = '1/day'
        response = self.client.post(self.url, {})
        self.assertEqual(
            response.status_code,
            status.HTTP_400_BAD_REQUEST,  # some fields are required
        )
        response = self.client.post(self.url, {})
        self.assertEqual(
            response.status_code,
            status.HTTP_429_TOO_MANY_REQUESTS,
        )

也可以修补 DRF 包中的方法以更改标准油门的行为 classes: @patch('rest_framework.throttling.SimpleRateThrottle.get_rate')

我实现了自己的缓存机制,用于根据用户和调用请求的参数进行限制。您可以覆盖 SimpleRateThrottle.get_cache_key 以获得此行为。

以此油门class为例:

class YourCustomThrottleClass(SimpleRateThrottle):
    rate = "1/d"
    scope = "your-custom-throttle"

    def get_cache_key(self, request: Request, view: viewsets.ModelViewSet):
        # we want to throttle the based on the request user as well as the parameter
        # `foo` (i.e. the user can make a request with a different `foo` as many times
        # as they want in a day, but only once a day for a given `foo`).
        foo_request_param = view.kwargs["foo"]
        ident = f"{request.user.pk}_{foo_request_param}"
        # below format is copied from `UserRateThrottle.get_cache_key`.
        return self.cache_format % {"scope": self.scope, "ident": ident}

为了在 TestCase 中清除此问题,我根据需要在每个测试方法中调用以下方法:

def _clear_throttle_cache(self, request_user, foo_param):
    # we need to clear the cache of the throttle limits already stored there.
    throttle = YourCustomThrottleClass()

    # in the below two lines mock whatever attributes on the request and
    # view instances are used to calculate the cache key in `.get_cache_key`
    # which you overrode. Here we use `request.user` and `view.kwargs["foo"]` 
    # to calculate the throttle key, so we mock those.
    pretend_view = MagicMock(kwargs={foo: foo_param})
    pretend_request = MagicMock(user=request_user)
    
    # this is the method you overrode in `YourCustomThrottleClass`.
    throttle_key = throttle.get_cache_key(pretend_request, pretend_view)
    throttle.cache.delete(user_key)