Merge lp:~djfroofy/txaws/modernize-924459 into lp:txaws

[Drew Smathers]

Need to care how much we care about compatibility since: (1) This requires Twisted >= 11.1.0 so far as I know and (2) Some public members on BaseQuery no longer exist since they are not applicable when using Agent.

[lp:~djfroofy/txaws/modernize-924459]

145. By Drew Smathers on 2012-02-10

cherry pick: change StringIOReceiver to generic StreamingBodyReceiver

146. By Drew Smathers on 2012-03-15

check for response code in _handle_response and errback if >= 400 with response body

[Duncan McGreggor (oubiwann)]

On Thu, Feb 9, 2012 at 10:23 PM, Drew Smathers <email address hidden> wrote:
> Drew Smathers has proposed merging lp:~djfroofy/txaws/modernize-924459 into lp:txaws.
>
> Requested reviews:
>  Duncan McGreggor (oubiwann)
> Related bugs:
>  Bug #924459 in txAWS: "Modernize txaws.client to use twisted.web.client.Agent"
>  https://bugs.launchpad.net/txaws/+bug/924459
>
> For more details, see:
> https://code.launchpad.net/~djfroofy/txaws/modernize-924459/+merge/92404
>
> Need to care how much we care about compatibility since: (1) This requires Twisted >= 11.1.0 so far as I know and (2) Some public members on BaseQuery no longer exist since they are not applicable when using Agent.

I've taken a quick look at the latest code tonight, and first glance
is a happy one :-)

I've branched the code on my laptop so that I can review it on the
plane tomorrow, run the unit tests, etc.

I hope to have more for you soon.

[Duncan McGreggor (oubiwann)]

Merge away!

review: Approve

=== added file 'txaws/client/_producers.py'
--- txaws/client/_producers.py	1970-01-01 00:00:00 +0000
+++ txaws/client/_producers.py	2012-03-15 05:30:25 +0000
@@ -0,0 +1,122 @@
+import os
+
+from zope.interface import implements
+
+from twisted.internet import defer, task
+from twisted.web.iweb import UNKNOWN_LENGTH, IBodyProducer
+
+
+# Code below for FileBodyProducer cut-and-paste from twisted source.
+# Currently this is not released so here temporarily for forward compat.
+
+
+class FileBodyProducer(object):
+    """
+    L{FileBodyProducer} produces bytes from an input file object incrementally
+    and writes them to a consumer.
+
+    Since file-like objects cannot be read from in an event-driven manner,
+    L{FileBodyProducer} uses a L{Cooperator} instance to schedule reads from
+    the file.  This process is also paused and resumed based on notifications
+    from the L{IConsumer} provider being written to.
+
+    The file is closed after it has been read, or if the producer is stopped
+    early.
+
+    @ivar _inputFile: Any file-like object, bytes read from which will be
+        written to a consumer.
+
+    @ivar _cooperate: A method like L{Cooperator.cooperate} which is used to
+        schedule all reads.
+
+    @ivar _readSize: The number of bytes to read from C{_inputFile} at a time.
+    """
+    implements(IBodyProducer)
+
+    # Python 2.4 doesn't have these symbolic constants
+    _SEEK_SET = getattr(os, 'SEEK_SET', 0)
+    _SEEK_END = getattr(os, 'SEEK_END', 2)
+
+    def __init__(self, inputFile, cooperator=task, readSize=2 ** 16):
+        self._inputFile = inputFile
+        self._cooperate = cooperator.cooperate
+        self._readSize = readSize
+        self.length = self._determineLength(inputFile)
+
+
+    def _determineLength(self, fObj):
+        """
+        Determine how many bytes can be read out of C{fObj} (assuming it is not
+        modified from this point on).  If the determination cannot be made,
+        return C{UNKNOWN_LENGTH}.
+        """
+        try:
+            seek = fObj.seek
+            tell = fObj.tell
+        except AttributeError:
+            return UNKNOWN_LENGTH
+        originalPosition = tell()
+        seek(0, self._SEEK_END)
+        end = tell()
+        seek(originalPosition, self._SEEK_SET)
+        return end - originalPosition
+
+
+    def stopProducing(self):
+        """
+        Permanently stop writing bytes from the file to the consumer by
+        stopping the underlying L{CooperativeTask}.
+        """
+        self._inputFile.close()
+        self._task.stop()
+
+
+    def startProducing(self, consumer):
+        """
+        Start a cooperative task which will read bytes from the input file and
+        write them to C{consumer}.  Return a L{Deferred} which fires after all
+        bytes have been written.
+
+        @param consumer: Any L{IConsumer} provider
+        """
+        self._task = self._cooperate(self._writeloop(consumer))
+        d = self._task.whenDone()
+        def maybeStopped(reason):
+            # IBodyProducer.startProducing's Deferred isn't support to fire if
+            # stopProducing is called.
+            reason.trap(task.TaskStopped)
+            return defer.Deferred()
+        d.addCallbacks(lambda ignored: None, maybeStopped)
+        return d
+
+
+    def _writeloop(self, consumer):
+        """
+        Return an iterator which reads one chunk of bytes from the input file
+        and writes them to the consumer for each time it is iterated.
+        """
+        while True:
+            bytes = self._inputFile.read(self._readSize)
+            if not bytes:
+                self._inputFile.close()
+                break
+            consumer.write(bytes)
+            yield None
+
+
+    def pauseProducing(self):
+        """
+        Temporarily suspend copying bytes from the input file to the consumer
+        by pausing the L{CooperativeTask} which drives that activity.
+        """
+        self._task.pause()
+
+
+    def resumeProducing(self):
+        """
+        Undo the effects of a previous C{pauseProducing} and resume copying
+        bytes to the consumer by resuming the L{CooperativeTask} which drives
+        the write activity.
+        """
+        self._task.resume()
+

=== modified file 'txaws/client/base.py'
--- txaws/client/base.py	2011-11-29 08:17:54 +0000
+++ txaws/client/base.py	2012-03-15 05:30:25 +0000
@@ -3,10 +3,25 @@
 except ImportError:
     from xml.parsers.expat import ExpatError as ParseError
 
+import warnings
+from StringIO import StringIO
+
 from twisted.internet.ssl import ClientContextFactory
+from twisted.internet.protocol import Protocol
+from twisted.internet.defer import Deferred, succeed, fail
+from twisted.python import failure
 from twisted.web import http
+from twisted.web.iweb import UNKNOWN_LENGTH
 from twisted.web.client import HTTPClientFactory
+from twisted.web.client import Agent
+from twisted.web.client import ResponseDone
+from twisted.web.http import NO_CONTENT
+from twisted.web.http_headers import Headers
 from twisted.web.error import Error as TwistedWebError
+try:
+    from twisted.web.client import FileBodyProducer
+except ImportError:
+    from txaws.client._producers import FileBodyProducer
 
 from txaws.util import parse
 from txaws.credentials import AWSCredentials
@@ -71,20 +86,122 @@
         self.query_factory = query_factory
         self.parser = parser
 
+class StreamingError(Exception):
+    """
+    Raised if more data or less data is received than expected.
+    """
+
+
+class StreamingBodyReceiver(Protocol):
+    """
+    Streaming HTTP response body receiver.
+
+    TODO: perhaps there should be an interface specifying why
+    finished (Deferred) and content_length are necessary and
+    how to used them; eg. callback/errback finished on completion.
+    """
+    finished = None
+    content_length = None
+
+    def __init__(self, fd=None, readback=True):
+        """
+        @param fd: a file descriptor to write to
+        @param readback: if True read back data from fd to callback finished
+            with, otherwise we call back finish with fd itself
+        with
+        """
+        if fd is None:
+            fd = StringIO()
+        self._fd = fd
+        self._received = 0
+        self._readback = readback
+
+    def dataReceived(self, bytes):
+        streaming = self.content_length is UNKNOWN_LENGTH
+        if not streaming and (self._received > self.content_length):
+            self.transport.loseConnection()
+            raise StreamingError(
+                "Buffer overflow - received more data than "
+                "Content-Length dictated: %d" % self.content_length)
+        # TODO should be some limit on how much we receive
+        self._fd.write(bytes)
+        self._received += len(bytes)
+
+    def connectionLost(self, reason):
+        reason.trap(ResponseDone)
+        d = self.finished
+        self.finished = None
+        streaming = self.content_length is UNKNOWN_LENGTH
+        if streaming or (self._received == self.content_length):
+            if self._readback:
+                self._fd.seek(0)
+                data = self._fd.read()
+                self._fd.close()
+                self._fd = None
+                d.callback(data)
+            else:
+                d.callback(self._fd)
+        else:
+            f = failure.Failure(StreamingError("Connection lost before "
+                "receiving all data"))
+            d.errback(f)
+
+
+class WebClientContextFactory(ClientContextFactory):
+
+    def getContext(self, hostname, port):
+        return ClientContextFactory.getContext(self)
+
+
+class WebVerifyingContextFactory(VerifyingContextFactory):
+
+    def getContext(self, hostname, port):
+        return VerifyingContextFactory.getContext(self)
+
+
+class FakeClient(object):
+    """
+    XXX
+    A fake client object for some degree of backwards compatability for
+    code using the client attibute on BaseQuery to check url, status
+    etc.
+    """
+    url = None
+    status = None
 
 class BaseQuery(object):
 
-    def __init__(self, action=None, creds=None, endpoint=None, reactor=None):
+    def __init__(self, action=None, creds=None, endpoint=None, reactor=None,
+        body_producer=None, receiver_factory=None):
         if not action:
             raise TypeError("The query requires an action parameter.")
-        self.factory = HTTPClientFactory
         self.action = action
         self.creds = creds
         self.endpoint = endpoint
         if reactor is None:
             from twisted.internet import reactor
         self.reactor = reactor
-        self.client = None
+        self._client = None
+        self.request_headers = None
+        self.response_headers = None
+        self.body_producer = body_producer
+        self.receiver_factory = receiver_factory or StreamingBodyReceiver
+
+    @property
+    def client(self):
+        if self._client is None:
+            self._client_deprecation_warning()
+            self._client = FakeClient()
+        return self._client
+
+    @client.setter
+    def client(self, value):
+        self._client_deprecation_warning()
+        self._client = value
+
+    def _client_deprecation_warning(self):
+        warnings.warn('The client attribute on BaseQuery is deprecated and'
+                      ' will go away in future release.')
 
     def get_page(self, url, *args, **kwds):
         """
@@ -95,16 +212,39 @@
         """
         contextFactory = None
         scheme, host, port, path = parse(url)
-        self.client = self.factory(url, *args, **kwds)
+        data = kwds.get('postdata', None)
+        self._method = method = kwds.get('method', 'GET')
+        self.request_headers = self._headers(kwds.get('headers', {}))
+        if (self.body_producer is None) and (data is not None):
+            self.body_producer = FileBodyProducer(StringIO(data))
         if scheme == "https":
             if self.endpoint.ssl_hostname_verification:
-                contextFactory = VerifyingContextFactory(host)
+                contextFactory = WebVerifyingContextFactory(host)
             else:
-                contextFactory = ClientContextFactory()
-            self.reactor.connectSSL(host, port, self.client, contextFactory)
+                contextFactory = WebClientContextFactory()
+            agent = Agent(self.reactor, contextFactory)
+            self.client.url = url
+            d = agent.request(method, url, self.request_headers,
+                self.body_producer)
         else:
-            self.reactor.connectTCP(host, port, self.client)
-        return self.client.deferred
+            agent = Agent(self.reactor)
+            d = agent.request(method, url, self.request_headers,
+                self.body_producer)
+        d.addCallback(self._handle_response)
+        return d
+
+    def _headers(self, headers_dict):
+        """
+        Convert dictionary of headers into twisted.web.client.Headers object.
+        """
+        return Headers(dict((k,[v]) for (k,v) in headers_dict.items()))
+
+    def _unpack_headers(self, headers):
+        """
+        Unpack twisted.web.client.Headers object to dict. This is to provide
+        backwards compatability.
+        """
+        return dict((k,v[0]) for (k,v) in headers.getAllRawHeaders())
 
     def get_request_headers(self, *args, **kwds):
         """
@@ -114,8 +254,32 @@
         The AWS S3 API depends upon setting headers. This method is provided as
         a convenience for debugging issues with the S3 communications.
         """
-        if self.client:
-            return self.client.headers
+        if self.request_headers:
+            return self._unpack_headers(self.request_headers)
+
+    def _handle_response(self, response):
+        """
+        Handle the HTTP response by memoing the headers and then delivering
+        bytes.
+        """
+        self.client.status = response.code
+        self.response_headers = headers = response.headers
+        # XXX This workaround (which needs to be improved at that) for possible
+        # bug in Twisted with new client:
+        # http://twistedmatrix.com/trac/ticket/5476
+        if self._method.upper() == 'HEAD' or response.code == NO_CONTENT:
+            return succeed('')
+        receiver = self.receiver_factory()
+        receiver.finished = d = Deferred()
+        receiver.content_length = response.length
+        response.deliverBody(receiver)
+        if response.code >= 400:
+            d.addCallback(self._fail_response, response)
+        return d
+
+    def _fail_response(self, data, response):
+       return fail(failure.Failure(
+           TwistedWebError(response.code, response=data)))
 
     def get_response_headers(self, *args, **kwargs):
         """
@@ -125,5 +289,6 @@
         The AWS S3 API depends upon setting headers. This method is used by the
         head_object API call for getting a S3 object's metadata.
         """
-        if self.client:
-            return self.client.response_headers
+        if self.response_headers:
+            return self._unpack_headers(self.response_headers)
+

=== modified file 'txaws/client/tests/test_base.py'
--- txaws/client/tests/test_base.py	2012-01-26 18:43:48 +0000
+++ txaws/client/tests/test_base.py	2012-03-15 05:30:25 +0000
@@ -1,6 +1,11 @@
 import os
 
+from StringIO import StringIO
+
+from zope.interface import implements
+
 from twisted.internet import reactor
+from twisted.internet.defer import succeed, Deferred
 from twisted.internet.error import ConnectionRefusedError
 from twisted.protocols.policies import WrappingFactory
 from twisted.python import log
@@ -8,14 +13,18 @@
 from twisted.python.failure import Failure
 from twisted.test.test_sslverify import makeCertificate
 from twisted.web import server, static
+from twisted.web.iweb import IBodyProducer
 from twisted.web.client import HTTPClientFactory
+from twisted.web.client import ResponseDone
+from twisted.web.resource import Resource
 from twisted.web.error import Error as TwistedWebError
 
 from txaws.client import ssl
 from txaws.client.base import BaseClient, BaseQuery, error_wrapper
+from txaws.client.base import StreamingBodyReceiver
 from txaws.service import AWSServiceEndpoint
 from txaws.testing.base import TXAWSTestCase
-
+from txaws.testing.producers import StringBodyProducer
 
 class ErrorWrapperTestCase(TXAWSTestCase):
 
@@ -63,6 +72,12 @@
         self.assertEquals(client.parser, "parser")
 
 
+class PuttableResource(Resource):
+
+    def render_PUT(self, reuqest):
+        return ''
+
+
 class BaseQueryTestCase(TXAWSTestCase):
 
     def setUp(self):
@@ -71,6 +86,7 @@
         os.mkdir(name)
         FilePath(name).child("file").setContent("0123456789")
         r = static.File(name)
+        r.putChild('thing_to_put', PuttableResource())
         self.site = server.Site(r, timeout=None)
         self.wrapper = WrappingFactory(self.site)
         self.port = self._listen(self.wrapper)
@@ -99,7 +115,6 @@
 
     def test_creation(self):
         query = BaseQuery("an action", "creds", "http://endpoint")
-        self.assertEquals(query.factory, HTTPClientFactory)
         self.assertEquals(query.action, "an action")
         self.assertEquals(query.creds, "creds")
         self.assertEquals(query.endpoint, "http://endpoint")
@@ -142,16 +157,58 @@
     def test_get_response_headers_with_client(self):
 
         def check_results(results):
+            #self.assertEquals(sorted(results.keys()), [
+            #    "accept-ranges", "content-length", "content-type", "date",
+            #    "last-modified", "server"])
+            # XXX I think newclient exludes content-length from headers?
+            # Also the header names are capitalized ... do we need to worry
+            # about backwards compat?
             self.assertEquals(sorted(results.keys()), [
-                "accept-ranges", "content-length", "content-type", "date",
-                "last-modified", "server"])
-            self.assertEquals(len(results.values()), 6)
+                "Accept-Ranges", "Content-Type", "Date",
+                "Last-Modified", "Server"])
+            self.assertEquals(len(results.values()), 5)
 
         query = BaseQuery("an action", "creds", "http://endpoint")
         d = query.get_page(self._get_url("file"))
         d.addCallback(query.get_response_headers)
         return d.addCallback(check_results)
 
+    def test_errors(self):
+        query = BaseQuery("an action", "creds", "http://endpoint")
+        d = query.get_page(self._get_url("not_there"))
+        self.assertFailure(d, TwistedWebError)
+        return d
+
+    def test_custom_body_producer(self):
+
+        def check_producer_was_used(ignore):
+            self.assertEqual(producer.written, 'test data')
+
+        producer = StringBodyProducer('test data')
+        query = BaseQuery("an action", "creds", "http://endpoint",
+            body_producer=producer)
+        d = query.get_page(self._get_url("thing_to_put"), method='PUT')
+        return d.addCallback(check_producer_was_used)
+
+    def test_custom_receiver_factory(self):
+
+        class TestReceiverProtocol(StreamingBodyReceiver):
+            used = False
+
+            def __init__(self):
+                StreamingBodyReceiver.__init__(self)
+                TestReceiverProtocol.used = True
+
+        def check_used(ignore):
+            self.assert_(TestReceiverProtocol.used)
+
+        query = BaseQuery("an action", "creds", "http://endpoint",
+            receiver_factory=TestReceiverProtocol)
+        d = query.get_page(self._get_url("file"))
+        d.addCallback(self.assertEquals, "0123456789")
+        d.addCallback(check_used)
+        return d
+
     # XXX for systems that don't have certs in the DEFAULT_CERT_PATH, this test
     # will fail; instead, let's create some certs in a temp directory and set
     # the DEFAULT_CERT_PATH to point there.
@@ -167,8 +224,9 @@
             def __init__(self):
                 self.connects = []
 
-            def connectSSL(self, host, port, client, factory):
-                self.connects.append((host, port, client, factory))
+            def connectSSL(self, host, port, factory, contextFactory, timeout,
+                bindAddress):
+                self.connects.append((host, port, factory, contextFactory))
 
         certs = makeCertificate(O="Test Certificate", CN="something")[1]
         self.patch(ssl, "_ca_certs", certs)
@@ -176,9 +234,56 @@
         endpoint = AWSServiceEndpoint(ssl_hostname_verification=True)
         query = BaseQuery("an action", "creds", endpoint, fake_reactor)
         query.get_page("https://example.com/file")
-        [(host, port, client, factory)] = fake_reactor.connects
+        [(host, port, factory, contextFactory)] = fake_reactor.connects
         self.assertEqual("example.com", host)
         self.assertEqual(443, port)
-        self.assertTrue(isinstance(factory, ssl.VerifyingContextFactory))
-        self.assertEqual("example.com", factory.host)
-        self.assertNotEqual([], factory.caCerts)
+        wrappedFactory = contextFactory._webContext
+        self.assertTrue(isinstance(wrappedFactory, ssl.VerifyingContextFactory))
+        self.assertEqual("example.com", wrappedFactory.host)
+        self.assertNotEqual([], wrappedFactory.caCerts)
+
+class StreamingBodyReceiverTestCase(TXAWSTestCase):
+
+    def test_readback_mode_on(self):
+        """
+        Test that when readback mode is on inside connectionLost() data will
+        be read back from the start of the file we're streaming and results
+        passed to finished callback.
+        """
+
+        receiver = StreamingBodyReceiver()
+        d = Deferred()
+        receiver.finished = d
+        receiver.content_length = 5
+        fd = receiver._fd
+        receiver.dataReceived('hello')
+        why = Failure(ResponseDone('done'))
+        receiver.connectionLost(why)
+        self.assertEqual(d.result, 'hello')
+        self.assert_(fd.closed)
+
+    def test_readback_mode_off(self):
+        """
+        Test that when readback mode is off connectionLost() will simply
+        callback finished with the fd.
+        """
+
+        receiver = StreamingBodyReceiver(readback=False)
+        d = Deferred()
+        receiver.finished = d
+        receiver.content_length = 5
+        fd = receiver._fd
+        receiver.dataReceived('hello')
+        why = Failure(ResponseDone('done'))
+        receiver.connectionLost(why)
+        self.assertIdentical(d.result, fd)
+        self.assertIdentical(receiver._fd, fd)
+        self.failIf(fd.closed)
+
+    def test_user_fd(self):
+        """
+        Test that user's own file descriptor can be passed to init
+        """
+        user_fd = StringIO()
+        receiver = StreamingBodyReceiver(user_fd)
+        self.assertIdentical(receiver._fd, user_fd)

=== modified file 'txaws/client/tests/test_ssl.py'
--- txaws/client/tests/test_ssl.py	2012-01-26 22:54:44 +0000
+++ txaws/client/tests/test_ssl.py	2012-03-15 05:30:25 +0000
@@ -12,6 +12,10 @@
 from twisted.python.filepath import FilePath
 from twisted.test.test_sslverify import makeCertificate
 from twisted.web import server, static
+try:
+    from twisted.web.client import ResponseFailed
+except ImportError:
+    from twisted.web._newclient import ResponseFailed
 
 from txaws import exception
 from txaws.client import ssl
@@ -32,6 +36,11 @@
 PUBSANKEY = sibpath("public_san.ssl")
 
 
+class WebDefaultOpenSSLContextFactory(DefaultOpenSSLContextFactory):
+    def getContext(self, hostname=None, port=None):
+        return DefaultOpenSSLContextFactory.getContext(self)
+
+
 class BaseQuerySSLTestCase(TXAWSTestCase):
 
     def setUp(self):
@@ -75,7 +84,7 @@
         The L{VerifyingContextFactory} properly allows to connect to the
         endpoint if the certificates match.
         """
-        context_factory = DefaultOpenSSLContextFactory(PRIVKEY, PUBKEY)
+        context_factory = WebDefaultOpenSSLContextFactory(PRIVKEY, PUBKEY)
         self.port = reactor.listenSSL(
             0, self.site, context_factory, interface="127.0.0.1")
         self.portno = self.port.getHost().port
@@ -90,7 +99,7 @@
         The L{VerifyingContextFactory} fails with a SSL error the certificates
         can't be checked.
         """
-        context_factory = DefaultOpenSSLContextFactory(BADPRIVKEY, BADPUBKEY)
+        context_factory = WebDefaultOpenSSLContextFactory(BADPRIVKEY, BADPUBKEY)
         self.port = reactor.listenSSL(
             0, self.site, context_factory, interface="127.0.0.1")
         self.portno = self.port.getHost().port
@@ -98,7 +107,14 @@
         endpoint = AWSServiceEndpoint(ssl_hostname_verification=True)
         query = BaseQuery("an action", "creds", endpoint)
         d = query.get_page(self._get_url("file"))
-        return self.assertFailure(d, SSLError)
+        def fail(ignore):
+            self.fail('Expected SSLError')
+        def check_exception(why):
+            # XXX kind of a mess here ... need to unwrap the
+            # exception and check
+            root_exc = why.value[0][0].value
+            self.assert_(isinstance(root_exc, SSLError))
+        return d.addCallbacks(fail, check_exception)
 
     def test_ssl_verification_bypassed(self):
         """
@@ -121,7 +137,7 @@
         L{VerifyingContextFactory} supports checking C{subjectAltName} in the
         certificate if it's available.
         """
-        context_factory = DefaultOpenSSLContextFactory(PRIVSANKEY, PUBSANKEY)
+        context_factory = WebDefaultOpenSSLContextFactory(PRIVSANKEY, PUBSANKEY)
         self.port = reactor.listenSSL(
             0, self.site, context_factory, interface="127.0.0.1")
         self.portno = self.port.getHost().port

=== added file 'txaws/testing/producers.py'
--- txaws/testing/producers.py	1970-01-01 00:00:00 +0000
+++ txaws/testing/producers.py	2012-03-15 05:30:25 +0000
@@ -0,0 +1,23 @@
+from zope.interface import implements
+
+from twisted.internet.defer import succeed
+from twisted.web.iweb import IBodyProducer
+
+class StringBodyProducer(object):
+    implements(IBodyProducer)
+
+    def __init__(self, data):
+        self.data = data
+        self.length = len(data)
+        self.written = None
+
+    def startProducing(self, consumer):
+        consumer.write(self.data)
+        self.written = self.data
+        return succeed(None)
+
+    def pauseProducing(self):
+        pass
+
+    def stopProducing(self):
+        pass