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
|
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2010 OpenStack LLC.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import webob.dec
from nova import test
from nova import context
from nova import flags
from nova.api.openstack.ratelimiting import RateLimitingMiddleware
from nova.api.openstack.common import limited
from nova.tests.api.openstack import fakes
from webob import Request
FLAGS = flags.FLAGS
@webob.dec.wsgify
def simple_wsgi(req):
return ""
class RateLimitingMiddlewareTest(test.TestCase):
def test_get_action_name(self):
middleware = RateLimitingMiddleware(simple_wsgi)
def verify(method, url, action_name):
req = Request.blank(url)
req.method = method
action = middleware.get_action_name(req)
self.assertEqual(action, action_name)
verify('PUT', '/servers/4', 'PUT')
verify('DELETE', '/servers/4', 'DELETE')
verify('POST', '/images/4', 'POST')
verify('POST', '/servers/4', 'POST servers')
verify('GET', '/foo?a=4&changes-since=never&b=5', 'GET changes-since')
verify('GET', '/foo?a=4&monkeys-since=never&b=5', None)
verify('GET', '/servers/4', None)
verify('HEAD', '/servers/4', None)
def exhaust(self, middleware, method, url, username, times):
req = Request.blank(url, dict(REQUEST_METHOD=method),
headers={'X-Auth-User': username})
req.environ['nova.context'] = context.RequestContext(username,
username)
for i in range(times):
resp = req.get_response(middleware)
self.assertEqual(resp.status_int, 200)
resp = req.get_response(middleware)
self.assertEqual(resp.status_int, 413)
self.assertTrue('Retry-After' in resp.headers)
def test_single_action(self):
middleware = RateLimitingMiddleware(simple_wsgi)
self.exhaust(middleware, 'DELETE', '/servers/4', 'usr1', 100)
self.exhaust(middleware, 'DELETE', '/servers/4', 'usr2', 100)
def test_POST_servers_action_implies_POST_action(self):
middleware = RateLimitingMiddleware(simple_wsgi)
self.exhaust(middleware, 'POST', '/servers/4', 'usr1', 10)
self.exhaust(middleware, 'POST', '/images/4', 'usr2', 10)
self.assertTrue(set(middleware.limiter._levels) == \
set(['usr1:POST', 'usr1:POST servers', 'usr2:POST']))
def test_POST_servers_action_correctly_ratelimited(self):
middleware = RateLimitingMiddleware(simple_wsgi)
# Use up all of our "POST" allowance for the minute, 5 times
for i in range(5):
self.exhaust(middleware, 'POST', '/servers/4', 'usr1', 10)
# Reset the 'POST' action counter.
del middleware.limiter._levels['usr1:POST']
# All 50 daily "POST servers" actions should be all used up
self.exhaust(middleware, 'POST', '/servers/4', 'usr1', 0)
def test_proxy_ctor_works(self):
middleware = RateLimitingMiddleware(simple_wsgi)
self.assertEqual(middleware.limiter.__class__.__name__, "Limiter")
middleware = RateLimitingMiddleware(simple_wsgi, service_host='foobar')
self.assertEqual(middleware.limiter.__class__.__name__, "WSGIAppProxy")
|
