test_session.py 文件源码

def test_wrapProtocol(self):
        """
        L{wrapProtocol}, when passed a L{Protocol} should return something that
        has write(), writeSequence(), loseConnection() methods which call the
        Protocol's dataReceived() and connectionLost() methods, respectively.
        """
        protocol = MockProtocol()
        protocol.transport = StubTransport()
        protocol.connectionMade()
        wrapped = session.wrapProtocol(protocol)
        wrapped.dataReceived(b'dataReceived')
        self.assertEqual(protocol.transport.buf, b'dataReceived')
        wrapped.write(b'data')
        wrapped.writeSequence([b'1', b'2'])
        wrapped.loseConnection()
        self.assertEqual(protocol.data, b'data12')
        protocol.reason.trap(error.ConnectionDone)