Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/api/test_throttling.py
| # Copyright (C) 2017-2021 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 User | |||||
| 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, | ||||
| ▲ Show 20 Lines • Show All 136 Lines • ▼ Show 20 Lines | def test_scope3_requests_are_throttled_exempted(api_client): | ||||
| 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__) | @override_settings(ROOT_URLCONF=__name__) | ||||
| @pytest.mark.django_db | @pytest.mark.django_db | ||||
| def test_staff_users_are_not_rate_limited(api_client): | def test_staff_users_are_not_rate_limited(api_client, staff_user): | ||||
| staff_user = User.objects.create_user( | |||||
| username="johndoe", password="", is_staff=True | |||||
| ) | |||||
| api_client.force_login(staff_user) | api_client.force_login(staff_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) | ||||
| @override_settings(ROOT_URLCONF=__name__) | @override_settings(ROOT_URLCONF=__name__) | ||||
| @pytest.mark.django_db | @pytest.mark.django_db | ||||
| def test_non_staff_users_are_rate_limited(api_client): | def test_non_staff_users_are_rate_limited(api_client, regular_user): | ||||
| user = User.objects.create_user(username="johndoe", password="", is_staff=False) | |||||
| api_client.force_login(user) | api_client.force_login(regular_user) | ||||
| scope2_limiter_rate_user = ( | scope2_limiter_rate_user = ( | ||||
| scope2_limiter_rate * SwhWebUserRateThrottle.NUM_REQUESTS_FACTOR | scope2_limiter_rate * SwhWebUserRateThrottle.NUM_REQUESTS_FACTOR | ||||
| ) | ) | ||||
| for i in range(scope2_limiter_rate_user): | for i in range(scope2_limiter_rate_user): | ||||
| response = api_client.get("/scope2_func") | response = api_client.get("/scope2_func") | ||||
| check_response( | check_response( | ||||
| Show All 17 Lines | for i in range(scope2_limiter_rate_post_user): | ||||
| ) | ) | ||||
| 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( | ||||
| user = User.objects.create_user(username="johndoe", password="") | api_client, regular_user | ||||
| user.user_permissions.add(create_django_permission(API_THROTTLING_EXEMPTED_PERM)) | ): | ||||
| regular_user.user_permissions.add( | |||||
| create_django_permission(API_THROTTLING_EXEMPTED_PERM) | |||||
| ) | |||||
| assert user.has_perm(API_THROTTLING_EXEMPTED_PERM) | assert regular_user.has_perm(API_THROTTLING_EXEMPTED_PERM) | ||||
| api_client.force_login(user) | api_client.force_login(regular_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) | ||||