Ticket #5192: 5192_4.patch

twisted/web/_newclient.py

@@ -573 +573 @@
         self.persistent = persistent
 
 
-    def _writeHeaders(self, transport, TEorCL):
+    def _writeHeaders(self, transport, bodyFramingHeader):
         hosts = self.headers.getRawHeaders('host', ())
         if len(hosts) != 1:
             raise BadHeaders("Exactly one Host header required")
@@ -586 +586 @@
             '%s %s HTTP/1.1\r\n' % (self.method, self.uri))
         if not self.persistent:
             requestLines.append('Connection: close\r\n')
-        if TEorCL is not None:
-            requestLines.append(TEorCL)
+        if bodyFramingHeader is not None:
+            requestLines.append(bodyFramingHeader)
         for name, values in self.headers.getAllRawHeaders():
             requestLines.extend(['%s: %s\r\n' % (name, v) for v in values])
         requestLines.append('\r\n')
         transport.writeSequence(requestLines)
 
-
-    def _writeToChunked(self, transport):
+    def _writeBodyToChunked(self, transport):
         """
-        Write this request to the given transport using chunked
+        Write this request's body to the given transport using chunked
         transfer-encoding to frame the body.
         """
-        self._writeHeaders(transport, 'Transfer-Encoding: chunked\r\n')
         encoder = ChunkedEncoder(transport)
         encoder.registerProducer(self.bodyProducer, True)
         d = self.bodyProducer.startProducing(encoder)
@@ -618 +616 @@
         return d
 
 
-    def _writeToContentLength(self, transport):
+    def _writeBodyToContentLength(self, transport):
         """
-        Write this request to the given transport using content-length to frame
-        the body.
+        Write this request's body to the given transport using content-length
+        to frame the body.
         """
-        self._writeHeaders(
-            transport,
-            'Content-Length: %d\r\n' % (self.bodyProducer.length,))
 
         # This Deferred is used to signal an error in the data written to the
         # encoder below.  It can only errback and it will only do so before too
@@ -737 +732 @@
             been completely written to the transport or with a L{Failure} if
             there is any problem generating the request bytes.
         """
+        self._writeHeadersTo(transport)
+        return self._writeBodyTo(transport)
+
+    def _writeHeadersTo(self, transport):
+        """
+        Format this L{Request}'s headers as HTTP/1.1 and write them
+        to the given transport.
+        """
+        bodyFramingHeader = None
         if self.bodyProducer is not None:
+           if self.bodyProducer.length is UNKNOWN_LENGTH:
+               bodyFramingHeader = 'Transfer-Encoding: chunked\r\n'
+           else:
+               bodyFramingHeader = 'Content-Length: %d\r\n' % (
+                       self.bodyProducer.length,)
+
+        self._writeHeaders(transport, bodyFramingHeader)
+
+    def _writeBodyTo(self, transport):
+        """
+        Write this L{Request}'s body to the given transport, framing it
+        according to the given headers('Transport-Encoding' or
+        'Content-Length').
+
+        @return: A L{Deferred} which fires with C{None} when the request has
+            been completely written the transport or with a L{Failure} if there
+            is any problem generating the request body bytes. If bodyProducer
+            is None the returned L{Deferred} is already fired.
+        """
+        if self.bodyProducer is not None:
             if self.bodyProducer.length is UNKNOWN_LENGTH:
-                return self._writeToChunked(transport)
+                return self._writeBodyToChunked(transport)
             else:
-                return self._writeToContentLength(transport)
+                return self._writeBodyToContentLength(transport)
         else:
-            self._writeHeaders(transport, None)
             return succeed(None)
 
 
@@ -1186 +1209 @@
             self._producer.pauseProducing()
 
 
+class DiscardWithDeferred(Protocol):
+    """
+    A L{Protocol} that discards all received data and that fires a L{Deferred}
+    when all data has been received.
 
+    @ivar finishedDeferred: L{Deferred} which fires with C{None} when all data
+        has been received and with L{Failure} on error.
+
+    """
+
+    def __init__(self):
+        self.finishedDeferred = Deferred()
+
+    def dataReceived(self, data):
+        pass
+
+    def connectionLost(self, reason):
+        if reason.type == ResponseDone:
+            self.finishedDeferred.callback(None)
+        else:
+            self.finishedDeferred.errback(reason)
+
+
+TIMEOUT_100_CONTINUE = 1
+
+
 class HTTP11ClientProtocol(Protocol):
     """
     L{HTTP11ClientProtocol} is an implementation of the HTTP 1.1 client
-    protocol.  It supports as few features as possible.
+    protocol. It supports as few features as possible.
 
     @ivar _parser: After a request is issued, the L{HTTPClientParser} to
         which received data making up the response to that request is
         delivered.
 
+    @ivar _reactor: The reactor used for eventual callLater calls.
+
     @ivar _finishedRequest: After a request is issued, the L{Deferred} which
         will fire when a L{Response} object corresponding to that request is
         available.  This allows L{HTTP11ClientProtocol} to fail the request
@@ -1216 +1266 @@
         received.  This is eventually chained with C{_finishedRequest}, but
         only in certain cases to avoid double firing that Deferred.
 
+    @ivar _responseBodyDeferred: After a request is issued, the L{Deferred}
+        that fires when the C{_parser} has done parsing the entire L{Response},
+        including body, with the data that came after the current L{Response}.
+        This can be used to set another C{_parser} but usually this decision
+        is done on C{_responseDeferred}'s callback.
+
+    @ivar _forcedRequestBody: True if we had an 100-Continue L{Request} whose
+        body was forcefully written to the transport because the server
+        did not respond in time with a L{Response} (possibly because of a
+        buggy server that doesn't implement expectations correctly).
+
+    @ivar _firstResponseTimer: A L{Delayed} that fires after
+        TIMEOUT_100_CONTINUE seconds and forcefully sends the L{Request} body
+        to the server.
+
+    @ivar _firstResponseDeferred: A L{Deferred} that fires with the first
+        L{Response} to an 100-Continue L{Request}. This may be an 100-Continue
+        response or a response with a final status. It may fire in the
+        WAITING_100_CONTINUE_RESPONSE, TRANSMITTING or WAITING states.
+
+    @ivar _firstResponseBodyDeferred: A L{Deferred} that fires when the body
+        of the first L{Response} to an 100-Continue L{Request} has been
+        successfully parsed. It signals that we can start sending the
+        L{Request} body. It fires in the WAITING_100_CONTINUE_RESPONSE_BODY,
+        TRANSMITTING and WAITING states.
+
     @ivar _state: Indicates what state this L{HTTP11ClientProtocol} instance
         is in with respect to transmission of a request and reception of a
         response.  This may be one of the following strings:
@@ -1236 +1312 @@
           - GENERATION_FAILED: There was an error while the request.  The
             request was not fully sent to the network.
 
+          - WAITING_100_CONTINUE_RESPONSE: We're waiting for a L{Response} to a
+            L{Request} that expects 100-Continue.
+
+          - WAITING_100_CONTINUE_RESPONSE_BODY: Got an 100 Continue
+            L{Response} and we're discarding its body before sending the
+            L{Request}.
+
           - WAITING: The request was fully sent to the network.  The
             instance is now waiting for the response to be fully received.
 
@@ -1255 +1338 @@
     _responseDeferred = None
 
 
-    def __init__(self, quiescentCallback=lambda c: None):
+    def __init__(self, quiescentCallback=lambda c: None, reactor=None):
+        """
+        Initialize this L{HTTP11ClientProtocol}. Optionally a reactor can be
+        given. Otherwise use the global reactor.
+        """
+        if reactor is None:
+            from twisted.internet import reactor
+        self._reactor = reactor
         self._quiescentCallback = quiescentCallback
         self._abortDeferreds = []
 
@@ -1284 +1374 @@
             may errback with L{RequestNotSent} if it is not possible to send
             any more requests using this L{HTTP11ClientProtocol}.
         """
+
         if self._state != 'QUIESCENT':
             return fail(RequestNotSent())
 
-        self._state = 'TRANSMITTING'
-        _requestDeferred = maybeDeferred(request.writeTo, self.transport)
+        if request.headers.hasHeader('expect'):
+            _expectations = request.headers.getRawHeaders('expect')
+            _expects100Continue = '100-continue' in [x.lower() for x in
+                                                     _expectations]
+        else:
+            _expects100Continue = False
+
         self._finishedRequest = Deferred()
 
-        # Keep track of the Request object in case we need to call stopWriting
-        # on it.
         self._currentRequest = request
 
-        self._transportProxy = TransportProxyProducer(self.transport)
-        self._parser = HTTPClientParser(request, self._finishResponse)
-        self._parser.makeConnection(self._transportProxy)
-        self._responseDeferred = self._parser._responseDeferred
+        if _expects100Continue:
+            self._handle100ContinueRequest(request)
+        else:
+            self._handleRequest(request)
 
         def cbRequestWrotten(ignored):
-            if self._state == 'TRANSMITTING':
+            if self._state in ['TRANSMITTING',
+                               'WAITING_100_CONTINUE_RESPONSE',
+                               'WAITING_100_CONTINUE_RESPONSE_BODY']:
                 self._state = 'WAITING'
                 self._responseDeferred.chainDeferred(self._finishedRequest)
 
@@ -1315 +1411 @@
                 log.err(err, 'Error writing request, but not in valid state '
                              'to finalize request: %s' % self._state)
 
-        _requestDeferred.addCallbacks(cbRequestWrotten, ebRequestWriting)
+        self._requestDeferred.addCallbacks(cbRequestWrotten, ebRequestWriting)
 
+        self._responseBodyDeferred.addCallback(self._finishResponse)
+
         return self._finishedRequest
 
 
@@ -1336 +1434 @@
         # Currently the rest parameter is ignored. Don't forget to use it if
         # we ever add support for pipelining. And maybe check what trailers
         # mean.
-        if self._state == 'WAITING':
+        if self._state in ['WAITING',
+                'WAITING_100_CONTINUE_RESPONSE']:
             self._state = 'QUIESCENT'
         else:
             # The server sent the entire response before we could send the
@@ -1369 +1468 @@
                 log.err()
                 self.transport.loseConnection()
             self._disconnectParser(reason)
+            self._cleanupRequest()
 
 
     _finishResponse_TRANSMITTING = _finishResponse_WAITING
+    _finishResponse_WAITING_100_CONTINUE_RESPONSE = _finishResponse_WAITING
 
+    def _handleRequest(self, request):
+        """
+        Send a non-100-Continue-expecting L{Request} to the transport.
 
+        @param request: The L{Request} to be sent.
+
+        """
+        self._state = 'TRANSMITTING'
+        self._requestDeferred = maybeDeferred(request.writeTo, self.transport)
+        (self._responseDeferred,
+             self._responseBodyDeferred) = self._setupParser(request)
+
+
+    def _handle100ContinueRequest(self, request):
+        """
+        Send an 100-Continue L{Request} to the transport.
+
+        @param request: The L{Request} to be sent.
+        """
+        request._writeHeadersTo(self.transport)
+        self._state = 'WAITING_100_CONTINUE_RESPONSE'
+        self._forcedRequestBody = False
+        self._requestDeferred = Deferred()
+        self._responseDeferred = Deferred()
+        self._responseBodyDeferred = Deferred()
+        (self._firstResponseDeferred,
+             self._firstResponseBodyDeferred) = self._setupParser(request)
+        self._firstResponseTimer = self._reactor.callLater(
+            TIMEOUT_100_CONTINUE,
+            self._forceRequestBody)
+
+        self._firstResponseDeferred.addCallbacks(
+            self._handleFirstResponse, self._handle100ContinueError)
+
+
+    def _handleFirstResponse(self, response):
+        """
+        Handle the first L{Response} to an 100-Continue L{Request}. This may
+        be an 100-Continue or a final status L{Response}. If it is an 100
+        response, consume it's body and then send the L{Request} body, else
+        forward the response to the user by means of self._finishedRequest.
+
+        @param response: The L{Response} to the current L{Request}.
+        """
+        # This may be inactive if this is a response that came after the timer
+        # fired.
+        if self._firstResponseTimer.active():
+            self._firstResponseTimer.cancel()
+
+        if self._state == 'WAITING_100_CONTINUE_RESPONSE':
+
+            if response.code == 100:
+                self._state = 'WAITING_100_CONTINUE_RESPONSE_BODY'
+                self._discardResponseBody(response,
+                    self._handleFirstResponseBody,
+                    self._handle100ContinueError)
+            else:
+                # We're done with this request.
+                self._requestDeferred.callback(None)
+                self._forwardFirstResponse(response)
+
+        else:
+
+            if self._forcedRequestBody and response.code == 100:
+                # Late arrival, eat it.
+                self._discardResponseBody(response,
+                    self._handleFirstResponseBody,
+                    self._handle100ContinueError)
+
+            else:
+                # A late response that isn't 100-Continue; could be from a
+                # server that doesn't implement expectations correctly.
+                self._forcedRequestBody = False
+                self._forwardFirstResponse(response)
+
+
+    def _forwardFirstResponse(self, response):
+        """
+        Forward a L{Response} and its body to the caller.
+
+        @param response: The L{Response} to be forwarded.
+        """
+
+        # Save this since it may get cleaned up in _finishResponse
+        responseDeferred = self._responseDeferred
+        self._firstResponseBodyDeferred.chainDeferred(
+                self._responseBodyDeferred)
+        responseDeferred.callback(response)
+
+
+    def _handleFirstResponseBody(self, rest):
+        """
+        The body of the first L{Response} to the current 100-Continue
+        L{Request} has been parsed. If the L{Response} wasn't an 100-Continue
+        forward to self._responseBodyDeferred. Otherwise create a new parser
+        for the second L{Response}.
+
+        @param rest: Data that wasn't parsed by the parser because it came
+            after the L{Response}. If we reload the parser, initialize it
+            with this data.
+        """
+        # We've just done discarding an 100-Continue response's body. Might
+        # be because we're waiting to send the request body or it might be
+        # that we've ignored a late response.
+
+        self._forcedRequestBody = False
+
+        if self._state == 'WAITING_100_CONTINUE_RESPONSE_BODY':
+            self._state = 'TRANSMITTING'
+
+            # Send the request body.
+
+            _requestBodyDeferred = maybeDeferred(
+                self._currentRequest._writeBodyTo,
+                self.transport)
+
+            _requestBodyDeferred.chainDeferred(self._requestDeferred)
+
+        # In both cases create a new parser.
+
+        self._disconnectParser(None)
+
+        (_secondResponseDeferred, _secondResponseBodyDeferred) =\
+            self._setupParser(self._currentRequest, data=rest)
+
+        _secondResponseDeferred.chainDeferred(self._responseDeferred)
+        _secondResponseBodyDeferred.chainDeferred(self._responseBodyDeferred)
+
+
+    def _discardResponseBody(self, response, callback, errback):
+        """
+        Discard a L{Response}'s body and call callback when done and errback
+        on error.
+
+        @param response: L{Response} that needs to be discarded.
+        @param callback: function to be called when done.
+        @param errback: function to be called on error
+        """
+        discarder = DiscardWithDeferred()
+
+        # We use discarder.finishedDeferred to catch body parsing
+        # errors and self._firstResponseBodyDeferred to catch success.
+
+        discarder.finishedDeferred.addErrback(errback)
+
+        response.deliverBody(discarder)
+
+        self._firstResponseBodyDeferred.addCallback(callback)
+
+
+    def _forceRequestBody(self):
+        """
+        Send the current L{Request} body even though we were expecting an
+        100 or final status L{Response}. It may just be a broken server that
+        doesn't implement correctly expectations.
+        """
+        self._state = 'TRANSMITTING'
+
+        self._forcedRequestBody = True
+
+        _requestBodyDeferred = maybeDeferred(self._currentRequest._writeBodyTo,
+            self.transport)
+
+        _requestBodyDeferred.chainDeferred(self._requestDeferred)
+
+
+    def _setupParser(self, request, data=''):
+        """
+        Setup a L{HTTPClientParser} for a L{Response} to a L{Request}. If this
+        is not the first parser associated with this protocol, call
+        L{HTTP11ClientProtocol._disconnectParser} first. Pass the given C{data}
+        to the newly created parser.
+
+        @param request: L{Request} waiting for a L{Response}.
+        @param data: Data to initialize the L{HTTPClientParser} with.
+        """
+        self._transportProxy = TransportProxyProducer(self.transport)
+
+        _responseBodyDeferred = Deferred()
+
+        def cbOnBodyFinish(rest):
+            _responseBodyDeferred.callback(rest)
+
+        self._parser = HTTPClientParser(request, cbOnBodyFinish)
+
+        self._parser.makeConnection(self._transportProxy)
+
+        # Grab this before passing data, since it might disappear if data is a
+        # complete response.
+
+        _responseDeferred = self._parser._responseDeferred
+
+        self._parser.dataReceived(data)
+
+        return (_responseDeferred, _responseBodyDeferred)
+
+
+    def _cleanupOn100ContinueError(self):
+        """
+        State-dependent cleanup on parsing errors while handling an
+        100-Continue-expecting L{Request}.
+        """
+    _cleanupOn100ContinueError = makeStatefulDispatcher(
+            'cleanupOn100ContinueError', _cleanupOn100ContinueError)
+
+
+    def _cleanupOn100ContinueError_WAITING_100_CONTINUE_RESPONSE(self):
+        """
+        The L{Request} body no sent yet. Fire self._requestDeferred because
+        we've basically finished dealing with this L{Request}. Also, this
+        forwards the L{Failure} to the user.
+        """
+        self._requestDeferred.callback(None)
+
+
+    def _cleanupOn100ContinueError_WAITING_100_CONTINUE_RESPONSE_BODY(self):
+        """
+        The L{Request} body no sent yet. Fire self._requestDeferred because
+        we've basically finished dealing with this L{Request}. Also, this
+        forwards the L{Failure} to the user.
+        """
+        self._requestDeferred.callback(None)
+
+
+    def _cleanupOn100ContinueError_TRANSMITTING(self):
+        """
+        We're currently sending the L{Request} body. The error will be sent to
+        the user after the body has been sent. No cleanup needed.
+        """
+
+
+    def _cleanupOn100ContinueError_WAITING(self):
+        """
+        No cleanup needed.
+        """
+
+
+    def _handle100ContinueError(self, err):
+        """
+        Handle any L{Failure} that could occur while handling a L{Request} that
+        expects 100-Continue. These are errors on parsing the first response
+        and the first response's body and can occur in the
+        WAITING_100_CONTINUE_RESPONSE/WAITING_100_CONTINUE_RESPONSE_BODY if
+        the server supports expectations or TRANSMITTING/WAITING if the
+        L{Request} body was sent after TIMEOUT_100_CONTINUE. Depending on
+        the current state some cleanup needs to be performed and the L{Failure}
+        is forwarded to self._responseDeferred.
+
+        @param err: L{Failure} to be forwarded.
+        """
+        self._cleanupOn100ContinueError()
+        self._responseDeferred.errback(err)
+
+
+    def _cleanupRequest(self):
+        """
+        Clean up the L{HTTP11ClientProtocol}'s state after a L{Request} has
+        been done with.
+        """
+        self._currentRequest = None
+        self._finishedRequest = None
+        self._responseDeferred = None
+        self._firstResponseDeferred = None
+        self._firstResponseBodyDeferred = None
+
+
     def _disconnectParser(self, reason):
         """
         If there is still a parser, call its C{connectionLost} method with the
@@ -1384 +1750 @@
         if self._parser is not None:
             parser = self._parser
             self._parser = None
-            self._currentRequest = None
-            self._finishedRequest = None
-            self._responseDeferred = None
 
             # The parser is no longer allowed to do anything to the real
             # transport.  Stop proxying from the parser's transport to the real
@@ -1488 +1851 @@
         self._abortDeferreds = []
 
 
+    def _connectionLost_WAITING_100_CONTINUE_RESPONSE(self, reason):
+        """
+        Disconnect the parser so that it can propagate the event and move to
+        the C{'CONNECTION_LOST'} state.
+        """
+        self._disconnectParser(reason)
+        self._state = 'CONNECTION_LOST'
+
+
+    def _connectionLost_WAITING_100_CONTINUE_RESPONSE_BODY(self, reason):
+        """
+        Disconnect the parser so that it can propagate the event and move to
+        the C{'CONNECTION_LOST'} state.
+        """
+        self._disconnectParser(reason)
+        self._state = 'CONNECTION_LOST'
+
+
     def abort(self):
         """
         Close the connection and cause all outstanding L{request} L{Deferred}s

twisted/web/test/test_newclient.py

@@ -7 +7 @@
 
 __metaclass__ = type
 
+from StringIO import StringIO
+
 from zope.interface import implements
 from zope.interface.verify import verifyObject
 
@@ -16 +18 @@
 from twisted.internet.error import ConnectionDone, ConnectionLost
 from twisted.internet.defer import Deferred, succeed, fail
 from twisted.internet.protocol import Protocol
+from twisted.internet.task import Clock, Cooperator
 from twisted.trial.unittest import TestCase
 from twisted.test.proto_helpers import StringTransport, AccumulatingProtocol
 from twisted.web._newclient import UNKNOWN_LENGTH, STATUS, HEADER, BODY, DONE
@@ -27 +30 @@
 from twisted.web._newclient import ConnectionAborted, ResponseNeverReceived
 from twisted.web._newclient import BadHeaders, ResponseDone, PotentialDataLoss, ExcessWrite
 from twisted.web._newclient import TransportProxyProducer, LengthEnforcingConsumer, makeStatefulDispatcher
+from twisted.web._newclient import TIMEOUT_100_CONTINUE
 from twisted.web.http_headers import Headers
 from twisted.web.http import _DataLoss
 from twisted.web.iweb import IBodyProducer, IResponse
+from twisted.web.client import FileBodyProducer
 
 
 
@@ -827 +832 @@
     method = 'GET'
     stopped = False
     persistent = False
+    headers = Headers()
 
     def writeTo(self, transport):
         self.finished = Deferred()
@@ -846 +852 @@
     L{Request} with no body producer.
     """
     persistent = False
+    headers = Headers()
 
     def writeTo(self, transport):
         transport.write('SOME BYTES')
@@ -863 +870 @@
         Create an L{HTTP11ClientProtocol} connected to a fake transport.
         """
         self.transport = StringTransport()
-        self.protocol = HTTP11ClientProtocol()
+        self.clock = Clock()
+        self.protocol = HTTP11ClientProtocol(reactor=self.clock)
         self.protocol.makeConnection(self.transport)
 
 
@@ -916 +924 @@
         """
         class BrokenRequest:
             persistent = False
+            headers = Headers()
             def writeTo(self, transport):
                 return fail(ArbitraryException())
 
@@ -939 +948 @@
         """
         class BrokenRequest:
             persistent = False
+            headers = Headers()
             def writeTo(self, transport):
                 raise ArbitraryException()
 
@@ -1566 +1576 @@
         self.assertTrue(transport.disconnecting)
 
 
+    def _send100ContinueRequest(self, body, persistent=False):
+        """
+        Send a L{Request} that expects 100-Continue with the given body.
+        """
+        def _immediateScheduler(x):
+            return succeed(x())
 
+        cooperator = Cooperator(scheduler=_immediateScheduler, started=False)
+        producer = FileBodyProducer(StringIO(body), cooperator=cooperator)
+
+        headers = Headers({'host': ['example.com'], 'expect': ['100-Continue']})
+
+        d = self.protocol.request(Request('POST', '/foo', headers, producer,
+            persistent=persistent))
+
+        self.transport.clear()
+
+        cooperator.start()
+
+        self.assertEqual(self.transport.value(), '')
+
+        return d
+
+
+    def test_expect100ContinueGetFinalStatus(self):
+        """
+        When we expect 100-Continue and get a final status L{Response} we don't
+        send the L{Request} body and return the first L{Response} to the user.
+        """
+        d = self._send100ContinueRequest('x' * 10)
+
+        def cbResponse(response):
+            self.assertEqual(response.code, 200)
+
+        d.addCallback(cbResponse)
+
+        self.protocol.dataReceived(
+            "HTTP/1.1 200 OK\r\n"
+            "Content-length: 0\r\n"
+            "\r\n")
+
+        self.assertEqual(self.transport.value(), '')
+
+        return d
+
+
+    def test_expect100ContinueGet100Continue(self):
+        """
+        When we expect 100-Continue and get an 100-Continue L{Response} we send
+        the L{Request} body and return the second L{Response} to the user.
+        """
+        d = self._send100ContinueRequest('x' * 10)
+
+        def cbResponse(response):
+            self.assertEqual(response.code, 200)
+
+        d.addCallback(cbResponse)
+
+        self.protocol.dataReceived(
+            "HTTP/1.1 100 Continue\r\n"
+            "Content-Length: 3\r\n"
+            "\r\n"
+            "123")
+
+        self.protocol.dataReceived(
+            "HTTP/1.1 200 OK\r\n"
+            "Content-Length: 0\r\n"
+            "\r\n")
+
+        self.assertEqual(self.transport.value(), 'x' * 10)
+
+        return d
+
+
+    def test_expect100ContinueGet100ContinueBackToBack(self):
+        """
+        When we expect 100-Continue and we get 2 response back to back (100 and
+        final status) we should act as if they came separately.
+        """
+        d = self._send100ContinueRequest('x' * 10)
+
+        def cbResponse(response):
+            self.assertEqual(response.code, 200)
+
+        d.addCallback(cbResponse)
+
+        self.protocol.dataReceived(
+            "HTTP/1.1 100 Continue\r\n"
+            "Content-Length: 3\r\n"
+            "\r\n"
+            "123"
+            "HTTP/1.1 200 OK\r\n"
+            "Content-Length: 0\r\n"
+            "\r\n")
+
+        self.assertEqual(self.transport.value(), 'x' * 10)
+
+        return d
+
+
+    def test_expect100ContinueServerBroken(self):
+        """
+        When we expect 100-Continue and the server is broken and waits for the
+        L{Request} body we wait for a limited amount and then send the body.
+        """
+        d = self._send100ContinueRequest('x' * 10)
+
+        def cbResponse(response):
+            self.assertEqual(response.code, 200)
+
+        d.addCallback(cbResponse)
+
+        self.clock.advance(TIMEOUT_100_CONTINUE + 1)
+
+        self.assertEqual(self.transport.value(), 'x' * 10)
+
+        self.protocol.dataReceived(
+            "HTTP/1.1 200 OK\r\n"
+            "Content-Length: 0\r\n"
+            "\r\n")
+
+        return d
+
+
+    def test_expect100ContinueTimerFiresLate100ContinueResponse(self):
+        """
+        When we expect 100-Continue and the server is slow and sends an
+        100-Continue after we sent the body we consume the 100-Continue
+        L{Response} and return the second L{Response} to the user.
+        """
+        d = self._send100ContinueRequest('x' * 10)
+
+        def cbResponse(response):
+            self.assertEqual(response.code, 200)
+
+        d.addCallback(cbResponse)
+
+        self.clock.advance(TIMEOUT_100_CONTINUE + 1)
+
+        self.assertEqual(self.transport.value(), 'x' * 10)
+
+        self.protocol.dataReceived(
+            "HTTP/1.1 100 Continue\r\n"
+            "Content-length: 3\r\n"
+            "\r\n"
+            "123")
+
+        self.protocol.dataReceived(
+            "HTTP/1.1 200 OK\r\n"
+            "Content-length: 0\r\n"
+            "\r\n")
+
+        return d
+
+
+    _garbageResponse = "unparseable garbage goes here\r\n"
+
+
+    def test_expect100ContinueBrokenFirstResponse(self):
+        """
+        When we expect 100-Continue and the first L{Response} is broken, return
+        the error to the user.
+        """
+        d = self._send100ContinueRequest('x' * 10)
+
+        self.protocol.dataReceived(self._garbageResponse)
+
+        self.assertEqual(self.transport.value(), '')
+
+        return assertResponseFailed(self, d, [ParseError])
+
+
+    def test_expect100ContinueBrokenFirstResponseChunkedBody(self):
+        """
+        When we expect 100-Continue and the 100-Continue L{Response} has a
+        chunked body and it is broken, return the error to the user.
+        """
+        d = self._send100ContinueRequest('x' * 10)
+
+        self.protocol.dataReceived(
+            "HTTP/1.1 100 Continue\r\n"
+            "Transfer-Encoding: chunked\r\n"
+            "\r\n")
+
+        self.protocol.dataReceived("3\r\nzzz\r\n")
+        self.protocol.dataReceived("3\r\nzzz\r\nzzz\r\n") #incorrect chunk
+
+        self.assertEqual(self.transport.value(), '')
+
+        return assertResponseFailed(self, d, [ValueError, _DataLoss])
+
+
+    def test_expect100ContinueBrokenSecondResponse(self):
+        """
+        When we expect 100-Continue and the 100-Continue L{Response} is ok but
+        the second L{Response} is broken, return the error to the user.
+        """
+        d = self._send100ContinueRequest('x' * 10)
+
+        self.protocol.dataReceived(
+            "HTTP/1.1 100 Continue\r\n"
+            "Content-length: 3\r\n"
+            "\r\n"
+            "123")
+
+        self.protocol.dataReceived(self._garbageResponse)
+
+        self.assertEqual(self.transport.value(), 'x' * 10)
+
+        return assertResponseFailed(self, d, [ParseError])
+
+
+    def _setupForQuiescent(self):
+        self.quiescentResult = []
+
+        def callback(p):
+            self.assertEqual(p, self.protocol)
+            self.assertEqual(p.state, "QUIESCENT")
+            self.quiescentResult.append(p)
+
+        self.transport = StringTransport()
+        self.clock = Clock()
+        self.protocol = HTTP11ClientProtocol(callback, reactor=self.clock)
+        self.protocol.makeConnection(self.transport)
+
+
+    def _checkQuiescentCalled(self, requestDeferred, body=''):
+        # Headers done, but still no quiescent callback:
+        self.assertEqual(self.quiescentResult, [])
+
+        result = []
+        requestDeferred.addCallback(result.append)
+        response = result[0]
+
+        # When response body is done (i.e. connectionLost is called), note the
+        # fact in quiescentResult:
+        bodyProtocol = AccumulatingProtocol()
+        bodyProtocol.closedDeferred = Deferred()
+        bodyProtocol.closedDeferred.addCallback(
+            lambda ign: self.quiescentResult.append("response done"))
+
+        response.deliverBody(bodyProtocol)
+        self.protocol.dataReceived(body)
+        bodyProtocol.closedReason.trap(ResponseDone)
+        # Quiescent callback called *before* self.protocol handling the response
+        # body gets its connectionLost called:
+        self.assertEqual(self.quiescentResult, [self.protocol, "response done"])
+
+        # Make sure everything was cleaned up:
+        self.assertEqual(self.protocol._parser, None)
+        self.assertEqual(self.protocol._finishedRequest, None)
+        self.assertEqual(self.protocol._currentRequest, None)
+        self.assertEqual(self.protocol._transportProxy, None)
+        self.assertEqual(self.protocol._responseDeferred, None)
+
+
+    def test_expect100ContinueGot100ContinueQuiescentCallbackCalled(self):
+        """
+        We have a persistent connection and we expect 100-Continue. When we
+        get an 100-Continue L{Response} the quiescent callback needs to be
+        called.
+        """
+        self._setupForQuiescent()
+
+        d = self._send100ContinueRequest('x' * 10, persistent=True)
+
+        def cbResponse(response):
+            self.assertEqual(response.code, 200)
+            return response
+
+        d.addCallback(cbResponse)
+
+        self.protocol.dataReceived(
+            "HTTP/1.1 100 Continue\r\n"
+            "Content-Length: 3\r\n"
+            "\r\n"
+            "123")
+
+        self.protocol.dataReceived(
+            "HTTP/1.1 200 OK\r\n"
+            "Content-Length: 3\r\n"
+            "\r\n")
+
+        self._checkQuiescentCalled(d, body="abc")
+
+        self.assertEqual(self.transport.value(), 'x' * 10)
+
+        return d
+
+
+    def test_expect100ContinueGotFinalStatusQuiescentCallbackCalled(self):
+        """
+        We have a persistent connection and we expect 100-Continue. When we
+        get a L{Response} with a final status the quiescent callback needs to
+        be called.
+        """
+        self._setupForQuiescent()
+
+        d = self._send100ContinueRequest('x' * 10, persistent=True)
+
+        def cbResponse(response):
+            self.assertEqual(response.code, 200)
+            return response
+
+        d.addCallback(cbResponse)
+
+        self.protocol.dataReceived(
+            "HTTP/1.1 200 OK\r\n"
+            "Content-Length: 3\r\n"
+            "\r\n")
+
+        self._checkQuiescentCalled(d, body='abc')
+
+        self.assertEqual(self.transport.value(), '')
+
+        return d
+
+    
+    def test_expect100ContinueConnectionLostWhileWaitingFirstResponse(self):
+        """
+        When we expect 100-Continue and we get disconnected while waiting for
+        the first L{Response}, the L{Deferred} return by request() must errback
+        a ResponseFailed wrapping the underlying failure.
+        """
+        d = self._send100ContinueRequest('x' * 10)
+
+        self.assertEqual(self.transport.value(), '')
+
+        self.protocol.connectionLost(Failure(ArbitraryException()))
+
+        self.assertEqual(self.protocol._state, 'CONNECTION_LOST')
+
+        return assertResponseFailed(
+            self, d, [ArbitraryException])
+
+
+    def test_expect100ContinueConnectionLostWhileWaitingFirstResponseBody(self):
+        """
+        When we expect 100-Continue and we get an 100-Continue L{Response}
+        and we get disconnected while waiting for its body, the L{Deferred}
+        returned by request() must errback a ResponseFailed wrapping the
+        underlying failure.
+        """
+        d = self._send100ContinueRequest('x' * 10)
+
+        self.assertEqual(self.transport.value(), '')
+
+        self.protocol.dataReceived(
+            "HTTP/1.1 100 Continue\r\n"
+            "Content-length: 10\r\n"
+            "\r\n")
+
+        self.protocol.connectionLost(Failure(ArbitraryException()))
+
+        self.assertEqual(self.protocol._state, 'CONNECTION_LOST')
+
+        return assertResponseFailed(
+            self, d, [ArbitraryException, _DataLoss])
+
+
 class StringProducer:
     """
     L{StringProducer} is a dummy body producer.