test_newclient.py 文件源码

def test_dataReceived(self):
        """
        The L{IProtocol} provider passed to L{Response.deliverBody} has its
        C{dataReceived} method called with bytes received as part of the
        response body.
        """
        bytes = []
        class ListConsumer(Protocol):
            def dataReceived(self, data):
                bytes.append(data)


        consumer = ListConsumer()
        response = justTransportResponse(StringTransport())
        response.deliverBody(consumer)

        response._bodyDataReceived(b'foo')
        self.assertEqual(bytes, [b'foo'])