Merge lp:~alecu/ubuntuone-client/proxy-tunnel-webcalls into lp:ubuntuone-client

[Alejandro J. Cura]

- Use the txweb webclient from sso for webcalls, so they can be proxied too (LP: #929207, LP: #929212).

- Use the txweb webclient from sso for webcalls, so they can be proxied too. (LP: #929207, LP: #929212)

**NOTE** This branch depends on this SSO branch: lp:~alecu/ubuntu-sso-client/updated-txweb

[lp:~alecu/ubuntuone-client/proxy-tunnel-webcalls]

1231. By Alejandro J. Cura on 2012-03-13

timestamp is already corrected by txweb in sso

[Alejandro J. Cura (alecu)]

*** NOTE ***

This branch depends on this SSO branch: lp:~alecu/ubuntu-sso-client/updated-txweb

[Diego Sarmentero (diegosarmentero)]

GREAT WORK!
+1

review: Approve

[Eric Casteleijn (thisfred)]

Looks good to me

review: Approve

[Roberto Alsina (ralsina)]

looks good, seems to work...

review: Approve

=== modified file 'tests/syncdaemon/test_action_queue.py'
--- tests/syncdaemon/test_action_queue.py	2012-03-06 20:32:14 +0000
+++ tests/syncdaemon/test_action_queue.py	2012-03-13 03:12:18 +0000
@@ -23,7 +23,6 @@
 import operator
 import os
 import unittest
-import urllib2
 import uuid
 
 from functools import wraps
@@ -33,7 +32,7 @@
 
 from mocker import Mocker, MockerTestCase, ANY, expect
 from oauth import oauth
-from twisted.internet import defer, threads, reactor
+from twisted.internet import defer, reactor
 from twisted.internet import error as twisted_error
 from twisted.python.failure import DefaultException, Failure
 from twisted.web import server
@@ -399,6 +398,21 @@
                 self.assertEqual(defined_args[0], 'self')
                 self.assertEqual(set(defined_args[1:]), set(evtargs))
 
+    @defer.inlineCallbacks
+    def test_get_webclient(self):
+        """The webclient is created if it does not exist."""
+        self.assertEqual(self.action_queue.webclient, None)
+        webclient = yield self.action_queue.get_webclient()
+        self.assertNotEqual(webclient, None)
+
+    @defer.inlineCallbacks
+    def test_get_webclient_existing(self):
+        """The webclient is not created again if it exists."""
+        fake_wc = object()
+        self.patch(self.action_queue, "webclient", fake_wc)
+        webclient = yield self.action_queue.get_webclient()
+        self.assertEqual(webclient, fake_wc)
+
 
 class TestLoggingStorageClient(TwistedTestCase):
     """Tests for ensuring magic hash dont show in logs."""
@@ -2519,67 +2533,26 @@
         self.assertEqual(NODE, self.command.node_id)
         self.assertEqual(True, self.command.is_public)
 
-    def test_run_defers_work_to_thread(self):
-        """Test that work is deferred to a thread."""
-        original = threads.deferToThread
-        self.called = False
-
-        def check(function):
-            self.called = True
-            self.assertEqual(
-                self.command._change_public_access_http, function)
-            return defer.Deferred()
-
-        threads.deferToThread = check
-        try:
-            res = self.command._run()
-        finally:
-            threads.deferToThread = original
-
-        self.assertIsInstance(res, defer.Deferred)
-        self.assertTrue(self.called, "deferToThread was called")
-
+    @defer.inlineCallbacks
     def test_change_public_access_http(self):
-        """Test the blocking portion of the command."""
-        self.called = False
-        def check(request):
-            self.called = True
-            url = 'https://one.ubuntu.com/files/api/set_public/%s:%s' % (
+        """Test the command."""
+
+        def check_webcall(request_iri, method=None, post_content=None):
+            """Check the webcall made by this command."""
+            iri = u'https://one.ubuntu.com/files/api/set_public/%s:%s' % (
                 base64.urlsafe_b64encode(VOLUME.bytes).strip("="),
                 base64.urlsafe_b64encode(NODE.bytes).strip("="))
-            self.assertEqual(url, request.get_full_url())
-            self.assertEqual("is_public=True", request.get_data())
-            return StringIO(
-                '{"is_public": true, "public_url": "http://example.com"}')
-
-        from ubuntuone.syncdaemon import action_queue
-        self.patch(action_queue.timestamp_checker, "get_faithful_time",
-                   lambda: 1)
-        action_queue.urlopen = check
-        try:
-            res = self.command._change_public_access_http()
-        finally:
-            action_queue.urlopen = urllib2.urlopen
-
+            self.assertEqual(iri, request_iri)
+            self.assertEqual("is_public=True", post_content)
+            content = '{"is_public": true, "public_url": "http://example.com"}'
+            response = action_queue.txweb.Response(content)
+            return defer.succeed(response)
+
+        self.patch(self.action_queue, "webcall", check_webcall)
+        res = yield self.command._run()
         self.assertEqual(
             {'is_public': True, 'public_url': 'http://example.com'}, res)
 
-    def test_change_public_access_http_uses_timestamp(self):
-        """The timestamp is used for oauth signing."""
-        fake_timestamp = 12345678
-
-        def fake_urlopen(request):
-            """A fake urlopen."""
-            auth = request.headers["Authorization"]
-            expected = 'oauth_timestamp="%d"' % fake_timestamp
-            self.assertIn(expected, auth)
-            return StringIO("[]")
-
-        self.patch(action_queue.timestamp_checker, "get_faithful_time",
-                   lambda: fake_timestamp)
-        self.patch(action_queue, "urlopen", fake_urlopen)
-        self.command._change_public_access_http()
-
     def test_handle_success_push_event(self):
         """Test AQ_CHANGE_PUBLIC_ACCESS_OK is pushed on success."""
         response = {'is_public': True, 'public_url': 'http://example.com'}
@@ -2592,8 +2565,7 @@
     def test_handle_failure_push_event(self):
         """Test AQ_CHANGE_PUBLIC_ACCESS_ERROR is pushed on failure."""
         msg = 'Something went wrong'
-        failure = Failure(urllib2.HTTPError(
-                "http://example.com", 500, "Error", [], StringIO(msg)))
+        failure = Failure(action_queue.txweb.WebClientError("Misc Error", msg))
         self.command.handle_failure(failure=failure)
         event = ('AQ_CHANGE_PUBLIC_ACCESS_ERROR',
                  {'share_id': VOLUME, 'node_id': NODE, 'error': msg})
@@ -2620,11 +2592,11 @@
         default_url = 'https://one.ubuntu.com/files/api/public_files'
         request_queue = RequestQueue(action_queue=self.action_queue)
         command = GetPublicFiles(request_queue)
-        self.assertEqual(command._url, default_url)
+        self.assertEqual(command._iri, default_url)
         custom_url = 'http://example.com:1234/files/api/public_files'
         command_2 = GetPublicFiles(request_queue,
-                                   base_url='http://example.com:1234')
-        self.assertEqual(command_2._url, custom_url)
+                                   base_iri=u'http://example.com:1234')
+        self.assertEqual(command_2._iri, custom_url)
 
     def test_change_public_access(self):
         """Test the get_public_files method.."""
@@ -2634,75 +2606,39 @@
         """Test proper inheritance."""
         self.assertTrue(isinstance(self.command, ActionQueueCommand))
 
-    def test_run_defers_work_to_thread(self):
-        """Test that work is deferred to a thread."""
-        original = threads.deferToThread
-        self.called = False
-
-        def check(function):
-            self.called = True
-            self.assertEqual(
-                self.command._get_public_files_http, function)
-            return defer.Deferred()
-
-        threads.deferToThread = check
-        try:
-            res = self.command._run()
-        finally:
-            threads.deferToThread = original
-
-        self.assertIsInstance(res, defer.Deferred)
-        self.assertTrue(self.called, "deferToThread was called")
-
+    @defer.inlineCallbacks
     def test_get_public_files_http(self):
-        """Test the blocking portion of the command."""
-        self.called = False
+        """Test the _run method of the command."""
         node_id = uuid.uuid4()
         nodekey = '%s' % (base64.urlsafe_b64encode(node_id.bytes).strip("="))
         node_id_2 = uuid.uuid4()
         nodekey_2 = '%s' % (base64.urlsafe_b64encode(
                                                 node_id_2.bytes).strip("="))
         volume_id = uuid.uuid4()
-        def check(request):
-            self.called = True
-            url = 'https://one.ubuntu.com/files/api/public_files'
-            self.assertEqual(url, request.get_full_url())
-            return StringIO(
+
+        def check_webcall(request_iri, method=None):
+            """Check the webcall made by this command."""
+            """Check the webcall made by this command."""
+            iri = u'https://one.ubuntu.com/files/api/public_files'
+            self.assertEqual(method.upper(), "GET")
+            self.assertEqual(iri, request_iri)
+            content = (
                 '[{"nodekey": "%s", "volume_id": null,"public_url": '
                 '"http://example.com"}, '
                 '{"nodekey": "%s", "volume_id": "%s", "public_url": '
                 '"http://example.com"}]' % (nodekey, nodekey_2, volume_id))
-
-        from ubuntuone.syncdaemon import action_queue
-        self.patch(action_queue.timestamp_checker, "get_faithful_time",
-                   lambda: 1)
-        action_queue.urlopen = check
-        try:
-            res = self.command._get_public_files_http()
-        finally:
-            action_queue.urlopen = urllib2.urlopen
+            response = action_queue.txweb.Response(content)
+            return defer.succeed(response)
+
+        self.patch(self.action_queue, "webcall", check_webcall)
+        res = yield self.command._run()
+
         self.assertEqual([{'node_id': str(node_id), 'volume_id': '',
                           'public_url': 'http://example.com'},
                           {'node_id': str(node_id_2),
                            'volume_id': str(volume_id),
                            'public_url': 'http://example.com'}], res)
 
-    def test_get_public_files_http_uses_timestamp(self):
-        """The timestamp is used for oauth signing."""
-        fake_timestamp = 12345678
-
-        def fake_urlopen(request):
-            """A fake urlopen."""
-            auth = request.headers["Authorization"]
-            expected = 'oauth_timestamp="%d"' % fake_timestamp
-            self.assertIn(expected, auth)
-            return StringIO("[]")
-
-        self.patch(action_queue.timestamp_checker, "get_faithful_time",
-                   lambda: fake_timestamp)
-        self.patch(action_queue, "urlopen", fake_urlopen)
-        self.command._get_public_files_http()
-
     def test_handle_success_push_event(self):
         """Test AQ_PUBLIC_FILES_LIST_OK is pushed on success."""
         response = [{'node_id': uuid.uuid4(), 'volume_id':None,
@@ -2714,8 +2650,7 @@
     def test_handle_failure_push_event(self):
         """Test AQ_PUBLIC_FILES_LIST_ERROR is pushed on failure."""
         msg = 'Something went wrong'
-        failure = Failure(urllib2.HTTPError(
-                "http://example.com", 500, "Error", [], StringIO(msg)))
+        failure = Failure(action_queue.txweb.WebClientError("Misc Error", msg))
         self.command.handle_failure(failure=failure)
         event = ('AQ_PUBLIC_FILES_LIST_ERROR', {'error': msg})
         self.assertIn(event, self.command.action_queue.event_queue.events)
@@ -3700,27 +3635,6 @@
         self.assertEqual('share_name', name)
         self.assertTrue(read_only)
 
-    @defer.inlineCallbacks
-    def test_create_share_http_uses_timestamp(self):
-        """The timestamp is used for oauth signing."""
-        fake_timestamp = 12345678
-
-        def fake_urlopen(request):
-            """A fake urlopen."""
-            auth = request.headers["Authorization"]
-            expected = 'oauth_timestamp="%d"' % fake_timestamp
-            self.assertIn(expected, auth)
-
-        self.patch(action_queue.timestamp_checker, "get_faithful_time",
-                   lambda: fake_timestamp)
-        self.patch(action_queue, "urlopen", fake_urlopen)
-        self.user_connect()
-        command = CreateShare(self.request_queue, 'node_id',
-                              '[e-mail]', 'share_name',
-                              ACCESS_LEVEL_RO, 'marker', 'path')
-        self.assertTrue(command.use_http, 'CreateShare should be in http mode')
-        yield command._run()
-
     def test_possible_markers(self):
         """Test that it returns the correct values."""
         cmd = CreateShare(self.request_queue, 'node_id', '[e-mail]',
 
=== modified file 'ubuntuone/syncdaemon/action_queue.py'
--- ubuntuone/syncdaemon/action_queue.py	2012-03-10 00:39:26 +0000
+++ ubuntuone/syncdaemon/action_queue.py	2012-03-13 03:12:18 +0000
@@ -31,19 +31,18 @@
 from collections import deque, defaultdict
 from functools import partial
 from urllib import urlencode
-from urllib2 import urlopen, Request, HTTPError
 from urlparse import urljoin
 
 import OpenSSL.SSL
 
 from zope.interface import implements
-from twisted.internet import reactor, defer, threads, task
+from twisted.internet import reactor, defer, task
 from twisted.internet import error as twisted_errors
 from twisted.names import client as dns_client
 from twisted.python.failure import Failure, DefaultException
 
 from oauth import oauth
-from ubuntu_sso.utils import timestamp_checker
+from ubuntu_sso.utils.webclient import txweb
 from ubuntuone import clientdefs
 from ubuntuone.platform import platform, remove_file
 from ubuntuone.storageprotocol import protocol_pb2, content_hash
@@ -685,6 +684,8 @@
         # credentials
         self.token = None
         self.consumer = None
+        self.credentials = None
+        self.webclient = None
 
         self.client = None  # an instance of self.protocol
 
@@ -730,6 +731,7 @@
 
     def handle_SYS_USER_CONNECT(self, access_token):
         """Stow the access token away for later use."""
+        self.credentials = access_token
         self.token = oauth.OAuthToken(access_token['token'],
                                       access_token['token_secret'])
         self.consumer = oauth.OAuthConsumer(access_token['consumer_key'],
@@ -818,6 +820,25 @@
         else:
             return defer.succeed((self.host, self.port))
 
+
+    @defer.inlineCallbacks
+    def webcall(self, iri, **kwargs):
+        """Perform a web call to the api servers."""
+        webclient = yield self.get_webclient()
+        response = yield webclient.request(iri,
+                             oauth_credentials=self.credentials, **kwargs)
+        defer.returnValue(response)
+
+    @defer.inlineCallbacks
+    def get_webclient(self):
+        """Get the webclient, creating it if needed."""
+        if self.webclient is None:
+            client = yield self.tunnel_runner.get_client()
+            self.webclient = txweb.WebClient(connector=client,
+                                             appname="Ubuntu One",
+                                             oauth_sign_plain=True)
+        defer.returnValue(self.webclient)
+
     @defer.inlineCallbacks
     def _make_connection(self, result):
         """Do the real connect call."""
@@ -1778,36 +1799,23 @@
         if share_to and re.match(EREGEX, share_to):
             self.use_http = True
 
+    @defer.inlineCallbacks
     def _create_share_http(self, node_id, user, name, read_only):
         """Create a share using the HTTP Web API method."""
 
-        url = "https://one.ubuntu.com/files/api/offer_share/"
-        method = oauth.OAuthSignatureMethod_PLAINTEXT()
-        timestamp = timestamp_checker.get_faithful_time()
-        parameters = {"oauth_timestamp": timestamp}
-        request = oauth.OAuthRequest.from_consumer_and_token(
-            http_url=url,
-            http_method="POST",
-            parameters=parameters,
-            oauth_consumer=self.action_queue.consumer,
-            token=self.action_queue.token)
-        request.sign_request(method, self.action_queue.consumer,
-                             self.action_queue.token)
+        iri = u"https://one.ubuntu.com/files/api/offer_share/"
         data = dict(offer_to_email=user,
                     read_only=read_only,
                     node_id=node_id,
                     share_name=name)
         pdata = urlencode(data)
-        headers = request.to_header()
-        req = Request(url, pdata, headers)
-        urlopen(req)
+        yield self.action_queue.webcall(iri, method="POST", post_content=pdata)
 
     def _run(self):
         """Do the actual running."""
         if self.use_http:
             # External user, do the HTTP REST method
-            return threads.deferToThread(self._create_share_http,
-                                         self.node_id, self.share_to,
+            return self._create_share_http(self.node_id, self.share_to,
                                          self.name,
                                          self.access_level != ACCESS_LEVEL_RW)
         else:
@@ -1831,7 +1839,7 @@
         """It didn't work! Push the event."""
         self.action_queue.event_queue.push('AQ_CREATE_SHARE_ERROR',
                                            marker=self.marker,
-                                           error=failure.getErrorMessage())
+                                           error=failure.value[1])
 
     def _acquire_pathlock(self):
         """Acquire pathlock."""
@@ -2113,6 +2121,7 @@
         self.node_id = node_id
         self.is_public = is_public
 
+    @defer.inlineCallbacks
     def _change_public_access_http(self):
         """Change public access using the HTTP Web API method."""
 
@@ -2123,28 +2132,16 @@
                 base64.urlsafe_b64encode(self.share_id.bytes).strip("="),
                 node_key)
 
-        url = "https://one.ubuntu.com/files/api/set_public/%s" % (node_key,)
-        method = oauth.OAuthSignatureMethod_PLAINTEXT()
-        timestamp = timestamp_checker.get_faithful_time()
-        parameters = {"oauth_timestamp": timestamp}
-        request = oauth.OAuthRequest.from_consumer_and_token(
-            http_url=url,
-            http_method="POST",
-            parameters=parameters,
-            oauth_consumer=self.action_queue.consumer,
-            token=self.action_queue.token)
-        request.sign_request(method, self.action_queue.consumer,
-                             self.action_queue.token)
+        iri = u"https://one.ubuntu.com/files/api/set_public/%s" % (node_key,)
         data = dict(is_public=bool(self.is_public))
         pdata = urlencode(data)
-        headers = request.to_header()
-        req = Request(url, pdata, headers)
-        response = urlopen(req)
-        return simplejson.load(response)
+        response = yield self.action_queue.webcall(iri, method="POST",
+                                                   post_content=pdata)
+        defer.returnValue(simplejson.loads(response.content))
 
     def _run(self):
         """See ActionQueueCommand."""
-        return threads.deferToThread(self._change_public_access_http)
+        return self._change_public_access_http()
 
     def handle_success(self, success):
         """See ActionQueueCommand."""
@@ -2156,51 +2153,36 @@
 
     def handle_failure(self, failure):
         """It didn't work! Push the event."""
-        if issubclass(failure.type, HTTPError):
-            message = failure.value.read()
-        else:
-            message = failure.getErrorMessage()
         self.action_queue.event_queue.push('AQ_CHANGE_PUBLIC_ACCESS_ERROR',
                                            share_id=self.share_id,
                                            node_id=self.node_id,
-                                           error=message)
+                                           error=failure.value[1])
 
 
 class GetPublicFiles(ActionQueueCommand):
     """Get the list of public files."""
 
-    __slots__ = ('_url',)
+    __slots__ = ('_iri',)
     logged_attrs = ActionQueueCommand.logged_attrs + __slots__
 
-    def __init__(self, request_queue, base_url='https://one.ubuntu.com'):
+    def __init__(self, request_queue, base_iri=u'https://one.ubuntu.com'):
         super(GetPublicFiles, self).__init__(request_queue)
-        self._url = urljoin(base_url, 'files/api/public_files')
+        self._iri = urljoin(base_iri, u'files/api/public_files')
 
+    @defer.inlineCallbacks
     def _get_public_files_http(self):
         """Get public files list using the HTTP Web API method."""
 
-        method = oauth.OAuthSignatureMethod_PLAINTEXT()
-        timestamp = timestamp_checker.get_faithful_time()
-        parameters = {"oauth_timestamp": timestamp}
-        request = oauth.OAuthRequest.from_consumer_and_token(
-            http_url=self._url,
-            http_method="GET",
-            parameters=parameters,
-            oauth_consumer=self.action_queue.consumer,
-            token=self.action_queue.token)
-        request.sign_request(method, self.action_queue.consumer,
-                             self.action_queue.token)
-        headers = request.to_header()
-        req = Request(self._url, headers=headers)
-        response = urlopen(req)
-        files = simplejson.load(response)
+        response = yield self.action_queue.webcall(self._iri, method="GET")
+
+        files = simplejson.loads(response.content)
         # translate nodekeys to (volume_id, node_id)
         for pf in files:
             _, node_id = self.split_nodekey(pf.pop('nodekey'))
             volume_id = pf['volume_id']
             pf['volume_id'] = '' if volume_id is None else volume_id
             pf['node_id'] = node_id
-        return files
+        defer.returnValue(files)
 
     @property
     def uniqueness(self):
@@ -2213,7 +2195,7 @@
 
     def _run(self):
         """See ActionQueueCommand."""
-        return threads.deferToThread(self._get_public_files_http)
+        return self._get_public_files_http()
 
     def handle_success(self, success):
         """See ActionQueueCommand."""
@@ -2222,12 +2204,8 @@
 
     def handle_failure(self, failure):
         """It didn't work! Push the event."""
-        if issubclass(failure.type, HTTPError):
-            message = failure.value.read()
-        else:
-            message = failure.getErrorMessage()
         self.action_queue.event_queue.push('AQ_PUBLIC_FILES_LIST_ERROR',
-                                           error=message)
+                                           error=failure.value[1])
 
     def split_nodekey(self, nodekey):
         """Split a node key into a share_id, node_id."""