Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/api/test_throttling.py
# Copyright (C) 2017-2019 The Software Heritage developers | # Copyright (C) 2017-2019 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU Affero General Public License version 3, or any later version | # License: GNU Affero General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
from swh.web.settings.tests import ( | import pytest | ||||
scope1_limiter_rate, scope1_limiter_rate_post, | |||||
scope2_limiter_rate, scope2_limiter_rate_post, | |||||
scope3_limiter_rate, scope3_limiter_rate_post | |||||
) | |||||
from django.conf.urls import url | from django.conf.urls import url | ||||
from django.contrib.auth.models import User | |||||
from django.test.utils import override_settings | from django.test.utils import override_settings | ||||
from rest_framework.views import APIView | from rest_framework.views import APIView | ||||
from rest_framework.response import Response | from rest_framework.response import Response | ||||
from rest_framework.decorators import api_view | from rest_framework.decorators import api_view | ||||
from swh.web.api.throttling import SwhWebRateThrottle, throttle_scope | from swh.web.api.throttling import SwhWebRateThrottle, throttle_scope | ||||
from swh.web.settings.tests import ( | |||||
scope1_limiter_rate, scope1_limiter_rate_post, | |||||
scope2_limiter_rate, scope2_limiter_rate_post, | |||||
scope3_limiter_rate, scope3_limiter_rate_post | |||||
) | |||||
class MockViewScope1(APIView): | class MockViewScope1(APIView): | ||||
throttle_classes = (SwhWebRateThrottle,) | throttle_classes = (SwhWebRateThrottle,) | ||||
throttle_scope = 'scope1' | throttle_scope = 'scope1' | ||||
def get(self, request): | def get(self, request): | ||||
return Response('foo_get') | return Response('foo_get') | ||||
▲ Show 20 Lines • Show All 112 Lines • ▼ Show 20 Lines | def test_scope3_requests_are_throttled_exempted(api_client): | ||||
for _ in range(scope3_limiter_rate+1): | for _ in range(scope3_limiter_rate+1): | ||||
response = api_client.get('/scope3_func') | response = api_client.get('/scope3_func') | ||||
check_response(response, 200) | check_response(response, 200) | ||||
for _ in range(scope3_limiter_rate_post+1): | for _ in range(scope3_limiter_rate_post+1): | ||||
response = api_client.post('/scope3_func') | response = api_client.post('/scope3_func') | ||||
check_response(response, 200) | check_response(response, 200) | ||||
@override_settings(ROOT_URLCONF=__name__) | |||||
@pytest.mark.django_db | |||||
def test_staff_users_are_not_rate_limited(api_client): | |||||
staff_user = User.objects.create_user( | |||||
username='johndoe', password='', is_staff=True) | |||||
api_client.force_login(staff_user) | |||||
for _ in range(scope2_limiter_rate+1): | |||||
response = api_client.get('/scope2_func') | |||||
check_response(response, 200) | |||||
for _ in range(scope2_limiter_rate_post+1): | |||||
response = api_client.post('/scope2_func') | |||||
check_response(response, 200) | |||||
@override_settings(ROOT_URLCONF=__name__) | |||||
@pytest.mark.django_db | |||||
def test_non_staff_users_are_rate_limited(api_client): | |||||
staff_user = User.objects.create_user( | |||||
username='johndoe', password='', is_staff=False) | |||||
api_client.force_login(staff_user) | |||||
for i in range(scope2_limiter_rate): | |||||
response = api_client.get('/scope2_func') | |||||
check_response(response, 200, scope2_limiter_rate, | |||||
scope2_limiter_rate - i - 1) | |||||
response = api_client.get('/scope2_func') | |||||
check_response(response, 429, scope2_limiter_rate, 0) | |||||
for i in range(scope2_limiter_rate_post): | |||||
response = api_client.post('/scope2_func') | |||||
check_response(response, 200, scope2_limiter_rate_post, | |||||
scope2_limiter_rate_post - i - 1) | |||||
response = api_client.post('/scope2_func') | |||||
check_response(response, 429, scope2_limiter_rate_post, 0) |