Merge lp:~djfroofy/txaws/921421-completemultipart into lp:txaws

[Drew Smathers]

Some details need to be worked out for full support, but I think a decent happy path to start with. Depends on 921419-uploadpart which depends on ... hehe.

[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-02-10 22:39:30 +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-02-10 22:39:30 +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
+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
@@ -59,9 +74,10 @@
     @param query_factory: The class or function that produces a query
         object for making requests to the EC2 service.
     @param parser: A parser object for parsing responses from the EC2 service.
+    @param receiver_factory: Factory for receiving responses from EC2 service.
     """
     def __init__(self, creds=None, endpoint=None, query_factory=None,
-                 parser=None):
+                 parser=None, receiver_factory=None):
         if creds is None:
             creds = AWSCredentials()
         if endpoint is None:
@@ -69,22 +85,109 @@
         self.creds = creds
         self.endpoint = endpoint
         self.query_factory = query_factory
+        self.receiver_factory = receiver_factory
         self.parser = parser
 
+class StreamingError(Exception):
+    """
+    Raised if more data or less data is received than expected.
+    """
+
+
+class StringIOBodyReceiver(Protocol):
+    """
+    Simple StringIO-based 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):
+        self._buffer = StringIO()
+        self._received = 0
+
+    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._buffer.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):
+            d.callback(self._buffer.getvalue())
+        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 StringIOBodyReceiver
+
+    @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 +198,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 +240,26 @@
         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)
+        return d
 
     def get_response_headers(self, *args, **kwargs):
         """
@@ -125,5 +269,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-02-10 22:39:30 +0000
@@ -1,6 +1,9 @@
 import os
 
+from zope.interface import implements
+
 from twisted.internet import reactor
+from twisted.internet.defer import succeed
 from twisted.internet.error import ConnectionRefusedError
 from twisted.protocols.policies import WrappingFactory
 from twisted.python import log
@@ -8,14 +11,16 @@
 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.error import Error as TwistedWebError
 
 from txaws.client import ssl
 from txaws.client.base import BaseClient, BaseQuery, error_wrapper
+from txaws.client.base import StringIOBodyReceiver
 from txaws.service import AWSServiceEndpoint
 from txaws.testing.base import TXAWSTestCase
-
+from txaws.testing.producers import StringBodyProducer
 
 class ErrorWrapperTestCase(TXAWSTestCase):
 
@@ -99,7 +104,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 +146,52 @@
     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_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("file"), method='PUT')
+        return d.addCallback(check_producer_was_used)
+
+    def test_custom_receiver_factory(self):
+
+        class TestReceiverProtocol(StringIOBodyReceiver):
+            used = False
+
+            def __init__(self):
+                StringIOBodyReceiver.__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 +207,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 +217,10 @@
         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)

=== 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-02-10 22:39:30 +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

=== modified file 'txaws/s3/client.py'
--- txaws/s3/client.py	2012-01-28 00:39:00 +0000
+++ txaws/s3/client.py	2012-02-10 22:39:30 +0000
@@ -23,7 +23,8 @@
 from txaws.s3.model import (
     Bucket, BucketItem, BucketListing, ItemOwner, LifecycleConfiguration,
     LifecycleConfigurationRule, NotificationConfiguration, RequestPayment,
-    VersioningConfiguration, WebsiteConfiguration)
+    VersioningConfiguration, WebsiteConfiguration, MultipartInitiationResponse,
+    MultipartCompletionResponse)
 from txaws.s3.exception import S3Error
 from txaws.service import AWSServiceEndpoint, S3_ENDPOINT
 from txaws.util import XML, calculate_md5
@@ -74,10 +75,12 @@
 class S3Client(BaseClient):
     """A client for S3."""
 
-    def __init__(self, creds=None, endpoint=None, query_factory=None):
+    def __init__(self, creds=None, endpoint=None, query_factory=None,
+        receiver_factory=None):
         if query_factory is None:
             query_factory = Query
-        super(S3Client, self).__init__(creds, endpoint, query_factory)
+        super(S3Client, self).__init__(creds, endpoint, query_factory,
+            receiver_factory=receiver_factory)
 
     def list_buckets(self):
         """
@@ -87,7 +90,8 @@
         the request.
         """
         query = self.query_factory(
-            action="GET", creds=self.creds, endpoint=self.endpoint)
+            action="GET", creds=self.creds, endpoint=self.endpoint,
+            receiver_factory=self.receiver_factory)
         d = query.submit()
         return d.addCallback(self._parse_list_buckets)
 
@@ -131,7 +135,7 @@
         """
         query = self.query_factory(
             action="GET", creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket)
+            bucket=bucket, receiver_factory=self.receiver_factory)
         d = query.submit()
         return d.addCallback(self._parse_get_bucket)
 
@@ -174,7 +178,8 @@
         """
         query = self.query_factory(action="GET", creds=self.creds,
                                    endpoint=self.endpoint, bucket=bucket,
-                                   object_name="?location")
+                                   object_name="?location",
+                                   receiver_factory=self.receiver_factory)
         d = query.submit()
         return d.addCallback(self._parse_bucket_location)
 
@@ -193,7 +198,8 @@
         """
         query = self.query_factory(
             action='GET', creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name='?lifecycle')
+            bucket=bucket, object_name='?lifecycle',
+            receiver_factory=self.receiver_factory)
         return query.submit().addCallback(self._parse_lifecycle_config)
 
     def _parse_lifecycle_config(self, xml_bytes):
@@ -221,7 +227,8 @@
         """
         query = self.query_factory(
             action='GET', creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name='?website')
+            bucket=bucket, object_name='?website',
+            receiver_factory=self.receiver_factory)
         return query.submit().addCallback(self._parse_website_config)
 
     def _parse_website_config(self, xml_bytes):
@@ -242,7 +249,8 @@
         """
         query = self.query_factory(
             action='GET', creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name='?notification')
+            bucket=bucket, object_name='?notification',
+            receiver_factory=self.receiver_factory)
         return query.submit().addCallback(self._parse_notification_config)
 
     def _parse_notification_config(self, xml_bytes):
@@ -262,7 +270,8 @@
         """
         query = self.query_factory(
             action='GET', creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name='?versioning')
+            bucket=bucket, object_name='?versioning',
+            receiver_factory=self.receiver_factory)
         return query.submit().addCallback(self._parse_versioning_config)
 
     def _parse_versioning_config(self, xml_bytes):
@@ -279,7 +288,8 @@
         """
         query = self.query_factory(
             action='GET', creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name='?acl')
+            bucket=bucket, object_name='?acl',
+            receiver_factory=self.receiver_factory)
         return query.submit().addCallback(self._parse_acl)
 
     def put_bucket_acl(self, bucket, access_control_policy):
@@ -289,7 +299,8 @@
         data = access_control_policy.to_xml()
         query = self.query_factory(
             action='PUT', creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name='?acl', data=data)
+            bucket=bucket, object_name='?acl', data=data,
+            receiver_factory=self.receiver_factory)
         return query.submit().addCallback(self._parse_acl)
 
     def _parse_acl(self, xml_bytes):
@@ -299,8 +310,8 @@
         """
         return AccessControlPolicy.from_xml(xml_bytes)
 
-    def put_object(self, bucket, object_name, data, content_type=None,
-                   metadata={}, amz_headers={}):
+    def put_object(self, bucket, object_name, data=None, content_type=None,
+                   metadata={}, amz_headers={}, body_producer=None):
         """
         Put an object in a bucket.
 
@@ -318,7 +329,8 @@
             action="PUT", creds=self.creds, endpoint=self.endpoint,
             bucket=bucket, object_name=object_name, data=data,
             content_type=content_type, metadata=metadata,
-            amz_headers=amz_headers)
+            amz_headers=amz_headers, body_producer=body_producer,
+            receiver_factory=self.receiver_factory)
         return query.submit()
 
     def copy_object(self, source_bucket, source_object_name, dest_bucket=None,
@@ -344,7 +356,8 @@
         query = self.query_factory(
             action="PUT", creds=self.creds, endpoint=self.endpoint,
             bucket=dest_bucket, object_name=dest_object_name,
-            metadata=metadata, amz_headers=amz_headers)
+            metadata=metadata, amz_headers=amz_headers,
+            receiver_factory=self.receiver_factory)
         return query.submit()
 
     def get_object(self, bucket, object_name):
@@ -353,7 +366,8 @@
         """
         query = self.query_factory(
             action="GET", creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name=object_name)
+            bucket=bucket, object_name=object_name,
+            receiver_factory=self.receiver_factory)
         return query.submit()
 
     def head_object(self, bucket, object_name):
@@ -384,7 +398,8 @@
         data = access_control_policy.to_xml()
         query = self.query_factory(
             action='PUT', creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name='%s?acl' % object_name, data=data)
+            bucket=bucket, object_name='%s?acl' % object_name, data=data,
+            receiver_factory=self.receiver_factory)
         return query.submit().addCallback(self._parse_acl)
 
     def get_object_acl(self, bucket, object_name):
@@ -393,7 +408,8 @@
         """
         query = self.query_factory(
             action='GET', creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name='%s?acl' % object_name)
+            bucket=bucket, object_name='%s?acl' % object_name,
+            receiver_factory=self.receiver_factory)
         return query.submit().addCallback(self._parse_acl)
 
     def put_request_payment(self, bucket, payer):
@@ -407,7 +423,8 @@
         data = RequestPayment(payer).to_xml()
         query = self.query_factory(
             action="PUT", creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name="?requestPayment", data=data)
+            bucket=bucket, object_name="?requestPayment", data=data,
+            receiver_factory=self.receiver_factory)
         return query.submit()
 
     def get_request_payment(self, bucket):
@@ -419,7 +436,8 @@
         """
         query = self.query_factory(
             action="GET", creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name="?requestPayment")
+            bucket=bucket, object_name="?requestPayment",
+            receiver_factory=self.receiver_factory)
         return query.submit().addCallback(self._parse_get_request_payment)
 
     def _parse_get_request_payment(self, xml_bytes):
@@ -429,17 +447,102 @@
         """
         return RequestPayment.from_xml(xml_bytes).payer
 
+    def init_multipart_upload(self, bucket, object_name, content_type=None,
+                         metadata={}):
+        """
+        Initiate a multipart upload to a bucket.
+
+        @param bucket: The name of the bucket
+        @param object_name: The object name
+        @param content_type: The Content-Type for the object
+        @param metadata: C{dict} containing additional metadata
+        @return: C{str} upload_id
+        """
+        objectname_plus = '%s?uploads' % object_name
+        query = self.query_factory(
+            action="POST", creds=self.creds, endpoint=self.endpoint,
+            bucket=bucket, object_name=objectname_plus, data='',
+            content_type=content_type, metadata=metadata)
+        d = query.submit()
+        return d.addCallback(MultipartInitiationResponse.from_xml)
+
+    def upload_part(self, bucket, object_name, upload_id, part_number,
+                    data=None, content_type=None, metadata={},
+                    body_producer=None):
+        """
+        Upload a part of data corresponding to a multipart upload.
+
+        @param bucket: The bucket name
+        @param object_name: The object name
+        @param upload_id: The multipart upload id
+        @param part_number: The part number
+        @param data: Data (optional, requires body_producer if not specified)
+        @param content_type: The Content-Type
+        @param metadata: Additional metadata
+        @param body_producer: an C{IBodyProducer} (optional, requires data if
+            not specified)
+        @return: the C{Deferred} from underlying query.submit() call
+        """
+        parms = 'partNumber=%s&uploadId=%s' % (str(part_number), upload_id)
+        objectname_plus = '%s?%s' % (object_name, parms)
+        query = self.query_factory(
+            action="PUT", creds=self.creds, endpoint=self.endpoint,
+            bucket=bucket, object_name=objectname_plus, data=data,
+            content_type=content_type, metadata=metadata,
+            body_producer=body_producer, receiver_factory=self.receiver_factory)
+        d = query.submit()
+        return d.addCallback(query.get_response_headers)
+
+    def complete_multipart_upload(self, bucket, object_name, upload_id,
+                                  parts_list, content_type=None, metadata={}):
+        """
+        Complete a multipart upload.
+
+        N.B. This can be possibly be a slow operation.
+
+        @param bucket: The bucket name
+        @param object_name: The object name
+        @param upload_id: The multipart upload id
+        @param parts_list: A List of all the parts
+            (2-tuples of part sequence number and etag)
+        @param content_type: The Content-Type of the object
+        @param metadata: C{dict} containing additional metadata
+        @return: a C{Deferred} that fires after request is complete
+        """
+        data = self._build_complete_multipart_upload_xml(parts_list)
+        objectname_plus = '%s?uploadId=%s' % (object_name, upload_id)
+        query = self.query_factory(
+            action="POST", creds=self.creds, endpoint=self.endpoint,
+            bucket=bucket, object_name=objectname_plus, data=data,
+            content_type=content_type, metadata=metadata)
+        d = query.submit()
+        # TODO - handle error responses
+        return d.addCallback(MultipartCompletionResponse.from_xml)
+
+    def _build_complete_multipart_upload_xml(self, parts_list):
+        xml = []
+        parts_list.sort(key=lambda p: int(p[0]))
+        xml.append('<CompleteMultipartUpload>')
+        for pt in parts_list:
+            xml.append('<Part>')
+            xml.append('<PartNumber>%s</PartNumber>' % pt[0])
+            xml.append('<ETag>%s</ETag>' % pt[1])
+            xml.append('</Part>')
+        xml.append('</CompleteMultipartUpload>')
+        return '\n'.join(xml)
+
 
 class Query(BaseQuery):
     """A query for submission to the S3 service."""
 
     def __init__(self, bucket=None, object_name=None, data="",
-                 content_type=None, metadata={}, amz_headers={}, *args,
-                 **kwargs):
+                 content_type=None, metadata={}, amz_headers={},
+                 body_producer=None, *args, **kwargs):
         super(Query, self).__init__(*args, **kwargs)
         self.bucket = bucket
         self.object_name = object_name
         self.data = data
+        self.body_producer = body_producer
         self.content_type = content_type
         self.metadata = metadata
         self.amz_headers = amz_headers
@@ -463,9 +566,14 @@
         """
         Build the list of headers needed in order to perform S3 operations.
         """
-        headers = {"Content-Length": len(self.data),
-                   "Content-MD5": calculate_md5(self.data),
+        if self.body_producer:
+            content_length = self.body_producer.length
+        else:
+            content_length = len(self.data)
+        headers = {"Content-Length": content_length,
                    "Date": self.date}
+        if self.body_producer is None:
+            headers["Content-MD5"] = calculate_md5(self.data)
         for key, value in self.metadata.iteritems():
             headers["x-amz-meta-" + key] = value
         for key, value in self.amz_headers.iteritems():
@@ -529,5 +637,6 @@
                 self.endpoint, self.bucket, self.object_name)
         d = self.get_page(
             url_context.get_url(), method=self.action, postdata=self.data,
-            headers=self.get_headers())
+            headers=self.get_headers(), body_producer=self.body_producer,
+            receiver_factory=self.receiver_factory)
         return d.addErrback(s3_error_wrapper)

=== modified file 'txaws/s3/model.py'
--- txaws/s3/model.py	2012-01-28 00:42:38 +0000
+++ txaws/s3/model.py	2012-02-10 22:39:30 +0000
@@ -150,3 +150,65 @@
         """
         root = XML(xml_bytes)
         return cls(root.findtext("Payer"))
+
+
+class MultipartInitiationResponse(object):
+    """
+    A response to Initiate Multipart Upload
+    """
+
+    def __init__(self, bucket, object_name, upload_id):
+        """
+        @param bucket: The bucket name
+        @param object_name: The object name
+        @param upload_id: The upload id
+        """
+        self.bucket = bucket
+        self.object_name = object_name
+        self.upload_id = upload_id
+
+    @classmethod
+    def from_xml(cls, xml_bytes):
+        """
+        Create an instance of this from XML bytes.
+
+        @param xml_bytes: C{str} bytes of XML to parse
+        @return: an instance of L{MultipartInitiationResponse}
+        """
+        root = XML(xml_bytes)
+        return cls(root.findtext('Bucket'),
+                   root.findtext('Key'),
+                   root.findtext('UploadId'))
+
+
+class MultipartCompletionResponse(object):
+    """
+    Represents a response to Complete Multipart Upload
+    """
+
+    def __init__(self, location, bucket, object_name, etag):
+        """
+        @param location: The URI identifying newly created object
+        @param bucket: The bucket name
+        @param object_name: The object name / key
+        @param etag: The entity tag
+        """
+        self.location = location
+        self.bucket = bucket
+        self.object_name = object_name
+        self.etag = etag
+
+    @classmethod
+    def from_xml(cls, xml_bytes):
+        """
+        Create an instance of this class from XML bytes.
+
+        @param xml_bytes: C{str} bytes of XML to parse
+        @return: an instance of L{MultipartCompletionResponse}
+        """
+        root = XML(xml_bytes)
+        return cls(root.findtext('Location'),
+                   root.findtext('Bucket'),
+                   root.findtext('Key'),
+                   root.findtext('ETag'))
+

=== modified file 'txaws/s3/tests/test_client.py'
--- txaws/s3/tests/test_client.py	2012-01-28 00:44:53 +0000
+++ txaws/s3/tests/test_client.py	2012-02-10 22:39:30 +0000
@@ -9,7 +9,9 @@
 else:
     s3clientSkip = None
 from txaws.s3.acls import AccessControlPolicy
-from txaws.s3.model import RequestPayment
+from txaws.s3.model import (RequestPayment, MultipartInitiationResponse,
+    MultipartCompletionResponse)
+from txaws.testing.producers import StringBodyProducer
 from txaws.service import AWSServiceEndpoint
 from txaws.testing import payload
 from txaws.testing.base import TXAWSTestCase
@@ -100,7 +102,8 @@
 
         class StubQuery(client.Query):
 
-            def __init__(query, action, creds, endpoint):
+            def __init__(query, action, creds, endpoint,
+                body_producer=None, receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds)
                 self.assertEquals(action, "GET")
@@ -134,7 +137,8 @@
 
         class StubQuery(client.Query):
 
-            def __init__(query, action, creds, endpoint, bucket=None):
+            def __init__(query, action, creds, endpoint, bucket=None,
+                body_producer=None, receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket)
                 self.assertEquals(action, "PUT")
@@ -156,7 +160,8 @@
 
         class StubQuery(client.Query):
 
-            def __init__(query, action, creds, endpoint, bucket=None):
+            def __init__(query, action, creds, endpoint, bucket=None,
+                body_producer=None, receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket)
                 self.assertEquals(action, "GET")
@@ -208,7 +213,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -243,7 +249,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -284,7 +291,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -323,7 +331,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -360,7 +369,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -396,7 +406,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -433,7 +444,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -473,7 +485,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -509,7 +522,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -546,7 +560,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -576,7 +591,8 @@
 
         class StubQuery(client.Query):
 
-            def __init__(query, action, creds, endpoint, bucket=None):
+            def __init__(query, action, creds, endpoint, bucket=None,
+                body_producer=None, receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket)
                 self.assertEquals(action, "DELETE")
@@ -599,7 +615,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None, data=""):
+                         object_name=None, data="", body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name,
@@ -630,7 +647,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None, data=""):
+                         object_name=None, data="", receiver_factory=None,
+                         body_producer=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name,
@@ -665,7 +683,7 @@
 
             def __init__(query, action, creds, endpoint, bucket=None,
                 object_name=None, data=None, content_type=None,
-                metadata=None):
+                metadata=None, body_producer=None, receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket,
                     object_name=object_name, data=data,
@@ -701,7 +719,7 @@
 
             def __init__(query, action, creds, endpoint, bucket=None,
                 object_name=None, data=None, content_type=None,
-                metadata=None):
+                metadata=None, body_producer=None, receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket,
                     object_name=object_name, data=data,
@@ -730,7 +748,8 @@
 
             def __init__(query, action, creds, endpoint, bucket=None,
                 object_name=None, data=None, content_type=None,
-                metadata=None, amz_headers=None):
+                metadata=None, amz_headers=None, body_producer=None,
+                receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket,
                     object_name=object_name, data=data,
@@ -756,6 +775,42 @@
                              metadata={"key": "some meta data"},
                              amz_headers={"acl": "public-read"})
 
+    def test_put_object_with_custom_body_producer(self):
+
+        class StubQuery(client.Query):
+
+            def __init__(query, action, creds, endpoint, bucket=None,
+                object_name=None, data=None, content_type=None,
+                metadata=None, amz_headers=None, body_producer=None,
+                receiver_factory=None):
+                super(StubQuery, query).__init__(
+                    action=action, creds=creds, bucket=bucket,
+                    object_name=object_name, data=data,
+                    content_type=content_type, metadata=metadata,
+                    amz_headers=amz_headers, body_producer=body_producer)
+                self.assertEqual(action, "PUT")
+                self.assertEqual(creds.access_key, "foo")
+                self.assertEqual(creds.secret_key, "bar")
+                self.assertEqual(query.bucket, "mybucket")
+                self.assertEqual(query.object_name, "objectname")
+                self.assertEqual(query.content_type, "text/plain")
+                self.assertEqual(query.metadata, {"key": "some meta data"})
+                self.assertEqual(query.amz_headers, {"acl": "public-read"})
+                self.assertIdentical(body_producer, string_producer)
+
+            def submit(query):
+                return succeed(None)
+
+
+        string_producer = StringBodyProducer("some data")
+        creds = AWSCredentials("foo", "bar")
+        s3 = client.S3Client(creds, query_factory=StubQuery)
+        return s3.put_object("mybucket", "objectname",
+                             content_type="text/plain",
+                             metadata={"key": "some meta data"},
+                             amz_headers={"acl": "public-read"},
+                             body_producer=string_producer)
+
     def test_copy_object(self):
         """
         L{S3Client.copy_object} creates a L{Query} to copy an object from one
@@ -766,7 +821,8 @@
 
             def __init__(query, action, creds, endpoint, bucket=None,
                 object_name=None, data=None, content_type=None,
-                metadata=None, amz_headers=None):
+                metadata=None, amz_headers=None, body_producer=None,
+                receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket,
                     object_name=object_name, data=data,
@@ -798,7 +854,8 @@
 
             def __init__(query, action, creds, endpoint, bucket=None,
                 object_name=None, data=None, content_type=None,
-                metadata=None, amz_headers=None):
+                metadata=None, amz_headers=None, body_producer=None,
+                receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket,
                     object_name=object_name, data=data,
@@ -822,7 +879,7 @@
 
             def __init__(query, action, creds, endpoint, bucket=None,
                 object_name=None, data=None, content_type=None,
-                metadata=None):
+                metadata=None, body_producer=None, receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket,
                     object_name=object_name, data=data,
@@ -846,7 +903,7 @@
 
             def __init__(query, action, creds, endpoint, bucket=None,
                 object_name=None, data=None, content_type=None,
-                metadata=None):
+                metadata=None, body_producer=None, receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket,
                     object_name=object_name, data=data,
@@ -869,7 +926,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None, data=""):
+                         object_name=None, data="", body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name,
@@ -902,7 +960,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None, data=""):
+                         object_name=None, data="", body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name,
@@ -926,6 +985,113 @@
         deferred = s3.get_object_acl("mybucket", "myobject")
         return deferred.addCallback(check_result)
 
+    def test_init_multipart_upload(self):
+
+        class StubQuery(client.Query):
+
+            def __init__(query, action, creds, endpoint, bucket=None,
+                         object_name=None, data="", body_producer=None,
+                         content_type=None, receiver_factory=None, metadata={}):
+                super(StubQuery, query).__init__(action=action, creds=creds,
+                                                 bucket=bucket,
+                                                 object_name=object_name,
+                                                 data=data)
+                self.assertEquals(action, "POST")
+                self.assertEqual(creds.access_key, "foo")
+                self.assertEqual(creds.secret_key, "bar")
+                self.assertEqual(query.bucket, "example-bucket")
+                self.assertEqual(query.object_name, "example-object?uploads")
+                self.assertEqual(query.data, "")
+                self.assertEqual(query.metadata, {})
+
+            def submit(query, url_context=None):
+                return succeed(payload.sample_s3_init_multipart_upload_result)
+
+
+        def check_result(result):
+            self.assert_(isinstance(result, MultipartInitiationResponse))
+            self.assertEqual(result.bucket, "example-bucket")
+            self.assertEqual(result.object_name, "example-object")
+            self.assertEqual(result.upload_id, "deadbeef")
+
+        creds = AWSCredentials("foo", "bar")
+        s3 = client.S3Client(creds, query_factory=StubQuery)
+        deferred = s3.init_multipart_upload("example-bucket", "example-object")
+        return deferred.addCallback(check_result)
+
+    def test_upload_part(self):
+
+        class StubQuery(client.Query):
+
+            def __init__(query, action, creds, endpoint, bucket=None,
+                         object_name=None, data="", body_producer=None,
+                         content_type=None, receiver_factory=None, metadata={}):
+                super(StubQuery, query).__init__(action=action, creds=creds,
+                                                 bucket=bucket,
+                                                 object_name=object_name,
+                                                 data=data)
+                self.assertEquals(action, "PUT")
+                self.assertEqual(creds.access_key, "foo")
+                self.assertEqual(creds.secret_key, "bar")
+                self.assertEqual(query.bucket, "example-bucket")
+                self.assertEqual(query.object_name,
+                    "example-object?partNumber=3&uploadId=testid")
+                self.assertEqual(query.data, "some data")
+                self.assertEqual(query.metadata, {})
+
+            def submit(query, url_context=None):
+                return succeed(None)
+
+        creds = AWSCredentials("foo", "bar")
+        s3 = client.S3Client(creds, query_factory=StubQuery)
+        return s3.upload_part("example-bucket", "example-object", "testid", 3,
+                              "some data")
+
+    def test_complete_multipart_upload(self):
+
+        class StubQuery(client.Query):
+
+            def __init__(query, action, creds, endpoint, bucket=None,
+                         object_name=None, data="", body_producer=None,
+                         content_type=None, receiver_factory=None, metadata={}):
+                super(StubQuery, query).__init__(action=action, creds=creds,
+                                                 bucket=bucket,
+                                                 object_name=object_name,
+                                                 data=data)
+                self.assertEquals(action, "POST")
+                self.assertEqual(creds.access_key, "foo")
+                self.assertEqual(creds.secret_key, "bar")
+                self.assertEqual(query.bucket, "example-bucket")
+                self.assertEqual(query.object_name,
+                    "example-object?uploadId=testid")
+                self.assertEqual(query.data, "<CompleteMultipartUpload>\n"
+                    "<Part>\n<PartNumber>1</PartNumber>\n<ETag>a</ETag>\n"
+                    "</Part>\n<Part>\n<PartNumber>2</PartNumber>\n"
+                    "<ETag>b</ETag>\n</Part>\n</CompleteMultipartUpload>")
+                self.assertEqual(query.metadata, {})
+
+            def submit(query, url_context=None):
+                return succeed(
+                    payload.sample_s3_complete_multipart_upload_result)
+
+
+        def check_result(result):
+            self.assert_(isinstance(result, MultipartCompletionResponse))
+            self.assertEqual(result.bucket, "example-bucket")
+            self.assertEqual(result.object_name, "example-object")
+            self.assertEqual(result.location,
+                "http://example-bucket.s3.amazonaws.com/example-object")
+            self.assertEqual(result.etag,
+                '"3858f62230ac3c915f300c664312c11f-9"')
+
+        creds = AWSCredentials("foo", "bar")
+        s3 = client.S3Client(creds, query_factory=StubQuery)
+        deferred = s3.complete_multipart_upload("example-bucket",
+                                                "example-object",
+                                                "testid", [(1, "a"), (2, "b")])
+        return deferred.addCallback(check_result)
+
+
 S3ClientTestCase.skip = s3clientSkip
 
 
@@ -1077,7 +1243,8 @@
         """
         class StubQuery(client.Query):
 
-            def __init__(query, action, creds, endpoint, bucket):
+            def __init__(query, action, creds, endpoint, bucket,
+                body_producer=None, receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket)
                 self.assertEquals(action, "GET")

=== modified file 'txaws/testing/payload.py'
--- txaws/testing/payload.py	2012-01-28 00:39:00 +0000
+++ txaws/testing/payload.py	2012-02-10 22:39:30 +0000
@@ -1085,3 +1085,19 @@
   <Status>Enabled</Status>
   <MfaDelete>Disabled</MfaDelete>
 </VersioningConfiguration>"""
+
+sample_s3_init_multipart_upload_result = """\
+<InitiateMultipartUploadResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+  <Bucket>example-bucket</Bucket>
+  <Key>example-object</Key>
+  <UploadId>deadbeef</UploadId>
+</InitiateMultipartUploadResult>"""
+
+sample_s3_complete_multipart_upload_result = """\
+<?xml version="1.0" encoding="UTF-8"?>
+<CompleteMultipartUploadResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+  <Location>http://example-bucket.s3.amazonaws.com/example-object</Location>
+  <Bucket>example-bucket</Bucket>
+  <Key>example-object</Key>
+  <ETag>"3858f62230ac3c915f300c664312c11f-9"</ETag>
+</CompleteMultipartUploadResult>"""

=== added file 'txaws/testing/producers.py'
--- txaws/testing/producers.py	1970-01-01 00:00:00 +0000
+++ txaws/testing/producers.py	2012-02-10 22:39:30 +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