test_ratelimit.py
4.95 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
# -*- coding: utf8 -*-
# This file is part of PyBossa.
#
# Copyright (C) 2015 SciFabric LTD.
#
# PyBossa is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# PyBossa is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with PyBossa. If not, see <http://www.gnu.org/licenses/>.
"""
This module tests the RateLimit class and decorator for the API.
It tests all the actions: GET, POST, DEL and PUT, as well as the specific
API endpoints like userprogress or vmcp.
"""
import json
from default import flask_app, sentinel
from factories import ProjectFactory, UserFactory
from mock import patch
class TestAPI(object):
app = flask_app.test_client()
def setUp(self):
sentinel.connection.master_for('mymaster').flushall()
limit = flask_app.config.get('LIMIT')
def check_limit(self, url, action, obj, data=None):
# Set the limit
limit = self.limit - 1
# Start check
for i in range(limit, -1, -1):
if action == 'get':
res = self.app.get(url)
elif action == 'post':
if obj == 'project':
data = dict(name=i,
short_name=i,
long_description=u'something')
data = json.dumps(data)
res = self.app.post(url, data=data)
elif action == 'put':
_url = '/api/%s/%s' % (obj, i)
if obj == 'project':
data = dict(name=i,
short_name=i,
long_description=u'something')
data = json.dumps(data)
res = self.app.put(_url + url, data=data)
elif action == 'delete':
_url = '/api/%s/%s' % (obj, i)
res = self.app.delete(_url + url)
else:
raise Exception("action not found")
# Error message
err_msg = "GET X-RateLimit-Remaining not working"
# Tests
print "X-RateLimit-Remaining: %s" % res.headers['X-RateLimit-Remaining']
print "Expected value: %s" % i
assert int(res.headers['X-RateLimit-Remaining']) == i, err_msg
if res.headers['X-RateLimit-Remaining'] == 0:
error = json.loads(res.data)
err_msg = "The status_code should be 429"
assert error['status_code'] == 429, err_msg
err_msg = "The status should be failed"
assert error['status'] == 'failed', err_msg
err_msg = "The exception_cls should be TooManyRequests"
assert error['exception_cls'] == 'TooManyRequests', err_msg
def test_00_api_get(self):
"""Test API GET rate limit."""
# GET as Anonymous
url = '/api/'
action = 'get'
self.check_limit(url, action, 'project')
@patch('pybossa.api.api_base.APIBase._db_query')
def test_00_project_get(self, mock):
"""Test API.project GET rate limit."""
mock.return_value = {}
# GET as Anonymous
url = '/api/project'
action = 'get'
self.check_limit(url, action, 'project')
@patch('pybossa.api.api_base.APIBase._create_instance_from_request')
def test_01_project_post(self, mock):
"""Test API.project POST rate limit."""
mock.return_value = {}
url = '/api/project'
self.check_limit(url, 'post', 'project')
@patch('pybossa.api.api_base.APIBase._delete_instance')
def test_02_project_delete(self, mock):
"""Test API.project DELETE rate limit."""
mock.return_value = {}
url = ''
self.check_limit(url, 'delete', 'project')
@patch('pybossa.api.api_base.APIBase._update_instance')
def test_03_project_put(self, mock):
"""Test API.project PUT rate limit."""
mock.return_value = {}
url = ''
self.check_limit(url, 'put', 'project')
@patch('pybossa.api._retrieve_new_task')
def test_04_new_task(self, mock):
"""Test API.new_task(project_id) GET rate limit."""
mock.return_value = {}
url = '/api/project/1/newtask'
self.check_limit(url, 'get', 'project')
def test_05_vmcp(self):
"""Test API.vmcp GET rate limit."""
url = '/api/vmcp'
self.check_limit(url, 'get', 'project')
@patch('pybossa.api.project_repo')
def test_05_user_progress(self, mock):
"""Test API.user_progress GET rate limit."""
url = '/api/project/1/userprogress'
self.check_limit(url, 'get', 'project')