Added more tests for Client. Fixed a few bugs in the client for sending POST OAuth requests.

Author: joestump

oauth/__init__.py

@@ -563,8 +563,11 @@ def set_signature_method(method):
     def request(self, uri, method="GET", body=None, headers=None, 
         redirections=httplib2.DEFAULT_MAX_REDIRECTS, connection_type=None):
 
+        if not isinstance(headers, dict):
+            headers = {}
+
         if body and method == "POST":
-            parameters = urlparse.parse_qs(body)
+            parameters = dict(urlparse.parse_qsl(body))
         elif method == "GET":
             parsed = urlparse.urlparse(uri)
             parameters = urlparse.parse_qs(parsed.query)     
@@ -578,12 +581,10 @@ def request(self, uri, method="GET", body=None, headers=None,
 
         if method == "POST":
             body = req.to_postdata() 
+            headers['Content-Type'] = 'application/x-www-form-urlencoded'
         elif method == "GET":
             uri = req.to_url()
         else:
-            if headers is None:
-                headers = {}
-
             headers.update(req.to_header())
 
         return httplib2.Http.request(self, uri, method=method, body=body, 
@@ -651,7 +652,7 @@ def sign(self, request, consumer, token):
         try:
             import hashlib # 2.5
             hashed = hmac.new(key, raw, hashlib.sha1)
-        except:
+        except ImportError:
             import sha # Deprecated
             hashed = hmac.new(key, raw, sha)
 

tests/test_oauth.py

@@ -27,7 +27,6 @@
 import time
 import urllib
 import urlparse
-import cgi
 
 class TestError(unittest.TestCase):
     def test_message(self):
@@ -76,18 +75,27 @@ def test_gen_timestamp(self):
         self.assertEqual(exp, now) 
 
 class TestConsumer(unittest.TestCase):
+    def setUp(self):
+        self.key = 'my-key'
+        self.secret = 'my-secret'
+        self.consumer = oauth.Consumer(key=self.key, secret=self.secret)
+
     def test_init(self):
-        key = 'my-key'
-        secret = 'my-secret'
-        consumer = oauth.Consumer(key, secret)
-        self.assertEqual(consumer.key, key)
-        self.assertEqual(consumer.secret, secret)
+        self.assertEqual(self.consumer.key, self.key)
+        self.assertEqual(self.consumer.secret, self.secret)
 
     def test_basic(self):
         self.assertRaises(ValueError, lambda: oauth.Consumer(None, None))
         self.assertRaises(ValueError, lambda: oauth.Consumer('asf', None))
         self.assertRaises(ValueError, lambda: oauth.Consumer(None, 'dasf'))
 
+    def test_str(self):
+        res = dict(urlparse.parse_qsl(str(self.consumer)))
+        self.assertTrue('oauth_consumer_key' in res)
+        self.assertTrue('oauth_consumer_secret' in res)
+        self.assertEquals(res['oauth_consumer_key'], self.consumer.key)
+        self.assertEquals(res['oauth_consumer_secret'], self.consumer.secret)
+
 class TestToken(unittest.TestCase):
     def setUp(self):
         self.key = 'my-key'
@@ -420,7 +428,7 @@ def test_from_request(self):
         qs = urllib.urlencode(params)
         req = oauth.Request.from_request("GET", url, query_string=qs)
 
-        exp = cgi.parse_qs(qs, keep_blank_values=False)
+        exp = urlparse.parse_qs(qs, keep_blank_values=False)
         for k, v in exp.iteritems():
             exp[k] = urllib.unquote(v[0])
 
@@ -669,30 +677,59 @@ class TestClient(unittest.TestCase):
     consumer_secret = '0e9e6413a9ef49510a4f68ed02cd'
     host = 'http://oauth-sandbox.sevengoslings.net'
 
-#    host = "http://term.ie/oauth/example"
-#    consumer_key = 'key'
-#    consumer_secret = 'secret'
-
     def setUp(self):
         self.consumer = oauth.Consumer(key=self.consumer_key, 
             secret=self.consumer_secret)
 
+        self.body = {
+            'foo': 'bar',
+            'bar': 'foo',
+            'blah': 599999
+        }
+
     def _uri(self, type):
         uri = self.oauth_uris.get(type)
         if uri is None:
             raise KeyError("%s is not a valid OAuth URI type." % type)
 
         return "%s%s" % (self.host, uri)
 
-    def test_get_access_token(self):
+    def test_access_token_get(self):
+        """Test getting an access token via GET."""
         client = oauth.Client(self.consumer, None)
         resp, content = client.request(self._uri('request_token'), "GET")
 
         self.assertEquals(int(resp['status']), 200)
 
-    def test_get_access_token(self):
+    def test_access_token_post(self):
+        """Test getting an access token via POST."""
         client = oauth.Client(self.consumer, None)
         resp, content = client.request(self._uri('request_token'), "POST")
+        print resp
+        print content
 
         self.assertEquals(int(resp['status']), 200)
 
+        res = dict(urlparse.parse_qsl(content))
+        self.assertTrue('oauth_token_key' in res)
+        self.assertTrue('oauth_token_secret' in res)
+
+    def _two_legged(self, method):
+        client = oauth.Client(self.consumer, None)
+
+        return client.request(self._uri('two_legged'), method, 
+            body=urllib.urlencode(self.body))
+
+    def test_two_legged_post(self):
+        """A test of a two-legged OAuth POST request."""
+        resp, content = self._two_legged("POST")
+        print resp
+        print content
+
+        self.assertEquals(int(resp['status']), 200)
+
+    def test_two_legged_get(self):
+        """A test of a two-legged OAuth GET request."""
+        resp, content = self._two_legged("GET")
+        self.assertEquals(int(resp['status']), 200)
+