Test the WSGIApp

2 files changed, 68 insertions, 3 deletions

diff --git a/nova/api/rackspace/ratelimiting/__init__.py b/nova/api/rackspace/ratelimiting/__init__.py
index 176e7d66e..64d5fff2c 100644
--- a/nova/api/rackspace/ratelimiting/__init__.py
+++ b/nova/api/rackspace/ratelimiting/__init__.py
@@ -86,7 +86,7 @@ class WSGIApp(object):
         self.limiter = limiter
 
     @webob.dec.wsgify
-    def __call__(req):
+    def __call__(self, req):
         parts = req.path_info.split('/')
         # format: /limiter/<username>/<urlencoded action>
         if req.method != 'POST':
diff --git a/nova/api/rackspace/ratelimiting/tests.py b/nova/api/rackspace/ratelimiting/tests.py
index 545e1d1b6..f924e7805 100644
--- a/nova/api/rackspace/ratelimiting/tests.py
+++ b/nova/api/rackspace/ratelimiting/tests.py
@@ -1,8 +1,10 @@
-import ratelimiting
 import time
 import unittest
+import webob
 
-class Test(unittest.TestCase):
+import nova.api.rackspace.ratelimiting as ratelimiting
+
+class LimiterTest(unittest.TestCase):
 
     def setUp(self):
         self.limits = {
@@ -59,5 +61,68 @@ class Test(unittest.TestCase):
         self.exhaust('c', 0, username='bob')
         self.exhaust('c', 0, username='alice')
 
+
+class WSGIAppTest(unittest.TestCase):
+
+    def setUp(self):
+        test = self
+        class FakeLimiter(object):
+            def __init__(self):
+                self._action = self._username = self._delay = None
+            def mock(self, action, username, delay):
+                self._action = action
+                self._username = username
+                self._delay = delay
+            def perform(self, action, username):
+                test.assertEqual(action, self._action)
+                test.assertEqual(username, self._username)
+                return self._delay
+        self.limiter = FakeLimiter()
+        self.app = ratelimiting.WSGIApp(self.limiter)
+
+    def test_invalid_methods(self):
+        requests = []
+        for method in ['GET', 'PUT', 'DELETE']:
+            req = webob.Request.blank('/limits/michael/breakdance', 
+                                      dict(REQUEST_METHOD=method))
+            requests.append(req)
+        for req in requests:
+            self.assertEqual(req.get_response(self.app).status_int, 405)
+
+    def test_invalid_urls(self):
+        requests = []
+        for prefix in ['limit', '', 'limiter2', 'limiter/limits', 'limiter/1']:
+            req = webob.Request.blank('/%s/michael/breakdance' % prefix,
+                                      dict(REQUEST_METHOD='POST'))
+        requests.append(req)
+        for req in requests:
+            self.assertEqual(req.get_response(self.app).status_int, 404)
+
+    def verify(self, url, username, action, delay=None):
+        """Make sure that POSTing to the given url causes the given username
+        to perform the given action.  Make the internal rate limiter return
+        delay and make sure that the WSGI app returns the correct response.
+        """
+        req = webob.Request.blank(url, dict(REQUEST_METHOD='POST'))
+        self.limiter.mock(action, username, delay)
+        resp = req.get_response(self.app)
+        if not delay:
+            self.assertEqual(resp.status_int, 200)
+        else:
+            self.assertEqual(resp.status_int, 403)
+            self.assertEqual(resp.headers['X-Wait-Seconds'], delay)
+
+    def test_good_urls(self):
+        self.verify('/limiter/michael/hoot', 'michael', 'hoot')
+
+    def test_escaping(self):
+        self.verify('/limiter/michael/jump%20up', 'michael', 'jump up')
+
+    def test_response_to_delays(self):
+        self.verify('/limiter/michael/hoot', 'michael', 'hoot', 1)
+        self.verify('/limiter/michael/hoot', 'michael', 'hoot', 1.56)
+        self.verify('/limiter/michael/hoot', 'michael', 'hoot', 1000)
+
+
 if __name__ == '__main__':
     unittest.main()