Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/api/test_throttling.py
| # Copyright (C) 2017-2020 The Software Heritage developers | # Copyright (C) 2017-2021 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 | ||||
| import pytest | import pytest | ||||
| from django.conf.urls import url | from django.conf.urls import url | ||||
| from django.contrib.auth.models import Permission, User | from django.contrib.auth.models import User | ||||
| from django.contrib.contenttypes.models import ContentType | |||||
| from django.test.utils import override_settings | from django.test.utils import override_settings | ||||
| from rest_framework.decorators import api_view | from rest_framework.decorators import api_view | ||||
| from rest_framework.response import Response | from rest_framework.response import Response | ||||
| from rest_framework.views import APIView | from rest_framework.views import APIView | ||||
| from swh.web.api.throttling import ( | from swh.web.api.throttling import ( | ||||
| API_THROTTLING_EXEMPTED_PERM, | API_THROTTLING_EXEMPTED_PERM, | ||||
| SwhWebRateThrottle, | SwhWebRateThrottle, | ||||
| SwhWebUserRateThrottle, | SwhWebUserRateThrottle, | ||||
| throttle_scope, | throttle_scope, | ||||
| ) | ) | ||||
| from swh.web.settings.tests import ( | from swh.web.settings.tests import ( | ||||
| scope1_limiter_rate, | scope1_limiter_rate, | ||||
| scope1_limiter_rate_post, | scope1_limiter_rate_post, | ||||
| scope2_limiter_rate, | scope2_limiter_rate, | ||||
| scope2_limiter_rate_post, | scope2_limiter_rate_post, | ||||
| scope3_limiter_rate, | scope3_limiter_rate, | ||||
| scope3_limiter_rate_post, | scope3_limiter_rate_post, | ||||
| ) | ) | ||||
| from swh.web.tests.utils import create_django_permission | |||||
| from swh.web.urls import urlpatterns | from swh.web.urls import urlpatterns | ||||
| 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): | ||||
| ▲ Show 20 Lines • Show All 174 Lines • ▼ Show 20 Lines | def test_non_staff_users_are_rate_limited(api_client): | ||||
| response = api_client.post("/scope2_func") | response = api_client.post("/scope2_func") | ||||
| check_response(response, 429, scope2_limiter_rate_post_user, 0) | check_response(response, 429, scope2_limiter_rate_post_user, 0) | ||||
| @override_settings(ROOT_URLCONF=__name__) | @override_settings(ROOT_URLCONF=__name__) | ||||
| @pytest.mark.django_db | @pytest.mark.django_db | ||||
| def test_users_with_throttling_exempted_perm_are_not_rate_limited(api_client): | def test_users_with_throttling_exempted_perm_are_not_rate_limited(api_client): | ||||
| user = User.objects.create_user(username="johndoe", password="") | user = User.objects.create_user(username="johndoe", password="") | ||||
| perm_splitted = API_THROTTLING_EXEMPTED_PERM.split(".") | user.user_permissions.add(create_django_permission(API_THROTTLING_EXEMPTED_PERM)) | ||||
| app_label = ".".join(perm_splitted[:-1]) | |||||
| perm_name = perm_splitted[-1] | |||||
| content_type = ContentType.objects.create(app_label=app_label, model="dummy") | |||||
| permission = Permission.objects.create( | |||||
| codename=perm_name, name=perm_name, content_type=content_type, | |||||
| ) | |||||
| user.user_permissions.add(permission) | |||||
| assert user.has_perm(API_THROTTLING_EXEMPTED_PERM) | assert user.has_perm(API_THROTTLING_EXEMPTED_PERM) | ||||
| api_client.force_login(user) | api_client.force_login(user) | ||||
| for _ in range(scope2_limiter_rate + 1): | for _ in range(scope2_limiter_rate + 1): | ||||
| response = api_client.get("/scope2_func") | response = api_client.get("/scope2_func") | ||||
| check_response(response, 200) | check_response(response, 200) | ||||
| for _ in range(scope2_limiter_rate_post + 1): | for _ in range(scope2_limiter_rate_post + 1): | ||||
| response = api_client.post("/scope2_func") | response = api_client.post("/scope2_func") | ||||
| check_response(response, 200) | check_response(response, 200) | ||||