Merge lp:~nataliabidart/ubuntu-sso-client/add-oauth-to-url into lp:ubuntu-sso-client

[Natalia Bidart]

- Add a method to obtain an OAuth signed uri.

[Alejandro J. Cura (alecu)]

Looks great!

review: Approve

=== modified file 'ubuntu_sso/utils/webclient/common.py'
--- ubuntu_sso/utils/webclient/common.py	2012-01-25 15:02:50 +0000
+++ ubuntu_sso/utils/webclient/common.py	2012-01-27 20:24:24 +0000
@@ -19,6 +19,7 @@
 
 from httplib2 import iri2uri
 from oauth import oauth
+from twisted.internet import defer
 
 from ubuntu_sso.utils.webclient.timestamp import TimestampChecker
 
@@ -103,18 +104,21 @@
     def iri_to_uri(self, iri):
         """Transform a unicode iri into a ascii uri."""
         if not isinstance(iri, unicode):
-            raise TypeError
+            raise TypeError('iri %r should be unicode.' % iri)
         return bytes(iri2uri(iri))
 
-    def build_oauth_headers(self, method, uri, credentials, timestamp):
+    def build_oauth_request(self, method, uri, credentials, timestamp,
+                            parameters=None):
         """Build an oauth request given some credentials."""
         consumer = oauth.OAuthConsumer(credentials["consumer_key"],
                                        credentials["consumer_secret"])
         token = oauth.OAuthToken(credentials["token"],
                                  credentials["token_secret"])
-        parameters = {}
+        if parameters is None:
+            parameters = {}
         if timestamp:
             parameters["oauth_timestamp"] = timestamp
+
         request = oauth.OAuthRequest.from_consumer_and_token(
                                             http_url=uri,
                                             http_method=method,
@@ -126,7 +130,23 @@
         else:
             sig_method = oauth.OAuthSignatureMethod_HMAC_SHA1()
         request.sign_request(sig_method, consumer, token)
+        return request
+
+    def build_oauth_headers(self, method, uri, credentials, timestamp):
+        """Build oauth request headers given some credentials."""
+        request = self.build_oauth_request(method, uri, credentials, timestamp)
         return request.to_header()
 
+    @defer.inlineCallbacks
+    def build_signed_iri(self, iri, credentials, parameters=None):
+        """Build a new iri signing 'iri' with 'credentials'."""
+        uri = self.iri_to_uri(iri)
+        timestamp = yield self.get_timestamp()
+        request = self.build_oauth_request(method='GET', uri=uri,
+                                           credentials=credentials,
+                                           timestamp=timestamp,
+                                           parameters=parameters)
+        defer.returnValue(request.to_url())
+
     def shutdown(self):
         """Shut down all pending requests (if possible)."""

=== modified file 'ubuntu_sso/utils/webclient/tests/test_webclient.py'
--- ubuntu_sso/utils/webclient/tests/test_webclient.py	2012-01-27 16:09:40 +0000
+++ ubuntu_sso/utils/webclient/tests/test_webclient.py	2012-01-27 20:24:24 +0000
@@ -23,6 +23,7 @@
 from twisted.cred import checkers, portal
 from twisted.internet import defer
 from twisted.web import http, resource, server, guard
+from urllib2 import urlparse
 
 from ubuntuone.devtools.testcases import TestCase
 from ubuntuone.devtools.testcases.squid import SquidTestCase
@@ -470,8 +471,28 @@
         self.assertNotIn("cLaVe", hd)
 
 
-class OAuthTestCase(TestCase):
-    """Test for the oauth signing code."""
+class OAuthPlainTextTestCase(TestCase):
+    """Test for the oauth signing code using PLAINTEXT."""
+
+    oauth_sign = "PLAINTEXT"
+    sample_method = "GET"
+    sample_url = "http://one.ubuntu.com/"
+    sample_params = {'next': 'yadda-yadda-yo', 'foo': 'naranja fanta'}
+    timestamp = 1
+    expected_params = {
+        "oauth_timestamp": str(timestamp),
+        "oauth_consumer_key": SAMPLE_CREDENTIALS["consumer_key"],
+        "oauth_token": SAMPLE_CREDENTIALS["token"],
+        "oauth_nonce": ANY_VALUE,
+        "oauth_signature": ANY_VALUE,
+    }
+
+    @defer.inlineCallbacks
+    def setUp(self):
+        yield super(OAuthPlainTextTestCase, self).setUp()
+        self.expected_params["oauth_signature_method"] = self.oauth_sign
+        oauth_sign_plain = self.oauth_sign == "PLAINTEXT"
+        self.wc = BaseWebClient(oauth_sign_plain=oauth_sign_plain)
 
     def parse_oauth_header(self, header):
         """Parse an oauth header into a tuple of (method, params_dict)."""
@@ -484,38 +505,72 @@
 
         return method, params
 
-    def do_build_oauth_headers(self, expected_sign_method, oauth_sign_plain):
-        """Build the oauth headers for a sample request."""
-
-        sample_method = "GET"
-        sample_url = "http://one.ubuntu.com/"
-        timestamp = 1
-        expected_params = {
-            "oauth_timestamp": str(timestamp),
-            "oauth_consumer_key": SAMPLE_CREDENTIALS["consumer_key"],
-            "oauth_signature_method": expected_sign_method,
-            "oauth_token": SAMPLE_CREDENTIALS["token"],
-            "oauth_nonce": ANY_VALUE,
-            "oauth_signature": ANY_VALUE,
-        }
-
-        wc = BaseWebClient(oauth_sign_plain=oauth_sign_plain)
-        headers = wc.build_oauth_headers(sample_method,
-                                    sample_url, SAMPLE_CREDENTIALS, timestamp)
-
+    def assert_headers_correct(self, headers):
+        """Check that 'headers' match expected_params."""
         method, params = self.parse_oauth_header(headers["Authorization"])
 
         self.assertEqual(method, "OAuth")
 
-        for k, expected_value in expected_params.iteritems():
+        for k, expected_value in self.expected_params.iteritems():
             self.assertIn(k, params)
             if expected_value is not ANY_VALUE:
                 self.assertEqual(params[k], expected_value)
 
-    def test_build_oauth_headers_hmac_sha1(self):
-        """Build the oauth headers with a sha1 signature."""
-        self.do_build_oauth_headers("HMAC-SHA1", oauth_sign_plain=False)
-
-    def test_build_oauth_headers_plaintext(self):
-        """Build the oauth headers with a sha1 signature."""
-        self.do_build_oauth_headers("PLAINTEXT", oauth_sign_plain=True)
+    def test_build_oauth_headers(self):
+        """Check that the oauth headers are properly built."""
+        headers = self.wc.build_oauth_headers(self.sample_method,
+                        self.sample_url, SAMPLE_CREDENTIALS, self.timestamp)
+        self.assert_headers_correct(headers)
+
+    def test_build_oauth_request(self, params=None):
+        """Check that the oauth request are properly built."""
+        request = self.wc.build_oauth_request(self.sample_method,
+                           self.sample_url, SAMPLE_CREDENTIALS, self.timestamp,
+                           parameters=params)
+
+        self.assert_headers_correct(request.to_header())
+        self.assertEqual(request.http_url, self.sample_url)
+        if params is not None:
+            for param, value in params.iteritems():
+                self.assertIn(param, request.parameters)
+                actual = request.parameters[param]
+                self.assertEqual(value, actual)
+
+    def test_build_oauth_request_with_params(self):
+        """Check that the oauth request are properly with params."""
+        self.test_build_oauth_request(params=self.sample_params)
+
+    @defer.inlineCallbacks
+    def test_build_signed_iri(self, params=None):
+        """Check that the oauth signed iri is properly built."""
+        self.patch(self.wc, 'get_timestamp', lambda: self.timestamp)
+        iri = u'http://foo.bar.baz/ñandú'
+        uri = self.wc.iri_to_uri(iri)
+        request = self.wc.build_oauth_request(self.sample_method,
+                           uri, SAMPLE_CREDENTIALS,
+                           self.timestamp, parameters=params)
+
+        result = yield self.wc.build_signed_iri(iri, SAMPLE_CREDENTIALS,
+                                          parameters=params)
+
+        scheme, netloc, path, params, query, _ = urlparse.urlparse(result)
+
+        self.assertEqual(urlparse.urljoin(scheme + '://' + netloc, path), uri)
+
+        expected = request.parameters
+        actual = dict(urlparse.parse_qsl(query))
+        for i in ('oauth_nonce', 'oauth_signature'):
+            expected.pop(i)
+            actual.pop(i)
+        actual['oauth_timestamp'] = int(actual['oauth_timestamp'])
+        self.assertEqual(expected, actual)
+
+    def test_build_signed_iri_with_params(self, params=None):
+        """Check that the oauth signed iri is properly built with params."""
+        return self.test_build_signed_iri(params=self.sample_params)
+
+
+class OAuthHmacSha1TestCase(OAuthPlainTextTestCase):
+    """Test for the oauth signing code using HMAC-SHA1."""
+
+    oauth_sign = "HMAC-SHA1"