Differential D2242 Diff 7803 swh/web/tests/common/test_throttling.py

Changeset View

Standalone View

swh/web/tests/common/test_throttling.py

	# Copyright (C) 2017-2018 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 (			from swh.web.settings.tests import (
	scope1_limiter_rate, scope1_limiter_rate_post,			scope1_limiter_rate, scope1_limiter_rate_post,
	scope2_limiter_rate, scope2_limiter_rate_post,			scope2_limiter_rate, scope2_limiter_rate_post,
	scope3_limiter_rate, scope3_limiter_rate_post			scope3_limiter_rate, scope3_limiter_rate_post
	)			)

	from django.conf.urls import url			from django.conf.urls import url
	from django.core.cache import cache
	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.test import APIRequestFactory			from rest_framework.test import APIRequestFactory
	from rest_framework.decorators import api_view			from rest_framework.decorators import api_view


	from swh.web.common.throttling import (			from swh.web.common.throttling import SwhWebRateThrottle, throttle_scope
	SwhWebRateThrottle, throttle_scope
	)
	from swh.web.tests.testcase import WebTestCase


	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 All 34 Lines
	urlpatterns = [			urlpatterns = [
	url(r'^scope1_class$', MockViewScope1.as_view()),			url(r'^scope1_class$', MockViewScope1.as_view()),
	url(r'^scope2_func$', mock_view_scope2),			url(r'^scope2_func$', mock_view_scope2),
	url(r'^scope3_class$', MockViewScope3.as_view()),			url(r'^scope3_class$', MockViewScope3.as_view()),
	url(r'^scope3_func$', mock_view_scope3)			url(r'^scope3_func$', mock_view_scope3)
	]			]


	@override_settings(ROOT_URLCONF=__name__)			factory = APIRequestFactory()
				vlorentzUnsubmitted Not Done Inline Actions is this used? vlorentz: is this used?
				anlambertAuthorUnsubmitted Done Inline Actions Nope and it should have been removed before this refactoring, well seen anlambert: Nope and it should have been removed before this refactoring, well seen
	class ThrottlingTests(WebTestCase):
	def setUp(self):
	"""
	Reset the cache so that no throttles will be active
	"""
	cache.clear()
	self.factory = APIRequestFactory()

	def check_response(self, response, status_code,			def check_response(response, status_code,
	limit=None, remaining=None):			limit=None, remaining=None):
	assert response.status_code == status_code			assert response.status_code == status_code
	if limit is not None:			if limit is not None:
	assert response['X-RateLimit-Limit'] == str(limit)			assert response['X-RateLimit-Limit'] == str(limit)
	else:			else:
	assert 'X-RateLimit-Limit' not in response			assert 'X-RateLimit-Limit' not in response
	if remaining is not None:			if remaining is not None:
	assert response['X-RateLimit-Remaining'] == str(remaining)			assert response['X-RateLimit-Remaining'] == str(remaining)
	else:			else:
	assert 'X-RateLimit-Remaining' not in response			assert 'X-RateLimit-Remaining' not in response

	def test_scope1_requests_are_throttled(self):
				@override_settings(ROOT_URLCONF=__name__)
				def test_scope1_requests_are_throttled(api_client):
	"""			"""
	Ensure request rate is limited in scope1			Ensure request rate is limited in scope1
	"""			"""
	for i in range(scope1_limiter_rate):			for i in range(scope1_limiter_rate):
	response = self.client.get('/scope1_class')			response = api_client.get('/scope1_class')
	self.check_response(response, 200, scope1_limiter_rate,			check_response(response, 200, scope1_limiter_rate,
	scope1_limiter_rate - i - 1)			scope1_limiter_rate - i - 1)

	response = self.client.get('/scope1_class')			response = api_client.get('/scope1_class')
	self.check_response(response, 429, scope1_limiter_rate, 0)			check_response(response, 429, scope1_limiter_rate, 0)

	for i in range(scope1_limiter_rate_post):			for i in range(scope1_limiter_rate_post):
	response = self.client.post('/scope1_class')			response = api_client.post('/scope1_class')
	self.check_response(response, 200, scope1_limiter_rate_post,			check_response(response, 200, scope1_limiter_rate_post,
	scope1_limiter_rate_post - i - 1)			scope1_limiter_rate_post - i - 1)

	response = self.client.post('/scope1_class')			response = api_client.post('/scope1_class')
	self.check_response(response, 429, scope1_limiter_rate_post, 0)			check_response(response, 429, scope1_limiter_rate_post, 0)


	def test_scope2_requests_are_throttled(self):			@override_settings(ROOT_URLCONF=__name__)
				def test_scope2_requests_are_throttled(api_client):
	"""			"""
	Ensure request rate is limited in scope2			Ensure request rate is limited in scope2
	"""			"""
	for i in range(scope2_limiter_rate):			for i in range(scope2_limiter_rate):
	response = self.client.get('/scope2_func')			response = api_client.get('/scope2_func')
	self.check_response(response, 200, scope2_limiter_rate,			check_response(response, 200, scope2_limiter_rate,
	scope2_limiter_rate - i - 1)			scope2_limiter_rate - i - 1)

	response = self.client.get('/scope2_func')			response = api_client.get('/scope2_func')
	self.check_response(response, 429, scope2_limiter_rate, 0)			check_response(response, 429, scope2_limiter_rate, 0)

	for i in range(scope2_limiter_rate_post):			for i in range(scope2_limiter_rate_post):
	response = self.client.post('/scope2_func')			response = api_client.post('/scope2_func')
	self.check_response(response, 200, scope2_limiter_rate_post,			check_response(response, 200, scope2_limiter_rate_post,
	scope2_limiter_rate_post - i - 1)			scope2_limiter_rate_post - i - 1)

	response = self.client.post('/scope2_func')			response = api_client.post('/scope2_func')
	self.check_response(response, 429, scope2_limiter_rate_post, 0)			check_response(response, 429, scope2_limiter_rate_post, 0)

	def test_scope3_requests_are_throttled_exempted(self):
				@override_settings(ROOT_URLCONF=__name__)
				def test_scope3_requests_are_throttled_exempted(api_client):
	"""			"""
	Ensure request rate is not limited in scope3 as			Ensure request rate is not limited in scope3 as
	requests coming from localhost are exempted from rate limit.			requests coming from localhost are exempted from rate limit.
	"""			"""
	for _ in range(scope3_limiter_rate+1):			for _ in range(scope3_limiter_rate+1):
	response = self.client.get('/scope3_class')			response = api_client.get('/scope3_class')
	self.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 = self.client.post('/scope3_class')			response = api_client.post('/scope3_class')
	self.check_response(response, 200)			check_response(response, 200)

	for _ in range(scope3_limiter_rate+1):			for _ in range(scope3_limiter_rate+1):
	response = self.client.get('/scope3_func')			response = api_client.get('/scope3_func')
	self.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 = self.client.post('/scope3_func')			response = api_client.post('/scope3_func')
	self.check_response(response, 200)			check_response(response, 200)